HDDS-12185. Enhance FileSizeCountTask for Faster Processing.

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

@@ -18,6 +18,8 @@
 
 package org.apache.hadoop.ozone.recon;
 
+import java.util.concurrent.atomic.AtomicBoolean;
+
 /**
  * Recon Server constants file.
  */
@@ -90,4 +92,15 @@ private ReconConstants() {
       (double) MAX_CONTAINER_SIZE_UPPER_BOUND /
           MIN_CONTAINER_SIZE_UPPER_BOUND) /
       Math.log(2)) + 1;
+
+  // For file-size count reprocessing: ensure only one task truncates the table.
+  public static final AtomicBoolean TABLE_TRUNCATED = new AtomicBoolean(false);
+
+  /**
+   * Resets the table-truncated flag. This should be called once per reprocess cycle,
+   * for example by the OM task controller, before the tasks run.
+   */
+  public static void resetTableTruncatedFlag() {
+    TABLE_TRUNCATED.set(false);
+  }
 }

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

@@ -49,7 +49,8 @@
 import org.apache.hadoop.ozone.recon.spi.impl.ReconNamespaceSummaryManagerImpl;
 import org.apache.hadoop.ozone.recon.spi.impl.StorageContainerServiceProviderImpl;
 import org.apache.hadoop.ozone.recon.tasks.ContainerKeyMapperTask;
-import org.apache.hadoop.ozone.recon.tasks.FileSizeCountTask;
+import org.apache.hadoop.ozone.recon.tasks.FileSizeCountTaskFSO;
+import org.apache.hadoop.ozone.recon.tasks.FileSizeCountTaskOBS;
 import org.apache.hadoop.ozone.recon.tasks.NSSummaryTask;
 import org.apache.hadoop.ozone.recon.tasks.ReconOmTask;
 import org.apache.hadoop.ozone.recon.tasks.ReconTaskController;
@@ -74,6 +75,8 @@
 import java.io.IOException;
 import java.lang.reflect.Constructor;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 import static org.apache.hadoop.hdds.scm.cli.ContainerOperationClient.newContainerRpcClient;
 import static org.apache.hadoop.ozone.OmUtils.getOzoneManagerServiceId;
@@ -130,12 +133,19 @@ protected void configure() {
       Multibinder<ReconOmTask> taskBinder =
           Multibinder.newSetBinder(binder(), ReconOmTask.class);
       taskBinder.addBinding().to(ContainerKeyMapperTask.class);
-      taskBinder.addBinding().to(FileSizeCountTask.class);
+      taskBinder.addBinding().to(FileSizeCountTaskFSO.class);
+      taskBinder.addBinding().to(FileSizeCountTaskOBS.class);
       taskBinder.addBinding().to(OmTableInsightTask.class);
       taskBinder.addBinding().to(NSSummaryTask.class);
     }
   }
 
+  @Provides
+  @Singleton
+  public ExecutorService provideReconExecutorService() {
+    return Executors.newFixedThreadPool(5);
+  }
+
   /**
    * Class that has all the DAO bindings in Recon.
    */

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/FileSizeCountTask.java → hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/FileSizeCountTaskFSO.java

@@ -26,6 +26,7 @@
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.hdds.utils.db.Table;
 import org.apache.hadoop.hdds.utils.db.TableIterator;
+import org.apache.hadoop.ozone.recon.ReconConstants;
 import org.apache.hadoop.ozone.recon.ReconUtils;
 import org.hadoop.ozone.recon.schema.UtilizationSchemaDefinition;
 import org.hadoop.ozone.recon.schema.tables.daos.FileCountBySizeDao;
@@ -44,25 +45,23 @@
 import java.util.Map;
 
 import static org.apache.hadoop.ozone.om.OmMetadataManagerImpl.FILE_TABLE;
-import static org.apache.hadoop.ozone.om.OmMetadataManagerImpl.KEY_TABLE;
 import static org.hadoop.ozone.recon.schema.tables.FileCountBySizeTable.FILE_COUNT_BY_SIZE;
 
 /**
  * Class to iterate over the OM DB and store the counts of existing/new
  * files binned into ranges (1KB, 2Kb..,4MB,.., 1TB,..1PB) to the Recon
  * fileSize DB.
  */
-public class FileSizeCountTask implements ReconOmTask {
+public class FileSizeCountTaskFSO implements ReconOmTask {
   private static final Logger LOG =
-      LoggerFactory.getLogger(FileSizeCountTask.class);
+      LoggerFactory.getLogger(FileSizeCountTaskFSO.class);
 
   private FileCountBySizeDao fileCountBySizeDao;
   private DSLContext dslContext;
 
   @Inject
-  public FileSizeCountTask(FileCountBySizeDao fileCountBySizeDao,
-                           UtilizationSchemaDefinition
-                               utilizationSchemaDefinition) {
+  public FileSizeCountTaskFSO(FileCountBySizeDao fileCountBySizeDao,
+                              UtilizationSchemaDefinition utilizationSchemaDefinition) {
     this.fileCountBySizeDao = fileCountBySizeDao;
     this.dslContext = utilizationSchemaDefinition.getDSLContext();
   }
@@ -76,27 +75,25 @@ public FileSizeCountTask(FileCountBySizeDao fileCountBySizeDao,
    */
   @Override
   public Pair<String, Boolean> reprocess(OMMetadataManager omMetadataManager) {
-    // Map to store the count of files based on file size
     Map<FileSizeCountKey, Long> fileSizeCountMap = new HashMap<>();
+    long startTime = System.currentTimeMillis(); // Start time for execution
 
-    // Delete all records from FILE_COUNT_BY_SIZE table
-    int execute = dslContext.delete(FILE_COUNT_BY_SIZE).execute();
-    LOG.debug("Deleted {} records from {}", execute, FILE_COUNT_BY_SIZE);
-
-    // Call reprocessBucket method for FILE_SYSTEM_OPTIMIZED bucket layout
-    boolean statusFSO =
-        reprocessBucketLayout(BucketLayout.FILE_SYSTEM_OPTIMIZED,
-            omMetadataManager,
-            fileSizeCountMap);
-    // Call reprocessBucket method for LEGACY bucket layout
-    boolean statusOBS =
-        reprocessBucketLayout(BucketLayout.LEGACY, omMetadataManager,
-            fileSizeCountMap);
-    if (!statusFSO && !statusOBS) {
+    // Use the shared atomic flag to ensure only one task truncates the table.
+    if (ReconConstants.TABLE_TRUNCATED.compareAndSet(false, true)) {
+      int execute = dslContext.delete(FILE_COUNT_BY_SIZE).execute();
+      LOG.info("Deleted {} records from {}", execute, FILE_COUNT_BY_SIZE);
+    } else {
+      LOG.info("Table already truncated by another task; skipping deletion.");
+    }
+
+    // Process only the FILE_SYSTEM_OPTIMIZED bucket layout using the FILE_TABLE.
+    boolean status = reprocessBucketLayout(BucketLayout.FILE_SYSTEM_OPTIMIZED, omMetadataManager, fileSizeCountMap);
+    if (!status) {
       return new ImmutablePair<>(getTaskName(), false);
     }
     writeCountsToDB(fileSizeCountMap);
-    LOG.debug("Completed a 'reprocess' run of FileSizeCountTask.");
+    long endTime = System.currentTimeMillis(); // End time for execution
+    LOG.info("FileSizeCountFSOTask completed Reprocess in {} ms.", (endTime - startTime));
     return new ImmutablePair<>(getTaskName(), true);
   }
 
@@ -105,11 +102,15 @@ private boolean reprocessBucketLayout(BucketLayout bucketLayout,
                                Map<FileSizeCountKey, Long> fileSizeCountMap) {
     Table<String, OmKeyInfo> omKeyInfoTable =
         omMetadataManager.getKeyTable(bucketLayout);
+    int totalKeysProcessed = 0;
+
     try (TableIterator<String, ? extends Table.KeyValue<String, OmKeyInfo>>
              keyIter = omKeyInfoTable.iterator()) {
       while (keyIter.hasNext()) {
         Table.KeyValue<String, OmKeyInfo> kv = keyIter.next();
         handlePutKeyEvent(kv.getValue(), fileSizeCountMap);
+        totalKeysProcessed++; // Increment key count
+
         //  The time complexity of .size() method is constant time, O(1)
         if (fileSizeCountMap.size() >= 100000) {
           writeCountsToDB(fileSizeCountMap);
@@ -121,17 +122,17 @@ private boolean reprocessBucketLayout(BucketLayout bucketLayout,
           " in Recon DB. ", ioEx);
       return false;
     }
+    LOG.info("Reprocessed {} keys for {} bucket layout.", totalKeysProcessed, bucketLayout);
     return true;
   }
 
   @Override
   public String getTaskName() {
-    return "FileSizeCountTask";
+    return "FileSizeCountTaskFSO";
   }
 
   public Collection<String> getTaskTables() {
     List<String> taskTables = new ArrayList<>();
-    taskTables.add(KEY_TABLE);
     taskTables.add(FILE_TABLE);
     return taskTables;
   }