HDDS-12062. Recon - Error handling in NSSummaryTask to avoid data inconsistencies.

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

@@ -510,7 +510,9 @@ boolean innerGetAndApplyDeltaUpdatesFromOM(long fromSequenceNumber,
       }
       for (byte[] data : dbUpdates.getData()) {
         try (ManagedWriteBatch writeBatch = new ManagedWriteBatch(data)) {
+          // Events gets populated in events list in OMDBUpdatesHandler with call back for put/delete/update
           writeBatch.iterate(omdbUpdatesHandler);
+          // Commit the OM DB transactions in recon rocks DB and sync here.
           try (RDBBatchOperation rdbBatchOperation =
                    new RDBBatchOperation(writeBatch)) {
             try (ManagedWriteOptions wOpts = new ManagedWriteOptions()) {

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

@@ -85,7 +85,7 @@ public ContainerKeyMapperTask(ReconContainerMetadataManager
    * (container, key) -> count to Recon Container DB.
    */
   @Override
-  public Pair<String, Boolean> reprocess(OMMetadataManager omMetadataManager) {
+  public Pair<String, Pair<Integer, Boolean>> reprocess(OMMetadataManager omMetadataManager) {
     long omKeyCount = 0;
 
     // In-memory maps for fast look up and batch write
@@ -121,7 +121,7 @@ public Pair<String, Boolean> reprocess(OMMetadataManager omMetadataManager) {
                 containerKeyCountMap);
             if (!checkAndCallFlushToDB(containerKeyMap)) {
               LOG.error("Unable to flush containerKey information to the DB");
-              return new ImmutablePair<>(getTaskName(), false);
+              return new ImmutablePair<>(getTaskName(), new ImmutablePair<>(0, false));
             }
             omKeyCount++;
           }
@@ -134,7 +134,7 @@ public Pair<String, Boolean> reprocess(OMMetadataManager omMetadataManager) {
           containerKeyCountMap)) {
         LOG.error("Unable to flush Container Key Count and " +
             "remaining Container Key information to the DB");
-        return new ImmutablePair<>(getTaskName(), false);
+        return new ImmutablePair<>(getTaskName(), new ImmutablePair<>(0, false));
       }
 
       LOG.debug("Completed 'reprocess' of ContainerKeyMapperTask.");
@@ -145,9 +145,9 @@ public Pair<String, Boolean> reprocess(OMMetadataManager omMetadataManager) {
     } catch (IOException ioEx) {
       LOG.error("Unable to populate Container Key data in Recon DB. ",
           ioEx);
-      return new ImmutablePair<>(getTaskName(), false);
+      return new ImmutablePair<>(getTaskName(), new ImmutablePair<>(0, false));
     }
-    return new ImmutablePair<>(getTaskName(), true);
+    return new ImmutablePair<>(getTaskName(), new ImmutablePair<>(0, true));
   }
 
   private boolean flushAndCommitContainerKeyInfoToDB(
@@ -192,7 +192,8 @@ public Collection<String> getTaskTables() {
   }
 
   @Override
-  public Pair<String, Boolean> process(OMUpdateEventBatch events) {
+  public Pair<String, Pair<Integer, Boolean>> process(OMUpdateEventBatch events,
+                                                      int seekPosition) {
     Iterator<OMDBUpdateEvent> eventIterator = events.getIterator();
     int eventCount = 0;
     final Collection<String> taskTables = getTaskTables();
@@ -249,18 +250,18 @@ public Pair<String, Boolean> process(OMUpdateEventBatch events) {
       } catch (IOException e) {
         LOG.error("Unexpected exception while updating key data : {} ",
             updatedKey, e);
-        return new ImmutablePair<>(getTaskName(), false);
+        return new ImmutablePair<>(getTaskName(), new ImmutablePair<>(0, false));
       }
     }
     try {
       writeToTheDB(containerKeyMap, containerKeyCountMap, deletedKeyCountList);
     } catch (IOException e) {
       LOG.error("Unable to write Container Key Prefix data in Recon DB.", e);
-      return new ImmutablePair<>(getTaskName(), false);
+      return new ImmutablePair<>(getTaskName(), new ImmutablePair<>(0, false));
     }
     LOG.debug("{} successfully processed {} OM DB update event(s) in {} milliseconds.",
         getTaskName(), eventCount, (System.currentTimeMillis() - startTime));
-    return new ImmutablePair<>(getTaskName(), true);
+    return new ImmutablePair<>(getTaskName(), new ImmutablePair<>(0, true));
   }
 
   private void writeToTheDB(Map<ContainerKeyPrefix, Integer> containerKeyMap,

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

@@ -75,7 +75,7 @@ public FileSizeCountTask(FileCountBySizeDao fileCountBySizeDao,
    * @return Pair
    */
   @Override
-  public Pair<String, Boolean> reprocess(OMMetadataManager omMetadataManager) {
+  public Pair<String, Pair<Integer, Boolean>> reprocess(OMMetadataManager omMetadataManager) {
     // Map to store the count of files based on file size
     Map<FileSizeCountKey, Long> fileSizeCountMap = new HashMap<>();
 
@@ -93,11 +93,11 @@ public Pair<String, Boolean> reprocess(OMMetadataManager omMetadataManager) {
         reprocessBucketLayout(BucketLayout.LEGACY, omMetadataManager,
             fileSizeCountMap);
     if (!statusFSO && !statusOBS) {
-      return new ImmutablePair<>(getTaskName(), false);
+      return new ImmutablePair<>(getTaskName(), new ImmutablePair<>(0, false));
     }
     writeCountsToDB(true, fileSizeCountMap);
     LOG.debug("Completed a 'reprocess' run of FileSizeCountTask.");
-    return new ImmutablePair<>(getTaskName(), true);
+    return new ImmutablePair<>(getTaskName(), new ImmutablePair<>(0, true));
   }
 
   private boolean reprocessBucketLayout(BucketLayout bucketLayout,
@@ -144,7 +144,8 @@ public Collection<String> getTaskTables() {
    * @return Pair
    */
   @Override
-  public Pair<String, Boolean> process(OMUpdateEventBatch events) {
+  public Pair<String, Pair<Integer, Boolean>> process(OMUpdateEventBatch events,
+                                                      int seekPosition) {
     Iterator<OMDBUpdateEvent> eventIterator = events.getIterator();
     Map<FileSizeCountKey, Long> fileSizeCountMap = new HashMap<>();
     final Collection<String> taskTables = getTaskTables();
@@ -191,7 +192,7 @@ public Pair<String, Boolean> process(OMUpdateEventBatch events) {
         } catch (Exception e) {
           LOG.error("Unexpected exception while processing key {}.",
               updatedKey, e);
-          return new ImmutablePair<>(getTaskName(), false);
+          return new ImmutablePair<>(getTaskName(), new ImmutablePair<>(0, false));
         }
       } else {
         LOG.warn("Unexpected value type {} for key {}. Skipping processing.",
@@ -201,7 +202,7 @@ public Pair<String, Boolean> process(OMUpdateEventBatch events) {
     writeCountsToDB(false, fileSizeCountMap);
     LOG.debug("{} successfully processed in {} milliseconds",
         getTaskName(), (System.currentTimeMillis() - startTime));
-    return new ImmutablePair<>(getTaskName(), true);
+    return new ImmutablePair<>(getTaskName(), new ImmutablePair<>(0, true));
   }
 
   /**

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

@@ -40,6 +40,9 @@
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 
+import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_NSSUMMARY_FLUSH_TO_DB_MAX_THRESHOLD;
+import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_NSSUMMARY_FLUSH_TO_DB_MAX_THRESHOLD_DEFAULT;
+
 /**
  * Task to query data from OMDB and write into Recon RocksDB.
  * Reprocess() will take a snapshots on OMDB, and iterate the keyTable,
@@ -70,7 +73,6 @@ public class NSSummaryTask implements ReconOmTask {
   private final NSSummaryTaskWithFSO nsSummaryTaskWithFSO;
   private final NSSummaryTaskWithLegacy nsSummaryTaskWithLegacy;
   private final NSSummaryTaskWithOBS nsSummaryTaskWithOBS;
-  private final OzoneConfiguration ozoneConfiguration;
 
   @Inject
   public NSSummaryTask(ReconNamespaceSummaryManager
@@ -81,16 +83,19 @@ public NSSummaryTask(ReconNamespaceSummaryManager
                        ozoneConfiguration) {
     this.reconNamespaceSummaryManager = reconNamespaceSummaryManager;
     this.reconOMMetadataManager = reconOMMetadataManager;
-    this.ozoneConfiguration = ozoneConfiguration;
+    long nsSummaryFlushToDBMaxThreshold = ozoneConfiguration.getLong(
+        OZONE_RECON_NSSUMMARY_FLUSH_TO_DB_MAX_THRESHOLD,
+        OZONE_RECON_NSSUMMARY_FLUSH_TO_DB_MAX_THRESHOLD_DEFAULT);
+
     this.nsSummaryTaskWithFSO = new NSSummaryTaskWithFSO(
-        reconNamespaceSummaryManager,
-        reconOMMetadataManager, ozoneConfiguration);
+        reconNamespaceSummaryManager, reconOMMetadataManager,
+        ozoneConfiguration, nsSummaryFlushToDBMaxThreshold);
     this.nsSummaryTaskWithLegacy = new NSSummaryTaskWithLegacy(
-        reconNamespaceSummaryManager,
-        reconOMMetadataManager, ozoneConfiguration);
+        reconNamespaceSummaryManager, reconOMMetadataManager,
+        ozoneConfiguration, nsSummaryFlushToDBMaxThreshold);
     this.nsSummaryTaskWithOBS = new NSSummaryTaskWithOBS(
-        reconNamespaceSummaryManager,
-        reconOMMetadataManager, ozoneConfiguration);
+        reconNamespaceSummaryManager, reconOMMetadataManager,
+        ozoneConfiguration, nsSummaryFlushToDBMaxThreshold);
   }
 
   @Override
@@ -99,27 +104,28 @@ public String getTaskName() {
   }
 
   @Override
-  public Pair<String, Boolean> process(OMUpdateEventBatch events) {
-    long startTime = System.currentTimeMillis();
-    boolean success = nsSummaryTaskWithFSO.processWithFSO(events);
-    if (!success) {
+  public Pair<String, Pair<Integer, Boolean>> process(
+      OMUpdateEventBatch events, int seekPosition) {
+    Pair<Integer, Boolean> itrPosStatusPair =
+        nsSummaryTaskWithFSO.processWithFSO(events, seekPosition);
+    if (!itrPosStatusPair.getRight()) {
       LOG.error("processWithFSO failed.");
     }
-    success = nsSummaryTaskWithLegacy.processWithLegacy(events);
-    if (!success) {
+    itrPosStatusPair =
+        nsSummaryTaskWithLegacy.processWithLegacy(events, seekPosition);
+    if (!itrPosStatusPair.getRight()) {
       LOG.error("processWithLegacy failed.");
     }
-    success = nsSummaryTaskWithOBS.processWithOBS(events);
-    if (!success) {
+    itrPosStatusPair =
+        nsSummaryTaskWithOBS.processWithOBS(events, seekPosition);
+    if (!itrPosStatusPair.getRight()) {
       LOG.error("processWithOBS failed.");
     }
-    LOG.debug("{} successfully processed in {} milliseconds",
-        getTaskName(), (System.currentTimeMillis() - startTime));
-    return new ImmutablePair<>(getTaskName(), success);
+    return new ImmutablePair<>(getTaskName(), itrPosStatusPair);
   }
 
   @Override
-  public Pair<String, Boolean> reprocess(OMMetadataManager omMetadataManager) {
+  public Pair<String, Pair<Integer, Boolean>> reprocess(OMMetadataManager omMetadataManager) {
     // Initialize a list of tasks to run in parallel
     Collection<Callable<Boolean>> tasks = new ArrayList<>();
 
@@ -131,7 +137,7 @@ public Pair<String, Boolean> reprocess(OMMetadataManager omMetadataManager) {
     } catch (IOException ioEx) {
       LOG.error("Unable to clear NSSummary table in Recon DB. ",
           ioEx);
-      return new ImmutablePair<>(getTaskName(), false);
+      return new ImmutablePair<>(getTaskName(), new ImmutablePair<>(0, false));
     }
 
     tasks.add(() -> nsSummaryTaskWithFSO
@@ -151,15 +157,12 @@ public Pair<String, Boolean> reprocess(OMMetadataManager omMetadataManager) {
       results = executorService.invokeAll(tasks);
       for (int i = 0; i < results.size(); i++) {
         if (results.get(i).get().equals(false)) {
-          return new ImmutablePair<>(getTaskName(), false);
+          return new ImmutablePair<>(getTaskName(), new ImmutablePair<>(0, false));
         }
       }
-    } catch (InterruptedException ex) {
+    } catch (InterruptedException | ExecutionException ex) {
       LOG.error("Error while reprocessing NSSummary table in Recon DB.", ex);
-      return new ImmutablePair<>(getTaskName(), false);
-    } catch (ExecutionException ex2) {
-      LOG.error("Error while reprocessing NSSummary table in Recon DB.", ex2);
-      return new ImmutablePair<>(getTaskName(), false);
+      return new ImmutablePair<>(getTaskName(), new ImmutablePair<>(0, false));
     } finally {
       executorService.shutdown();
 
@@ -172,7 +175,7 @@ public Pair<String, Boolean> reprocess(OMMetadataManager omMetadataManager) {
       LOG.debug("Task execution time: {} milliseconds", durationInMillis);
     }
 
-    return new ImmutablePair<>(getTaskName(), true);
+    return new ImmutablePair<>(getTaskName(), new ImmutablePair<>(0, true));
   }
 
 }

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

@@ -73,15 +73,16 @@ public ReconOMMetadataManager getReconOMMetadataManager() {
   protected void writeNSSummariesToDB(Map<Long, NSSummary> nsSummaryMap)
       throws IOException {
     try (RDBBatchOperation rdbBatchOperation = new RDBBatchOperation()) {
-      nsSummaryMap.keySet().forEach((Long key) -> {
+      for (Map.Entry<Long, NSSummary> entry : nsSummaryMap.entrySet()) {
         try {
           reconNamespaceSummaryManager.batchStoreNSSummaries(rdbBatchOperation,
-              key, nsSummaryMap.get(key));
+              entry.getKey(), entry.getValue());
         } catch (IOException e) {
           LOG.error("Unable to write Namespace Summary data in Recon DB.",
               e);
+          throw e;
         }
-      });
+      }
       reconNamespaceSummaryManager.commitBatchOperation(rdbBatchOperation);
     }
   }
@@ -210,21 +211,22 @@ protected void handleDeleteDirEvent(OmDirectoryInfo directoryInfo,
   protected boolean flushAndCommitNSToDB(Map<Long, NSSummary> nsSummaryMap) {
     try {
       writeNSSummariesToDB(nsSummaryMap);
-      nsSummaryMap.clear();
     } catch (IOException e) {
       LOG.error("Unable to write Namespace Summary data in Recon DB.", e);
       return false;
+    } finally {
+      nsSummaryMap.clear();
     }
     return true;
   }
 
-  protected boolean checkAndCallFlushToDB(
+/*  protected boolean checkAndCallFlushToDB(
       Map<Long, NSSummary> nsSummaryMap) {
     // if map contains more than entries, flush to DB and clear the map
     if (null != nsSummaryMap && nsSummaryMap.size() >=
         nsSummaryFlushToDBMaxThreshold) {
       return flushAndCommitNSToDB(nsSummaryMap);
     }
     return true;
-  }
+  }*/
 }