adress comments

Author: xyuanlu

meta-client/src/main/java/org/apache/helix/metaclient/recipes/leaderelection/LeaderElectionClient.java

@@ -195,7 +195,8 @@ private void createPathIfNotExists(String path) {
   }
 
   private void subscribeAndTryCreateLeaderEntry(String leaderPath) {
-    _metaClient.subscribeDataChange(leaderPath + LEADER_ENTRY_KEY, _reElectListener, false);
+    _leaderGroups.add(leaderPath + LEADER_ENTRY_KEY);
+    registerAllListeners();
     LeaderInfo leaderInfo = new LeaderInfo(LEADER_ENTRY_KEY);
     leaderInfo.setLeaderName(_participant);
     leaderInfo.setAcquiredTime();
@@ -215,8 +216,6 @@ private void subscribeAndTryCreateLeaderEntry(String leaderPath) {
     } catch (MetaClientNodeExistsException ex) {
       LOG.info("Already a leader in leader group {}.", leaderPath);
     }
-
-    _leaderGroups.add(leaderPath + LEADER_ENTRY_KEY);
   }
 
   /**
@@ -356,7 +355,7 @@ public List<String> getParticipants(String leaderPath) {
    */
   public boolean subscribeLeadershipChanges(String leaderPath, LeaderElectionListenerInterface listener) {
     LeaderElectionListenerInterfaceAdapter adapter = new LeaderElectionListenerInterfaceAdapter(leaderPath, listener);
-    _leaderChangeListeners.computeIfAbsent(leaderPath, k -> ConcurrentHashMap.newKeySet()).add(adapter);
+    _leaderChangeListeners.computeIfAbsent(leaderPath + LEADER_ENTRY_KEY, k -> ConcurrentHashMap.newKeySet()).add(adapter);
     _metaClient.subscribeDataChange(leaderPath + LEADER_ENTRY_KEY,
         adapter, false /*skipWatchingNonExistNode*/); // we need to subscribe event when path is not there
     _metaClient.subscribeStateChanges(adapter);
@@ -372,7 +371,7 @@ public void unsubscribeLeadershipChanges(String leaderPath, LeaderElectionListen
     _metaClient.unsubscribeDataChange(leaderPath + LEADER_ENTRY_KEY, adapter
         );
     _metaClient.unsubscribeConnectStateChanges(adapter);
-    _leaderChangeListeners.get(leaderPath).remove(adapter);
+    _leaderChangeListeners.get(leaderPath + LEADER_ENTRY_KEY).remove(adapter);
   }
 
   @Override
@@ -430,23 +429,8 @@ public void handleConnectStateChanged(MetaClientInterface.ConnectState prevState
             LOG.info("Participant {} already in leader group {}.", _participant, leaderPath);
           }
         }
-        // resubscribe the  re-elect listener
-        for (String leaderPath : _leaderGroups) {
-            LOG.info("Resubscribe re-elect listener for leaderPath {}.", leaderPath);
-          _metaClient.subscribeDataChange(leaderPath + LEADER_ENTRY_KEY, _reElectListener, false);
-        }
 
-        // resubscribe to leader entry change since we are reconnected
-        for (Map.Entry<String, Set<LeaderElectionListenerInterfaceAdapter>> entry: _leaderChangeListeners.entrySet()) {
-          LOG.info("Resubscribe leader change listener for leaderPath {}.", entry.getKey());
-          String leaderPath = entry.getKey();
-          Set<LeaderElectionListenerInterfaceAdapter> listeners = entry.getValue();
-          for (LeaderElectionListenerInterfaceAdapter listener : listeners) {
-            _metaClient.subscribeDataChange(leaderPath + LEADER_ENTRY_KEY,
-                    listener, false /*skipWatchingNonExistNode*/); // we need to subscribe event when path is not there
-            _metaClient.subscribeStateChanges(listener);
-          }
-        }
+        registerAllListeners();
         // touch leader node to renew session ID
         touchLeaderNode();
       }
@@ -482,6 +466,26 @@ private void touchLeaderNode() {
     }
   }
 
+  private void registerAllListeners(){
+    // resubscribe the  re-elect listener
+    for (String leaderPath : _leaderGroups) {
+      LOG.info("Subscribe re-elect listener for leaderPath {}.", leaderPath);
+      _metaClient.subscribeDataChange(leaderPath, _reElectListener, false);
+    }
+
+    // resubscribe to leader entry change since we are reconnected
+    for (Map.Entry<String, Set<LeaderElectionListenerInterfaceAdapter>> entry: _leaderChangeListeners.entrySet()) {
+      String leaderPath = entry.getKey();
+      LOG.info("Subscribe leader change listener for leaderPath {}.",leaderPath);
+      Set<LeaderElectionListenerInterfaceAdapter> listeners = entry.getValue();
+      for (LeaderElectionListenerInterfaceAdapter listener : listeners) {
+        _metaClient.subscribeDataChange(leaderPath,
+                listener, false /*skipWatchingNonExistNode*/); // we need to subscribe event when path is not there
+        _metaClient.subscribeStateChanges(listener);
+      }
+    }
+  }
+
   public MetaClientInterface getMetaClient() {
     return _metaClient;
   }

meta-client/src/test/java/org/apache/helix/metaclient/recipes/leaderelection/TestLeaderElection.java

@@ -446,6 +446,26 @@ public void onLeadershipChange(String leaderPath, ChangeType type, String curLea
       return (clt1.getLeader(leaderPath) != null);
     }, MetaClientTestUtil.WAIT_DURATION));
 
+    // have clt2 join and clt1 leave and join again, and verify listener still works
+    clt2.joinLeaderElectionParticipantPool(leaderPath, participantInfo2);
+    clt1.exitLeaderElectionParticipantPool(leaderPath);
+
+    Assert.assertTrue(MetaClientTestUtil.verify(() -> {
+      return (clt2.getLeader(leaderPath) != null);
+    }, MetaClientTestUtil.WAIT_DURATION));
+    Assert.assertTrue(MetaClientTestUtil.verify(() -> {
+      return (clt2.getLeader(leaderPath).equals(PARTICIPANT_NAME2));
+    }, MetaClientTestUtil.WAIT_DURATION));
+
+    // now clt1 join again, and verify listener still works
+    clt1.joinLeaderElectionParticipantPool(leaderPath, participantInfo);
+    clt2.exitLeaderElectionParticipantPool(leaderPath);
+    // verify clt1 is leader
+    Assert.assertTrue(MetaClientTestUtil.verify(() -> {
+      return (clt1.getLeader(leaderPath) != null);
+    }, MetaClientTestUtil.WAIT_DURATION));
+
+
     ((ZkMetaClient<?>) clt1.getMetaClient()).close();
     System.out.println("END TestLeaderElection.testClientClosedAndReconnectAfterExpire");
   }