HDDS-13648. Update NSSummary rebuilding implementation to queue based approach. (#9009)

Author: devmadhuu

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconUtils.java

@@ -49,8 +49,6 @@
 import java.util.Set;
 import java.util.TimeZone;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.stream.Collectors;
 import javax.ws.rs.core.Response;
 import org.apache.commons.io.FileUtils;
@@ -97,15 +95,6 @@ public class ReconUtils {
   private static Logger log = LoggerFactory.getLogger(
       ReconUtils.class);
 
-  // Use NSSummaryTask's unified rebuild control instead of separate tracking
-  private static final ExecutorService NSSUMMARY_REBUILD_EXECUTOR =
-      Executors.newSingleThreadExecutor(r -> {
-        Thread t = new Thread(r);
-        t.setName("RebuildNSSummaryThread");
-        t.setDaemon(true); // Optional: allows JVM to exit without waiting
-        return t;
-      });
-
   public ReconUtils() {
   }
 
@@ -119,33 +108,6 @@ public static org.apache.hadoop.ozone.recon.tasks.NSSummaryTask.RebuildState get
     return org.apache.hadoop.ozone.recon.tasks.NSSummaryTask.getRebuildState();
   }
 
-  /**
-   * Convenience method to trigger asynchronous NSSummary tree rebuild.
-   * Uses the unified control mechanism in NSSummaryTask.
-   * 
-   * @param reconNamespaceSummaryManager The namespace summary manager
-   * @param omMetadataManager The OM metadata manager
-   * @return true if rebuild was triggered successfully, false otherwise
-   */
-  public static boolean triggerAsyncNSSummaryRebuild(
-      ReconNamespaceSummaryManager reconNamespaceSummaryManager,
-      ReconOMMetadataManager omMetadataManager) {
-
-    // Submit rebuild task to single thread executor for async execution
-    NSSUMMARY_REBUILD_EXECUTOR.submit(() -> {
-      try {
-
-        // This will go through NSSummaryTask's unified control mechanism
-        reconNamespaceSummaryManager.rebuildNSSummaryTree(omMetadataManager);
-        log.info("Async NSSummary tree rebuild completed successfully.");
-      } catch (Exception e) {
-        log.error("Async NSSummary tree rebuild failed.", e);
-      }
-    });
-    
-    return true;
-  }
-
   public static File getReconScmDbDir(ConfigurationSource conf) {
     return new ReconUtils().getReconDbDir(conf, OZONE_RECON_SCM_DB_DIR);
   }
@@ -299,13 +261,6 @@ public static StringBuilder constructFullPathPrefix(long initialParentId, String
             "deletion, returning empty string for path construction.");
         throw new ServiceNotReadyException("Service is initializing. Please try again later.");
       }
-      if (nsSummary.getParentId() == -1) {
-        // Trigger async rebuild using unified control mechanism
-        triggerAsyncNSSummaryRebuild(reconNamespaceSummaryManager, omMetadataManager);
-        log.warn(
-            "NSSummary tree corruption detected, rebuild triggered. Returning empty string for path construction.");
-        throw new ServiceNotReadyException("Service is initializing. Please try again later.");
-      }
       // On the last pass, dir-name will be empty and parent will be zero, indicating the loop should end.
       if (!nsSummary.getDirName().isEmpty()) {
         pathSegments.add(nsSummary.getDirName());

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/ReconNamespaceSummaryManager.java

@@ -22,7 +22,6 @@
 import org.apache.hadoop.hdds.utils.db.BatchOperation;
 import org.apache.hadoop.hdds.utils.db.DBStore;
 import org.apache.hadoop.hdds.utils.db.RDBBatchOperation;
-import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.recon.api.types.NSSummary;
 import org.apache.hadoop.ozone.recon.spi.impl.ReconDBProvider;
 
@@ -52,6 +51,4 @@ void batchStoreNSSummaries(BatchOperation batch, long objectId,
 
   void commitBatchOperation(RDBBatchOperation rdbBatchOperation)
       throws IOException;
-
-  void rebuildNSSummaryTree(OMMetadataManager omMetadataManager);
 }

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconNamespaceSummaryManagerImpl.java

@@ -26,7 +26,6 @@
 import org.apache.hadoop.hdds.utils.db.DBStore;
 import org.apache.hadoop.hdds.utils.db.RDBBatchOperation;
 import org.apache.hadoop.hdds.utils.db.Table;
-import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.recon.api.types.NSSummary;
 import org.apache.hadoop.ozone.recon.spi.ReconNamespaceSummaryManager;
 import org.apache.hadoop.ozone.recon.tasks.NSSummaryTask;
@@ -105,13 +104,6 @@ public void commitBatchOperation(RDBBatchOperation rdbBatchOperation)
     this.namespaceDbStore.commitBatchOperation(rdbBatchOperation);
   }
 
-  @Override
-  public void rebuildNSSummaryTree(OMMetadataManager omMetadataManager) {
-    // This method is called by the unified ReconUtils.triggerNSSummaryRebuild
-    // It should only handle the actual rebuild logic without state management
-    nsSummaryTask.reprocess(omMetadataManager);
-  }
-
   public Table getNSSummaryTable() {
     return nsSummaryTable;
   }

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

@@ -205,9 +205,14 @@ public TaskResult reprocess(OMMetadataManager omMetadataManager) {
       LOG.info("NSSummary tree rebuild is already in progress, skipping duplicate request.");
       return buildTaskResult(false);
     }
-    
+
     if (!REBUILD_STATE.compareAndSet(currentState, RebuildState.RUNNING)) {
-      LOG.info("Failed to acquire rebuild lock, another thread may have started rebuild.");
+      // Check if another thread successfully started the rebuild
+      if (REBUILD_STATE.get() == RebuildState.RUNNING) {
+        LOG.info("Rebuild already in progress by another thread, returning success");
+        return buildTaskResult(true);
+      }
+      LOG.info("Failed to acquire rebuild lock, unknown state");
       return buildTaskResult(false);
     }
 
@@ -250,7 +255,7 @@ protected TaskResult executeReprocess(OMMetadataManager omMetadataManager, long
     ThreadFactory threadFactory = new ThreadFactoryBuilder()
         .setNameFormat("Recon-NSSummaryTask-%d")
         .build();
-    ExecutorService executorService = Executors.newFixedThreadPool(2,
+    ExecutorService executorService = Executors.newFixedThreadPool(3,
         threadFactory);
     boolean success = false;
     try {
@@ -263,14 +268,31 @@ protected TaskResult executeReprocess(OMMetadataManager omMetadataManager, long
         }
       }
       success = true;
-      
-    } catch (InterruptedException | ExecutionException ex) {
-      LOG.error("Error while reprocessing NSSummary table in Recon DB.", ex);
+
+    } catch (InterruptedException ex) {
+      Thread.currentThread().interrupt();
+      LOG.error("NSSummaryTask was interrupted.", ex);
+      REBUILD_STATE.set(RebuildState.FAILED);
+      return buildTaskResult(false);
+    } catch (ExecutionException ex) {
+      LOG.error("Error while reprocessing NSSummary table in Recon DB.", ex.getCause());
       REBUILD_STATE.set(RebuildState.FAILED);
       return buildTaskResult(false);
-      
     } finally {
       executorService.shutdown();
+      // Deterministic resource cleanup with timeout
+      try {
+        // get() ensures the work is done. awaitTermination ensures the workers are also verifiably gone.
+        // It turns an asynchronous shutdown into a synchronous, deterministic one
+        if (!executorService.awaitTermination(5, TimeUnit.MINUTES)) {
+          LOG.warn("Executor service for NSSummaryTask did not terminate in the specified time.");
+          executorService.shutdownNow();
+        }
+      } catch (InterruptedException ex) {
+        LOG.error("NSSummaryTask executor service termination was interrupted.", ex);
+        executorService.shutdownNow();
+        Thread.currentThread().interrupt();
+      }
 
       long endTime = System.nanoTime();
       // Convert to milliseconds

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

@@ -261,7 +261,7 @@ protected boolean flushAndCommitUpdatedNSToDB(Map<Long, NSSummary> nsSummaryMap,
     try {
       updateNSSummariesToDB(nsSummaryMap, objectIdsToBeDeleted);
     } catch (IOException e) {
-      LOG.error("Unable to write Namespace Summary data in Recon DB.", e);
+      LOG.error("Unable to write Namespace Summary data in Recon DB. batchSize={}", nsSummaryMap.size(), e);
       return false;
     } finally {
       nsSummaryMap.clear();

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/NSSummaryAggregatedTotalsUpgrade.java

@@ -22,9 +22,8 @@
 import com.google.inject.Injector;
 import javax.sql.DataSource;
 import org.apache.hadoop.ozone.recon.ReconGuiceServletContextListener;
-import org.apache.hadoop.ozone.recon.ReconUtils;
-import org.apache.hadoop.ozone.recon.recovery.ReconOMMetadataManager;
-import org.apache.hadoop.ozone.recon.spi.ReconNamespaceSummaryManager;
+import org.apache.hadoop.ozone.recon.tasks.ReconTaskController;
+import org.apache.hadoop.ozone.recon.tasks.ReconTaskReInitializationEvent;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -50,12 +49,16 @@ public void execute(DataSource source) throws Exception {
           "Guice injector not initialized. NSSummary rebuild cannot proceed during upgrade.");
     }
 
-    ReconNamespaceSummaryManager nsMgr = injector.getInstance(ReconNamespaceSummaryManager.class);
-    ReconOMMetadataManager omMgr = injector.getInstance(ReconOMMetadataManager.class);
-
-    // Fire and forget: unified control using ReconUtils -> NSSummaryTask
+    ReconTaskController reconTaskController = injector.getInstance(ReconTaskController.class);
     LOG.info("Triggering asynchronous NSSummary tree rebuild for materialized totals (upgrade action).");
-    ReconUtils.triggerAsyncNSSummaryRebuild(nsMgr, omMgr);
+    ReconTaskController.ReInitializationResult result = reconTaskController.queueReInitializationEvent(
+        ReconTaskReInitializationEvent.ReInitializationReason.MANUAL_TRIGGER);
+    if (result != ReconTaskController.ReInitializationResult.SUCCESS) {
+      LOG.error(
+          "Failed to queue reinitialization event for manual trigger (result: {}), failing the reinitialization " +
+              "during NSSummaryAggregatedTotalsUpgrade action, will be retried as part of syncDataFromOM " +
+              "scheduler task.", result);
+    }
   }
 
   @Override

hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestNSSummaryEndpointWithFSO.java

@@ -29,9 +29,7 @@
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.mockito.Mockito.anyLong;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.io.File;
@@ -88,8 +86,6 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
-import org.mockito.ArgumentCaptor;
-import org.slf4j.Logger;
 
 /**
  * Test for NSSummary REST APIs with FSO.
@@ -813,40 +809,6 @@ public void testConstructFullPathWithNegativeParentIdTriggersRebuild() throws IO
         ReconUtils.constructFullPath(keyInfo, mockSummaryManager, mockMetadataManager));
   }
 
-  @Test
-  public void testLoggingWhenParentIdIsNegative() throws IOException {
-    ReconNamespaceSummaryManager mockManager =
-        mock(ReconNamespaceSummaryManager.class);
-    Logger mockLogger = mock(Logger.class);
-    ReconUtils.setLogger(mockLogger);
-
-    NSSummary mockSummary = new NSSummary();
-    mockSummary.setParentId(-1);
-    when(mockManager.getNSSummary(anyLong())).thenReturn(mockSummary);
-
-    OmKeyInfo keyInfo = new OmKeyInfo.Builder()
-        .setKeyName("testKey")
-        .setVolumeName("vol")
-        .setBucketName("bucket")
-        .setObjectID(1L)
-        .setParentObjectID(1L)
-        .build();
-
-    assertThrows(ServiceNotReadyException.class, () ->
-        ReconUtils.constructFullPath(keyInfo, mockManager, null));
-
-    // Assert
-    ArgumentCaptor<String> logCaptor = ArgumentCaptor.forClass(String.class);
-    verify(mockLogger).warn(logCaptor.capture());
-    String loggedMessage = logCaptor.getValue();
-
-    // Here we can assert the exact message we expect to see in the logs.
-    // Since we set parentId = -1, this triggers the corruption detection path
-    assertEquals(
-        "NSSummary tree corruption detected, rebuild triggered. Returning empty string " +
-            "for path construction.", loggedMessage);
-  }
-
   /**
    * Write directories and keys info into OM DB.
    * @throws Exception

hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryUnifiedControl.java

@@ -434,54 +434,6 @@ void testReconUtilsIntegration() throws Exception {
         "State should return to IDLE after completion");
   }
 
-  /**
-   * Test that ReconUtils async trigger respects unified control.
-   */
-  @Test
-  void testReconUtilsRespectsUnifiedControl() throws Exception {
-    CountDownLatch firstRebuildStarted = new CountDownLatch(1);
-    CountDownLatch firstRebuildCanFinish = new CountDownLatch(1);
-
-    // Setup first rebuild to block
-    doAnswer(invocation -> {
-      firstRebuildStarted.countDown();
-      boolean awaitSuccess = firstRebuildCanFinish.await(5, TimeUnit.SECONDS);
-      if (!awaitSuccess) {
-        LOG.warn("firstRebuildCanFinish.await() timed out");
-      }
-      return null;
-    }).when(mockNamespaceSummaryManager).clearNSSummaryTable();
-
-    // Start first rebuild via NSSummaryTask directly
-    CompletableFuture<TaskResult> directRebuild = CompletableFuture.supplyAsync(() -> 
-        nsSummaryTask.reprocess(mockOMMetadataManager));
-
-    // Wait for first rebuild to start
-    assertTrue(firstRebuildStarted.await(5, TimeUnit.SECONDS),
-        "First rebuild should start");
-    assertEquals(RebuildState.RUNNING, NSSummaryTask.getRebuildState(),
-        "State should be RUNNING");
-
-    // Try to trigger via ReconUtils - should still respect the running state
-    // (The async execution will be queued but the actual rebuild will be rejected)
-    boolean triggered = ReconUtils.triggerAsyncNSSummaryRebuild(
-        mockNamespaceSummaryManager, mockReconOMMetadataManager);
-    assertTrue(triggered, "ReconUtils should queue the async request");
-
-    // State should still be RUNNING from the first rebuild
-    assertEquals(RebuildState.RUNNING, NSSummaryTask.getRebuildState(),
-        "State should still be RUNNING from first rebuild");
-
-    // Complete first rebuild
-    firstRebuildCanFinish.countDown();
-    TaskResult result = directRebuild.get(5, TimeUnit.SECONDS);
-    assertTrue(result.isTaskSuccess(), "First rebuild should succeed");
-
-    // Final state should be IDLE
-    assertEquals(RebuildState.IDLE, NSSummaryTask.getRebuildState(),
-        "Final state should be IDLE");
-  }
-
   /**
    * Test state transitions during exception scenarios.
    */