Hz: remove listeners on deactivate

Change-Id: I544917508cd4b9513e3fcd3a100c44928954f413

@@ -64,6 +64,8 @@ public class DistributedClusterStore
     private final MembershipListener listener = new InternalMembershipListener();
     private final Map<NodeId, State> states = new ConcurrentHashMap<>();
 
+    private String nodesListenerId;
+
     @Override
     @Activate
     public void activate() {
@@ -74,7 +76,7 @@ public class DistributedClusterStore
         OptionalCacheLoader<NodeId, DefaultControllerNode> nodeLoader
                 = new OptionalCacheLoader<>(serializer, rawNodes);
         nodes = new AbsentInvalidatingLoadingCache<>(newBuilder().build(nodeLoader));
-        rawNodes.addEntryListener(new RemoteCacheEventHandler<>(nodes), true);
+        nodesListenerId = rawNodes.addEntryListener(new RemoteCacheEventHandler<>(nodes), true);
 
         loadClusterNodes();
 
@@ -90,6 +92,7 @@ public class DistributedClusterStore
 
     @Deactivate
     public void deactivate() {
+        rawNodes.removeEntryListener(nodesListenerId);
         theInstance.getCluster().removeMembershipListener(listenerId);
         log.info("Stopped");
     }

@@ -22,6 +22,7 @@ import com.hazelcast.core.EntryEvent;
 import com.hazelcast.core.EntryListener;
 import com.hazelcast.core.IAtomicLong;
 import com.hazelcast.core.MapEvent;
+
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
@@ -54,6 +55,8 @@ public class DistributedApplicationIdStore
 
     protected Map<Short, DefaultApplicationId> appIds = new ConcurrentHashMap<>();
 
+    private String listenerId;
+
 
     @Override
     @Activate
@@ -73,7 +76,7 @@ public class DistributedApplicationIdStore
         lastAppId = theInstance.getAtomicLong("applicationId");
 
         appIdsByName = new SMap<>(theInstance.<byte[], byte[]>getMap("appIdsByName"), this.serializer);
-        appIdsByName.addEntryListener((new RemoteAppIdEventHandler()), true);
+        listenerId = appIdsByName.addEntryListener((new RemoteAppIdEventHandler()), true);
 
         primeAppIds();
 
@@ -82,6 +85,7 @@ public class DistributedApplicationIdStore
 
     @Deactivate
     public void deactivate() {
+        appIdsByName.removeEntryListener(listenerId);
         log.info("Stopped");
     }
 

@@ -107,6 +107,8 @@ public class HazelcastIntentStore
     private Timer getIntentTimer;
     private Timer getIntentStateTimer;
 
+    private String listenerId;
+
     private Timer createResponseTimer(String methodName) {
         return createTimer("IntentStore", methodName, "responseTime");
     }
@@ -150,7 +152,7 @@ public class HazelcastIntentStore
         IMap<byte[], byte[]> rawStates = super.theInstance.getMap("intent-states");
         states = new SMap<>(rawStates , super.serializer);
         EntryListener<IntentId, IntentState> listener = new RemoteIntentStateListener();
-        states.addEntryListener(listener , true);
+        listenerId = states.addEntryListener(listener , true);
 
         transientStates.clear();
 
@@ -163,6 +165,7 @@ public class HazelcastIntentStore
 
     @Deactivate
     public void deactivate() {
+        states.removeEntryListener(listenerId);
         log.info("Stopped");
     }
 

@@ -74,6 +74,8 @@ public class DistributedMastershipStore
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected ClusterService clusterService;
 
+    private String listenerId;
+
     @Override
     @Activate
     public void activate() {
@@ -91,7 +93,7 @@ public class DistributedMastershipStore
         };
 
         roleMap = new SMap<>(theInstance.<byte[], byte[]>getMap("nodeRoles"), this.serializer);
-        roleMap.addEntryListener((new RemoteMasterShipEventHandler()), true);
+        listenerId = roleMap.addEntryListener((new RemoteMasterShipEventHandler()), true);
         terms = new SMap<>(theInstance.<byte[], byte[]>getMap("terms"), this.serializer);
 
         log.info("Started");
@@ -99,6 +101,7 @@ public class DistributedMastershipStore
 
     @Deactivate
     public void deactivate() {
+        roleMap.removeEntryListener(listenerId);
         log.info("Stopped");
     }