HDDS-16091. Recon fills disk with leaked checkpoints and crashes on startup when local OM DB is missing.

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

@@ -294,7 +294,16 @@ public void start() {
                   deltaTaskStatusUpdater.getLastUpdatedSeqNumber()) < 0; // Condition 3
         })
         .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));  // Collect into desired Map
-    if (!reconOmTaskMap.isEmpty()) {
+    if (!reconOmTaskMap.isEmpty() && omMetadataManager.getStore() == null) {
+      // Fresh start (or the local OM snapshot DB is missing) while stale task
+      // status rows still exist in the Recon SQL DB. There is no local OM DB to
+      // checkpoint/reprocess yet, so attempting reinitialization here would fail
+      // (checkpoint creation dereferences a null DB store). Skip it; the full
+      // snapshot sync scheduled below will download the OM DB and initialize tasks.
+      LOG.info("Skipping startup task reinitialization because the local OM DB store " +
+          "is not initialized yet (no OM snapshot present). The scheduled full snapshot " +
+          "sync will download the OM DB and initialize tasks.");
+    } else if (!reconOmTaskMap.isEmpty()) {
       LOG.info("Task name and last updated sequence number of tasks, that are not matching with " +
           "the last updated sequence number of OmDeltaRequest task:\n");
       LOG.info("{} -> {}", deltaTaskStatusUpdater.getTaskName(), deltaTaskStatusUpdater.getLastUpdatedSeqNumber());

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

@@ -106,6 +106,9 @@ public class ReconTaskControllerImpl implements ReconTaskController {
   // Clock for the retry-delay gate; overridable in tests via the
   // @VisibleForTesting constructor to drive the gate with a MockClock.
   private Clock clock = Clock.systemUTC();
+  // Log the 1st cleanup and every Nth after that at INFO; the rest at DEBUG.
+  private static final int CHECKPOINT_CLEANUP_LOG_SAMPLE_RATE = 20;
+  private final AtomicLong checkpointCleanupCount = new AtomicLong(0);
 
   @Inject
   @SuppressWarnings("checkstyle:ParameterNumber")
@@ -633,30 +636,45 @@ public synchronized ReconTaskController.ReInitializationResult queueReInitializa
 
     // Try checkpoint creation (single attempt per iteration)
     ReconOMMetadataManager checkpointedOMMetadataManager = null;
-
+    // Whether the checkpoint has been handed off to the event buffer. If not,
+    // this method owns its cleanup (the finally block below).
+    boolean handedOff = false;
+
     try {
       LOG.info("Attempting checkpoint creation (retry attempt: {})", eventProcessRetryCount.get() + 1);
-      checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
-      LOG.info("Checkpoint creation succeeded");
-    } catch (IOException e) {
-      LOG.error("Checkpoint creation failed: {}", e.getMessage());
+      try {
+        checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+        LOG.info("Checkpoint creation succeeded");
+      } catch (IOException e) {
+        LOG.error("Checkpoint creation failed: {}", e.getMessage());
+        handleEventFailure();
+        return ReInitializationResult.RETRY_LATER;
+      }
+
+      // Create and queue the reinitialization event with checkpointed metadata manager
+      ReconTaskReInitializationEvent reinitEvent =
+          new ReconTaskReInitializationEvent(reason, checkpointedOMMetadataManager);
+      // If reinitialization event queued successfully, reset event buffer overflow flag and task failure flag,
+      // so that we can resume queuing the delta events.
+      if (eventBuffer.offer(reinitEvent)) {
+        // The downstream consumer now owns the checkpoint and its cleanup.
+        handedOff = true;
+        resetEventFlags();
+        LOG.info("Successfully queued reinitialization event after {} retries", eventProcessRetryCount.get() + 1);
+        return ReconTaskController.ReInitializationResult.SUCCESS;
+      }
+
+      // Buffer full - drop the event and clean up the fresh checkpoint (in finally) to avoid leaking it.
+      LOG.warn("Failed to queue reinitialization event (buffer full); discarding fresh checkpoint at {}",
+          checkpointedOMMetadataManager.getStore() != null
+              ? checkpointedOMMetadataManager.getStore().getDbLocation() : "<unknown>");
       handleEventFailure();
       return ReInitializationResult.RETRY_LATER;
+    } finally {
+      if (!handedOff && checkpointedOMMetadataManager != null) {
+        cleanupCheckpoint(checkpointedOMMetadataManager);
+      }
     }
-
-    // Create and queue the reinitialization event with checkpointed metadata manager
-    ReconTaskReInitializationEvent reinitEvent =
-        new ReconTaskReInitializationEvent(reason, checkpointedOMMetadataManager);
-    boolean queued = eventBuffer.offer(reinitEvent);
-    // If reinitialization event queued successfully, reset event buffer overflow flag and task failure flag,
-    // so that we can resume queuing the delta events.
-    if (queued) {
-      resetEventFlags();
-      // Success - reset retry counters and flags
-      LOG.info("Successfully queued reinitialization event after {} retries", eventProcessRetryCount.get() + 1);
-      return ReconTaskController.ReInitializationResult.SUCCESS;
-    }
-    return null;
   }
 
   private ReconTaskController.ReInitializationResult validateRetryCountAndDelay() {
@@ -715,15 +733,7 @@ public void drainEventBufferAndCleanExistingCheckpoints() {
           ReconOMMetadataManager checkpointedManager = reinitEvent.getCheckpointedOMMetadataManager();
           if (checkpointedManager != null) {
             LOG.info("Cleaning up unprocessed checkpoint from drained ReconTaskReInitializationEvent");
-            // Close the database connections first
-            try {
-              checkpointedManager.close();
-              LOG.debug("Closed checkpointed OM metadata manager database connections");
-            } catch (Exception e) {
-              LOG.warn("Failed to close checkpointed OM metadata manager", e);
-            }
-            // Then clean up the files
-            cleanupCheckpointFiles(checkpointedManager);
+            cleanupCheckpoint(checkpointedManager);
           }
         }
       }
@@ -770,6 +780,10 @@ public ReconOMMetadataManager createOMCheckpoint(ReconOMMetadataManager omMetaMa
    * @throws IOException if directory operations fail
    */
   private String cleanTempCheckPointPath(ReconOMMetadataManager omMetaManager) throws IOException {
+    if (omMetaManager == null || omMetaManager.getStore() == null) {
+      throw new IOException("OM DB store is not initialized yet; cannot create "
+          + "reinitialization checkpoint. A full OM snapshot must be fetched first.");
+    }
     File dbLocation = omMetaManager.getStore().getDbLocation();
     if (dbLocation == null) {
       throw new IOException("OM DB location is null");
@@ -793,9 +807,9 @@ private void processReInitializationEvent(ReconTaskReInitializationEvent event)
         event.getReason(), event.getTimestamp());
     resetTasksFailureFlag();
     // Use the checkpointed OM metadata manager for reinitialization to prevent data inconsistency
-    ReconOMMetadataManager checkpointedOMMetadataManager = null;
-    try (ReconOMMetadataManager manager = event.getCheckpointedOMMetadataManager()) {
-      checkpointedOMMetadataManager = manager;
+    ReconOMMetadataManager checkpointedOMMetadataManager =
+        event.getCheckpointedOMMetadataManager();
+    try {
       if (checkpointedOMMetadataManager != null) {
         LOG.info("Starting async task reinitialization with checkpointed OM metadata manager due to: {}",
                  event.getReason());
@@ -817,9 +831,8 @@ private void processReInitializationEvent(ReconTaskReInitializationEvent event)
     } catch (Exception e) {
       LOG.error("Error processing reinitialization event", e);
     } finally {
-      if (checkpointedOMMetadataManager != null) {
-        cleanupCheckpointFiles(checkpointedOMMetadataManager);
-      }
+      // Clean up the checkpointed metadata manager and its files after use
+      cleanupCheckpoint(checkpointedOMMetadataManager);
     }
   }
 
@@ -877,11 +890,14 @@ AtomicBoolean getTasksFailedFlag() {
    */
   private void cleanupPreExistingCheckpoints() {
     try {
+      // The DB store is only initialized after Recon downloads its first DB
+      // snapshot from the OM. On a fresh startup it may still be null.
       if (currentOMMetadataManager == null || currentOMMetadataManager.getStore() == null) {
-        LOG.debug("No current OM metadata manager or store, skipping pre-existing checkpoint cleanup");
+        LOG.debug("No current OM metadata manager or DB store not yet initialized, "
+            + "skipping pre-existing checkpoint cleanup");
         return;
       }
-      
+
       // Get the base directory where checkpoints are created
       File dbLocation = currentOMMetadataManager.getStore().getDbLocation();
       if (dbLocation == null || dbLocation.getParent() == null) {
@@ -924,41 +940,50 @@ private void cleanupPreExistingCheckpoints() {
   }
 
   /**
-   * Cleanup checkpoint files for a checkpointed OM metadata manager.
-   * This method only removes the temporary checkpoint files without closing database connections.
-   * Used when the manager is closed via try-with-resources.
-   * 
-   * @param checkpointedManager the checkpointed OM metadata manager
+   * Cleanup checkpointed OM metadata manager and associated checkpoint files.
+   * This method closes the database connections and removes the temporary checkpoint files.
+   *
+   * @param checkpointedManager the checkpointed OM metadata manager to clean up
    */
-  private void cleanupCheckpointFiles(ReconOMMetadataManager checkpointedManager) {
+  private void cleanupCheckpoint(ReconOMMetadataManager checkpointedManager) {
     if (checkpointedManager == null) {
       return;
     }
+    // Get the checkpoint location before closing.
+    File checkpointLocation = null;
     try {
-      // Get the checkpoint location
-      File checkpointLocation = null;
-      try {
-        if (checkpointedManager.getStore() != null && 
-            checkpointedManager.getStore().getDbLocation() != null) {
-          // The checkpoint location is typically the parent directory of the DB location
-          checkpointLocation = checkpointedManager.getStore().getDbLocation().getParentFile();
-        }
-      } catch (Exception e) {
-        LOG.warn("Failed to get checkpoint location for cleanup", e);
+      if (checkpointedManager.getStore() != null &&
+          checkpointedManager.getStore().getDbLocation() != null) {
+        // The checkpoint location is typically the parent directory of the DB location
+        checkpointLocation = checkpointedManager.getStore().getDbLocation().getParentFile();
       }
-
-      // Clean up the checkpoint files if we have the location
+    } catch (Exception e) {
+      LOG.warn("Failed to get checkpoint location for cleanup", e);
+    }
+
+    // Close the database connections first, but always attempt to delete the
+    // checkpoint files afterwards - even if stop() throws - so the directory
+    // (a full copy of the OM DB) is never leaked.
+    try {
+      checkpointedManager.stop();
+      LOG.debug("Closed checkpointed OM metadata manager database connections");
+    } catch (Exception e) {
+      LOG.warn("Failed to stop checkpointed OM metadata manager", e);
+    } finally {
       if (checkpointLocation != null && checkpointLocation.exists()) {
         try {
           FileUtils.deleteDirectory(checkpointLocation);
-          LOG.debug("Cleaned up checkpoint directory: {}", checkpointLocation);
+          long cleaned = checkpointCleanupCount.incrementAndGet();
+          if (cleaned % CHECKPOINT_CLEANUP_LOG_SAMPLE_RATE == 1) {
+            LOG.info("Cleaned up checkpoint directory: {} (total cleaned so far: {})",
+                checkpointLocation, cleaned);
+          } else {
+            LOG.debug("Cleaned up checkpoint directory: {}", checkpointLocation);
+          }
         } catch (IOException e) {
           LOG.warn("Failed to cleanup checkpoint directory: {}", checkpointLocation, e);
         }
       }
-
-    } catch (Exception e) {
-      LOG.warn("Failed to cleanup checkpoint files", e);
     }
   }