When relinquishing mastership, wait for LeadershipService::withdraw to complete …

…before returning a MastershipEvent

Change-Id: Ieb0816bf3d17fee28815ea59959ed6d4a4efb478

@@ -127,7 +127,7 @@ public class ConsistentDeviceMastershipStore
         transferExecutor =
         transferExecutor =
                 Executors.newSingleThreadScheduledExecutor(
                 Executors.newSingleThreadScheduledExecutor(
                         groupedThreads("onos/store/device/mastership", "mastership-transfer-executor"));
                         groupedThreads("onos/store/device/mastership", "mastership-transfer-executor"));
-        clusterCommunicator.<DeviceId, MastershipEvent>addSubscriber(ROLE_RELINQUISH_SUBJECT,
+        clusterCommunicator.addSubscriber(ROLE_RELINQUISH_SUBJECT,
                 SERIALIZER::decode,
                 SERIALIZER::decode,
                 this::relinquishLocalRole,
                 this::relinquishLocalRole,
                 SERIALIZER::encode,
                 SERIALIZER::encode,
@@ -310,25 +310,26 @@ public class ConsistentDeviceMastershipStore
         checkArgument(nodeId != null, NODE_ID_NULL);
         checkArgument(nodeId != null, NODE_ID_NULL);
         checkArgument(deviceId != null, DEVICE_ID_NULL);
         checkArgument(deviceId != null, DEVICE_ID_NULL);
 
 
-        if (!nodeId.equals(localNodeId)) {
-            log.debug("Forwarding request to relinquish "
-                    + "role for device {} to {}", deviceId, nodeId);
-            return clusterCommunicator.sendAndReceive(
-                    deviceId,
-                    ROLE_RELINQUISH_SUBJECT,
-                    SERIALIZER::encode,
-                    SERIALIZER::decode,
-                    nodeId);
+        if (nodeId.equals(localNodeId)) {
+            return relinquishLocalRole(deviceId);
         }
         }
-        return CompletableFuture.completedFuture(relinquishLocalRole(deviceId));
+
+        log.debug("Forwarding request to relinquish "
+                + "role for device {} to {}", deviceId, nodeId);
+        return clusterCommunicator.sendAndReceive(
+                deviceId,
+                ROLE_RELINQUISH_SUBJECT,
+                SERIALIZER::encode,
+                SERIALIZER::decode,
+                nodeId);
     }
     }
 
 
-    private MastershipEvent relinquishLocalRole(DeviceId deviceId) {
+    private CompletableFuture<MastershipEvent> relinquishLocalRole(DeviceId deviceId) {
         checkArgument(deviceId != null, DEVICE_ID_NULL);
         checkArgument(deviceId != null, DEVICE_ID_NULL);
 
 
         // Check if this node is can be managed by this node.
         // Check if this node is can be managed by this node.
         if (!connectedDevices.contains(deviceId)) {
         if (!connectedDevices.contains(deviceId)) {
-            return null;
+            return CompletableFuture.completedFuture(null);
         }
         }
 
 
         String leadershipTopic = createDeviceMastershipTopic(deviceId);
         String leadershipTopic = createDeviceMastershipTopic(deviceId);
@@ -339,9 +340,8 @@ public class ConsistentDeviceMastershipStore
             : MastershipEvent.Type.BACKUPS_CHANGED;
             : MastershipEvent.Type.BACKUPS_CHANGED;
 
 
         connectedDevices.remove(deviceId);
         connectedDevices.remove(deviceId);
-        leadershipService.withdraw(leadershipTopic);
-
-        return new MastershipEvent(eventType, deviceId, getNodes(deviceId));
+        return leadershipService.withdraw(leadershipTopic)
+                                .thenApply(v -> new MastershipEvent(eventType, deviceId, getNodes(deviceId)));
     }
     }
 
 
     private MastershipEvent transitionFromMasterToStandby(DeviceId deviceId) {
     private MastershipEvent transitionFromMasterToStandby(DeviceId deviceId) {