FlowRuleStore: Consider errors when updating state of all current backups

Change-Id: I3bf4d20d79dc37c7040648ec6379794b8c93aad2

@@ -290,7 +290,7 @@ public class NewDistributedFlowRuleStore
         clusterCommunicator.addSubscriber(
         clusterCommunicator.addSubscriber(
                 REMOVE_FLOW_ENTRY, SERIALIZER::decode, this::removeFlowRuleInternal, SERIALIZER::encode, executor);
                 REMOVE_FLOW_ENTRY, SERIALIZER::decode, this::removeFlowRuleInternal, SERIALIZER::encode, executor);
         clusterCommunicator.addSubscriber(
         clusterCommunicator.addSubscriber(
-                FLOW_TABLE_BACKUP, SERIALIZER::decode, flowTable::onBackupReceipt, executor);
+                FLOW_TABLE_BACKUP, SERIALIZER::decode, flowTable::onBackupReceipt, SERIALIZER::encode, executor);
     }
     }
 
 
     private void unregisterMessageHandlers() {
     private void unregisterMessageHandlers() {
@@ -644,22 +644,34 @@ public class NewDistributedFlowRuleStore
 
 
         private void backupFlowEntries(NodeId nodeId, Set<DeviceId> deviceIds) {
         private void backupFlowEntries(NodeId nodeId, Set<DeviceId> deviceIds) {
             log.debug("Sending flowEntries for devices {} to {} as backup.", deviceIds, nodeId);
             log.debug("Sending flowEntries for devices {} to {} as backup.", deviceIds, nodeId);
-            Map<DeviceId, ConcurrentMap<FlowId, Set<StoredFlowEntry>>> deviceFlowEntries =
+            Map<DeviceId, Map<FlowId, Set<StoredFlowEntry>>> deviceFlowEntries =
                     Maps.newConcurrentMap();
                     Maps.newConcurrentMap();
             flowEntries.forEach((key, value) -> {
             flowEntries.forEach((key, value) -> {
                 if (deviceIds.contains(key)) {
                 if (deviceIds.contains(key)) {
                     deviceFlowEntries.put(key, value);
                     deviceFlowEntries.put(key, value);
                 }
                 }
             });
             });
-            clusterCommunicator.unicast(deviceFlowEntries,
+            clusterCommunicator.<Map<DeviceId, Map<FlowId, Set<StoredFlowEntry>>>, Set<DeviceId>>sendAndReceive(
+                                        deviceFlowEntries,
                                         FLOW_TABLE_BACKUP,
                                         FLOW_TABLE_BACKUP,
                                         SERIALIZER::encode,
                                         SERIALIZER::encode,
-                    nodeId);
-            deviceIds.forEach(id -> {
+                                        SERIALIZER::decode,
+                                        nodeId)
+                               .whenComplete((backedupDevices, error) -> {
+                                   Set<DeviceId> devicesNotBackedup = error != null ?
+                                           deviceFlowEntries.keySet() :
+                                           Sets.difference(deviceFlowEntries.keySet(), backedupDevices);
+                                   if (devicesNotBackedup.size() > 0) {
+                                       log.warn("Failed to backup devices: {}", devicesNotBackedup, error);
+                                   }
+                                   if (backedupDevices != null) {
+                                       backedupDevices.forEach(id -> {
                                            lastBackupTimes.put(id, System.currentTimeMillis());
                                            lastBackupTimes.put(id, System.currentTimeMillis());
                                            lastBackupNodes.put(id, nodeId);
                                            lastBackupNodes.put(id, nodeId);
                                        });
                                        });
                                    }
                                    }
+                               });
+        }
 
 
         /**
         /**
          * Returns the flow table for specified device.
          * Returns the flow table for specified device.
@@ -751,8 +763,10 @@ public class NewDistributedFlowRuleStore
             }
             }
         }
         }
 
 
-        private void onBackupReceipt(Map<DeviceId, Map<FlowId, Set<StoredFlowEntry>>> flowTables) {
+        private Set<DeviceId> onBackupReceipt(Map<DeviceId, Map<FlowId, Set<StoredFlowEntry>>> flowTables) {
             log.debug("Received flowEntries for {} to backup", flowTables.keySet());
             log.debug("Received flowEntries for {} to backup", flowTables.keySet());
+            Set<DeviceId> backedupDevices = Sets.newHashSet();
+            try {
                 Set<DeviceId> managedDevices = mastershipService.getDevicesOf(local);
                 Set<DeviceId> managedDevices = mastershipService.getDevicesOf(local);
                 // Only process those devices are that not managed by the local node.
                 // Only process those devices are that not managed by the local node.
                 Maps.filterKeys(flowTables, deviceId -> !managedDevices.contains(deviceId))
                 Maps.filterKeys(flowTables, deviceId -> !managedDevices.contains(deviceId))
@@ -760,7 +774,12 @@ public class NewDistributedFlowRuleStore
                     Map<FlowId, Set<StoredFlowEntry>> deviceFlowTable = getFlowTable(deviceId);
                     Map<FlowId, Set<StoredFlowEntry>> deviceFlowTable = getFlowTable(deviceId);
                     deviceFlowTable.clear();
                     deviceFlowTable.clear();
                     deviceFlowTable.putAll(flowTable);
                     deviceFlowTable.putAll(flowTable);
+                    backedupDevices.add(deviceId);
                 });
                 });
+            } catch (Exception e) {
+                log.warn("Failure processing backup request", e);
+            }
+            return backedupDevices;
         }
         }
     }
     }
 }
 }