Repository: telstra/open-kilda
Issue: Unhandled exception in org.openkilda.wfm.topology.nbworker.bolts.FlowOperationsBolt

Sometimes we receive the below exception.
[error11153.txt](https://github.com/telstra/open-kilda/files/4884154/error11153.txt)
search reference:
```
Duplicate key Flow(entityId=22849, flowId=03Jul135507_722_capsicum2093, srcSwitch=Switch(entityId=83, switchId=00:00:00:00:00:00:00:01, status=INACTIVE, socketAddress=/172.18.0.24:40782, hostname=lab_service-1.kilda-controller_default, controller=null, description=Nicira, Inc. OF_13 2.12.0, ofVersion=OF_13, ofDescriptionManufacturer=Nicira, Inc., ofDescriptionHardware=Open vSwitch, ofDescriptionSoftware=2.12.0, ofDescriptionSerialNumber=None, ofDescriptionDatapath=None, features=[RESET_COUNTS_FLAG, GROUP_PACKET_OUT_CONTROLLER, PKTPS_FLAG, METERS, MULTI_TABLE, MATCH_UDP_PORT], underMaintenance=false, timeCreate=null, timeModify=null, pop=null, latitude=0.0, longitude=0.0, street=null, city=null, country=null), destSwitch=Switch(entityId=83, switchId=00:00:00:00:00:00:00:01, status=INACTIVE, socketAddress=/172.18.0.24:40782, hostname...
```

Changed files:
src-java/nbworker-topology/nbworker-storm-topology/src/main/java/org/openkilda/wfm/topology/nbworker/services/FlowOperationsService.java

diff --git a/src-java/nbworker-topology/nbworker-storm-topology/src/main/java/org/openkilda/wfm/topology/nbworker/services/FlowOperationsService.java b/src-java/nbworker-topology/nbworker-storm-topology/src/main/java/org/openkilda/wfm/topology/nbworker/services/FlowOperationsService.java
index a34dd59e6..3621c6c3d 100644
--- a/src-java/nbworker-topology/nbworker-storm-topology/src/main/java/org/openkilda/wfm/topology/nbworker/services/FlowOperationsService.java
+++ b/src-java/nbworker-topology/nbworker-storm-topology/src/main/java/org/openkilda/wfm/topology/nbworker/services/FlowOperationsService.java
@@ -72,6 +72,7 @@ import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 @Slf4j
 public class FlowOperationsService {
@@ -182,26 +183,27 @@ public class FlowOperationsService {
         if (!switchRepository.findById(switchId).isPresent()) {
             throw new SwitchNotFoundException(switchId);
         }
-        Set<Flow> flows = new HashSet<>();
 
         if (port != null) {
-            flowPathRepository.findBySegmentEndpoint(switchId, port).stream()
-                    // NOTE(tdurakov): filter out paths here that are orphaned for the flow
-                    .filter(flowPath -> flowPath.getFlow().isActualPathId(flowPath.getPathId()))
-                    .map(FlowPath::getFlow)
-                    .forEach(flows::add);
-            flows.addAll(flowRepository.findByEndpoint(switchId, port));
+            return getFlowsForEndpoint(flowPathRepository.findBySegmentEndpoint(switchId, port),
+                    flowRepository.findByEndpoint(switchId, port));
         } else {
-            flowPathRepository.findBySegmentSwitch(switchId).stream()
-                    // NOTE(tdurakov): filter out paths here that are orphaned for the flow
-                    .filter(flowPath -> flowPath.getFlow().isActualPathId(flowPath.getPathId()))
-                    .map(FlowPath::getFlow)
-                    .forEach(flows::add);
-            flows.addAll(flowRepository.findByEndpointSwitch(switchId));
+            return getFlowsForEndpoint(flowPathRepository.findBySegmentSwitch(switchId),
+                    flowRepository.findByEndpointSwitch(switchId));
         }
+    }
+
+    private Collection<Flow> getFlowsForEndpoint(Collection<FlowPath> flowPaths,
+                                                 Collection<Flow> flows) {
+        Stream<Flow> flowBySegment = flowPaths.stream()
+                // NOTE(tdurakov): filter out paths here that are orphaned for the flow
+                .filter(flowPath -> flowPath.getFlow().isActualPathId(flowPath.getPathId()))
+                .map(FlowPath::getFlow);
         // need to return Flows unique by id
-        return flows.stream()
-                .collect(Collectors.toMap(Flow::getFlowId, Function.identity()))
+        // Due to possible race condition we can have one flow with different flow paths
+        // In this case we should get the last one
+        return Stream.concat(flowBySegment, flows.stream())
+                .collect(Collectors.toMap(Flow::getFlowId, Function.identity(), (flow1, flow2) -> flow2))
                 .values();
     }