Repository: telstra/open-kilda
Issue: Can't patch flow with empty vlan stats and non zeros src/dst vlans

**Steps to reproduce:**
1. Create a flow with non empty Vlans stats and with 0 src and dst vlans via v2 API `POST /v2/flows`
```json
{
  "flow_id": "flow1",
  "source": {
    "inner_vlan_id": 0,
    "port_number": 10,
    "switch_id": "00:00:00:00:00:00:00:01",
    "vlan_id": 0
  },
  "destination": {
    "inner_vlan_id": 0,
    "port_number": 0,
    "switch_id": "00:00:00:00:00:00:00:02",
    "vlan_id": 0
  },
  "maximum_bandwidth": 10000,
  "statistics": {
    "vlans": [
      1,2,3
    ]
  }
}
```
2. Patch the flow with set vlans and empty stat vlans `PATCH /v2/flows/{flow_id}`
```json
{
  "source": {
    "vlan_id": 15
  },
  "destination": {
     "vlan_id": 17
  },
  "statistics": {
    "vlans": []
  }
}
```

**Expected result:**
Successful response:

**Actual result:**
`To collect vlan statistics you need to set source or destination vlan_id to zero`

Changed files:
src-java/nbworker-topology/nbworker-storm-topology/src/main/java/org/openkilda/wfm/topology/nbworker/services/FlowOperationsService.java
src-java/nbworker-topology/nbworker-storm-topology/src/test/java/org/openkilda/wfm/topology/nbworker/services/FlowOperationsServiceTest.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 baed2cba6..c1cd1ed1c 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
@@ -572,8 +572,7 @@ public class FlowOperationsService {
                     + "at the same time");
         }
 
-        if ((flow.getVlanStatistics() != null && !flow.getVlanStatistics().isEmpty())
-                || (flowPatch.getVlanStatistics() != null && !flowPatch.getVlanStatistics().isEmpty())) {
+        if (!isVlanStatisticsEmpty(flowPatch, flow)) {
             boolean zeroResultSrcVlan = isResultingVlanValueIsZero(flowPatch.getSource(), flow.getSrcVlan());
             boolean zeroResultDstVlan = isResultingVlanValueIsZero(flowPatch.getDestination(), flow.getDestVlan());
 
@@ -733,6 +732,13 @@ public class FlowOperationsService {
                         .collect(Collectors.toSet());
     }
 
+    private static boolean isVlanStatisticsEmpty(FlowPatch flowPatch, Flow flow) {
+        if (flowPatch.getVlanStatistics() != null) {
+            return flowPatch.getVlanStatistics().isEmpty();
+        }
+        return flow.getVlanStatistics() == null || flow.getVlanStatistics().isEmpty();
+    }
+
     @Data
     @Builder
     static class UpdateFlowResult {
diff --git a/src-java/nbworker-topology/nbworker-storm-topology/src/test/java/org/openkilda/wfm/topology/nbworker/services/FlowOperationsServiceTest.java b/src-java/nbworker-topology/nbworker-storm-topology/src/test/java/org/openkilda/wfm/topology/nbworker/services/FlowOperationsServiceTest.java
index d8db615fc..2eb4d40f3 100644
--- a/src-java/nbworker-topology/nbworker-storm-topology/src/test/java/org/openkilda/wfm/topology/nbworker/services/FlowOperationsServiceTest.java
+++ b/src-java/nbworker-topology/nbworker-storm-topology/src/test/java/org/openkilda/wfm/topology/nbworker/services/FlowOperationsServiceTest.java
@@ -80,8 +80,10 @@ public class FlowOperationsServiceTest extends InMemoryGraphBasedTest {
     public static final SwitchId SWITCH_ID_3 = new SwitchId(3);
     public static final SwitchId SWITCH_ID_4 = new SwitchId(4);
     public static final int VLAN_1 = 1;
-    public static final int PORT_1 = 1;
-    public static final int PORT_2 = 2;
+    public static final int PORT_1 = 2;
+    public static final int PORT_2 = 3;
+    public static final int VLAN_2 = 4;
+    public static final int VLAN_3 = 5;
 
     private static FlowOperationsService flowOperationsService;
     private static FlowRepository flowRepository;
@@ -255,28 +257,111 @@ public class FlowOperationsServiceTest extends InMemoryGraphBasedTest {
     }
 
     @Test
-    public void updateVlanStatisticsToZeroDstVlanIsZeroTest() throws FlowNotFoundException, InvalidFlowException {
-        runUpdateVlanStatisticsToZero(VLAN_1, 0);
+    public void updateVlanStatisticsToZeroOldSrcAndDstVlanAreZeroTest()
+            throws FlowNotFoundException, InvalidFlowException {
+        runUpdateVlanStatisticsToZeroTest(0, 0, VLAN_1, VLAN_2);
+    }
+
+    @Test
+    public void updateVlanStatisticsToZeroOldDstVlanAreNotZeroTest()
+            throws FlowNotFoundException, InvalidFlowException {
+        runUpdateVlanStatisticsToZeroTest(VLAN_3, 0, VLAN_1, VLAN_2);
+    }
+
+    @Test
+    public void updateVlanStatisticsToZeroOldSrcVlanAreNotZeroTest()
+            throws FlowNotFoundException, InvalidFlowException {
+        runUpdateVlanStatisticsToZeroTest(0, VLAN_3, VLAN_1, VLAN_2);
+    }
+
+    @Test
+    public void updateVlanStatisticsToZeroDstVlanIsZeroNewVlansNullTest()
+            throws FlowNotFoundException, InvalidFlowException {
+        runUpdateVlanStatisticsToZeroTest(VLAN_1, 0, null, null);
     }
 
     @Test
-    public void updateVlanStatisticsToZeroSrcVlanIsZeroTest() throws FlowNotFoundException, InvalidFlowException {
-        runUpdateVlanStatisticsToZero(0, VLAN_1);
+    public void updateVlanStatisticsToZeroSrcVlanIsZeroNewVlansNullTest()
+            throws FlowNotFoundException, InvalidFlowException {
+        runUpdateVlanStatisticsToZeroTest(0, VLAN_1, null, null);
     }
 
     @Test
-    public void updateVlanStatisticsToZeroSrcAndVlanAreZeroTest() throws FlowNotFoundException, InvalidFlowException {
-        runUpdateVlanStatisticsToZero(0, 0);
+    public void updateVlanStatisticsToZeroSrcAndDstVlanAreZeroNewVlansNullTest()
+            throws FlowNotFoundException, InvalidFlowException {
+        runUpdateVlanStatisticsToZeroTest(0, 0, null, null);
+    }
+
+    @Test
+    public void updateVlanStatisticsToZeroDstVlanIsZeroNewVlansZerosTest()
+            throws FlowNotFoundException, InvalidFlowException {
+        runUpdateVlanStatisticsToZeroTest(VLAN_1, 0, 0, 0);
+    }
+
+    @Test
+    public void updateVlanStatisticsToZeroSrcVlanIsZeroNewVlansZerosTest()
+            throws FlowNotFoundException, InvalidFlowException {
+        runUpdateVlanStatisticsToZeroTest(0, VLAN_1, 0, 0);
+    }
+
+    @Test
+    public void updateVlanStatisticsToZeroSrcAndDstVlanAreZeroNewVlansZerosTest()
+            throws FlowNotFoundException, InvalidFlowException {
+        runUpdateVlanStatisticsToZeroTest(0, 0, 0, 0);
+    }
+
+    private void runUpdateVlanStatisticsToZeroTest(
+            int oldSrcVlan, int oldDstVlan, Integer newSrcVlan, Integer newDstVLan)
+            throws FlowNotFoundException, InvalidFlowException {
+        Set<Integer> originalVlanStatistics = Sets.newHashSet(1, 2, 3);
+        Flow flow = new TestFlowBuilder()
+                .flowId(FLOW_ID_1)
+                .srcSwitch(switchA)
+                .srcVlan(oldSrcVlan)
+                .destSwitch(switchB)
+                .destVlan(oldDstVlan)
+                .vlanStatistics(originalVlanStatistics)
+                .build();
+        flowRepository.add(flow);
+
+        FlowPatch receivedFlow = FlowPatch.builder()
+                .flowId(FLOW_ID_1)
+                .vlanStatistics(new HashSet<>())
+                .source(buildPathEndpoint(newSrcVlan))
+                .destination(buildPathEndpoint(newDstVLan))
+                .build();
+
+        Flow updatedFlow = flowOperationsService.updateFlow(new FlowCarrierImpl(), receivedFlow);
+        assertTrue(updatedFlow.getVlanStatistics().isEmpty());
     }
 
     @Test(expected = IllegalArgumentException.class)
-    public void unableToUpdateVlanStatisticsTest() throws FlowNotFoundException, InvalidFlowException {
+    public void unableToUpdateVlanStatisticsOldVlansSetNewVlansNullTest()
+            throws FlowNotFoundException, InvalidFlowException {
+        runUnableToUpdateVlanStatisticsTest(VLAN_1, VLAN_2, null, null);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void unableToUpdateVlanStatisticsOldVlansSetNewVlansSetTest()
+            throws FlowNotFoundException, InvalidFlowException {
+        runUnableToUpdateVlanStatisticsTest(VLAN_1, VLAN_2, VLAN_2, VLAN_3);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void unableToUpdateVlanStatisticsOldVlansZeroNewVlansSetTest()
+            throws FlowNotFoundException, InvalidFlowException {
+        runUnableToUpdateVlanStatisticsTest(0, 0, VLAN_1, VLAN_2);
+    }
+
+    private void runUnableToUpdateVlanStatisticsTest(
+            int oldSrcVlan, int oldDstVlan, Integer newSrcVlan, Integer newDstVLan)
+            throws FlowNotFoundException, InvalidFlowException {
         Flow flow = new TestFlowBuilder()
                 .flowId(FLOW_ID_1)
                 .srcSwitch(switchA)
-                .srcVlan(VLAN_1)
+                .srcVlan(oldSrcVlan)
                 .destSwitch(switchB)
-                .destVlan(VLAN_1)
+                .destVlan(oldDstVlan)
                 .vlanStatistics(new HashSet<>())
                 .build();
         flowRepository.add(flow);
@@ -284,6 +369,8 @@ public class FlowOperationsServiceTest extends InMemoryGraphBasedTest {
         FlowPatch receivedFlow = FlowPatch.builder()
                 .flowId(FLOW_ID_1)
                 .vlanStatistics(Sets.newHashSet(1, 2, 3))
+                .source(PatchEndpoint.builder().vlanId(newSrcVlan).build())
+                .destination(PatchEndpoint.builder().vlanId(newDstVLan).build())
                 .build();
 
         flowOperationsService.updateFlow(new FlowCarrierImpl(), receivedFlow);
@@ -916,6 +1003,13 @@ public class FlowOperationsServiceTest extends InMemoryGraphBasedTest {
         assertTrue(updatedFlow.getVlanStatistics().isEmpty());
     }
 
+    private static PatchEndpoint buildPathEndpoint(Integer vlan) {
+        if (vlan == null) {
+            return null;
+        }
+        return PatchEndpoint.builder().vlanId(vlan).build();
+    }
+
     private static class FlowCarrierImpl implements FlowOperationsCarrier {
         @Override
         public void emitPeriodicPingUpdate(String flowId, boolean enabled) {