You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@ozone.apache.org by "sumitagrawl (via GitHub)" <gi...@apache.org> on 2025/09/02 06:59:14 UTC
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
sumitagrawl commented on code in PR #8962:
URL: https://github.com/apache/ozone/pull/8962#discussion_r2300415543
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -431,4 +457,156 @@ public boolean hasDeltaTasksFailed() {
public void resetDeltaTasksFailureFlag() {
deltaTasksFailed.set(false);
}
+
+ @Override
+ public boolean queueReInitializationEvent(ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {}", reason);
+
+ try {
+ // Create checkpoint of current OM metadata manager before queueing reinitialization event
+ // This prevents data inconsistency as parallel DB updates will continue on the original
+ // omMetadataManager reference while reinitialization uses the checkpointed snapshot
+ ReconOMMetadataManager checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
Review Comment:
we can close in finally block
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -431,4 +457,156 @@ public boolean hasDeltaTasksFailed() {
public void resetDeltaTasksFailureFlag() {
deltaTasksFailed.set(false);
}
+
+ @Override
+ public boolean queueReInitializationEvent(ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {}", reason);
+
+ try {
+ // Create checkpoint of current OM metadata manager before queueing reinitialization event
+ // This prevents data inconsistency as parallel DB updates will continue on the original
+ // omMetadataManager reference while reinitialization uses the checkpointed snapshot
+ ReconOMMetadataManager checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+
+ resetEventBuffer();
+
+ // Create and queue the reinitialization event with checkpointed metadata manager
+ ReconTaskReInitializationEvent reinitEvent =
+ new ReconTaskReInitializationEvent(reason, checkpointedOMMetadataManager);
+ boolean queued = eventBuffer.offer(reinitEvent);
+ if (!queued) {
+ // CAUTION: If this failure happens, something is seriously wrong as we just cleared the buffer
+ // and are trying to queue a single event, also since we cleared all other delta events, next delta sync cycle
+ // will not have these lost events as new OMDBUpdatesHandler will be created with last sequence number
+ LOG.error(
+ "Failed to queue reinitialization event. Ideally should not happen, so dropping event and " +
+ "checkpointed data.");
+ // Close the checkpointed metadata manager as it won't be used
+ try {
+ checkpointedOMMetadataManager.stop();
+ } catch (Exception e) {
+ LOG.warn("Failed to close checkpointed OM metadata manager", e);
+ }
+ return false;
+ }
+ resetEventBufferOverflowFlag();
+ resetDeltaTasksFailureFlag();
+ return true;
+ } catch (IOException e) {
+ LOG.error("Failed to create checkpoint for reinitialization", e);
+ return false;
+ }
+ }
+
+ @Override
+ public void resetEventBuffer() {
+ // Clear the buffer to discard all pending events
+ eventBuffer.clear();
+ }
+
+ @Override
+ public void updateOMMetadataManager(ReconOMMetadataManager omMetadataManager) {
+ this.currentOMMetadataManager = omMetadataManager;
+ }
+
+ @Override
+ public ReconOMMetadataManager createOMCheckpoint(ReconOMMetadataManager omMetaManager)
+ throws IOException {
+ // Create temporary directory for checkpoint
+ String parentPath = cleanTempCheckPointPath(omMetaManager);
+
+ // Create checkpoint
+ DBCheckpoint checkpoint = omMetaManager.getStore().getCheckpoint(parentPath, true);
+
+ // Create a new ReconOmMetadataManagerImpl instance with checkpointed data
+ // Note: We need ReconUtils but it's not available in this class
+ // We'll create the ReconOmMetadataManagerImpl with minimal dependencies
+ ReconOmMetadataManagerImpl checkpointedManager = new ReconOmMetadataManagerImpl(configuration, null);
+
+ // Initialize the checkpointed manager with the checkpoint location
+ File checkpointDir = checkpoint.getCheckpointLocation().toFile();
+ checkpointedManager.updateOmDB(checkpointDir);
+
+ return checkpointedManager;
+ }
+
+ /**
+ * Clean and prepare temporary checkpoint path.
+ * Similar to QuotaRepairTask.cleanTempCheckPointPath.
+ *
+ * @param omMetaManager the OM metadata manager
+ * @return path to temporary checkpoint directory
+ * @throws IOException if directory operations fail
+ */
+ private String cleanTempCheckPointPath(ReconOMMetadataManager omMetaManager) throws IOException {
+ File dbLocation = omMetaManager.getStore().getDbLocation();
+ if (dbLocation == null) {
+ throw new NullPointerException("OM DB location is null");
+ }
+ String tempData = dbLocation.getParent();
+ if (tempData == null) {
+ throw new NullPointerException("Parent OM DB dir is null");
+ }
+ File reinitTmpPath = Paths.get(tempData, "temp-recon-reinit-checkpoint").toFile();
+ FileUtils.deleteDirectory(reinitTmpPath);
+ FileUtils.forceMkdir(reinitTmpPath);
+ return reinitTmpPath.toString();
+ }
+
+ /**
+ * Process a task reinitialization event asynchronously.
+ */
+ private void processReInitializationEvent(ReconTaskReInitializationEvent event) {
+ LOG.info("Processing reinitialization event: reason={}, timestamp={}",
+ event.getReason(), event.getTimestamp());
+
+ try {
+ // Use the checkpointed OM metadata manager for reinitialization to prevent data inconsistency
+ ReconOMMetadataManager checkpointedOMMetadataManager = event.getCheckpointedOMMetadataManager();
+ if (checkpointedOMMetadataManager != null) {
+ LOG.info("Starting async task reinitialization with checkpointed OM metadata manager due to: {}",
+ event.getReason());
+ reInitializeTasks(checkpointedOMMetadataManager, null);
+ LOG.info("Completed async task reinitialization");
+
+ // Close the checkpointed metadata manager after use
+ try {
+ checkpointedOMMetadataManager.stop();
Review Comment:
also delete the checkpoint directory in finally
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -431,4 +468,359 @@ public boolean hasDeltaTasksFailed() {
public void resetDeltaTasksFailureFlag() {
deltaTasksFailed.set(false);
}
+
+ @Override
+ public ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (iteration retry count: {})", reason,
+ iterationRetryCount);
+
+ // Check if we should retry based on timing for iteration-based retries
+ long currentTime = System.currentTimeMillis();
+ if (iterationRetryCount > 0) {
+ // Check if 2 seconds have passed since last iteration
+ long timeSinceLastRetry = currentTime - lastCheckpointRetryTimestamp;
+ if (timeSinceLastRetry < RETRY_DELAY_MS) {
+ LOG.debug("Skipping checkpoint creation retry, only {}ms since last iteration (need {}ms)",
+ timeSinceLastRetry, RETRY_DELAY_MS);
+ return ReconTaskController.ReInitializationResult.RETRY_LATER;
+ }
+ LOG.info("Attempting retry for checkpoint creation (iteration: {}, delay: {}ms)",
+ iterationRetryCount + 1, timeSinceLastRetry);
+ }
+
+ // Try checkpoint creation with immediate retry capability
+ ReconOMMetadataManager checkpointedOMMetadataManager = null;
+ boolean checkpointCreated = false;
+
+ // First attempt
+ try {
+ LOG.info("Attempting checkpoint creation (iteration: {}, attempt: 1)", iterationRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ checkpointCreated = true;
+ LOG.info("Checkpoint creation succeeded on first attempt");
+ } catch (IOException e) {
+ LOG.warn("First checkpoint creation attempt failed: {}", e.getMessage());
+
+ // Immediate retry within the same iteration
+ try {
+ LOG.info("Attempting immediate retry for checkpoint creation (iteration: {}, attempt: 2)",
+ iterationRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ checkpointCreated = true;
+ LOG.info("Checkpoint creation succeeded on immediate retry");
+ } catch (IOException retryException) {
+ LOG.error("Immediate retry for checkpoint creation also failed: {}", retryException.getMessage());
+ // Both attempts in this iteration failed
+ handleIterationFailure(currentTime);
+ return getIterationRetryResult();
+ }
+ }
+ // Proceed with queueing the event
+ resetEventBuffer();
+
+ // Create and queue the reinitialization event with checkpointed metadata manager
+ ReconTaskReInitializationEvent reinitEvent =
+ new ReconTaskReInitializationEvent(reason, checkpointedOMMetadataManager);
+ boolean queued = eventBuffer.offer(reinitEvent);
+ if (!queued) {
+ // CAUTION: If this failure happens, something is seriously wrong as we just cleared the buffer
+ // and are trying to queue a single event, also since we cleared all other delta events, next delta sync cycle
+ // will not have these lost events as new OMDBUpdatesHandler will be created with last sequence number
+ LOG.error("Failed to queue reinitialization event. Ideally should not happen, so dropping event and " +
+ "checkpointed data.");
+ // Clean up the checkpointed metadata manager and its files as it won't be used
+ cleanupCheckpoint(checkpointedOMMetadataManager);
+ handleIterationFailure(currentTime);
+ return getIterationRetryResult();
+ }
+
+ // Success - reset retry counters and flags
+ LOG.info("Successfully queued reinitialization event after {} iteration(s)", iterationRetryCount + 1);
+ iterationRetryCount = 0;
+ lastCheckpointRetryTimestamp = 0;
+ resetEventBufferOverflowFlag();
+ resetDeltaTasksFailureFlag();
+ return ReconTaskController.ReInitializationResult.SUCCESS;
+ }
+
+ /**
+ * Handle iteration failure by updating retry counters.
+ */
+ private void handleIterationFailure(long currentTime) {
Review Comment:
this should be done when continuous reprocess is failing also. not only just during add to queue task.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java:
##########
@@ -776,13 +796,22 @@ private void executeFullSnapshot(ReconTaskStatusUpdater fullSnapshotReconTaskUpd
fullSnapshotReconTaskUpdater.recordRunCompletion();
deltaReconTaskStatusUpdater.updateDetails();
+ // Update the current OM metadata manager in task controller
+ reconTaskController.updateOMMetadataManager(omMetadataManager);
+
// Reinitialize tasks that are listening.
- LOG.info("Calling reprocess on Recon tasks.");
- reconTaskController.reInitializeTasks(omMetadataManager, null);
+ LOG.info("Queueing async reinitialization event instead of blocking call");
+ 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 snapshot operation",
+ result);
+ metrics.incrNumSnapshotRequestsFailed();
+ fullSnapshotReconTaskUpdater.setLastTaskRunStatus(-1);
Review Comment:
recordCompletion is missing
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -431,4 +457,156 @@ public boolean hasDeltaTasksFailed() {
public void resetDeltaTasksFailureFlag() {
deltaTasksFailed.set(false);
}
+
+ @Override
+ public boolean queueReInitializationEvent(ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {}", reason);
+
+ try {
+ // Create checkpoint of current OM metadata manager before queueing reinitialization event
+ // This prevents data inconsistency as parallel DB updates will continue on the original
+ // omMetadataManager reference while reinitialization uses the checkpointed snapshot
+ ReconOMMetadataManager checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+
+ resetEventBuffer();
+
+ // Create and queue the reinitialization event with checkpointed metadata manager
+ ReconTaskReInitializationEvent reinitEvent =
+ new ReconTaskReInitializationEvent(reason, checkpointedOMMetadataManager);
+ boolean queued = eventBuffer.offer(reinitEvent);
+ if (!queued) {
+ // CAUTION: If this failure happens, something is seriously wrong as we just cleared the buffer
+ // and are trying to queue a single event, also since we cleared all other delta events, next delta sync cycle
+ // will not have these lost events as new OMDBUpdatesHandler will be created with last sequence number
+ LOG.error(
+ "Failed to queue reinitialization event. Ideally should not happen, so dropping event and " +
+ "checkpointed data.");
+ // Close the checkpointed metadata manager as it won't be used
+ try {
+ checkpointedOMMetadataManager.stop();
+ } catch (Exception e) {
+ LOG.warn("Failed to close checkpointed OM metadata manager", e);
+ }
+ return false;
+ }
+ resetEventBufferOverflowFlag();
+ resetDeltaTasksFailureFlag();
+ return true;
+ } catch (IOException e) {
+ LOG.error("Failed to create checkpoint for reinitialization", e);
Review Comment:
can set task failure here, so that it can be picked up later.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -431,4 +468,359 @@ public boolean hasDeltaTasksFailed() {
public void resetDeltaTasksFailureFlag() {
deltaTasksFailed.set(false);
}
+
+ @Override
+ public ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (iteration retry count: {})", reason,
+ iterationRetryCount);
+
+ // Check if we should retry based on timing for iteration-based retries
+ long currentTime = System.currentTimeMillis();
+ if (iterationRetryCount > 0) {
+ // Check if 2 seconds have passed since last iteration
+ long timeSinceLastRetry = currentTime - lastCheckpointRetryTimestamp;
+ if (timeSinceLastRetry < RETRY_DELAY_MS) {
+ LOG.debug("Skipping checkpoint creation retry, only {}ms since last iteration (need {}ms)",
+ timeSinceLastRetry, RETRY_DELAY_MS);
+ return ReconTaskController.ReInitializationResult.RETRY_LATER;
+ }
+ LOG.info("Attempting retry for checkpoint creation (iteration: {}, delay: {}ms)",
+ iterationRetryCount + 1, timeSinceLastRetry);
+ }
+
+ // Try checkpoint creation with immediate retry capability
+ ReconOMMetadataManager checkpointedOMMetadataManager = null;
+ boolean checkpointCreated = false;
+
+ // First attempt
+ try {
+ LOG.info("Attempting checkpoint creation (iteration: {}, attempt: 1)", iterationRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ checkpointCreated = true;
+ LOG.info("Checkpoint creation succeeded on first attempt");
+ } catch (IOException e) {
+ LOG.warn("First checkpoint creation attempt failed: {}", e.getMessage());
+
+ // Immediate retry within the same iteration
+ try {
+ LOG.info("Attempting immediate retry for checkpoint creation (iteration: {}, attempt: 2)",
Review Comment:
we can avoid immediate retry of creating checkpoint, as its on same file-system will have same issue, current retry after 2 sec is ok.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -431,4 +468,359 @@ public boolean hasDeltaTasksFailed() {
public void resetDeltaTasksFailureFlag() {
deltaTasksFailed.set(false);
}
+
+ @Override
+ public ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (iteration retry count: {})", reason,
+ iterationRetryCount);
+
+ // Check if we should retry based on timing for iteration-based retries
+ long currentTime = System.currentTimeMillis();
+ if (iterationRetryCount > 0) {
+ // Check if 2 seconds have passed since last iteration
+ long timeSinceLastRetry = currentTime - lastCheckpointRetryTimestamp;
+ if (timeSinceLastRetry < RETRY_DELAY_MS) {
+ LOG.debug("Skipping checkpoint creation retry, only {}ms since last iteration (need {}ms)",
+ timeSinceLastRetry, RETRY_DELAY_MS);
+ return ReconTaskController.ReInitializationResult.RETRY_LATER;
+ }
+ LOG.info("Attempting retry for checkpoint creation (iteration: {}, delay: {}ms)",
+ iterationRetryCount + 1, timeSinceLastRetry);
+ }
+
+ // Try checkpoint creation with immediate retry capability
+ ReconOMMetadataManager checkpointedOMMetadataManager = null;
+ boolean checkpointCreated = false;
+
+ // First attempt
+ try {
+ LOG.info("Attempting checkpoint creation (iteration: {}, attempt: 1)", iterationRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ checkpointCreated = true;
+ LOG.info("Checkpoint creation succeeded on first attempt");
+ } catch (IOException e) {
+ LOG.warn("First checkpoint creation attempt failed: {}", e.getMessage());
+
+ // Immediate retry within the same iteration
+ try {
+ LOG.info("Attempting immediate retry for checkpoint creation (iteration: {}, attempt: 2)",
+ iterationRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ checkpointCreated = true;
+ LOG.info("Checkpoint creation succeeded on immediate retry");
+ } catch (IOException retryException) {
+ LOG.error("Immediate retry for checkpoint creation also failed: {}", retryException.getMessage());
+ // Both attempts in this iteration failed
+ handleIterationFailure(currentTime);
+ return getIterationRetryResult();
+ }
+ }
+ // Proceed with queueing the event
+ resetEventBuffer();
+
+ // Create and queue the reinitialization event with checkpointed metadata manager
+ ReconTaskReInitializationEvent reinitEvent =
+ new ReconTaskReInitializationEvent(reason, checkpointedOMMetadataManager);
+ boolean queued = eventBuffer.offer(reinitEvent);
+ if (!queued) {
+ // CAUTION: If this failure happens, something is seriously wrong as we just cleared the buffer
+ // and are trying to queue a single event, also since we cleared all other delta events, next delta sync cycle
+ // will not have these lost events as new OMDBUpdatesHandler will be created with last sequence number
+ LOG.error("Failed to queue reinitialization event. Ideally should not happen, so dropping event and " +
+ "checkpointed data.");
+ // Clean up the checkpointed metadata manager and its files as it won't be used
+ cleanupCheckpoint(checkpointedOMMetadataManager);
+ handleIterationFailure(currentTime);
+ return getIterationRetryResult();
+ }
+
+ // Success - reset retry counters and flags
+ LOG.info("Successfully queued reinitialization event after {} iteration(s)", iterationRetryCount + 1);
+ iterationRetryCount = 0;
+ lastCheckpointRetryTimestamp = 0;
+ resetEventBufferOverflowFlag();
+ resetDeltaTasksFailureFlag();
+ return ReconTaskController.ReInitializationResult.SUCCESS;
+ }
+
+ /**
+ * Handle iteration failure by updating retry counters.
+ */
+ private void handleIterationFailure(long currentTime) {
+ iterationRetryCount++;
+ lastCheckpointRetryTimestamp = currentTime;
+ LOG.error("Checkpoint creation failed in iteration {} (both immediate attempts failed)", iterationRetryCount);
+ }
+
+ /**
+ * Determine the appropriate retry result based on current iteration retry count.
+ */
+ private ReconTaskController.ReInitializationResult getIterationRetryResult() {
+ if (iterationRetryCount >= MAX_ITERATION_RETRIES) {
+ LOG.warn("Maximum iteration retries ({}) exceeded, resetting counters and signaling full snapshot fallback",
+ MAX_ITERATION_RETRIES);
+ iterationRetryCount = 0;
+ lastCheckpointRetryTimestamp = 0;
+ return ReconTaskController.ReInitializationResult.MAX_RETRIES_EXCEEDED;
+ } else {
+ return ReconTaskController.ReInitializationResult.RETRY_LATER;
+ }
+ }
+
+ @Override
+ public void resetEventBuffer() {
+ // Clear the buffer to discard all pending events
+ eventBuffer.clear();
Review Comment:
as clear() implementation, need to use drainTo() to move all elements to temp list, and if any event is Reinit, need close and remove checkpoint dbs which is not yet scheduled.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -431,4 +468,359 @@ public boolean hasDeltaTasksFailed() {
public void resetDeltaTasksFailureFlag() {
deltaTasksFailed.set(false);
}
+
+ @Override
+ public ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (iteration retry count: {})", reason,
+ iterationRetryCount);
+
+ // Check if we should retry based on timing for iteration-based retries
+ long currentTime = System.currentTimeMillis();
+ if (iterationRetryCount > 0) {
+ // Check if 2 seconds have passed since last iteration
+ long timeSinceLastRetry = currentTime - lastCheckpointRetryTimestamp;
+ if (timeSinceLastRetry < RETRY_DELAY_MS) {
+ LOG.debug("Skipping checkpoint creation retry, only {}ms since last iteration (need {}ms)",
+ timeSinceLastRetry, RETRY_DELAY_MS);
+ return ReconTaskController.ReInitializationResult.RETRY_LATER;
+ }
+ LOG.info("Attempting retry for checkpoint creation (iteration: {}, delay: {}ms)",
+ iterationRetryCount + 1, timeSinceLastRetry);
+ }
+
+ // Try checkpoint creation with immediate retry capability
+ ReconOMMetadataManager checkpointedOMMetadataManager = null;
+ boolean checkpointCreated = false;
+
+ // First attempt
+ try {
+ LOG.info("Attempting checkpoint creation (iteration: {}, attempt: 1)", iterationRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ checkpointCreated = true;
+ LOG.info("Checkpoint creation succeeded on first attempt");
+ } catch (IOException e) {
+ LOG.warn("First checkpoint creation attempt failed: {}", e.getMessage());
+
+ // Immediate retry within the same iteration
+ try {
+ LOG.info("Attempting immediate retry for checkpoint creation (iteration: {}, attempt: 2)",
+ iterationRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ checkpointCreated = true;
+ LOG.info("Checkpoint creation succeeded on immediate retry");
+ } catch (IOException retryException) {
+ LOG.error("Immediate retry for checkpoint creation also failed: {}", retryException.getMessage());
+ // Both attempts in this iteration failed
+ handleIterationFailure(currentTime);
+ return getIterationRetryResult();
+ }
+ }
+ // Proceed with queueing the event
+ resetEventBuffer();
+
+ // Create and queue the reinitialization event with checkpointed metadata manager
+ ReconTaskReInitializationEvent reinitEvent =
+ new ReconTaskReInitializationEvent(reason, checkpointedOMMetadataManager);
+ boolean queued = eventBuffer.offer(reinitEvent);
+ if (!queued) {
+ // CAUTION: If this failure happens, something is seriously wrong as we just cleared the buffer
+ // and are trying to queue a single event, also since we cleared all other delta events, next delta sync cycle
+ // will not have these lost events as new OMDBUpdatesHandler will be created with last sequence number
+ LOG.error("Failed to queue reinitialization event. Ideally should not happen, so dropping event and " +
+ "checkpointed data.");
+ // Clean up the checkpointed metadata manager and its files as it won't be used
+ cleanupCheckpoint(checkpointedOMMetadataManager);
+ handleIterationFailure(currentTime);
+ return getIterationRetryResult();
+ }
+
+ // Success - reset retry counters and flags
+ LOG.info("Successfully queued reinitialization event after {} iteration(s)", iterationRetryCount + 1);
+ iterationRetryCount = 0;
+ lastCheckpointRetryTimestamp = 0;
+ resetEventBufferOverflowFlag();
+ resetDeltaTasksFailureFlag();
+ return ReconTaskController.ReInitializationResult.SUCCESS;
+ }
+
+ /**
+ * Handle iteration failure by updating retry counters.
+ */
+ private void handleIterationFailure(long currentTime) {
+ iterationRetryCount++;
+ lastCheckpointRetryTimestamp = currentTime;
+ LOG.error("Checkpoint creation failed in iteration {} (both immediate attempts failed)", iterationRetryCount);
+ }
+
+ /**
+ * Determine the appropriate retry result based on current iteration retry count.
+ */
+ private ReconTaskController.ReInitializationResult getIterationRetryResult() {
+ if (iterationRetryCount >= MAX_ITERATION_RETRIES) {
+ LOG.warn("Maximum iteration retries ({}) exceeded, resetting counters and signaling full snapshot fallback",
+ MAX_ITERATION_RETRIES);
+ iterationRetryCount = 0;
+ lastCheckpointRetryTimestamp = 0;
+ return ReconTaskController.ReInitializationResult.MAX_RETRIES_EXCEEDED;
+ } else {
+ return ReconTaskController.ReInitializationResult.RETRY_LATER;
+ }
+ }
+
+ @Override
+ public void resetEventBuffer() {
+ // Clear the buffer to discard all pending events
+ eventBuffer.clear();
+ }
+
+ @Override
+ public void updateOMMetadataManager(ReconOMMetadataManager omMetadataManager) {
+ LOG.info("Updating OM metadata manager");
+
+ // Clean up any pre-existing checkpoint directories when setting new manager
+ if (omMetadataManager != null) {
+ cleanupPreExistingCheckpoints();
+ }
+
+ this.currentOMMetadataManager = omMetadataManager;
+ }
+
+ @Override
+ public ReconOMMetadataManager createOMCheckpoint(ReconOMMetadataManager omMetaManager)
+ throws IOException {
+ // Create temporary directory for checkpoint
+ String parentPath = cleanTempCheckPointPath(omMetaManager);
+
+ // Create checkpoint
+ DBCheckpoint checkpoint = omMetaManager.getStore().getCheckpoint(parentPath, true);
+
+ // Create a new ReconOmMetadataManagerImpl instance with checkpointed data
+ // Note: We need ReconUtils but it's not available in this class
+ // We'll create the ReconOmMetadataManagerImpl with minimal dependencies
+ ReconOmMetadataManagerImpl checkpointedManager = new ReconOmMetadataManagerImpl(configuration);
+
+ // Initialize the checkpointed manager with the checkpoint location
+ File checkpointDir = checkpoint.getCheckpointLocation().toFile();
+ checkpointedManager.updateOmDB(checkpointDir, false);
+
+ return checkpointedManager;
+ }
+
+ /**
+ * Clean and prepare temporary checkpoint path.
+ * Similar to QuotaRepairTask.cleanTempCheckPointPath.
+ *
+ * @param omMetaManager the OM metadata manager
+ * @return path to temporary checkpoint directory
+ * @throws IOException if directory operations fail
+ */
+ private String cleanTempCheckPointPath(ReconOMMetadataManager omMetaManager) throws IOException {
+ File dbLocation = omMetaManager.getStore().getDbLocation();
+ if (dbLocation == null) {
+ throw new NullPointerException("OM DB location is null");
+ }
+ String tempData = dbLocation.getParent();
+ if (tempData == null) {
+ throw new NullPointerException("Parent OM DB dir is null");
+ }
+ File reinitTmpPath = Paths.get(tempData, "temp-recon-reinit-checkpoint").toFile();
+ FileUtils.deleteDirectory(reinitTmpPath);
+ FileUtils.forceMkdir(reinitTmpPath);
+ return reinitTmpPath.toString();
+ }
+
+ /**
+ * Process a task reinitialization event asynchronously.
+ */
+ private void processReInitializationEvent(ReconTaskReInitializationEvent event) {
+ LOG.info("Processing reinitialization event: reason={}, timestamp={}",
+ event.getReason(), event.getTimestamp());
+
+ try {
+ // Use the checkpointed OM metadata manager for reinitialization to prevent data inconsistency
+ ReconOMMetadataManager checkpointedOMMetadataManager = event.getCheckpointedOMMetadataManager();
+ if (checkpointedOMMetadataManager != null) {
+ LOG.info("Starting async task reinitialization with checkpointed OM metadata manager due to: {}",
+ event.getReason());
+ reInitializeTasks(checkpointedOMMetadataManager, null);
+ LOG.info("Completed async task reinitialization");
+
+ // Clean up the checkpointed metadata manager and its files after use
+ cleanupCheckpoint(checkpointedOMMetadataManager);
+ } else {
+ LOG.error("Checkpointed OM metadata manager is null, cannot perform reinitialization");
+ return;
+ }
+
+ resetEventFlags(event.getReason());
+
+ LOG.info("Completed processing reinitialization event: {}", event.getReason());
+
+ } catch (Exception e) {
+ LOG.error("Error processing reinitialization event", e);
+ }
+ }
+
+ @Override
+ public void resetEventFlags(ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ // Reset appropriate flags based on the reason
+ if (reason == ReconTaskReInitializationEvent.ReInitializationReason.BUFFER_OVERFLOW ||
+ reason == ReconTaskReInitializationEvent.ReInitializationReason.TASK_FAILURES ||
+ reason == ReconTaskReInitializationEvent.ReInitializationReason.MANUAL_TRIGGER) {
+ resetEventBufferOverflowFlag();
+ resetDeltaTasksFailureFlag();
+ }
+ }
+
+ @Override
+ public int getEventBufferSize() {
+ return eventBuffer.getQueueSize();
+ }
+
+ @Override
+ public long getDroppedBatches() {
+ return eventBuffer.getDroppedBatches();
+ }
+
+ /**
+ * Reset retry counters - for testing purposes.
+ */
+ @VisibleForTesting
+ void resetRetryCounters() {
+ iterationRetryCount = 0;
+ lastCheckpointRetryTimestamp = 0;
+ }
+
+ /**
+ * Get current iteration retry count - for testing purposes.
+ */
+ @VisibleForTesting
+ int getIterationRetryCount() {
+ return iterationRetryCount;
+ }
+
+ /**
+ * Wait for all currently buffered events to be processed asynchronously.
+ * This method returns a CompletableFuture that completes when the event buffer becomes empty.
+ * Useful for testing to ensure async processing is complete before assertions.
+ *
+ * @return CompletableFuture that completes when buffer is empty
+ */
+ @VisibleForTesting
+ @Override
+ public CompletableFuture<Void> waitForEventBufferEmpty() {
Review Comment:
this can be part of test code util
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -431,4 +457,156 @@ public boolean hasDeltaTasksFailed() {
public void resetDeltaTasksFailureFlag() {
deltaTasksFailed.set(false);
}
+
+ @Override
+ public boolean queueReInitializationEvent(ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {}", reason);
+
+ try {
+ // Create checkpoint of current OM metadata manager before queueing reinitialization event
+ // This prevents data inconsistency as parallel DB updates will continue on the original
+ // omMetadataManager reference while reinitialization uses the checkpointed snapshot
+ ReconOMMetadataManager checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+
+ resetEventBuffer();
+
+ // Create and queue the reinitialization event with checkpointed metadata manager
+ ReconTaskReInitializationEvent reinitEvent =
+ new ReconTaskReInitializationEvent(reason, checkpointedOMMetadataManager);
+ boolean queued = eventBuffer.offer(reinitEvent);
+ if (!queued) {
+ // CAUTION: If this failure happens, something is seriously wrong as we just cleared the buffer
+ // and are trying to queue a single event, also since we cleared all other delta events, next delta sync cycle
+ // will not have these lost events as new OMDBUpdatesHandler will be created with last sequence number
+ LOG.error(
+ "Failed to queue reinitialization event. Ideally should not happen, so dropping event and " +
+ "checkpointed data.");
+ // Close the checkpointed metadata manager as it won't be used
+ try {
+ checkpointedOMMetadataManager.stop();
+ } catch (Exception e) {
+ LOG.warn("Failed to close checkpointed OM metadata manager", e);
+ }
+ return false;
+ }
+ resetEventBufferOverflowFlag();
+ resetDeltaTasksFailureFlag();
+ return true;
+ } catch (IOException e) {
+ LOG.error("Failed to create checkpoint for reinitialization", e);
+ return false;
+ }
+ }
+
+ @Override
+ public void resetEventBuffer() {
+ // Clear the buffer to discard all pending events
+ eventBuffer.clear();
+ }
+
+ @Override
+ public void updateOMMetadataManager(ReconOMMetadataManager omMetadataManager) {
+ this.currentOMMetadataManager = omMetadataManager;
+ }
+
+ @Override
+ public ReconOMMetadataManager createOMCheckpoint(ReconOMMetadataManager omMetaManager)
+ throws IOException {
+ // Create temporary directory for checkpoint
+ String parentPath = cleanTempCheckPointPath(omMetaManager);
Review Comment:
we can get unique checkpoint, as there can be 2 task, 1 running for reprocess, another waiting in queue.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -431,4 +468,359 @@ public boolean hasDeltaTasksFailed() {
public void resetDeltaTasksFailureFlag() {
deltaTasksFailed.set(false);
}
+
+ @Override
+ public ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (iteration retry count: {})", reason,
+ iterationRetryCount);
+
+ // Check if we should retry based on timing for iteration-based retries
+ long currentTime = System.currentTimeMillis();
+ if (iterationRetryCount > 0) {
+ // Check if 2 seconds have passed since last iteration
+ long timeSinceLastRetry = currentTime - lastCheckpointRetryTimestamp;
+ if (timeSinceLastRetry < RETRY_DELAY_MS) {
+ LOG.debug("Skipping checkpoint creation retry, only {}ms since last iteration (need {}ms)",
+ timeSinceLastRetry, RETRY_DELAY_MS);
+ return ReconTaskController.ReInitializationResult.RETRY_LATER;
+ }
+ LOG.info("Attempting retry for checkpoint creation (iteration: {}, delay: {}ms)",
+ iterationRetryCount + 1, timeSinceLastRetry);
+ }
+
+ // Try checkpoint creation with immediate retry capability
+ ReconOMMetadataManager checkpointedOMMetadataManager = null;
+ boolean checkpointCreated = false;
+
+ // First attempt
+ try {
+ LOG.info("Attempting checkpoint creation (iteration: {}, attempt: 1)", iterationRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ checkpointCreated = true;
+ LOG.info("Checkpoint creation succeeded on first attempt");
+ } catch (IOException e) {
+ LOG.warn("First checkpoint creation attempt failed: {}", e.getMessage());
+
+ // Immediate retry within the same iteration
+ try {
+ LOG.info("Attempting immediate retry for checkpoint creation (iteration: {}, attempt: 2)",
+ iterationRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ checkpointCreated = true;
+ LOG.info("Checkpoint creation succeeded on immediate retry");
+ } catch (IOException retryException) {
+ LOG.error("Immediate retry for checkpoint creation also failed: {}", retryException.getMessage());
+ // Both attempts in this iteration failed
+ handleIterationFailure(currentTime);
+ return getIterationRetryResult();
+ }
+ }
+ // Proceed with queueing the event
+ resetEventBuffer();
+
+ // Create and queue the reinitialization event with checkpointed metadata manager
+ ReconTaskReInitializationEvent reinitEvent =
+ new ReconTaskReInitializationEvent(reason, checkpointedOMMetadataManager);
+ boolean queued = eventBuffer.offer(reinitEvent);
+ if (!queued) {
+ // CAUTION: If this failure happens, something is seriously wrong as we just cleared the buffer
+ // and are trying to queue a single event, also since we cleared all other delta events, next delta sync cycle
+ // will not have these lost events as new OMDBUpdatesHandler will be created with last sequence number
+ LOG.error("Failed to queue reinitialization event. Ideally should not happen, so dropping event and " +
+ "checkpointed data.");
+ // Clean up the checkpointed metadata manager and its files as it won't be used
+ cleanupCheckpoint(checkpointedOMMetadataManager);
+ handleIterationFailure(currentTime);
+ return getIterationRetryResult();
+ }
+
+ // Success - reset retry counters and flags
+ LOG.info("Successfully queued reinitialization event after {} iteration(s)", iterationRetryCount + 1);
+ iterationRetryCount = 0;
+ lastCheckpointRetryTimestamp = 0;
+ resetEventBufferOverflowFlag();
+ resetDeltaTasksFailureFlag();
+ return ReconTaskController.ReInitializationResult.SUCCESS;
+ }
+
+ /**
+ * Handle iteration failure by updating retry counters.
+ */
+ private void handleIterationFailure(long currentTime) {
+ iterationRetryCount++;
+ lastCheckpointRetryTimestamp = currentTime;
+ LOG.error("Checkpoint creation failed in iteration {} (both immediate attempts failed)", iterationRetryCount);
+ }
+
+ /**
+ * Determine the appropriate retry result based on current iteration retry count.
+ */
+ private ReconTaskController.ReInitializationResult getIterationRetryResult() {
+ if (iterationRetryCount >= MAX_ITERATION_RETRIES) {
+ LOG.warn("Maximum iteration retries ({}) exceeded, resetting counters and signaling full snapshot fallback",
+ MAX_ITERATION_RETRIES);
+ iterationRetryCount = 0;
+ lastCheckpointRetryTimestamp = 0;
+ return ReconTaskController.ReInitializationResult.MAX_RETRIES_EXCEEDED;
+ } else {
+ return ReconTaskController.ReInitializationResult.RETRY_LATER;
+ }
+ }
+
+ @Override
+ public void resetEventBuffer() {
+ // Clear the buffer to discard all pending events
+ eventBuffer.clear();
+ }
+
+ @Override
+ public void updateOMMetadataManager(ReconOMMetadataManager omMetadataManager) {
+ LOG.info("Updating OM metadata manager");
+
+ // Clean up any pre-existing checkpoint directories when setting new manager
+ if (omMetadataManager != null) {
+ cleanupPreExistingCheckpoints();
+ }
+
+ this.currentOMMetadataManager = omMetadataManager;
+ }
+
+ @Override
+ public ReconOMMetadataManager createOMCheckpoint(ReconOMMetadataManager omMetaManager)
+ throws IOException {
+ // Create temporary directory for checkpoint
+ String parentPath = cleanTempCheckPointPath(omMetaManager);
+
+ // Create checkpoint
+ DBCheckpoint checkpoint = omMetaManager.getStore().getCheckpoint(parentPath, true);
+
+ // Create a new ReconOmMetadataManagerImpl instance with checkpointed data
+ // Note: We need ReconUtils but it's not available in this class
+ // We'll create the ReconOmMetadataManagerImpl with minimal dependencies
+ ReconOmMetadataManagerImpl checkpointedManager = new ReconOmMetadataManagerImpl(configuration);
+
+ // Initialize the checkpointed manager with the checkpoint location
+ File checkpointDir = checkpoint.getCheckpointLocation().toFile();
+ checkpointedManager.updateOmDB(checkpointDir, false);
+
+ return checkpointedManager;
+ }
+
+ /**
+ * Clean and prepare temporary checkpoint path.
+ * Similar to QuotaRepairTask.cleanTempCheckPointPath.
+ *
+ * @param omMetaManager the OM metadata manager
+ * @return path to temporary checkpoint directory
+ * @throws IOException if directory operations fail
+ */
+ private String cleanTempCheckPointPath(ReconOMMetadataManager omMetaManager) throws IOException {
+ File dbLocation = omMetaManager.getStore().getDbLocation();
+ if (dbLocation == null) {
+ throw new NullPointerException("OM DB location is null");
+ }
+ String tempData = dbLocation.getParent();
+ if (tempData == null) {
+ throw new NullPointerException("Parent OM DB dir is null");
+ }
+ File reinitTmpPath = Paths.get(tempData, "temp-recon-reinit-checkpoint").toFile();
+ FileUtils.deleteDirectory(reinitTmpPath);
+ FileUtils.forceMkdir(reinitTmpPath);
+ return reinitTmpPath.toString();
+ }
+
+ /**
+ * Process a task reinitialization event asynchronously.
+ */
+ private void processReInitializationEvent(ReconTaskReInitializationEvent event) {
+ LOG.info("Processing reinitialization event: reason={}, timestamp={}",
+ event.getReason(), event.getTimestamp());
+
+ try {
+ // Use the checkpointed OM metadata manager for reinitialization to prevent data inconsistency
+ ReconOMMetadataManager checkpointedOMMetadataManager = event.getCheckpointedOMMetadataManager();
+ if (checkpointedOMMetadataManager != null) {
+ LOG.info("Starting async task reinitialization with checkpointed OM metadata manager due to: {}",
+ event.getReason());
+ reInitializeTasks(checkpointedOMMetadataManager, null);
+ LOG.info("Completed async task reinitialization");
+
+ // Clean up the checkpointed metadata manager and its files after use
+ cleanupCheckpoint(checkpointedOMMetadataManager);
+ } else {
+ LOG.error("Checkpointed OM metadata manager is null, cannot perform reinitialization");
+ return;
+ }
+
+ resetEventFlags(event.getReason());
Review Comment:
we should not reset this as might be happened again, should be reset when added to queue successfully
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org