wip for standardizing refreshAttemptForZkReconnect

Author: gabrieldrouin

clients/da-vinci-client/src/main/java/com/linkedin/davinci/config/VeniceClusterConfig.java

@@ -120,7 +120,7 @@ public VeniceClusterConfig(VeniceProperties clusterProps, Map<String, Map<String
     this.kafkaFetchQuotaUnorderedRecordPerSecond =
         clusterProps.getLong(KAFKA_FETCH_QUOTA_UNORDERED_RECORDS_PER_SECOND, -1);
 
-    this.refreshAttemptsForZkReconnect = clusterProps.getInt(REFRESH_ATTEMPTS_FOR_ZK_RECONNECT, 3);
+    this.refreshAttemptsForZkReconnect = clusterProps.getInt(REFRESH_ATTEMPTS_FOR_ZK_RECONNECT, 9);
     this.refreshIntervalForZkReconnectInMs =
         clusterProps.getLong(REFRESH_INTERVAL_FOR_ZK_RECONNECT_MS, TimeUnit.SECONDS.toMillis(10));
     this.kafkaReadCycleDelayMs = clusterProps.getLong(KAFKA_READ_CYCLE_DELAY_MS, 1000);

clients/da-vinci-client/src/main/java/com/linkedin/davinci/helix/HelixParticipationService.java

@@ -69,9 +69,6 @@ public class HelixParticipationService extends AbstractVeniceService
     implements StatusMessageHandler<KillOfflinePushMessage> {
   private static final Logger LOGGER = LogManager.getLogger(HelixParticipationService.class);
 
-  private static final int MAX_RETRY = 30;
-  private static final int RETRY_INTERVAL_SEC = 1;
-
   private final Instance instance;
   private final String clusterName;
   private final String participantName;
@@ -305,7 +302,7 @@ private void checkBeforeJoinInCluster() {
     HelixAdmin admin = new ZKHelixAdmin(zkAddress);
     try {
       // Check whether the cluster is ready or not at first to prevent zk no node exception.
-      HelixUtils.checkClusterSetup(admin, clusterName, MAX_RETRY);
+      HelixUtils.checkClusterSetup(admin, clusterName, 9);
       List<String> instances = admin.getInstancesInCluster(clusterName);
       if (instances.contains(instance.getNodeId())) {
         LOGGER.info("{} is not a new node to cluster: {}, skip the cleaning up.", instance.getNodeId(), clusterName);
@@ -365,9 +362,8 @@ private void asyncStart() {
         clusterName,
         zkClient,
         new HelixAdapterSerializer(),
-        veniceConfigLoader.getVeniceClusterConfig().getRefreshAttemptsForZkReconnect(),
-        veniceConfigLoader.getVeniceClusterConfig().getRefreshIntervalForZkReconnectInMs(),
-        veniceServerConfig.getRegionName());
+        veniceServerConfig.getRegionName(),
+        veniceConfigLoader.getCombinedProperties());
 
     /**
      * The accessor can only get created successfully after helix manager is created.

clients/da-vinci-client/src/main/java/com/linkedin/davinci/repository/VeniceMetadataRepositoryBuilder.java

@@ -133,12 +133,8 @@ private void initServerStoreAndSchemaRepository() {
     HelixReadOnlyZKSharedSystemStoreRepository readOnlyZKSharedSystemStoreRepository =
         new HelixReadOnlyZKSharedSystemStoreRepository(zkClient, adapter, systemSchemaClusterName);
 
-    HelixReadOnlyStoreRepository readOnlyStoreRepository = new HelixReadOnlyStoreRepository(
-        zkClient,
-        adapter,
-        clusterName,
-        clusterConfig.getRefreshAttemptsForZkReconnect(),
-        clusterConfig.getRefreshIntervalForZkReconnectInMs());
+    HelixReadOnlyStoreRepository readOnlyStoreRepository =
+        new HelixReadOnlyStoreRepository(zkClient, adapter, clusterName);
 
     storeRepo = new HelixReadOnlyStoreRepositoryAdapter(
         readOnlyZKSharedSystemStoreRepository,
@@ -155,17 +151,15 @@ private void initServerStoreAndSchemaRepository() {
         zkClient,
         adapter,
         systemSchemaClusterName,
-        clusterConfig.getRefreshAttemptsForZkReconnect(),
-        clusterConfig.getRefreshIntervalForZkReconnectInMs());
+        clusterConfig.getRefreshAttemptsForZkReconnect());
     schemaRepo = new HelixReadOnlySchemaRepositoryAdapter(
         readOnlyZKSharedSchemaRepository,
         new HelixReadOnlySchemaRepository(
             readOnlyStoreRepository,
             zkClient,
             adapter,
             clusterName,
-            clusterConfig.getRefreshAttemptsForZkReconnect(),
-            clusterConfig.getRefreshIntervalForZkReconnectInMs()));
+            clusterConfig.getRefreshAttemptsForZkReconnect()));
     schemaRepo.refresh();
 
     liveClusterConfigRepo = new HelixReadOnlyLiveClusterConfigRepository(zkClient, adapter, clusterName);

internal/venice-common/src/main/java/com/linkedin/venice/helix/CachedResourceZkStateListener.java

@@ -20,22 +20,19 @@
 public class CachedResourceZkStateListener implements IZkStateListener {
   private final Logger logger;
   public static final int DEFAULT_RETRY_LOAD_ATTEMPTS = 1;
-  public static final long DEFAULT_RETRY_LOAD_INTERVAL_IN_MS = TimeUnit.SECONDS.toMillis(10);
   private final VeniceResource resource;
-  private final int retryLoadAttempts;
-  private final long retryLoadIntervalInMs;
+  private final int refreshAttemptsForZkReconnect;
   private volatile boolean disconnected = false;
 
   public CachedResourceZkStateListener(VeniceResource resource) {
     // By default, we only retry once after connection is reconnected.
-    this(resource, DEFAULT_RETRY_LOAD_ATTEMPTS, DEFAULT_RETRY_LOAD_INTERVAL_IN_MS);
+    this(resource, DEFAULT_RETRY_LOAD_ATTEMPTS);
   }
 
-  public CachedResourceZkStateListener(VeniceResource resource, int retryLoadAttempts, long retryLoadIntervalInMs) {
+  public CachedResourceZkStateListener(VeniceResource resource, int refreshAttemptsForZkReconnect) {
     this.resource = resource;
     this.logger = LogManager.getLogger(this.getClass().getSimpleName() + " [" + getResourceName() + "]");
-    this.retryLoadAttempts = retryLoadAttempts;
-    this.retryLoadIntervalInMs = retryLoadIntervalInMs;
+    this.refreshAttemptsForZkReconnect = refreshAttemptsForZkReconnect;
   }
 
   /**
@@ -62,20 +59,22 @@ public void handleStateChanged(Watcher.Event.KeeperState state) throws Exception
           // retryLoadAttempts with retryLoadIntervalInMs between each two loading.
           // Sleep a random time(no more than retryLoadIntervalInMs) to avoid thunderstorm issue that all nodes are
           // trying to refresh resource at the same time if there is a network issue in that DC.
+          // TODO: refactor to use exponential backoff like implemented in HelixUtils
+          long retryLoadIntervalInMs = TimeUnit.SECONDS.toMillis(2);
           Utils.sleep((long) (Math.random() * retryLoadIntervalInMs));
           int attempt = 1;
-          while (attempt <= retryLoadAttempts) {
+          while (attempt <= refreshAttemptsForZkReconnect) {
             logger.info(
                 "Attempt #{} of {}: Refresh resource after connection is reconnected.",
                 attempt,
-                retryLoadAttempts);
+                refreshAttemptsForZkReconnect);
             try {
               resource.refresh();
-              logger.info("Attempt #{} of {}: Refresh completed.", attempt, retryLoadAttempts);
+              logger.info("Attempt #{} of {}: Refresh completed.", attempt, refreshAttemptsForZkReconnect);
               return;
             } catch (Exception e) {
               logger.error("Can not refresh resource correctly after client is reconnected", e);
-              if (attempt < retryLoadAttempts) {
+              if (attempt < refreshAttemptsForZkReconnect) {
                 logger.info("Will retry after {} ms", retryLoadIntervalInMs);
                 Utils.sleep(retryLoadIntervalInMs);
               }

internal/venice-common/src/main/java/com/linkedin/venice/helix/HelixReadOnlySchemaRepository.java

@@ -94,35 +94,26 @@ public HelixReadOnlySchemaRepository(
       ZkClient zkClient,
       HelixAdapterSerializer adapter,
       String clusterName,
-      int refreshAttemptsForZkReconnect,
-      long refreshIntervalForZkReconnectInMs) {
+      int refreshAttemptsForZkReconnect) {
     this.storeRepository = storeRepository;
     this.zkClient = zkClient;
-    this.accessor = new HelixSchemaAccessor(
-        zkClient,
-        adapter,
-        clusterName,
-        refreshAttemptsForZkReconnect,
-        refreshIntervalForZkReconnectInMs);
+    this.accessor = new HelixSchemaAccessor(zkClient, adapter, clusterName, refreshAttemptsForZkReconnect);
 
     storeRepository.registerStoreDataChangedListener(this);
-    this.zkStateListener =
-        new CachedResourceZkStateListener(this, refreshAttemptsForZkReconnect, refreshIntervalForZkReconnectInMs);
+    this.zkStateListener = new CachedResourceZkStateListener(this, refreshAttemptsForZkReconnect);
   }
 
   /** test-only */
   HelixReadOnlySchemaRepository(
       ReadOnlyStoreRepository storeRepository,
       ZkClient zkClient,
       HelixSchemaAccessor accessor,
-      int refreshAttemptsForZkReconnect,
-      long refreshIntervalForZkReconnectInMs) {
+      int refreshAttemptsForZkReconnect) {
     this.storeRepository = storeRepository;
     this.zkClient = zkClient;
     this.accessor = accessor;
     storeRepository.registerStoreDataChangedListener(this);
-    this.zkStateListener =
-        new CachedResourceZkStateListener(this, refreshAttemptsForZkReconnect, refreshIntervalForZkReconnectInMs);
+    this.zkStateListener = new CachedResourceZkStateListener(this, refreshAttemptsForZkReconnect);
   }
 
   /**

internal/venice-common/src/main/java/com/linkedin/venice/helix/HelixReadOnlyStoreRepository.java

@@ -22,9 +22,7 @@ public class HelixReadOnlyStoreRepository extends CachedReadOnlyStoreRepository
   public HelixReadOnlyStoreRepository(
       ZkClient zkClient,
       HelixAdapterSerializer compositeSerializer,
-      String clusterName,
-      int refreshAttemptsForZkReconnect,
-      long refreshIntervalForZkReconnectInMs) {
+      String clusterName) {
     /**
      * HelixReadOnlyStoreRepository is used in router, server, fast-client, da-vinci and system store.
      * Its centralized locking should NOT be shared with other classes. Create a new instance.
@@ -118,15 +116,12 @@ protected void onRepositoryChanged(Collection<String> newZkStoreNames) {
 
   private final CachedResourceZkStateListener zkStateListener = new CachedResourceZkStateListener(this);
 
-  private final IZkChildListener zkStoreRepositoryListener = new IZkChildListener() {
-    @Override
-    public void handleChildChange(String path, List<String> children) {
-      if (!path.equals(clusterStoreRepositoryPath)) {
-        LOGGER.warn("Notification path mismatch, path={}, expected={}.", path, clusterStoreRepositoryPath);
-        return;
-      }
-      onRepositoryChanged(children);
+  private final IZkChildListener zkStoreRepositoryListener = (path, children) -> {
+    if (!path.equals(clusterStoreRepositoryPath)) {
+      LOGGER.warn("Notification path mismatch, path={}, expected={}.", path, clusterStoreRepositoryPath);
+      return;
     }
+    onRepositoryChanged(children);
   };
 
   private final IZkDataListener zkStoreListener = new IZkDataListener() {

internal/venice-common/src/main/java/com/linkedin/venice/helix/HelixReadOnlyZKSharedSchemaRepository.java

@@ -16,14 +16,7 @@ public HelixReadOnlyZKSharedSchemaRepository(
       ZkClient zkClient,
       HelixAdapterSerializer adapter,
       String systemStoreClusterName,
-      int refreshAttemptsForZkReconnect,
-      long refreshIntervalForZkReconnectInMs) {
-    super(
-        storeRepository,
-        zkClient,
-        adapter,
-        systemStoreClusterName,
-        refreshAttemptsForZkReconnect,
-        refreshIntervalForZkReconnectInMs);
+      int refreshAttemptsForZkReconnect) {
+    super(storeRepository, zkClient, adapter, systemStoreClusterName, refreshAttemptsForZkReconnect);
   }
 }

internal/venice-common/src/main/java/com/linkedin/venice/helix/HelixReadOnlyZKSharedSystemStoreRepository.java

@@ -28,7 +28,7 @@ public HelixReadOnlyZKSharedSystemStoreRepository(
       ZkClient zkClient,
       HelixAdapterSerializer compositeSerializer,
       String systemStoreClusterName) {
-    super(zkClient, compositeSerializer, systemStoreClusterName, 0, 0);
+    super(zkClient, compositeSerializer, systemStoreClusterName);
     // Initialize the necessary zk shared system stores
     for (VeniceSystemStoreType type: VeniceSystemStoreType.values()) {
       if (type.isNewMedataRepositoryAdopted()) {

internal/venice-common/src/main/java/com/linkedin/venice/helix/HelixSchemaAccessor.java

@@ -7,7 +7,6 @@
 import com.linkedin.venice.utils.HelixUtils;
 import com.linkedin.venice.utils.PathResourceRegistry;
 import java.util.List;
-import java.util.concurrent.TimeUnit;
 import org.apache.helix.AccessOption;
 import org.apache.helix.manager.zk.ZkBaseDataAccessor;
 import org.apache.helix.zookeeper.impl.client.ZkClient;
@@ -19,9 +18,6 @@
 public class HelixSchemaAccessor {
   private static final Logger logger = LogManager.getLogger(HelixSchemaAccessor.class);
 
-  private static final int DEFAULT_ZK_REFRESH_ATTEMPTS = 3;
-  private static final long DEFAULT_ZK_REFRESH_INTERVAL = TimeUnit.SECONDS.toMillis(10);
-
   // Key schema path name
   private static final String KEY_SCHEMA_PATH = "key-schema";
   // Value schema path name
@@ -45,22 +41,20 @@ public class HelixSchemaAccessor {
   private final String clusterName;
 
   private final int refreshAttemptsForZkReconnect;
-  private final long refreshIntervalForZkReconnectInMs;
 
   public HelixSchemaAccessor(ZkClient zkClient, HelixAdapterSerializer helixAdapterSerializer, String clusterName) {
-    this(zkClient, helixAdapterSerializer, clusterName, DEFAULT_ZK_REFRESH_ATTEMPTS, DEFAULT_ZK_REFRESH_INTERVAL);
+    // TODO: refactor hard-coded refreshAttemptsForZkReconnect
+    this(zkClient, helixAdapterSerializer, clusterName, 9);
   }
 
   public HelixSchemaAccessor(
       ZkClient zkClient,
       HelixAdapterSerializer helixAdapterSerializer,
       String clusterName,
-      int refreshAttemptsForZkReconnect,
-      long refreshIntervalForZkReconnectInMs) {
+      int refreshAttemptsForZkReconnect) {
     this.clusterName = clusterName;
 
     this.refreshAttemptsForZkReconnect = refreshAttemptsForZkReconnect;
-    this.refreshIntervalForZkReconnectInMs = refreshIntervalForZkReconnectInMs;
 
     registerSerializerForSchema(zkClient, helixAdapterSerializer);
     schemaAccessor = new ZkBaseDataAccessor<>(zkClient);

internal/venice-common/src/main/java/com/linkedin/venice/helix/SubscriptionBasedStoreRepository.java

@@ -23,7 +23,7 @@ public SubscriptionBasedStoreRepository(
       ZkClient zkClient,
       HelixAdapterSerializer compositeSerializer,
       String clusterName) {
-    super(zkClient, compositeSerializer, clusterName, 0, 0);
+    super(zkClient, compositeSerializer, clusterName);
   }
 
   @Override

internal/venice-common/src/main/java/com/linkedin/venice/helix/VeniceOfflinePushMonitorAccessor.java

@@ -1,5 +1,6 @@
 package com.linkedin.venice.helix;
 
+import static com.linkedin.venice.ConfigKeys.REFRESH_ATTEMPTS_FOR_ZK_RECONNECT;
 import static com.linkedin.venice.zk.VeniceZkPaths.OFFLINE_PUSHES;
 
 import com.linkedin.venice.exceptions.VeniceException;
@@ -14,12 +15,12 @@
 import com.linkedin.venice.utils.HelixUtils;
 import com.linkedin.venice.utils.LogContext;
 import com.linkedin.venice.utils.PathResourceRegistry;
+import com.linkedin.venice.utils.VeniceProperties;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Optional;
-import java.util.concurrent.TimeUnit;
 import org.apache.helix.AccessOption;
 import org.apache.helix.manager.zk.ZkBaseDataAccessor;
 import org.apache.helix.zookeeper.impl.client.ZkClient;
@@ -46,8 +47,6 @@
  */
 public class VeniceOfflinePushMonitorAccessor implements OfflinePushAccessor {
   public static final String OFFLINE_PUSH_SUB_PATH = OFFLINE_PUSHES;
-  private static final int DEFAULT_ZK_REFRESH_ATTEMPTS = 3;
-  private static final long DEFAULT_ZK_REFRESH_INTERVAL = TimeUnit.SECONDS.toMillis(10);
 
   private static final Logger LOGGER = LogManager.getLogger(VeniceOfflinePushMonitorAccessor.class);
   private final String clusterName;
@@ -68,23 +67,20 @@ public class VeniceOfflinePushMonitorAccessor implements OfflinePushAccessor {
 
   private final int refreshAttemptsForZkReconnect;
 
-  private final long refreshIntervalForZkReconnectInMs;
-
   public VeniceOfflinePushMonitorAccessor(
       String clusterName,
       ZkClient zkClient,
       HelixAdapterSerializer adapter,
       LogContext logContext) {
-    this(clusterName, zkClient, adapter, DEFAULT_ZK_REFRESH_ATTEMPTS, DEFAULT_ZK_REFRESH_INTERVAL, logContext);
+    this(clusterName, zkClient, adapter, logContext, VeniceProperties.empty());
   }
 
   public VeniceOfflinePushMonitorAccessor(
       String clusterName,
       ZkClient zkClient,
       HelixAdapterSerializer adapter,
-      int refreshAttemptsForZkReconnect,
-      long refreshIntervalForZkReconnectInMs,
-      Object logContext) {
+      Object logContext,
+      VeniceProperties props) {
     this.clusterName = clusterName;
     this.offlinePushStatusParentPath = getOfflinePushStatuesParentPath();
     this.zkClient = zkClient;
@@ -94,8 +90,7 @@ public VeniceOfflinePushMonitorAccessor(
     this.partitionStatusAccessor = new ZkBaseDataAccessor<>(zkClient);
     this.listenerManager = new ListenerManager<>(logContext);
     this.partitionStatusZkListener = new PartitionStatusZkListener(logContext);
-    this.refreshAttemptsForZkReconnect = refreshAttemptsForZkReconnect;
-    this.refreshIntervalForZkReconnectInMs = refreshIntervalForZkReconnectInMs;
+    this.refreshAttemptsForZkReconnect = props.getInt(REFRESH_ATTEMPTS_FOR_ZK_RECONNECT, 9);
   }
 
   /**
@@ -105,16 +100,16 @@ public VeniceOfflinePushMonitorAccessor(
       String clusterName,
       ZkBaseDataAccessor<OfflinePushStatus> offlinePushStatusAccessor,
       ZkBaseDataAccessor<PartitionStatus> partitionStatusAccessor,
-      LogContext logContext) {
+      LogContext logContext,
+      VeniceProperties props) {
     this.clusterName = clusterName;
     this.offlinePushStatusAccessor = offlinePushStatusAccessor;
     this.partitionStatusAccessor = partitionStatusAccessor;
     this.offlinePushStatusParentPath = getOfflinePushStatuesParentPath();
     this.zkClient = null;
     this.listenerManager = new ListenerManager<>(logContext);
     this.partitionStatusZkListener = new PartitionStatusZkListener(logContext);
-    this.refreshAttemptsForZkReconnect = DEFAULT_ZK_REFRESH_ATTEMPTS;
-    this.refreshIntervalForZkReconnectInMs = DEFAULT_ZK_REFRESH_INTERVAL;
+    this.refreshAttemptsForZkReconnect = props.getInt(REFRESH_ATTEMPTS_FOR_ZK_RECONNECT, 9);
   }
 
   private void registerSerializers(HelixAdapterSerializer adapter) {

internal/venice-common/src/main/java/com/linkedin/venice/helix/ZkRoutersClusterManager.java

@@ -62,8 +62,7 @@ public ZkRoutersClusterManager(
     adapter.registerSerializer(getRouterRootPath(), new RouterClusterConfigJSONSerializer());
     zkClient.setZkSerializer(adapter);
     dataAccessor = new ZkBaseDataAccessor<>(zkClient);
-    zkStateListener =
-        new CachedResourceZkStateListener(this, refreshAttemptsForZkReconnect, refreshIntervalForZkReconnectInMs);
+    zkStateListener = new CachedResourceZkStateListener(this, refreshAttemptsForZkReconnect);
     isConnected.set(zkClient.getConnection().getZookeeperState().isAlive());
     this.zkClient.subscribeStateChanges(this);
   }

internal/venice-common/src/main/java/com/linkedin/venice/utils/HelixUtils.java

@@ -240,6 +240,7 @@ public static <T> void compareAndUpdate(
    */
   private static void handleFailedHelixOperation(String path, String helixOperation, int attempt, int retryCount) {
     if (attempt < retryCount) {
+      // Is empty if the caller doesn't operate on a specific ZK path (like connectHelixManager and checkClusterSetup)
       if (!path.isEmpty()) {
         path = " with path " + path;
       }