diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java index d221168aa003..7453e7bcc63b 100644 --- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java +++ b/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()); diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java index 5b142c0189dc..6f142b57a692 100644 --- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java +++ b/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() : ""); 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); } } diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestReconTaskControllerImpl.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestReconTaskControllerImpl.java index e9da0f76a0ec..b5f742116d9d 100644 --- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestReconTaskControllerImpl.java +++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestReconTaskControllerImpl.java @@ -17,6 +17,8 @@ package org.apache.hadoop.ozone.recon.tasks; +import static org.apache.hadoop.ozone.recon.OMMetadataManagerTestUtils.getTestReconOmMetadataManager; +import static org.apache.hadoop.ozone.recon.OMMetadataManagerTestUtils.initializeNewOmMetadataManager; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -26,6 +28,8 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyMap; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; @@ -63,6 +67,7 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; /** * Class used to test ReconTaskControllerImpl. @@ -506,27 +511,124 @@ public void testUpdateOMMetadataManager() throws Exception { @Test public void testCheckpointManagerCleanupOnQueueFailure() throws Exception { - // Set up properly mocked ReconOMMetadataManager with required dependencies - ReconOMMetadataManager mockOMMetadataManager = mock(ReconOMMetadataManager.class); + // Verify the buffer-full branch of queueReInitializationEvent: when the freshly + // created checkpoint cannot be handed off to the event buffer, it must be cleaned + // up (not leaked) and the call must return RETRY_LATER. + + // Stop the async processor from setUp so it can't consume buffered events. + reconTaskController.stop(); + + // Build a controller with a capacity-1 event buffer so a single pre-filled event + // makes the internal eventBuffer.offer(...) return false (buffer full). + OzoneConfiguration ozoneConfiguration = new OzoneConfiguration(); + ozoneConfiguration.setInt("ozone.recon.om.event.buffer.capacity", 1); + ReconTaskStatusUpdaterManager reconTaskStatusUpdaterManagerMock = mock(ReconTaskStatusUpdaterManager.class); + when(reconTaskStatusUpdaterManagerMock.getTaskStatusUpdater(anyString())) + .thenAnswer(i -> { + String taskName = i.getArgument(0); + return new ReconTaskStatusUpdater(reconTaskStatusDao, taskName); + }); + ReconDBProvider reconDbProvider = mock(ReconDBProvider.class); + when(reconDbProvider.getDbStore()).thenReturn(mock(DBStore.class)); + when(reconDbProvider.getStagedReconDBProvider()).thenReturn(reconDbProvider); + ReconTaskControllerImpl controller = new ReconTaskControllerImpl(ozoneConfiguration, new HashSet<>(), + reconTaskStatusUpdaterManagerMock, reconDbProvider, mock(ReconContainerMetadataManager.class), + mock(ReconNamespaceSummaryManager.class), mock(ReconGlobalStatsManager.class), + mock(ReconFileMetadataManager.class)); + // Do not start async processing. + + // Checkpointed manager whose cleanup (stop()) we verify. + ReconOMMetadataManager mockCheckpointedManager = mock(ReconOMMetadataManager.class); DBStore mockDBStore = mock(DBStore.class); File mockDbLocation = mock(File.class); - DBCheckpoint mockCheckpoint = mock(DBCheckpoint.class); - Path mockCheckpointPath = Paths.get("/tmp/test/checkpoint"); - - when(mockOMMetadataManager.getStore()).thenReturn(mockDBStore); + when(mockCheckpointedManager.getStore()).thenReturn(mockDBStore); when(mockDBStore.getDbLocation()).thenReturn(mockDbLocation); - when(mockDbLocation.getParent()).thenReturn("/tmp/test"); - when(mockDBStore.getCheckpoint(any(String.class), any(Boolean.class))).thenReturn(mockCheckpoint); - when(mockCheckpoint.getCheckpointLocation()).thenReturn(mockCheckpointPath); - - reconTaskController.updateOMMetadataManager(mockOMMetadataManager); - reconTaskController.stop(); - - // This test verifies the successful path - in practice, queue failure after clear is very rare - // since we clear the buffer before queueing the reinitialization event - ReconTaskController.ReInitializationResult result = reconTaskController.queueReInitializationEvent( + when(mockDbLocation.getParentFile()).thenReturn(mockDbLocation); + + ReconTaskControllerImpl controllerSpy = spy(controller); + // Keep the buffer full through the drain step, and return our checkpoint without + // touching RocksDB. + doNothing().when(controllerSpy).drainEventBufferAndCleanExistingCheckpoints(); + doReturn(mockCheckpointedManager).when(controllerSpy).createOMCheckpoint(any()); + controllerSpy.updateOMMetadataManager(mock(ReconOMMetadataManager.class)); + + // Fill the capacity-1 buffer so the real offer(...) inside the method fails. + OMUpdateEventBatch fillerBatch = mock(OMUpdateEventBatch.class); + when(fillerBatch.getEventType()).thenReturn(ReconEvent.EventType.OM_UPDATE_BATCH); + when(fillerBatch.getEventCount()).thenReturn(1); + assertTrue(controllerSpy.getEventBuffer().offer(fillerBatch), "precondition: filler event queued"); + + ReconTaskController.ReInitializationResult result = controllerSpy.queueReInitializationEvent( ReconTaskReInitializationEvent.ReInitializationReason.BUFFER_OVERFLOW); - assertEquals(ReconTaskController.ReInitializationResult.SUCCESS, result, "Should succeed under normal conditions"); + + assertEquals(ReconTaskController.ReInitializationResult.RETRY_LATER, result, + "Buffer-full offer should result in RETRY_LATER"); + // The fresh checkpoint was never handed off, so it must be cleaned up. + verify(mockCheckpointedManager, times(1)).stop(); + } + + @Test + public void testCleanupCheckpointDeletesDirEvenWhenStopThrows(@TempDir File tempDir) throws Exception { + // Verify the cleanupCheckpoint try/finally: even if stop() throws, the checkpoint + // directory (a full copy of the OM DB) is still deleted and not leaked. + + // Halt the async processor so the buffered event isn't consumed before we drain. + reconTaskController.stop(); + ReconTaskControllerImpl controllerImpl = (ReconTaskControllerImpl) reconTaskController; + + // Real on-disk checkpoint directory (with content) that cleanup must delete. + File checkpointDir = new File(tempDir, "temp-recon-reinit-checkpoint_test"); + assertTrue(checkpointDir.mkdirs(), "precondition: checkpoint dir created"); + assertTrue(new File(checkpointDir, "CURRENT").createNewFile(), "precondition: dir has content"); + + ReconOMMetadataManager mockCheckpointedManager = mock(ReconOMMetadataManager.class); + DBStore mockDBStore = mock(DBStore.class); + when(mockCheckpointedManager.getStore()).thenReturn(mockDBStore); + // getDbLocation().getParentFile() resolves to the real checkpointDir. + when(mockDBStore.getDbLocation()).thenReturn(new File(checkpointDir, "om.db")); + // stop() throws - cleanup must still delete the directory. + doThrow(new IOException("stop failed")).when(mockCheckpointedManager).stop(); + + controllerImpl.getEventBuffer().offer(new ReconTaskReInitializationEvent( + ReconTaskReInitializationEvent.ReInitializationReason.BUFFER_OVERFLOW, mockCheckpointedManager)); + + controllerImpl.drainEventBufferAndCleanExistingCheckpoints(); + + verify(mockCheckpointedManager, times(1)).stop(); + assertFalse(checkpointDir.exists(), + "checkpoint directory must be deleted even when stop() throws"); + } + + @Test + public void testRealCheckpointCreatedThenCleanedUp( + @TempDir File dirOmMetadata, @TempDir File dirReconMetadata) throws Exception { + // End-to-end with real RocksDB (no mocked manager): create a real reinit + // checkpoint, then verify the real cleanup path deletes it from disk. + + // Halt the async processor so our explicit drain owns the enqueued event. + reconTaskController.stop(); + ReconTaskControllerImpl controllerImpl = (ReconTaskControllerImpl) reconTaskController; + + // Real source OM DB + real RocksDB-backed Recon OM metadata manager. + OMMetadataManager omMetadataManager = initializeNewOmMetadataManager(dirOmMetadata); + ReconOMMetadataManager reconOMMetadataManager = + getTestReconOmMetadataManager(omMetadataManager, dirReconMetadata); + controllerImpl.updateOMMetadataManager(reconOMMetadataManager); + + // Real checkpoint creation -> temp-recon-reinit-checkpoint_/ on disk. + ReconOMMetadataManager checkpointed = controllerImpl.createOMCheckpoint(reconOMMetadataManager); + File checkpointDir = checkpointed.getStore().getDbLocation().getParentFile(); + assertTrue(checkpointDir.exists(), "checkpoint dir should exist after creation"); + assertTrue(checkpointDir.getName().startsWith("temp-recon-reinit-checkpoint"), + "checkpoint dir should be the reinit temp dir"); + + // Real cleanup path: closes the real RocksDB and deletes the directory. + controllerImpl.getEventBuffer().offer(new ReconTaskReInitializationEvent( + ReconTaskReInitializationEvent.ReInitializationReason.BUFFER_OVERFLOW, checkpointed)); + controllerImpl.drainEventBufferAndCleanExistingCheckpoints(); + + assertFalse(checkpointDir.exists(), + "real checkpoint directory must be deleted by the cleanup path"); } @Test @@ -811,7 +913,7 @@ public void testProcessReInitializationEventWithCheckpointedManager() throws Exc assertFalse(controllerSpy.hasTasksFailed(), "tasksFailed should remain false after successful reinitialization"); // Verify cleanup was called on the checkpointed manager - verify(mockCheckpointedManager, times(1)).close(); + verify(mockCheckpointedManager, times(1)).stop(); } @Test