[controller] Made disable replica service interval configurable (#811)

* [controller] Made disable replica service interval configurable Made disable replica service interval configurable so that it can be changed based on need. Also some logging changes in store cleanup service and skipping stores without any version. --------- Co-authored-by: Sourav Maji <[email protected]>
linkedin · Jan 4, 2024 · 6827575 · 6827575
1 parent 16ed564
commit 6827575
Show file tree

Hide file tree

Showing 5 changed files with 30 additions and 8 deletions.
diff --git a/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java b/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java
@@ -377,10 +377,13 @@ private ConfigKeys() {
       "kme.registration.from.message.header.enabled";
 
   /**
-   * The following config is to control whether to enable backup version cleanup based on retention policy or not at cluster level.
+   * The following config is to control whether to turn on disabled replica enabler service.
    */
   public static final String CONTROLLER_ENABLE_DISABLED_REPLICA_ENABLED = "controller.enable.disabled.replica.enabled";
 
+  public static final String CONTROLLER_DISABLED_REPLICA_ENABLER_INTERVAL_MS =
+      "controller.disabled.replica.enabler.interval.ms";
+
   // Server specific configs
   public static final String LISTENER_PORT = "listener.port";
   public static final String GRPC_READ_SERVER_PORT = "grpc.read.server.port";

diff --git a/...troller/src/main/java/com/linkedin/venice/controller/DisabledPartitionEnablerService.java b/...troller/src/main/java/com/linkedin/venice/controller/DisabledPartitionEnablerService.java
@@ -4,7 +4,6 @@
 import com.linkedin.venice.utils.SystemTime;
 import com.linkedin.venice.utils.Time;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -36,7 +35,7 @@ public DisabledPartitionEnablerService(
     this.admin = admin;
     this.multiClusterConfig = multiClusterConfig;
     this.allClusters = multiClusterConfig.getClusters();
-    this.sleepInterval = TimeUnit.HOURS.toMillis(10);
+    this.sleepInterval = multiClusterConfig.getDisabledReplicaEnablerServiceIntervalMs();
     this.time = time;
     this.cleanupThread = new Thread(new DisabledPartitionEnablerTask(), "StoreBackupVersionCleanupTask");
   }

diff --git a/...roller/src/main/java/com/linkedin/venice/controller/StoreBackupVersionCleanupService.java b/...roller/src/main/java/com/linkedin/venice/controller/StoreBackupVersionCleanupService.java
@@ -1,5 +1,7 @@
 package com.linkedin.venice.controller;
 
+import static com.linkedin.venice.meta.Store.NON_EXISTING_VERSION;
+
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.linkedin.davinci.listener.response.ServerCurrentVersionResponse;
 import com.linkedin.venice.ConfigKeys;
@@ -146,11 +148,11 @@ private boolean validateAllRouterOnCurrentVersion(Store store, String clusterNam
     for (Instance routerInstance: liveRouterInstances) {
       try {
         HttpGet routerRequest =
-            new HttpGet(routerInstance.getUrl() + "/" + TYPE_CURRENT_VERSION + "/" + store.getName());
+            new HttpGet(routerInstance.getUrl(true) + "/" + TYPE_CURRENT_VERSION + "/" + store.getName());
         HttpResponse response = getHttpAsyncClient().execute(routerRequest, null).get(500, TimeUnit.MILLISECONDS);
         if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
           LOGGER.warn(
-              "Got status code {} from host {} while querying current version for store {}",
+              "Got status code {} from host {} while querying router current version for store {}",
               response.getStatusLine().getStatusCode(),
               routerInstance,
               store.getName());
@@ -166,7 +168,7 @@ private boolean validateAllRouterOnCurrentVersion(Store store, String clusterNam
           return false;
         }
       } catch (Exception e) {
-        LOGGER.error("Got exception while getting current version for store {}", store.getName(), e);
+        LOGGER.error("Got exception while getting router current version for store {}", store.getName(), e);
         return false;
       }
     }
@@ -179,11 +181,11 @@ private boolean validateAllServerOnCurrentVersion(Store store, String clusterNam
     for (Instance instance: instances) {
       try {
         HttpGet routerRequest = new HttpGet(
-            instance.getUrl() + "/" + QueryAction.CURRENT_VERSION.toString().toLowerCase() + "/" + store.getName());
+            instance.getUrl(true) + "/" + QueryAction.CURRENT_VERSION.toString().toLowerCase() + "/" + store.getName());
         HttpResponse response = getHttpAsyncClient().execute(routerRequest, null).get(500, TimeUnit.MILLISECONDS);
         if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
           LOGGER.warn(
-              "Got status code {} from host {} while querying current version for store {}",
+              "Got status code {} from host {} while querying server current version for store {}",
               response.getStatusLine().getStatusCode(),
               instance,
               store.getName());
@@ -220,6 +222,10 @@ protected boolean cleanupBackupVersion(Store store, String clusterName) {
     List<Version> readyToBeRemovedVersions = new ArrayList<>();
     int currentVersion = store.getCurrentVersion();
 
+    if (currentVersion == NON_EXISTING_VERSION) {
+      return false;
+    }
+
     // Do not delete version unless all routers and all servers are on same current version
     if (multiClusterConfig.getControllerConfig(clusterName).isBackupVersionMetadataFetchBasedCleanupEnabled()
         && (!validateAllRouterOnCurrentVersion(store, clusterName, currentVersion)

diff --git a/...enice-controller/src/main/java/com/linkedin/venice/controller/VeniceControllerConfig.java b/...enice-controller/src/main/java/com/linkedin/venice/controller/VeniceControllerConfig.java
@@ -33,6 +33,7 @@
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_CLUSTER_LEADER_HAAS;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_CLUSTER_REPLICA;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_CLUSTER_ZK_ADDRESSS;
+import static com.linkedin.venice.ConfigKeys.CONTROLLER_DISABLED_REPLICA_ENABLER_INTERVAL_MS;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_DISABLED_ROUTES;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_DISABLE_PARENT_TOPIC_TRUNCATION_UPON_COMPLETION;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_EARLY_DELETE_BACKUP_ENABLED;
@@ -170,6 +171,9 @@ public class VeniceControllerConfig extends VeniceControllerClusterConfig {
   private final long deprecatedJobTopicRetentionMs;
   private final long deprecatedJobTopicMaxRetentionMs;
   private final long topicCleanupSleepIntervalBetweenTopicListFetchMs;
+
+  private final long disabledReplicaEnablerServiceIntervalMs;
+
   private final int topicCleanupDelayFactor;
   private final int topicManagerKafkaOperationTimeOutMs;
   private final int minNumberOfUnusedKafkaTopicsToPreserve;
@@ -412,6 +416,8 @@ public VeniceControllerConfig(VeniceProperties props) {
                                                                                  // topicCleanupSleepIntervalBetweenTopicListFetchMs
                                                                                  // = delayBeforeTopicDeletion
 
+    this.disabledReplicaEnablerServiceIntervalMs =
+        props.getLong(CONTROLLER_DISABLED_REPLICA_ENABLER_INTERVAL_MS, TimeUnit.HOURS.toMillis(16));
     this.topicManagerKafkaOperationTimeOutMs =
         props.getInt(TOPIC_MANAGER_KAFKA_OPERATION_TIMEOUT_MS, 30 * Time.MS_PER_SECOND);
 
@@ -629,6 +635,10 @@ public long getTopicCleanupSleepIntervalBetweenTopicListFetchMs() {
     return topicCleanupSleepIntervalBetweenTopicListFetchMs;
   }
 
+  public long getDisabledReplicaEnablerServiceIntervalMs() {
+    return disabledReplicaEnablerServiceIntervalMs;
+  }
+
   public int getDaVinciPushStatusScanMaxOfflineInstance() {
     return daVinciPushStatusScanMaxOfflineInstance;
   }

diff --git a/...ller/src/main/java/com/linkedin/venice/controller/VeniceControllerMultiClusterConfig.java b/...ller/src/main/java/com/linkedin/venice/controller/VeniceControllerMultiClusterConfig.java
@@ -118,6 +118,10 @@ public long getTopicCleanupSleepIntervalBetweenTopicListFetchMs() {
     return getCommonConfig().getTopicCleanupSleepIntervalBetweenTopicListFetchMs();
   }
 
+  public long getDisabledReplicaEnablerServiceIntervalMs() {
+    return getCommonConfig().getDisabledReplicaEnablerServiceIntervalMs();
+  }
+
   public int getTopicCleanupDelayFactor() {
     return getCommonConfig().getTopicCleanupDelayFactor();
   }