Repository: telstra/open-kilda
Issue: Periodic fl router sync in rare cases may disrupt ISL discovery process on port

**Scenario:**
given: port is in DOWN state
1. Floodlight router periodic sync happens
2. simultaneously with sync a port UP event happens
At this point race appears. Because sync uses correlationId as key when writing to kafka and portUp event uses switchId, the order of events may disrupt, so that portUp event gets immediately followed buy portDown state (comming from sync), causing the related port to be considered by the system as 'Down'. This in turn will prevent a related ISL on that port to be discovered until next periodic sync happens.
**Expected:** Disruption of port events order should not happen

Changed files:
src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/switchmanager/SwitchTrackingService.java
src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/RecordHandler.java
src-java/base-topology/base-messaging/src/main/java/org/openkilda/messaging/info/ChunkedInfoMessage.java
src-java/floodlight-service/floodlight-modules/src/test/java/org/openkilda/floodlight/switchmanager/SwitchTrackingServiceTest.java

diff --git a/src-java/base-topology/base-messaging/src/main/java/org/openkilda/messaging/info/ChunkedInfoMessage.java b/src-java/base-topology/base-messaging/src/main/java/org/openkilda/messaging/info/ChunkedInfoMessage.java
index d3f8db22b..f3b9567a7 100644
--- a/src-java/base-topology/base-messaging/src/main/java/org/openkilda/messaging/info/ChunkedInfoMessage.java
+++ b/src-java/base-topology/base-messaging/src/main/java/org/openkilda/messaging/info/ChunkedInfoMessage.java
@@ -51,11 +51,4 @@ public class ChunkedInfoMessage extends InfoMessage {
         this.messageId = String.join(" : ", String.valueOf(messageIndex), correlationId);
         this.totalMessages = totalMessages;
     }
-
-    public ChunkedInfoMessage(InfoData data, long timestamp, String correlationId,
-                              int messageIndex, int totalMessages, String region) {
-        super(data, timestamp, correlationId, region);
-        this.messageId = String.join(" : ", String.valueOf(messageIndex), correlationId);
-        this.totalMessages = totalMessages;
-    }
 }
diff --git a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/RecordHandler.java b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/RecordHandler.java
index ba4760553..7a56ceac4 100644
--- a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/RecordHandler.java
+++ b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/RecordHandler.java
@@ -191,8 +191,6 @@ class RecordHandler implements Runnable {
 
         try {
             handleCommand(message, replyToTopic, replyDestination);
-        } catch (SwitchOperationException e) {
-            logger.error("Unable to handle request {}: {}", message.getData().getClass().getName(), e.getMessage());
         } catch (FlowCommandException e) {
             String errorMessage = e.getCause() != null ? e.getCause().getMessage() : e.getMessage();
             logger.error("Failed to handle message {}: {}", message, errorMessage);
@@ -205,7 +203,7 @@ class RecordHandler implements Runnable {
     }
 
     private void handleCommand(CommandMessage message, String replyToTopic, Destination replyDestination)
-            throws FlowCommandException, SwitchOperationException {
+            throws FlowCommandException {
         logger.debug("Handling message: '{}'. Reply topic: '{}'. Reply destination: '{}'.",
                 message, replyToTopic, replyDestination);
         CommandData data = message.getData();
@@ -775,11 +773,11 @@ class RecordHandler implements Runnable {
      *
      * @param message NetworkCommandData
      */
-    private void doNetworkDump(final CommandMessage message) throws SwitchOperationException {
+    private void doNetworkDump(final CommandMessage message) {
         logger.info("Processing request from WFM to dump switches. {}", message.getCorrelationId());
 
         SwitchTrackingService switchTracking = context.getModuleContext().getServiceImpl(SwitchTrackingService.class);
-        switchTracking.dumpAllSwitches(message.getCorrelationId());
+        switchTracking.dumpAllSwitches();
     }
 
     private void doInstallSwitchRules(final CommandMessage message) {
diff --git a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/switchmanager/SwitchTrackingService.java b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/switchmanager/SwitchTrackingService.java
index 17cb27d95..00dcdbb1d 100644
--- a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/switchmanager/SwitchTrackingService.java
+++ b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/switchmanager/SwitchTrackingService.java
@@ -28,7 +28,6 @@ import org.openkilda.floodlight.utils.CorrelationContext;
 import org.openkilda.floodlight.utils.FloodlightDashboardLogger;
 import org.openkilda.floodlight.utils.NewCorrelationContextRequired;
 import org.openkilda.messaging.Message;
-import org.openkilda.messaging.info.ChunkedInfoMessage;
 import org.openkilda.messaging.info.InfoData;
 import org.openkilda.messaging.info.InfoMessage;
 import org.openkilda.messaging.info.discovery.NetworkDumpSwitchData;
@@ -77,10 +76,10 @@ public class SwitchTrackingService implements IOFSwitchListener, IService {
     /**
      * Send dump contain all connected at this moment switches.
      */
-    public void dumpAllSwitches(String correlationId) {
+    public void dumpAllSwitches() {
         discoveryLock.writeLock().lock();
         try {
-            dumpAllSwitchesAction(correlationId);
+            dumpAllSwitchesAction();
         } finally {
             discoveryLock.writeLock().unlock();
         }
@@ -167,20 +166,11 @@ public class SwitchTrackingService implements IOFSwitchListener, IService {
         }
     }
 
-    private void dumpAllSwitchesAction(String correlationId) {
+    private void dumpAllSwitchesAction() {
         Collection<IOFSwitch> iofSwitches = switchManager.getAllSwitchMap(true).values();
-        int switchCounter = 0;
         for (IOFSwitch sw : iofSwitches) {
-            try {
-                NetworkDumpSwitchData swData = new NetworkDumpSwitchData(buildSwitch(sw));
-                producerService.sendMessageAndTrack(discoveryTopic,
-                                                correlationId,
-                                                new ChunkedInfoMessage(swData, System.currentTimeMillis(),
-                                                        correlationId, switchCounter, iofSwitches.size(), region));
-            } catch (Exception e) {
-                logger.error("Failed to send network dump for {}", sw.getId());
-            }
-            switchCounter++;
+            NetworkDumpSwitchData payload = new NetworkDumpSwitchData(buildSwitch(sw));
+            emitDiscoveryEvent(sw.getId(), payload);
         }
     }
 
@@ -204,28 +194,33 @@ public class SwitchTrackingService implements IOFSwitchListener, IService {
 
     private void switchDiscoveryAction(DatapathId dpId, SwitchChangeType event) {
         logger.info("Send switch discovery ({} - {})", dpId, event);
-        Message message = null;
+        SwitchInfoData payload = null;
         if (SwitchChangeType.DEACTIVATED != event && SwitchChangeType.REMOVED != event) {
             try {
                 IOFSwitch sw = switchManager.lookupSwitch(dpId);
                 SpeakerSwitchView switchView = buildSwitch(sw);
-                message = buildSwitchMessage(sw, switchView, event);
+                payload = buildSwitchMessage(sw, switchView, event);
             } catch (SwitchNotFoundException e) {
                 logger.error(
                         "Switch {} is not in management state now({}), switch ISL discovery details will be degraded.",
                         dpId, e.getMessage());
             }
         }
-        if (message == null) {
-            message = buildSwitchMessage(dpId, event);
+        if (payload == null) {
+            payload = buildSwitchMessage(dpId, event);
         }
 
-        producerService.sendMessageAndTrack(discoveryTopic, dpId.toString(), message);
+        emitDiscoveryEvent(dpId, payload);
     }
 
     private void portDiscoveryAction(DatapathId dpId, OFPortDesc portDesc, PortChangeType changeType) {
         logger.info("Send port discovery ({}-{} - {})", dpId, portDesc.getPortNo(), changeType);
-        Message message = buildPortMessage(dpId, portDesc, changeType);
+        InfoData payload = OfPortDescConverter.INSTANCE.toPortInfoData(dpId, portDesc, changeType);
+        emitDiscoveryEvent(dpId, payload);
+    }
+
+    private void emitDiscoveryEvent(DatapathId dpId, InfoData payload) {
+        Message message = buildMessage(payload);
         producerService.sendMessageAndTrack(discoveryTopic, dpId.toString(), message);
     }
 
@@ -236,8 +231,8 @@ public class SwitchTrackingService implements IOFSwitchListener, IService {
      * @param eventType type of event
      * @return Message
      */
-    private Message buildSwitchMessage(IOFSwitch sw, SpeakerSwitchView switchView, SwitchChangeType eventType) {
-        return buildMessage(IofSwitchConverter.buildSwitchInfoData(sw, switchView, eventType));
+    private SwitchInfoData buildSwitchMessage(IOFSwitch sw, SpeakerSwitchView switchView, SwitchChangeType eventType) {
+        return IofSwitchConverter.buildSwitchInfoData(sw, switchView, eventType);
     }
 
     /**
@@ -247,21 +242,8 @@ public class SwitchTrackingService implements IOFSwitchListener, IService {
      * @param eventType type of event
      * @return Message
      */
-    private Message buildSwitchMessage(DatapathId dpId, SwitchChangeType eventType) {
-        return buildMessage(new SwitchInfoData(new SwitchId(dpId.getLong()), eventType));
-    }
-
-    /**
-     * Builds a port state change message with port number.
-     *
-     * @param switchId datapathId of switch
-     * @param portDesc port that triggered the event
-     * @param type type of port event
-     * @return Message
-     */
-    private Message buildPortMessage(final DatapathId switchId, final OFPortDesc portDesc, final PortChangeType type) {
-        InfoData data = OfPortDescConverter.INSTANCE.toPortInfoData(switchId, portDesc, type);
-        return buildMessage(data);
+    private SwitchInfoData buildSwitchMessage(DatapathId dpId, SwitchChangeType eventType) {
+        return new SwitchInfoData(new SwitchId(dpId.getLong()), eventType);
     }
 
     /**
diff --git a/src-java/floodlight-service/floodlight-modules/src/test/java/org/openkilda/floodlight/switchmanager/SwitchTrackingServiceTest.java b/src-java/floodlight-service/floodlight-modules/src/test/java/org/openkilda/floodlight/switchmanager/SwitchTrackingServiceTest.java
index 385d06c40..74e7b41b5 100644
--- a/src-java/floodlight-service/floodlight-modules/src/test/java/org/openkilda/floodlight/switchmanager/SwitchTrackingServiceTest.java
+++ b/src-java/floodlight-service/floodlight-modules/src/test/java/org/openkilda/floodlight/switchmanager/SwitchTrackingServiceTest.java
@@ -33,8 +33,9 @@ import org.openkilda.floodlight.error.SwitchNotFoundException;
 import org.openkilda.floodlight.service.FeatureDetectorService;
 import org.openkilda.floodlight.service.kafka.IKafkaProducerService;
 import org.openkilda.floodlight.service.kafka.KafkaUtilityService;
+import org.openkilda.floodlight.utils.CorrelationContext;
+import org.openkilda.floodlight.utils.CorrelationContext.CorrelationContextClosable;
 import org.openkilda.messaging.Message;
-import org.openkilda.messaging.info.ChunkedInfoMessage;
 import org.openkilda.messaging.info.InfoData;
 import org.openkilda.messaging.info.InfoMessage;
 import org.openkilda.messaging.info.discovery.NetworkDumpSwitchData;
@@ -364,12 +365,14 @@ public class SwitchTrackingServiceTest extends EasyMockSupport {
         replayAll();
 
         String correlationId = "unit-test-correlation-id";
-        service.dumpAllSwitches(correlationId);
+        try (CorrelationContextClosable dummy = CorrelationContext.create(correlationId)) {
+            service.dumpAllSwitches();
+        }
 
         verify(producerService);
 
         ArrayList<Message> expectedMessages = new ArrayList<>();
-        expectedMessages.add(new ChunkedInfoMessage(
+        expectedMessages.add(new InfoMessage(
                         new NetworkDumpSwitchData(new SpeakerSwitchView(
                                 new SwitchId(swAid.getLong()),
                                 new InetSocketAddress(Inet4Address.getByName("127.0.1.1"), 32768),
@@ -380,8 +383,8 @@ public class SwitchTrackingServiceTest extends EasyMockSupport {
                                 ImmutableList.of(
                                         new SpeakerSwitchPortView(1, SpeakerSwitchPortView.State.UP),
                                         new SpeakerSwitchPortView(2, SpeakerSwitchPortView.State.UP)))),
-                0, correlationId, 0, 2, "1"));
-        expectedMessages.add(new ChunkedInfoMessage(
+                0, correlationId));
+        expectedMessages.add(new InfoMessage(
                 new NetworkDumpSwitchData(new SpeakerSwitchView(
                         new SwitchId(swBid.getLong()),
                         new InetSocketAddress(Inet4Address.getByName("127.0.1.2"), 32768),
@@ -393,7 +396,7 @@ public class SwitchTrackingServiceTest extends EasyMockSupport {
                                 new SpeakerSwitchPortView(3, SpeakerSwitchPortView.State.UP),
                                 new SpeakerSwitchPortView(4, SpeakerSwitchPortView.State.UP),
                                 new SpeakerSwitchPortView(5, SpeakerSwitchPortView.State.DOWN)))),
-                0, correlationId, 1, 2, "1"));
+                0, correlationId));
         assertEquals(expectedMessages, producedMessages);
     }