HDDS-15335. Recon: parallelize NSSummaryTask sub-tasks and cache OmBucketInfo lookups (#10321)

Author: smengcl

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/NSSummaryTask.java

@@ -83,6 +83,16 @@ public class NSSummaryTask implements ReconOmTask {
   private final NSSummaryTaskWithLegacy nsSummaryTaskWithLegacy;
   private final NSSummaryTaskWithOBS nsSummaryTaskWithOBS;
 
+  // Shared executor for the three FSO/Legacy/OBS sub-tasks during process().
+  // The sub-tasks operate on disjoint slices of the event stream (filtered by
+  // table and bucket layout) and write to disjoint NSSummary entries, so they
+  // are safe to run in parallel.
+  private static final ExecutorService SUB_TASK_EXECUTOR =
+      Executors.newFixedThreadPool(3, new ThreadFactoryBuilder()
+          .setNameFormat("NSSummarySubTask-%d")
+          .setDaemon(true)
+          .build());
+
   /**
    * Rebuild state enum to track NSSummary tree rebuild status.
    */
@@ -172,44 +182,58 @@ public String getDescription() {
   @Override
   public TaskResult process(
       OMUpdateEventBatch events, Map<String, Integer> subTaskSeekPosMap) {
-    boolean anyFailure = false; // Track if any bucket fails
     Map<String, Integer> updatedSeekPositions = new HashMap<>();
 
-    // Process FSO bucket
-    Integer bucketSeek = subTaskSeekPosMap.getOrDefault(BucketType.FSO.name(), 0);
-    Pair<Integer, Boolean> bucketResult = nsSummaryTaskWithFSO.processWithFSO(events, bucketSeek);
-    updatedSeekPositions.put(BucketType.FSO.name(), bucketResult.getLeft());
-    if (!bucketResult.getRight()) {
-      LOG.error("processWithFSO failed.");
-      anyFailure = true;
-    }
-
-    // Process Legacy bucket
-    bucketSeek = subTaskSeekPosMap.getOrDefault(BucketType.LEGACY.name(), 0);
-    bucketResult = nsSummaryTaskWithLegacy.processWithLegacy(events, bucketSeek);
-    updatedSeekPositions.put(BucketType.LEGACY.name(), bucketResult.getLeft());
-    if (!bucketResult.getRight()) {
-      LOG.error("processWithLegacy failed.");
-      anyFailure = true;
-    }
-
-    // Process OBS bucket
-    bucketSeek = subTaskSeekPosMap.getOrDefault(BucketType.OBS.name(), 0);
-    bucketResult = nsSummaryTaskWithOBS.processWithOBS(events, bucketSeek);
-    updatedSeekPositions.put(BucketType.OBS.name(), bucketResult.getLeft());
-    if (!bucketResult.getRight()) {
-      LOG.error("processWithOBS failed.");
-      anyFailure = true;
-    }
+    int fsoSeek = subTaskSeekPosMap.getOrDefault(BucketType.FSO.name(), 0);
+    int legacySeek = subTaskSeekPosMap.getOrDefault(BucketType.LEGACY.name(), 0);
+    int obsSeek = subTaskSeekPosMap.getOrDefault(BucketType.OBS.name(), 0);
+
+    Future<Pair<Integer, Boolean>> fsoFuture = SUB_TASK_EXECUTOR.submit(
+        () -> nsSummaryTaskWithFSO.processWithFSO(events, fsoSeek));
+    Future<Pair<Integer, Boolean>> legacyFuture = SUB_TASK_EXECUTOR.submit(
+        () -> nsSummaryTaskWithLegacy.processWithLegacy(events, legacySeek));
+    Future<Pair<Integer, Boolean>> obsFuture = SUB_TASK_EXECUTOR.submit(
+        () -> nsSummaryTaskWithOBS.processWithOBS(events, obsSeek));
+
+    boolean anyFailure = false;
+    anyFailure |= !awaitSubTask("processWithFSO", BucketType.FSO,
+        fsoFuture, fsoSeek, updatedSeekPositions);
+    anyFailure |= !awaitSubTask("processWithLegacy", BucketType.LEGACY,
+        legacyFuture, legacySeek, updatedSeekPositions);
+    anyFailure |= !awaitSubTask("processWithOBS", BucketType.OBS,
+        obsFuture, obsSeek, updatedSeekPositions);
 
-    // Return task failure if any bucket failed, while keeping each bucket's latest seek position
     return new TaskResult.Builder()
         .setTaskName(getTaskName())
         .setSubTaskSeekPositions(updatedSeekPositions)
         .setTaskSuccess(!anyFailure)
         .build();
   }
 
+  private boolean awaitSubTask(String name, BucketType type,
+                               Future<Pair<Integer, Boolean>> future,
+                               int fallbackSeek,
+                               Map<String, Integer> updatedSeekPositions) {
+    try {
+      Pair<Integer, Boolean> result = future.get();
+      updatedSeekPositions.put(type.name(), result.getLeft());
+      if (!result.getRight()) {
+        LOG.error("{} failed.", name);
+        return false;
+      }
+      return true;
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      LOG.error("{} interrupted.", name, e);
+      updatedSeekPositions.put(type.name(), fallbackSeek);
+      return false;
+    } catch (ExecutionException e) {
+      LOG.error("{} threw an exception.", name, e.getCause());
+      updatedSeekPositions.put(type.name(), fallbackSeek);
+      return false;
+    }
+  }
+
   @Override
   public TaskResult reprocess(OMMetadataManager omMetadataManager) {
     // Unified control for all NSS tree rebuild operations

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/NSSummaryTaskDbEventHandler.java

@@ -20,8 +20,10 @@
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import org.apache.hadoop.hdds.utils.db.RDBBatchOperation;
+import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
 import org.apache.hadoop.ozone.om.helpers.OmDirectoryInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.recon.ReconUtils;
@@ -43,6 +45,18 @@ public class NSSummaryTaskDbEventHandler {
   private ReconNamespaceSummaryManager reconNamespaceSummaryManager;
   private ReconOMMetadataManager reconOMMetadataManager;
 
+  // Cache OmBucketInfo lookups across process() calls so the Legacy and OBS
+  // sub-tasks don't pay a RocksDB point read per event. A bucket's objectID and
+  // layout are stable while the bucket exists, but a bucket can be deleted and
+  // recreated under the same volume/bucket name with a new objectID (same DB
+  // key, different identity). A recreate is always preceded by a delete, so the
+  // sub-tasks call invalidateBucketCache() when they observe a bucketTable
+  // delete event; a recreated bucket is then re-read instead of served stale.
+  //
+  // Single-thread access only (each sub-task runs on its own thread and owns
+  // its own cache instance). HashMap is fine.
+  private final Map<String, OmBucketInfo> bucketInfoCache = new HashMap<>();
+
   public NSSummaryTaskDbEventHandler(ReconNamespaceSummaryManager
                                      reconNamespaceSummaryManager,
                                      ReconOMMetadataManager
@@ -51,6 +65,34 @@ public NSSummaryTaskDbEventHandler(ReconNamespaceSummaryManager
     this.reconOMMetadataManager = reconOMMetadataManager;
   }
 
+  /** Look up an {@link OmBucketInfo} via {@code getBucketTable().getSkipCache}
+   *  and cache the result. Bucket layout/object-id are stable while a bucket
+   *  exists, so a field-level cache avoids one RocksDB point read per event in
+   *  the per-event sub-task loops. Entries are dropped via
+   *  {@link #invalidateBucketCache(String)} when a bucketTable delete event is
+   *  seen, so a bucket deleted and recreated under the same name is not served
+   *  stale. */
+  protected OmBucketInfo lookupBucketCached(String bucketDBKey) throws IOException {
+    OmBucketInfo cached = bucketInfoCache.get(bucketDBKey);
+    if (cached != null) {
+      return cached;
+    }
+    OmBucketInfo info = reconOMMetadataManager.getBucketTable().getSkipCache(bucketDBKey);
+    if (info != null) {
+      bucketInfoCache.put(bucketDBKey, info);
+    }
+    return info;
+  }
+
+  /** Drop the cached {@link OmBucketInfo} for the given bucket DB key. Invoked
+   *  when a bucketTable delete event is observed so the next key event re-reads
+   *  the current bucket info. This matters when a bucket is deleted and
+   *  recreated under the same volume/bucket name, which assigns a new objectID;
+   *  the recreate always follows the delete, so invalidating on delete suffices. */
+  protected void invalidateBucketCache(String bucketDBKey) {
+    bucketInfoCache.remove(bucketDBKey);
+  }
+
   public ReconNamespaceSummaryManager getReconNamespaceSummaryManager() {
     return reconNamespaceSummaryManager;
   }

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/NSSummaryTaskWithLegacy.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.ozone.recon.tasks;
 
 import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX;
+import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.BUCKET_TABLE;
 import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.KEY_TABLE;
 
 import java.io.IOException;
@@ -91,9 +92,20 @@ public Pair<Integer, Boolean> processWithLegacy(OMUpdateEventBatch events,
       OMDBUpdateEvent.OMDBUpdateAction action = omdbUpdateEvent.getAction();
       eventCounter++;
 
-      // we only process updates on OM's KeyTable
       String table = omdbUpdateEvent.getTable();
+      // A bucket can be deleted and recreated under the same name with a new
+      // objectID. A recreate is always preceded by a delete, so dropping the
+      // cached OmBucketInfo on the delete event is enough for a later key event
+      // to re-read the recreated bucket. Bucket property updates don't change
+      // objectID or layout, so they need not invalidate the cache.
+      if (table.equals(BUCKET_TABLE)) {
+        if (action == OMDBUpdateEvent.OMDBUpdateAction.DELETE) {
+          invalidateBucketCache(omdbUpdateEvent.getKey());
+        }
+        continue;
+      }
 
+      // we only process updates on OM's KeyTable
       if (!table.equals(KEY_TABLE)) {
         continue;
       }
@@ -363,8 +375,7 @@ private long setParentBucketId(OmKeyInfo keyInfo)
       throws IOException {
     String bucketKey = getReconOMMetadataManager()
         .getBucketKey(keyInfo.getVolumeName(), keyInfo.getBucketName());
-    OmBucketInfo parentBucketInfo =
-        getReconOMMetadataManager().getBucketTable().getSkipCache(bucketKey);
+    OmBucketInfo parentBucketInfo = lookupBucketCached(bucketKey);
 
     if (parentBucketInfo != null) {
       return parentBucketInfo.getObjectID();
@@ -388,8 +399,7 @@ private boolean isBucketLayoutValid(ReconOMMetadataManager metadataManager,
     String volumeName = keyInfo.getVolumeName();
     String bucketName = keyInfo.getBucketName();
     String bucketDBKey = metadataManager.getBucketKey(volumeName, bucketName);
-    OmBucketInfo omBucketInfo =
-        metadataManager.getBucketTable().getSkipCache(bucketDBKey);
+    OmBucketInfo omBucketInfo = lookupBucketCached(bucketDBKey);
 
     if (omBucketInfo.getBucketLayout() != LEGACY_BUCKET_LAYOUT) {
       LOG.debug(

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/NSSummaryTaskWithOBS.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.ozone.recon.tasks;
 
+import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.BUCKET_TABLE;
 import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.KEY_TABLE;
 
 import java.io.IOException;
@@ -201,10 +202,21 @@ public Pair<Integer, Boolean> processWithOBS(OMUpdateEventBatch events,
       OMDBUpdateEvent.OMDBUpdateAction action = omdbUpdateEvent.getAction();
       eventCounter++;
 
-      // We only process updates on OM's KeyTable
       String table = omdbUpdateEvent.getTable();
-      boolean updateOnKeyTable = table.equals(KEY_TABLE);
-      if (!updateOnKeyTable) {
+      // A bucket can be deleted and recreated under the same name with a new
+      // objectID. A recreate is always preceded by a delete, so dropping the
+      // cached OmBucketInfo on the delete event is enough for a later key event
+      // to re-read the recreated bucket. Bucket property updates don't change
+      // objectID or layout, so they need not invalidate the cache.
+      if (table.equals(BUCKET_TABLE)) {
+        if (action == OMDBUpdateEvent.OMDBUpdateAction.DELETE) {
+          invalidateBucketCache(omdbUpdateEvent.getKey());
+        }
+        continue;
+      }
+
+      // We only process updates on OM's KeyTable
+      if (!table.equals(KEY_TABLE)) {
         continue;
       }
 
@@ -234,15 +246,13 @@ public Pair<Integer, Boolean> processWithOBS(OMUpdateEventBatch events,
         String bucketName = updatedKeyInfo.getBucketName();
         String bucketDBKey =
             getReconOMMetadataManager().getBucketKey(volumeName, bucketName);
-        // Get bucket info from bucket table
-        OmBucketInfo omBucketInfo = getReconOMMetadataManager().getBucketTable()
-            .getSkipCache(bucketDBKey);
+        OmBucketInfo omBucketInfo = lookupBucketCached(bucketDBKey);
 
         if (omBucketInfo.getBucketLayout() != BUCKET_LAYOUT) {
           continue;
         }
 
-        long parentObjectID = getKeyParentID(updatedKeyInfo);
+        long parentObjectID = omBucketInfo.getObjectID();
 
         switch (action) {
         case PUT:
@@ -253,9 +263,10 @@ public Pair<Integer, Boolean> processWithOBS(OMUpdateEventBatch events,
           break;
         case UPDATE:
           if (oldKeyInfo != null) {
-            // delete first, then put
-            long oldKeyParentObjectID = getKeyParentID(oldKeyInfo);
-            handleDeleteKeyEvent(oldKeyInfo, nsSummaryMap, oldKeyParentObjectID);
+            // For OBS, parent is always the bucket, so same parentObjectID
+            // applies to old and new (a key cannot move between buckets via
+            // an UPDATE event — that would be a delete+put).
+            handleDeleteKeyEvent(oldKeyInfo, nsSummaryMap, parentObjectID);
           } else {
             LOG.warn("Update event does not have the old keyInfo for {}.",
                 updatedKey);