Repository: telstra/open-kilda
Issue: Switch Sync timeout if switch has 2 default rules with same cookie

**Steps to reproduce:**
1. Create excess default rule with existing default cookie. 
OVS: 
```
docker exec -it lab_service-1 bash
ovs-ofctl -O OpenFlow13 add-flow ofsw2 cookie=0x8000000000000002,priority=1,in_port=11,dl_src=00:05:95:41:ec:8c/ff:ff:ff:ff:ff:ff,actions=drop
```
2. validate switch
3. sync switch

**Expected result:**
(Optionally) excess rule after validation
Switch sync removes the rule

**Actual result:**
misconfigured rule
Switch sync timeout

Changed files:
src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfBatchHolder.java
src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfBatchExecutor.java
src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/BatchData.java

diff --git a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/BatchData.java b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/BatchData.java
index ac3c830e1..24dfc5338 100644
--- a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/BatchData.java
+++ b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/BatchData.java
@@ -15,6 +15,8 @@
 
 package org.openkilda.floodlight.command.rulemanager;
 
+import org.openkilda.rulemanager.SpeakerData;
+
 import lombok.Builder;
 import lombok.Data;
 import org.projectfloodlight.openflow.protocol.OFMessage;
@@ -26,5 +28,6 @@ public class BatchData {
     private boolean group;
     private boolean flow;
     private OFMessage message;
+    private SpeakerData origin;
     private boolean presenceBeVerified;
 }
diff --git a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfBatchExecutor.java b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfBatchExecutor.java
index 414c7ba77..a38f28041 100644
--- a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfBatchExecutor.java
+++ b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfBatchExecutor.java
@@ -43,16 +43,15 @@ import org.projectfloodlight.openflow.protocol.OFMessage;
 
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
 
 @Slf4j
 public class OfBatchExecutor {
@@ -106,7 +105,7 @@ public class OfBatchExecutor {
     public void executeBatch() {
         log.debug("Execute batch start (key={})", kafkaKey);
         List<UUID> stageCommandsUuids = holder.getCurrentStage();
-        Map<SpeakerData, OFMessage> stageMessages = new HashMap<>();
+        List<BatchData> stageMessages = new ArrayList<>();
         for (UUID uuid : stageCommandsUuids) {
             if (holder.canExecute(uuid)) {
                 BatchData batchData = holder.getByUUid(uuid);
@@ -116,10 +115,7 @@ public class OfBatchExecutor {
                 hasFlows |= batchData.isFlow();
                 hasMeters |= batchData.isMeter();
                 hasGroups |= batchData.isGroup();
-                OFMessage previous = stageMessages.put(holder.getSpeakerDataByUUid(uuid), batchData.getMessage());
-                if (previous != null) {
-                    log.warn("Command with uuid {} already processed.", uuid);
-                }
+                stageMessages.add(batchData);
             } else {
                 Map<UUID, String> blockingDependencies = holder.getBlockingDependencies(uuid);
                 if (!blockingDependencies.isEmpty()) {
@@ -137,7 +133,9 @@ public class OfBatchExecutor {
             removeAlreadyExists(stageMessages);
         }
 
-        Collection<OFMessage> ofMessages = stageMessages.values();
+        Collection<OFMessage> ofMessages = stageMessages.stream()
+                .map(BatchData::getMessage)
+                .collect(Collectors.toList());
         List<CompletableFuture<Optional<OFMessage>>> requests = new ArrayList<>();
         try (Session session = sessionService.open(messageContext, iofSwitch)) {
             for (OFMessage message : ofMessages) {
@@ -175,7 +173,7 @@ public class OfBatchExecutor {
                 });
     }
 
-    private void removeAlreadyExists(Map<SpeakerData, OFMessage> stageMessages) {
+    private void removeAlreadyExists(List<BatchData> stageMessages) {
         if (hasMeters) {
             meterStats = switchDataProvider.getMeters();
         }
@@ -186,15 +184,15 @@ public class OfBatchExecutor {
         try {
             List<SpeakerData> speakerData = new ArrayList<>(meterStats.get());
             speakerData.addAll(groupStats.get());
-            Set<SpeakerData> toRemove = new HashSet<>();
-            for (Entry<SpeakerData, OFMessage> entry : stageMessages.entrySet()) {
-                if (speakerData.contains(entry.getKey())) {
-                    toRemove.add(entry.getKey());
+            Set<BatchData> toRemove = new HashSet<>();
+            for (BatchData batchData : stageMessages) {
+                if (speakerData.contains(batchData.getOrigin())) {
+                    toRemove.add(batchData);
                 }
             }
             toRemove.forEach(data -> {
                 log.debug("OpenFlow entry is already exist. Skipping command {}", data);
-                holder.recordSuccessUuid(data.getUuid());
+                holder.recordSuccessUuid(data.getOrigin().getUuid());
                 stageMessages.remove(data);
             });
         } catch (ExecutionException | InterruptedException e) {
diff --git a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfBatchHolder.java b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfBatchHolder.java
index f39445a54..b5d430a9e 100644
--- a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfBatchHolder.java
+++ b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfBatchHolder.java
@@ -15,8 +15,6 @@
 
 package org.openkilda.floodlight.command.rulemanager;
 
-import static java.lang.String.format;
-
 import org.openkilda.floodlight.api.request.rulemanager.OfEntityBatch;
 import org.openkilda.floodlight.api.response.rulemanager.SpeakerCommandResponse;
 import org.openkilda.floodlight.converter.rulemanager.OfFlowConverter;
@@ -30,7 +28,6 @@ import org.openkilda.model.cookie.CookieBase;
 import org.openkilda.rulemanager.FlowSpeakerData;
 import org.openkilda.rulemanager.GroupSpeakerData;
 import org.openkilda.rulemanager.MeterSpeakerData;
-import org.openkilda.rulemanager.SpeakerData;
 
 import lombok.extern.slf4j.Slf4j;
 import net.floodlightcontroller.core.internal.IOFSwitchService;
@@ -45,7 +42,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
-import java.util.stream.Stream;
 
 @Slf4j
 public class OfBatchHolder implements OfEntityBatch {
@@ -142,17 +138,6 @@ public class OfBatchHolder implements OfEntityBatch {
         return commandMap.get(uuid);
     }
 
-    /**
-     * Get speaker data by UUID.
-     */
-    public SpeakerData getSpeakerDataByUUid(UUID uuid) {
-        return Stream.of(flowsMap.values(), metersMap.values(), groupsMap.values())
-                .flatMap(Collection::stream)
-                .filter(data -> uuid.equals(data.getUuid()))
-                .findFirst()
-                .orElseThrow(() -> new IllegalStateException(format("Can't find speaker data with uuid %s", uuid)));
-    }
-
     public MeterSpeakerData getByMeterId(MeterId meterId) {
         return metersMap.get(meterId);
     }
@@ -185,7 +170,12 @@ public class OfBatchHolder implements OfEntityBatch {
         OFMessage message = OfFlowConverter.INSTANCE.convertInstallFlowCommand(data, factory);
         xidMapping.put(message.getXid(), data.getUuid());
         flowsMap.put(data.getCookie(), data);
-        BatchData batchData = BatchData.builder().flow(true).message(message).presenceBeVerified(true).build();
+        BatchData batchData = BatchData.builder()
+                .flow(true)
+                .message(message)
+                .origin(data)
+                .presenceBeVerified(true)
+                .build();
         addBatchData(data.getUuid(), batchData, data.getDependsOn());
     }
 
@@ -195,10 +185,14 @@ public class OfBatchHolder implements OfEntityBatch {
         OFFactory factory = iofSwitchService.getSwitch(dpId).getOFFactory();
         OFMessage message = OfFlowConverter.INSTANCE.convertModifyFlowCommand(data, factory);
         xidMapping.put(message.getXid(), data.getUuid());
-        BatchData batchData = BatchData.builder().flow(true).message(message).presenceBeVerified(true).build();
-        commandMap.put(data.getUuid(), batchData);
+        BatchData batchData = BatchData.builder()
+                .flow(true)
+                .message(message)
+                .origin(data)
+                .presenceBeVerified(true)
+                .build();
         flowsMap.put(data.getCookie(), data);
-        executionGraph.add(data.getUuid(), data.getDependsOn());
+        addBatchData(data.getUuid(), batchData, data.getDependsOn());
     }
 
     @Override
@@ -208,7 +202,12 @@ public class OfBatchHolder implements OfEntityBatch {
         OFMessage message = OfFlowConverter.INSTANCE.convertDeleteFlowCommand(data, factory);
         xidMapping.put(message.getXid(), data.getUuid());
         flowsMap.put(data.getCookie(), data);
-        BatchData batchData = BatchData.builder().flow(true).message(message).presenceBeVerified(false).build();
+        BatchData batchData = BatchData.builder()
+                .flow(true)
+                .message(message)
+                .origin(data)
+                .presenceBeVerified(false)
+                .build();
         addBatchData(data.getUuid(), batchData, data.getDependsOn());
     }
 
@@ -219,7 +218,12 @@ public class OfBatchHolder implements OfEntityBatch {
         OFMessage message = OfMeterConverter.INSTANCE.convertInstallMeterCommand(data, factory);
         xidMapping.put(message.getXid(), data.getUuid());
         metersMap.put(data.getMeterId(), data);
-        BatchData batchData = BatchData.builder().meter(true).message(message).presenceBeVerified(true).build();
+        BatchData batchData = BatchData.builder()
+                .meter(true)
+                .message(message)
+                .origin(data)
+                .presenceBeVerified(true)
+                .build();
         addBatchData(data.getUuid(), batchData, data.getDependsOn());
     }
 
@@ -229,10 +233,14 @@ public class OfBatchHolder implements OfEntityBatch {
         OFFactory factory = iofSwitchService.getSwitch(dpId).getOFFactory();
         OFMessage message = OfMeterConverter.INSTANCE.convertModifyMeterCommand(data, factory);
         xidMapping.put(message.getXid(), data.getUuid());
-        BatchData batchData = BatchData.builder().meter(true).message(message).presenceBeVerified(true).build();
-        commandMap.put(data.getUuid(), batchData);
+        BatchData batchData = BatchData.builder()
+                .meter(true)
+                .message(message)
+                .origin(data)
+                .presenceBeVerified(true)
+                .build();
         metersMap.put(data.getMeterId(), data);
-        executionGraph.add(data.getUuid(), data.getDependsOn());
+        addBatchData(data.getUuid(), batchData, data.getDependsOn());
     }
 
     @Override
@@ -242,7 +250,12 @@ public class OfBatchHolder implements OfEntityBatch {
         OFMessage message = OfMeterConverter.INSTANCE.convertDeleteMeterCommand(data, factory);
         xidMapping.put(message.getXid(), data.getUuid());
         metersMap.put(data.getMeterId(), data);
-        BatchData batchData = BatchData.builder().meter(true).message(message).presenceBeVerified(false).build();
+        BatchData batchData = BatchData.builder()
+                .meter(true)
+                .message(message)
+                .origin(data)
+                .presenceBeVerified(false)
+                .build();
         addBatchData(data.getUuid(), batchData, data.getDependsOn());
     }
 
@@ -253,7 +266,12 @@ public class OfBatchHolder implements OfEntityBatch {
         OFMessage message = OfGroupConverter.INSTANCE.convertInstallGroupCommand(data, factory);
         xidMapping.put(message.getXid(), data.getUuid());
         groupsMap.put(data.getGroupId(), data);
-        BatchData batchData = BatchData.builder().group(true).message(message).presenceBeVerified(true).build();
+        BatchData batchData = BatchData.builder()
+                .group(true)
+                .message(message)
+                .origin(data)
+                .presenceBeVerified(true)
+                .build();
         addBatchData(data.getUuid(), batchData, data.getDependsOn());
     }
 
@@ -263,10 +281,14 @@ public class OfBatchHolder implements OfEntityBatch {
         OFFactory factory = iofSwitchService.getSwitch(dpId).getOFFactory();
         OFMessage message = OfGroupConverter.INSTANCE.convertModifyGroupCommand(data, factory);
         xidMapping.put(message.getXid(), data.getUuid());
-        BatchData batchData = BatchData.builder().group(true).message(message).presenceBeVerified(true).build();
+        BatchData batchData = BatchData.builder()
+                .group(true)
+                .message(message)
+                .origin(data)
+                .presenceBeVerified(true)
+                .build();
         commandMap.put(data.getUuid(), batchData);
-        groupsMap.put(data.getGroupId(), data);
-        executionGraph.add(data.getUuid(), data.getDependsOn());
+        addBatchData(data.getUuid(), batchData, data.getDependsOn());
     }
 
     @Override
@@ -276,7 +298,12 @@ public class OfBatchHolder implements OfEntityBatch {
         OFMessage message = OfGroupConverter.INSTANCE.convertDeleteGroupCommand(data, factory);
         xidMapping.put(message.getXid(), data.getUuid());
         groupsMap.put(data.getGroupId(), data);
-        BatchData batchData = BatchData.builder().group(true).message(message).presenceBeVerified(false).build();
+        BatchData batchData = BatchData.builder()
+                .group(true)
+                .message(message)
+                .origin(data)
+                .presenceBeVerified(false)
+                .build();
         addBatchData(data.getUuid(), batchData, data.getDependsOn());
     }