address comments

Author: ChenSammi

hadoop-hdds/docs/content/design/lifecycle-task-resume.md

@@ -13,7 +13,7 @@
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
-  */
+*/
 
 # Design for Resumable Lifecycle Scans(HDDS-8342)
 
@@ -46,21 +46,21 @@ message LifecycleScanState {
 ```
 
 ### OM DB Schema Updates
-Add a new table `lifecycleStateTable` to `OMMetadataManager` to store the scan states:
-- **Table Name:** `lifecycleStateTable`
+Add a new table `lifecycleScanStateTable` to `OMMetadataManager` to store the scan states:
+- **Table Name:** `lifecycleScanStateTable`
 - **Key:** `bucketKey` (String, e.g., `/volumeName/bucketName`)
 - **Value:** `LifecycleScanState`
 
 ### When to Persist the Pointer
 Persisting the pointer for every key would overwhelm Ratis and RocksDB. We should checkpoint periodically:
 
-1. **Piggybacking on Deletes:** Add an optional `LifecycleScanState` field to `DeleteKeysRequest`. When the OM state machine applies the deletion, it atomically updates the `lifecycleStateTable` with the new pointer. This guarantees exactly-once semantics for the scan pointer relative to deletions.
+1. **Piggybacking on Deletes:** Add an optional `LifecycleScanState` field to `DeleteKeysRequest`. When the OM state machine applies the deletion, it atomically updates the `lifecycleScanStateTable` with the new pointer. This guarantees exactly-once semantics for the scan pointer relative to deletions.
 2. **Move to trash**: Since there is no `RenameKeysRequest`, rename has be called multiple times for a batch of keys. We introduce a new OM request `SaveLifecycleScanStateRequest`. After a batch of keys are moved to trash, call `SaveLifecycleScanStateRequest` explicitly to persist the state.
 3. **Periodic Standalone Checkpoints:** If no keys are expired (e.g., scanning millions of valid keys), we still need to save progress. The `LifecycleActionTask` will send this request periodically (e.g., every 100,000 keys iterated, or every 1 minute of execution time).
 3. **End of Scan:** When the scan for a bucket finishes, a `SaveLifecycleScanStateRequest` is sent to mark state as completed by recording the completion time.
 
 ### How to Resume the Scan
-When `KeyLifecycleService` schedules a `LifecycleActionTask` for a bucket, it first reads the `LifecycleScanState` from the `lifecycleStateTable`.
+When `KeyLifecycleService` schedules a `LifecycleActionTask` for a bucket, it first reads the `LifecycleScanState` from the `lifecycleScanStateTable`.
 
 - **OBS/Legacy Resumption:** 
   The iterator for `keyTable` is initialized to seek to `lastScannedKey` instead of the bucket prefix.

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java

@@ -199,7 +199,6 @@ public final class OMConfigKeys {
       "ozone.lifecycle.service.delete.cached.directory.max-count";
   public static final long OZONE_KEY_LIFECYCLE_SERVICE_DELETE_CACHED_DIRECTORY_MAX_COUNT_DEFAULT = 1000000;
 
-  // Save task state for every 5m, or evaluated keys reaches 100k
   public static final String OZONE_KEY_LIFECYCLE_SERVICE_STATE_SAVE_INTERVAL_MS =
       "ozone.lifecycle.service.state.save.interval.ms";
   public static final long OZONE_KEY_LIFECYCLE_SERVICE_STATE_SAVE_INTERVAL_MS_DEFAULT = 5 * 60 * 1000;

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeysDeleteRequest.java

@@ -40,10 +40,12 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.hdds.utils.db.Table;
 import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
 import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
+import org.apache.hadoop.ozone.OmUtils;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.audit.AuditLogger;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
@@ -93,6 +95,25 @@ public OMKeysDeleteRequest(OMRequest omRequest, BucketLayout bucketLayout) {
     super(omRequest, bucketLayout);
   }
 
+  @Override
+  public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
+    DeleteKeysRequest deleteKeysRequest = super.preExecute(ozoneManager).getDeleteKeysRequest();
+    Objects.requireNonNull(deleteKeysRequest, "deleteKeysRequest == null");
+
+    if (deleteKeysRequest.getSourceType() == RequestSource.LIFECYCLE && deleteKeysRequest.hasScanState()) {
+      if (ozoneManager.getAclsEnabled()) {
+        UserGroupInformation ugi = createUGIForApi();
+        if (!ozoneManager.isAdmin(ugi)) {
+          throw new OMException("Access denied for user " + ugi + ". "
+              + "Superuser privilege is required to save Lifecycle Service task state.",
+              OMException.ResultCodes.ACCESS_DENIED);
+        }
+      }
+    }
+
+    return getOmRequest();
+  }
+
   @Override @SuppressWarnings("methodlength")
   public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, ExecutionContext context) {
     final long trxnLogIndex = context.getIndex();
@@ -164,17 +185,6 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       validateBucketAndVolume(omMetadataManager, volumeName, bucketName);
       String volumeOwner = getVolumeOwner(omMetadataManager, volumeName);
 
-      if (sourceType == RequestSource.LIFECYCLE && deleteKeyRequest.hasScanState()) {
-        if (ozoneManager.getAclsEnabled()) {
-          UserGroupInformation ugi = createUGIForApi();
-          if (!ozoneManager.isAdmin(ugi)) {
-            throw new OMException("Access denied for user " + ugi + ". "
-                + "Superuser privilege is required to save Lifecycle Service task state.",
-                OMException.ResultCodes.ACCESS_DENIED);
-          }
-        }
-      }
-
       for (indexFailed = 0; indexFailed < length; indexFailed++) {
         String keyName = deleteKeyArgs.getKeys(indexFailed);
         String objectKey =

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeysDeleteResponse.java

@@ -129,4 +129,8 @@ public OmBucketInfo getOmBucketInfo() {
   protected Map<String, OmKeyInfo> getOpenKeyInfoMap() {
     return openKeyInfoMap;
   }
+
+  public OmLifecycleScanState getScanState() {
+    return scanState;
+  }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeysDeleteResponseWithFSO.java

@@ -48,7 +48,6 @@ public class OMKeysDeleteResponseWithFSO extends OMKeysDeleteResponse {
 
   private List<OmKeyInfo> dirsList;
   private long volumeId;
-  private OmLifecycleScanState scanState;
 
   public OMKeysDeleteResponseWithFSO(
       @Nonnull OzoneManagerProtocolProtos.OMResponse omResponse,
@@ -60,7 +59,6 @@ public OMKeysDeleteResponseWithFSO(
     super(omResponse, keyDeleteList, omBucketInfo, openKeyInfoMap, scanState);
     this.dirsList = dirDeleteList;
     this.volumeId = volId;
-    this.scanState = scanState;
   }
 
   @Override
@@ -107,6 +105,7 @@ public void addToDBBatch(OMMetadataManager omMetadataManager,
       }
     }
 
+    OmLifecycleScanState scanState = getScanState();
     if (scanState != null) {
       omMetadataManager.getLifecycleScanStateTable().putWithBatch(
           batchOperation, scanState.getBucketKey(), scanState);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleService.java

@@ -171,7 +171,7 @@ public KeyLifecycleService(OzoneManager ozoneManager,
       LOG.warn("Illegal value {} for Property {}. Set {} to {}", stateSaveIntervalMs,
           OZONE_KEY_LIFECYCLE_SERVICE_STATE_SAVE_INTERVAL_MS, OZONE_KEY_LIFECYCLE_SERVICE_STATE_SAVE_INTERVAL_MS,
           OZONE_KEY_LIFECYCLE_SERVICE_STATE_SAVE_INTERVAL_MS_DEFAULT);
-      maxKeysProcessedPerState = OZONE_KEY_LIFECYCLE_SERVICE_STATE_SAVE_INTERVAL_MS_DEFAULT;
+      stateSaveIntervalMs = OZONE_KEY_LIFECYCLE_SERVICE_STATE_SAVE_INTERVAL_MS_DEFAULT;
     }
     this.maxKeysProcessedPerState = conf.getLong(OZONE_KEY_LIFECYCLE_SERVICE_STATE_SAVE_KEYS_PROCESSED,
         OZONE_KEY_LIFECYCLE_SERVICE_STATE_SAVE_KEYS_PROCESSED_DEFAULT);
@@ -181,6 +181,7 @@ public KeyLifecycleService(OzoneManager ozoneManager,
           OZONE_KEY_LIFECYCLE_SERVICE_STATE_SAVE_KEYS_PROCESSED_DEFAULT);
       maxKeysProcessedPerState = OZONE_KEY_LIFECYCLE_SERVICE_STATE_SAVE_KEYS_PROCESSED_DEFAULT;
     }
+    LOG.info("stateSaveIntervalMs = {}, maxKeysProcessedPerState = {}", stateSaveIntervalMs, maxKeysProcessedPerState);
     this.inFlight = new ConcurrentHashMap();
     this.omMetadataManager = ozoneManager.getMetadataManager();
     int limit = (int) conf.getStorageSize(
@@ -629,40 +630,21 @@ private List<RuleListWithDirectoryList> getRuleUnion(long volumeId, OmBucketInfo
       return sortedConsolidatedRules;
     }
 
-    private boolean canSkipDir(OmDirectoryInfo currentDir, String currentDirTableKey, DirectoryList dirList) {
-      // currentDir null is bucket root
-      if (currentDir == null) {
+    private boolean canSkipDir(String currentDirPath, String lastScannedDir) {
+      if (lastScannedDir == null || currentDirPath.isEmpty()) {
         return false;
       }
-
-      int count = dirList.getSubDirCount();
-      // if currentDir is equal to lastScannedDir
-      if (currentDir.getObjectID() == dirList.getSubDirList().get(count - 1).getObjectID()) {
-        return false;
-      }
-
-      // if currentDir is parent of lastScannedDir
-      long currentObjID = currentDir.getObjectID();
-      for (int i = 0; i < count; i++) {
-        OmDirectoryInfo dir = dirList.getSubDirList().get(i);
-        if (dir.getObjectID() == currentObjID) {
-          return false;
+      String[] cur = currentDirPath.split(OM_KEY_PREFIX);
+      String[] last = lastScannedDir.split(OM_KEY_PREFIX);
+      int n = Math.min(cur.length, last.length);
+      for (int i = 0; i < n; i++) {
+        int cmp = cur[i].compareTo(last[i]);
+        if (cmp != 0) {
+          // current name > last name -> skip
+          return cmp > 0;
         }
       }
-
-      // if currentDir and lastScannedDir has same parent
-      do {
-        long parentID = currentDir.getParentObjectID();
-        for (int i = 0; i < count; i++) {
-          OmDirectoryInfo dir = dirList.getSubDirList().get(i);
-          if (dir.getParentObjectID() == parentID) {
-            if (dirList.getSubDirKeyList().get(i).compareTo(currentDirTableKey) < 0) {
-              return true;
-            }
-          }
-        }
-        return false;
-      } while (true);
+      return false;
     }
 
     @SuppressWarnings({"checkstyle:parameternumber", "checkstyle:MethodLength"})
@@ -676,19 +658,6 @@ private void evaluateKeyAndDirTable(OmBucketInfo bucket, long volumeObjId, Table
       String lastScannedDirInState = scanStateBuilder == null ? null : scanStateBuilder.getLastScannedDir();
       String lastScannedDirKeyInState = scanStateBuilder == null ? null : scanStateBuilder.getLastScannedDirKey();
       String lastScannedKeyInState = scanStateBuilder == null ? null : scanStateBuilder.getLastScannedKey();
-      DirectoryList lastScannedDirList = null;
-      String bucketKey = omMetadataManager.getBucketKey(volumeName, bucketName);
-      if (lastScannedDirInState != null && lastScannedDirKeyInState != null) {
-        // find all parents of lastScannedSubDir
-        try {
-          lastScannedDirList = getDirList(volumeObjId, bucket, lastScannedDirInState, bucketKey);
-        } catch (IOException e) {
-          // Saved lastScannedDir in state could be deleted or renamed after it's saved.
-          // Fallback to no state saved status.
-          LOG.info("Failed to get DirList for lastScannedDirInState {}", lastScannedDirInState, e);
-          lastScannedDirInState = null;
-        }
-      }
       try {
         if (dir != null) {
           stack.push(new PendingEvaluateDirectory(dir, dirKey, directoryPath, null));
@@ -748,8 +717,7 @@ private void evaluateKeyAndDirTable(OmBucketInfo bucket, long volumeObjId, Table
          *   - dir3/dir8, lastScannedDir, partially scanned,
          *   - dir3/dir9, same the same parentID as lastScannedDir, and name order > lastScannedDir, scanned, skip
          */
-        if (lastScannedDirList != null &&
-            canSkipDir(currentDir, currentDirTableKey, lastScannedDirList)) {
+        if (canSkipDir(currentDirPath, lastScannedDirInState)) {
           LOG.info("Skip {} in LifecycleActionTask for bucket {}. ", currentDirPath, bucketName);
           continue;
         }
@@ -885,33 +853,12 @@ private void evaluateKeyAndDirTable(OmBucketInfo bucket, long volumeObjId, Table
 
         try (TableIterator<String, ? extends Table.KeyValue<String, OmKeyInfo>> keyTblItr =
                  keyTable.iterator(prefix)) {
+          boolean seekPerformed = false;
           if (lastScannedDirKeyInState != null && lastScannedDirKeyInState.compareTo(currentDirTableKey) == 0 &&
               lastScannedKeyInState != null && lastScannedKeyInState.startsWith(prefix)) {
             LOG.info("Seek to key {} under directory {}", scanStateBuilder.getLastScannedKey(), lastScannedDirInState);
             keyTblItr.seek(scanStateBuilder.getLastScannedKey());
-            if (keyTblItr.hasNext()) {
-              Table.KeyValue<String, OmKeyInfo> first = keyTblItr.next();
-              if (!first.getKey().equals(scanStateBuilder.getLastScannedKey())) {
-                OmKeyInfo key = first.getValue();
-                String keyPath = currentDirPath.isEmpty() ? key.getKeyName() :
-                    currentDirPath + OM_KEY_PREFIX + key.getKeyName();
-                if (!deletedKeySetInCache.remove(first.getKey()) && !keySetInCache.remove(first.getKey())) {
-                  numKeyIterated++;
-                  numKeysUnderDir++;
-                  for (OmLCRule rule : ruleList) {
-                    if (key.getParentObjectID() == currentDirObjID && rule.match(key, keyPath)) {
-                      if (keyList.isFull()) {
-                        handleAndClearFullList(bucket, keyList, false, scanStateBuilder, false);
-                      }
-                      keyList.add(keyPath, key.getReplicatedSize(), key.getUpdateID());
-                      numKeysExpired++;
-                      break;
-                    }
-                  }
-                  lastScannedKey = first.getKey();
-                }
-              }
-            }
+            seekPerformed = true;
           }
 
           while (keyTblItr.hasNext()) {
@@ -923,6 +870,9 @@ private void evaluateKeyAndDirTable(OmBucketInfo bucket, long volumeObjId, Table
             OmKeyInfo key = keyValue.getValue();
             String keyPath = currentDirPath.isEmpty() ? key.getKeyName() :
                 currentDirPath + OM_KEY_PREFIX + key.getKeyName();
+            if (seekPerformed && keyValue.getKey().equals(scanStateBuilder.getLastScannedKey())) {
+              continue;
+            }
             if (deletedKeySetInCache.remove(keyValue.getKey()) || keySetInCache.remove(keyValue.getKey())) {
               continue;
             }
@@ -1066,19 +1016,10 @@ private void evaluateBucket(OmBucketInfo bucketInfo,
 
       try (TableIterator<String, ? extends Table.KeyValue<String, OmKeyInfo>> keyTblItr =
                keyTable.iterator(bucketPrefix)) {
+        boolean seekPerformed = false;
         if (scanStateBuilder != null && scanStateBuilder.getLastScannedKey() != null) {
           keyTblItr.seek(scanStateBuilder.getLastScannedKey());
-          // Skip the exact match since it was already processed
-          if (keyTblItr.hasNext()) {
-            Table.KeyValue<String, OmKeyInfo> first = keyTblItr.next();
-            if (!first.getKey().equals(scanStateBuilder.getLastScannedKey())) {
-              // We seeked past it, so we need to process this one.
-              // We can't easily "push back" in TableIterator, so we handle it here.
-              processKey(bucketInfo, first.getValue(), ruleList, expiredKeyList, scanStateBuilder);
-              numKeyIterated++;
-              lastScannedKey = first.getKey();
-            }
-          }
+          seekPerformed = true;
         }
 
         while (keyTblItr.hasNext()) {
@@ -1091,6 +1032,9 @@ private void evaluateBucket(OmBucketInfo bucketInfo,
             flushAndSaveState(bucketInfo, expiredKeyList, null, scanStateBuilder);
           }
           Table.KeyValue<String, OmKeyInfo> keyValue = keyTblItr.next();
+          if (seekPerformed && keyValue.getKey().equals(scanStateBuilder.getLastScannedKey())) {
+            continue;
+          }
           processKey(bucketInfo, keyValue.getValue(), ruleList, expiredKeyList, scanStateBuilder);
           numKeyIterated++;
           lastScannedKey = keyValue.getKey();

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyLifecycleService.java

@@ -219,6 +219,7 @@ private void createConfig(File testDir) {
     conf.setLong(OZONE_KEY_LIFECYCLE_SERVICE_STATE_SAVE_INTERVAL_MS, stateSaveInternal);
     conf.setLong(OZONE_KEY_LIFECYCLE_SERVICE_STATE_SAVE_KEYS_PROCESSED, maxKeysProcessedPerState);
     OmLCExpiration.setTest(true);
+    KeyLifecycleService.setTest(true);
   }
 
   private void createSubject() throws Exception {
@@ -1083,7 +1084,7 @@ void testBucketRootScannedDirResume(BucketLayout layout) throws Exception {
       // So key1 and key2 are skipped, key3 is deleted.
       int expectedDeleted = 1;
       GenericTestUtils.waitFor(() ->
-          (getDeletedKeyCount() - initialDeletedKeyCount) >= expectedDeleted, WAIT_CHECK_INTERVAL, 10000);
+          (getDeletedKeyCount() - initialDeletedKeyCount) == expectedDeleted, WAIT_CHECK_INTERVAL, 10000);
 
       assertEquals(2, getKeyCount(layout) - initialKeyCount);
       GenericTestUtils.waitFor(() ->