addressed Bharat's comments

linkedin · Apr 23, 2024 · e87cf7e · e87cf7e
1 parent f8b854a
commit e87cf7e
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 24 deletions.
diff --git a/...roller/src/main/java/com/linkedin/venice/controller/StoreBackupVersionCleanupService.java b/...roller/src/main/java/com/linkedin/venice/controller/StoreBackupVersionCleanupService.java
@@ -21,14 +21,14 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.time.Duration;
-import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
 import org.apache.commons.io.IOUtils;
 import org.apache.http.HttpResponse;
 import org.apache.http.HttpStatus;
@@ -136,6 +136,9 @@ CloseableHttpAsyncClient getHttpAsyncClient() {
   }
 
   protected static boolean whetherStoreReadyToBeCleanup(Store store, long defaultBackupVersionRetentionMs, Time time) {
+    if (store.getCurrentVersion() == NON_EXISTING_VERSION) {
+      return false;
+    }
     long backupVersionRetentionMs = store.getBackupVersionRetentionMs();
     if (backupVersionRetentionMs < 0) {
       backupVersionRetentionMs = defaultBackupVersionRetentionMs;
@@ -225,7 +228,8 @@ private boolean validateAllServerOnCurrentVersion(Store store, String clusterNam
    * @return whether any backup version is removed or not
    */
   protected boolean cleanupBackupVersion(Store store, String clusterName) {
-    Version currentVersion = store.getVersion(store.getCurrentVersion()).get();
+    int currentVersionNum = store.getCurrentVersion();
+    Version currentVersion = store.getVersion(currentVersionNum).get();
     boolean isCurrentVersionFromRepush = Version.isPushIdRePush(currentVersion.getPushJobId());
 
     if (!isCurrentVersionFromRepush && !whetherStoreReadyToBeCleanup(store, defaultBackupVersionRetentionMs, time)) {
@@ -234,13 +238,6 @@ protected boolean cleanupBackupVersion(Store store, String clusterName) {
     }
 
     List<Version> versions = store.getVersions();
-    List<Version> readyToBeRemovedVersions = new ArrayList<>();
-    int currentVersionNum = store.getCurrentVersion();
-
-    if (currentVersionNum == 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, currentVersionNum)
@@ -250,19 +247,10 @@ protected boolean cleanupBackupVersion(Store store, String clusterName) {
       stats.recordBackupVersionMismatch();
       return false;
     }
-    if (isCurrentVersionFromRepush) {
-      versions.forEach(v -> {
-        if (v.getNumber() < currentVersionNum && v.getCreatedTime() == Duration.ZERO.toMillis()) {
-          readyToBeRemovedVersions.add(v);
-        }
-      });
-    } else {
-      versions.forEach(v -> {
-        if (v.getNumber() < currentVersionNum) {
-          readyToBeRemovedVersions.add(v);
-        }
-      });
-    }
+    List<Version> readyToBeRemovedVersions = versions.stream()
+        .filter(v -> readyToBeRemoved(v, isCurrentVersionFromRepush, currentVersionNum))
+        .collect(Collectors.toList());
+
     if (readyToBeRemovedVersions.isEmpty()) {
       return false;
     }
@@ -297,6 +285,13 @@ protected boolean cleanupBackupVersion(Store store, String clusterName) {
     return true;
   }
 
+  private boolean readyToBeRemoved(Version v, boolean isRepush, int currentVersionNum) {
+    if (isRepush) {
+      return v.getCreatedTime() == Duration.ZERO.toMillis();
+    }
+    return v.getNumber() < currentVersionNum;
+  }
+
   private class StoreBackupVersionCleanupTask implements Runnable {
     @Override
     public void run() {

diff --git a/...er/src/test/java/com/linkedin/venice/controller/TestStoreBackupVersionCleanupService.java b/...er/src/test/java/com/linkedin/venice/controller/TestStoreBackupVersionCleanupService.java
@@ -39,6 +39,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.http.HttpEntity;
 import org.apache.http.HttpResponse;
 import org.apache.http.HttpStatus;
@@ -75,12 +76,12 @@ private Store mockStore(
     doReturn(latestVersionPromoteToCurrentTimestamp).when(store).getLatestVersionPromoteToCurrentTimestamp();
     doReturn(currentVersion).when(store).getCurrentVersion();
     List<Version> versionList = new ArrayList<>();
+    AtomicInteger i = new AtomicInteger();
     versions.forEach((n, s) -> {
-      int i = 0;
       Version v = mock(Version.class);
       doReturn(n).when(v).getNumber();
       doReturn(s).when(v).getStatus();
-      doReturn("dummyID_" + i).when(v).getPushJobId();
+      doReturn("dummyID_" + i.getAndIncrement()).when(v).getPushJobId();
       versionList.add(v);
     });
     doReturn(versionList).when(store).getVersions();