Fix liveInstance provider call only in leader node (linkedin#879)

Co-authored-by: Vaibhav Maheshwari <vmaheshw@vmaheshw-mn1.linkedin.biz>
vmaheshw · Mar 1, 2022 · b12ba2a · b12ba2a
1 parent 3e8211d
commit b12ba2a
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 10 deletions.
diff --git a/datastream-server/src/main/java/com/linkedin/datastream/server/zk/ZkAdapter.java b/datastream-server/src/main/java/com/linkedin/datastream/server/zk/ZkAdapter.java
@@ -274,10 +274,6 @@ public void connect(boolean reinitOnNewSession) {
       _leaderElectionListener = new ZkLeaderElectionListener();
     }
 
-    if (_liveInstancesProvider == null) {
-      _liveInstancesProvider = new ZkBackedLiveInstanceListProvider();
-    }
-
     // create a globally unique instance name and create a live instance node in ZooKeeper
     _instanceName = createLiveInstanceNode();
 
@@ -353,18 +349,18 @@ private void closeZkListeners(boolean isDisconnect, boolean isSessionExpired) {
         _zkclient.unsubscribeDataChanges(KeyBuilder.liveInstance(_cluster, _currentSubscription), _leaderElectionListener);
         _currentSubscription = null;
       }
-
-      if (_liveInstancesProvider != null) {
-        _liveInstancesProvider.close();
-        _liveInstancesProvider = null;
-      }
     }
 
     if (_datastreamList != null) {
       _datastreamList.close();
       _datastreamList = null;
     }
 
+    if (_liveInstancesProvider != null) {
+      _liveInstancesProvider.close();
+      _liveInstancesProvider = null;
+    }
+
     if (_targetAssignmentProvider != null) {
       _targetAssignmentProvider.close();
       _targetAssignmentProvider = null;

diff --git a/datastream-server/src/test/java/com/linkedin/datastream/server/zk/TestZkAdapter.java b/datastream-server/src/test/java/com/linkedin/datastream/server/zk/TestZkAdapter.java
@@ -830,7 +830,7 @@ private void verifyZkListenersOfFollower(ZkClientInterceptingAdapter adapter2) {
     Assert.assertNotNull(adapter2.getLeaderElectionListener());
     Assert.assertNotNull(adapter2.getAssignmentListProvider());
     Assert.assertNotNull(adapter2.getStateChangeListener());
-    Assert.assertNotNull(adapter2.getLiveInstancesProvider());
+    Assert.assertNull(adapter2.getLiveInstancesProvider());
     Assert.assertNull(adapter2.getDatastreamList());
     Assert.assertNull(adapter2.getTargetAssignmentProvider());
   }