You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@ozone.apache.org by "devmadhuu (via GitHub)" <gi...@apache.org> on 2025/08/21 06:25:21 UTC
[PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
devmadhuu opened a new pull request, #8962:
URL: https://github.com/apache/ozone/pull/8962
## What changes were proposed in this pull request?
This PR change is to implement the `re-initialization` of OM tasks method as non-blocking and put in same `eventBuffer` which is used for pushing OM delta update events batch.
**Key Changes Made:**
```
1. Created Custom Event Types:
- ReconEvent interface for common event handling
- ReconTaskReInitializationEvent for async reinitialization triggers
- OMUpdateEventBatch now implements ReconEvent
2. Refactored Event Buffer:
- OMUpdateEventBuffer now handles generic ReconEvent instead of just OMUpdateEventBatch
- Supports both OM delta update events and custom control events
3. Enhanced ReconTaskController:
- Added queueReInitializationEvent() method for async reinitialization
- Added updateOMMetadataManager() to maintain current OM metadata reference
- Refactored processBufferedEventsAsync() to handle multiple event types
- Split event processing into processReconEvent() → processOMUpdateBatch() / processReInitializationEvent()
4. Updated OzoneManagerServiceProviderImpl:
- Non-blocking approach: Replaced blocking reInitializeTasks() calls with async queueReInitializationEvent()
- Event-driven: When buffer overflows or tasks fail, generates custom events instead of blocking
- Fallback handling: If event queueing fails, falls back to sync reinitialization
```
**How It Works Now:**
```
1. Normal Flow: OM delta events are buffered and processed asynchronously
2. Buffer Overflow Detection: When buffer is full:
- Generates BUFFER_OVERFLOW reinitialization event
- Clears buffer to discard pending events
- Queues reinitialization event for async processing
- Returns immediately (non-blocking!)
3. Task Failure Detection: When delta tasks fail after retry:
- Sets deltaTasksFailed flag
- Next sync cycle detects failure and generates TASK_FAILURES reinitialization event
- Returns immediately (non-blocking!)
4. Async Reinitialization: Dedicated thread processes reinitialization event:
- Calls reInitializeTasks() asynchronously
- Resets appropriate flags after completion
- Continues buffering and processing other delta events post the completion of reinitialization event.
```
## What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/HDDS-13576
## How was this patch tested?
Added new test: `TestOMUpdateEventBuffer.java` and did manual testing in local docker container using a python script
to generate bulk batch of FSO , non-FSO keys.
--
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
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
Posted by "devmadhuu (via GitHub)" <gi...@apache.org>.
devmadhuu commented on code in PR #8962:
URL: https://github.com/apache/ozone/pull/8962#discussion_r2315809210
##########
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:
ok done.
--
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
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
Posted by "sumitagrawl (via GitHub)" <gi...@apache.org>.
sumitagrawl commented on code in PR #8962:
URL: https://github.com/apache/ozone/pull/8962#discussion_r2318802159
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/recovery/ReconOmMetadataManagerImpl.java:
##########
@@ -72,6 +74,29 @@ public ReconOmMetadataManagerImpl(OzoneConfiguration configuration,
this.ozoneConfiguration = configuration;
}
+ private ReconOmMetadataManagerImpl(OzoneConfiguration configuration, File dir, String name) throws IOException {
Review Comment:
Need set reconUtils for completeness of constructor if used later on to avoid issue.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java:
##########
@@ -663,30 +665,45 @@ public boolean syncDataFromOM() {
fullSnapshotReconTaskUpdater.setLastUpdatedSeqNumber(getCurrentOMDBSequenceNumber());
deltaReconTaskStatusUpdater.recordRunCompletion();
fullSnapshotReconTaskUpdater.updateDetails();
+ // Update the current OM metadata manager in task controller
+ reconTaskController.updateOMMetadataManager(omMetadataManager);
Review Comment:
this call cleanup existing checkpoint also, but to be done only first time on startup, and its already done in ControllerImpl.start(). need change its implementation to remove this cleanup.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskController.java:
##########
@@ -78,14 +91,60 @@ void consumeOMEvents(OMUpdateEventBatch events,
void resetEventBufferOverflowFlag();
/**
- * Check if delta tasks have failed and need reinitialization.
+ * Check if task(s) have failed and need reinitialization.
*
- * @return true if delta tasks failed after retry
+ * @return true if task(s) failed after retry
+ */
+ boolean hasTasksFailed();
+
+ /**
+ * Reset the task(s) failure flag after reinitialization is completed.
*/
- boolean hasDeltaTasksFailed();
+ void resetTasksFailureFlag();
/**
- * Reset the delta tasks failure flag after reinitialization is completed.
+ * Queue a task reinitialization event to be processed asynchronously.
+ * This method creates a checkpoint of the current OM metadata manager,
+ * clears the event buffer and queues a reinitialization event.
+ * Includes internal retry logic with timing controls for checkpoint creation.
+ *
+ * @param reason the reason for reinitialization
+ * @return ReInitializationResult indicating success, retry needed, or max retries exceeded
+ */
+ ReInitializationResult queueReInitializationEvent(ReconTaskReInitializationEvent.ReInitializationReason reason);
+
+ /**
+ * Create a checkpoint of the current OM metadata manager.
+ * This method creates a snapshot of the current OM database state
+ * to prevent data inconsistency during reinitialization.
+ *
+ * @param omMetaManager the OM metadata manager to checkpoint
+ * @return a checkpointed ReconOMMetadataManager instance
+ * @throws IOException if checkpoint creation fails
*/
- void resetDeltaTasksFailureFlag();
+ ReconOMMetadataManager createOMCheckpoint(ReconOMMetadataManager omMetaManager) throws IOException;
+
+ /**
+ * Update the current OM metadata manager reference for reinitialization.
+ *
+ * @param omMetadataManager the current OM metadata manager
+ */
+ void updateOMMetadataManager(ReconOMMetadataManager omMetadataManager);
+
+ /**
+ * Get the current size of the event buffer.
+ *
+ * @return the number of events currently in the buffer
+ */
+ @VisibleForTesting
+ int getEventBufferSize();
+
+ /**
+ * Get the number of batches that have been dropped due to buffer overflow.
+ * This is used by the overflow detection logic.
+ *
+ * @return the number of dropped batches
+ */
+ long getDroppedBatches();
Review Comment:
this also not required to be exposed
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -71,7 +76,15 @@ public class ReconTaskControllerImpl implements ReconTaskController {
private final ReconTaskStatusUpdaterManager taskStatusUpdaterManager;
private final OMUpdateEventBuffer eventBuffer;
private ExecutorService eventProcessingExecutor;
- private final AtomicBoolean deltaTasksFailed = new AtomicBoolean(false);
+ private final AtomicBoolean tasksFailed = new AtomicBoolean(false);
+ private volatile ReconOMMetadataManager currentOMMetadataManager;
+ private final OzoneConfiguration configuration;
+
+ // Retry logic for checkpoint creation
+ private int eventProcessRetryCount = 0;
Review Comment:
This needs to be AtomicInteger, as well as timestamp.,
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -423,12 +463,348 @@ public void resetEventBufferOverflowFlag() {
}
@Override
- public boolean hasDeltaTasksFailed() {
- return deltaTasksFailed.get();
+ public boolean hasTasksFailed() {
+ return tasksFailed.get();
+ }
+
+ @Override
+ public void resetTasksFailureFlag() {
+ tasksFailed.compareAndSet(true, false);
+ }
+
+ @Override
+ public synchronized ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (retry attempt count: {})", reason,
+ eventProcessRetryCount);
+
+ Long currentTime = validateRetryCountAndDelay();
+ if (currentTime == null) {
+ return ReInitializationResult.RETRY_LATER;
+ }
+
+ // Drain all events in buffer and cleanup any existing checkpoints before falling back to full snapshot.
+ // Events can be present in queue when reinit checkpoint creation fails multiple times because only after
+ // successful creation of checkpoint, we are clearing the event buffer.
+ drainEventBufferAndCleanExistingCheckpoints();
+
+ // Try checkpoint creation (single attempt per iteration)
+ ReconOMMetadataManager checkpointedOMMetadataManager = null;
+
+ try {
+ LOG.info("Attempting checkpoint creation (retry attempt: {})", eventProcessRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ LOG.info("Checkpoint creation succeeded");
+ } catch (IOException e) {
+ LOG.error("Checkpoint creation failed: {}", e.getMessage());
+ handleEventFailure(currentTime);
+ return getEventRetryResult();
+ }
+
+ // 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();
+ resetRetryCounters();
+ // Success - reset retry counters and flags
+ LOG.info("Successfully queued reinitialization event after {} retries", eventProcessRetryCount + 1);
+ return ReconTaskController.ReInitializationResult.SUCCESS;
+ }
+ return null;
+ }
+
+ private Long validateRetryCountAndDelay() {
+ // Check if we should retry based on timing for iteration-based retries
+ long currentTime = System.currentTimeMillis();
+ if (eventProcessRetryCount > 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 retry attempt (need {}ms)",
+ timeSinceLastRetry, RETRY_DELAY_MS);
+ return null;
+ }
+ LOG.info("Attempting retry for checkpoint creation (retry attempt count: {}, delay: {}ms)",
+ eventProcessRetryCount + 1, timeSinceLastRetry);
+ }
+ return currentTime;
+ }
+
+ /**
+ * Handle iteration failure by updating retry counters.
+ */
+ private void handleEventFailure(long currentTime) {
+ eventProcessRetryCount++;
+ lastCheckpointRetryTimestamp = currentTime;
+ LOG.error("Event processing failed {} times.", eventProcessRetryCount);
+ }
+
+ /**
+ * Determine the appropriate retry result based on current event retry count.
+ */
+ private ReconTaskController.ReInitializationResult getEventRetryResult() {
+ if (eventProcessRetryCount >= MAX_EVENT_PROCESS_RETRIES) {
+ LOG.warn("Maximum iteration retries ({}) exceeded, resetting counters and signaling full snapshot fallback",
+ MAX_EVENT_PROCESS_RETRIES);
+ eventProcessRetryCount = 0;
Review Comment:
call resetRetryCounters() here also to have common place update
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -423,12 +463,348 @@ public void resetEventBufferOverflowFlag() {
}
@Override
- public boolean hasDeltaTasksFailed() {
- return deltaTasksFailed.get();
+ public boolean hasTasksFailed() {
+ return tasksFailed.get();
+ }
+
+ @Override
+ public void resetTasksFailureFlag() {
+ tasksFailed.compareAndSet(true, false);
+ }
+
+ @Override
+ public synchronized ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (retry attempt count: {})", reason,
+ eventProcessRetryCount);
+
+ Long currentTime = validateRetryCountAndDelay();
+ if (currentTime == null) {
+ return ReInitializationResult.RETRY_LATER;
+ }
+
+ // Drain all events in buffer and cleanup any existing checkpoints before falling back to full snapshot.
+ // Events can be present in queue when reinit checkpoint creation fails multiple times because only after
+ // successful creation of checkpoint, we are clearing the event buffer.
+ drainEventBufferAndCleanExistingCheckpoints();
+
+ // Try checkpoint creation (single attempt per iteration)
+ ReconOMMetadataManager checkpointedOMMetadataManager = null;
+
+ try {
+ LOG.info("Attempting checkpoint creation (retry attempt: {})", eventProcessRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ LOG.info("Checkpoint creation succeeded");
+ } catch (IOException e) {
+ LOG.error("Checkpoint creation failed: {}", e.getMessage());
+ handleEventFailure(currentTime);
+ return getEventRetryResult();
+ }
+
+ // 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();
+ resetRetryCounters();
Review Comment:
we should update resetRetryCounters() only when reprocess is completed successfully, not here.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskController.java:
##########
@@ -78,14 +91,60 @@ void consumeOMEvents(OMUpdateEventBatch events,
void resetEventBufferOverflowFlag();
/**
- * Check if delta tasks have failed and need reinitialization.
+ * Check if task(s) have failed and need reinitialization.
*
- * @return true if delta tasks failed after retry
+ * @return true if task(s) failed after retry
+ */
+ boolean hasTasksFailed();
+
+ /**
+ * Reset the task(s) failure flag after reinitialization is completed.
*/
- boolean hasDeltaTasksFailed();
+ void resetTasksFailureFlag();
/**
- * Reset the delta tasks failure flag after reinitialization is completed.
+ * Queue a task reinitialization event to be processed asynchronously.
+ * This method creates a checkpoint of the current OM metadata manager,
+ * clears the event buffer and queues a reinitialization event.
+ * Includes internal retry logic with timing controls for checkpoint creation.
+ *
+ * @param reason the reason for reinitialization
+ * @return ReInitializationResult indicating success, retry needed, or max retries exceeded
+ */
+ ReInitializationResult queueReInitializationEvent(ReconTaskReInitializationEvent.ReInitializationReason reason);
+
+ /**
+ * Create a checkpoint of the current OM metadata manager.
+ * This method creates a snapshot of the current OM database state
+ * to prevent data inconsistency during reinitialization.
+ *
+ * @param omMetaManager the OM metadata manager to checkpoint
+ * @return a checkpointed ReconOMMetadataManager instance
+ * @throws IOException if checkpoint creation fails
*/
- void resetDeltaTasksFailureFlag();
+ ReconOMMetadataManager createOMCheckpoint(ReconOMMetadataManager omMetaManager) throws IOException;
Review Comment:
this may not be exposed
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskController.java:
##########
@@ -78,14 +91,60 @@ void consumeOMEvents(OMUpdateEventBatch events,
void resetEventBufferOverflowFlag();
/**
- * Check if delta tasks have failed and need reinitialization.
+ * Check if task(s) have failed and need reinitialization.
*
- * @return true if delta tasks failed after retry
+ * @return true if task(s) failed after retry
+ */
+ boolean hasTasksFailed();
+
+ /**
+ * Reset the task(s) failure flag after reinitialization is completed.
*/
- boolean hasDeltaTasksFailed();
+ void resetTasksFailureFlag();
Review Comment:
resetTasksFailureFlag may not required to be exposed, and also resetEventBufferOverflowFlag
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -423,12 +463,348 @@ public void resetEventBufferOverflowFlag() {
}
@Override
- public boolean hasDeltaTasksFailed() {
- return deltaTasksFailed.get();
+ public boolean hasTasksFailed() {
+ return tasksFailed.get();
+ }
+
+ @Override
+ public void resetTasksFailureFlag() {
+ tasksFailed.compareAndSet(true, false);
+ }
+
+ @Override
+ public synchronized ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (retry attempt count: {})", reason,
+ eventProcessRetryCount);
+
+ Long currentTime = validateRetryCountAndDelay();
+ if (currentTime == null) {
+ return ReInitializationResult.RETRY_LATER;
+ }
+
+ // Drain all events in buffer and cleanup any existing checkpoints before falling back to full snapshot.
+ // Events can be present in queue when reinit checkpoint creation fails multiple times because only after
+ // successful creation of checkpoint, we are clearing the event buffer.
+ drainEventBufferAndCleanExistingCheckpoints();
+
+ // Try checkpoint creation (single attempt per iteration)
+ ReconOMMetadataManager checkpointedOMMetadataManager = null;
+
+ try {
+ LOG.info("Attempting checkpoint creation (retry attempt: {})", eventProcessRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ LOG.info("Checkpoint creation succeeded");
+ } catch (IOException e) {
+ LOG.error("Checkpoint creation failed: {}", e.getMessage());
+ handleEventFailure(currentTime);
+ return getEventRetryResult();
+ }
+
+ // 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();
+ resetRetryCounters();
+ // Success - reset retry counters and flags
+ LOG.info("Successfully queued reinitialization event after {} retries", eventProcessRetryCount + 1);
+ return ReconTaskController.ReInitializationResult.SUCCESS;
+ }
+ return null;
+ }
+
+ private Long validateRetryCountAndDelay() {
+ // Check if we should retry based on timing for iteration-based retries
+ long currentTime = System.currentTimeMillis();
+ if (eventProcessRetryCount > 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 retry attempt (need {}ms)",
+ timeSinceLastRetry, RETRY_DELAY_MS);
+ return null;
+ }
+ LOG.info("Attempting retry for checkpoint creation (retry attempt count: {}, delay: {}ms)",
+ eventProcessRetryCount + 1, timeSinceLastRetry);
+ }
+ return currentTime;
+ }
+
+ /**
+ * Handle iteration failure by updating retry counters.
+ */
+ private void handleEventFailure(long currentTime) {
+ eventProcessRetryCount++;
+ lastCheckpointRetryTimestamp = currentTime;
+ LOG.error("Event processing failed {} times.", eventProcessRetryCount);
+ }
+
+ /**
+ * Determine the appropriate retry result based on current event retry count.
+ */
+ private ReconTaskController.ReInitializationResult getEventRetryResult() {
+ if (eventProcessRetryCount >= MAX_EVENT_PROCESS_RETRIES) {
+ LOG.warn("Maximum iteration retries ({}) exceeded, resetting counters and signaling full snapshot fallback",
+ MAX_EVENT_PROCESS_RETRIES);
+ eventProcessRetryCount = 0;
+ lastCheckpointRetryTimestamp = 0;
+ return ReconTaskController.ReInitializationResult.MAX_RETRIES_EXCEEDED;
+ } else {
+ return ReconTaskController.ReInitializationResult.RETRY_LATER;
+ }
+ }
+
+ public void drainEventBufferAndCleanExistingCheckpoints() {
+ // First drain all events to check for any ReconTaskReInitializationEvent that need checkpoint cleanup
+ List<ReconEvent> drainedEvents = new ArrayList<>();
+ int drainedCount = eventBuffer.drainTo(drainedEvents);
+
+ if (drainedCount > 0) {
+ LOG.info("Drained {} events from buffer before clearing. Checking for checkpoint cleanup.", drainedCount);
+
+ // Check for any ReconTaskReInitializationEvent and cleanup their checkpoints
+ for (ReconEvent event : drainedEvents) {
+ if (event instanceof ReconTaskReInitializationEvent) {
+ ReconTaskReInitializationEvent reinitEvent = (ReconTaskReInitializationEvent) event;
+ ReconOMMetadataManager checkpointedManager = reinitEvent.getCheckpointedOMMetadataManager();
+ if (checkpointedManager != null) {
+ LOG.info("Cleaning up unprocessed checkpoint from drained ReconTaskReInitializationEvent");
+ cleanupCheckpoint(checkpointedManager);
+ }
+ }
+ }
+ }
+ }
+
+ @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);
+
+ return omMetaManager.createCheckpointReconMetadataManager(configuration, checkpoint);
+ }
+
+ /**
+ * 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" + "_" + System.currentTimeMillis()).toFile();
Review Comment:
time based checkpoint dir can fail, if trigger is quick. it does not ensure uniqueness
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -423,12 +463,348 @@ public void resetEventBufferOverflowFlag() {
}
@Override
- public boolean hasDeltaTasksFailed() {
- return deltaTasksFailed.get();
+ public boolean hasTasksFailed() {
+ return tasksFailed.get();
+ }
+
+ @Override
+ public void resetTasksFailureFlag() {
+ tasksFailed.compareAndSet(true, false);
+ }
+
+ @Override
+ public synchronized ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (retry attempt count: {})", reason,
+ eventProcessRetryCount);
+
+ Long currentTime = validateRetryCountAndDelay();
+ if (currentTime == null) {
+ return ReInitializationResult.RETRY_LATER;
+ }
+
+ // Drain all events in buffer and cleanup any existing checkpoints before falling back to full snapshot.
+ // Events can be present in queue when reinit checkpoint creation fails multiple times because only after
+ // successful creation of checkpoint, we are clearing the event buffer.
+ drainEventBufferAndCleanExistingCheckpoints();
+
+ // Try checkpoint creation (single attempt per iteration)
+ ReconOMMetadataManager checkpointedOMMetadataManager = null;
+
+ try {
+ LOG.info("Attempting checkpoint creation (retry attempt: {})", eventProcessRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ LOG.info("Checkpoint creation succeeded");
+ } catch (IOException e) {
+ LOG.error("Checkpoint creation failed: {}", e.getMessage());
+ handleEventFailure(currentTime);
+ return getEventRetryResult();
+ }
+
+ // 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();
+ resetRetryCounters();
+ // Success - reset retry counters and flags
+ LOG.info("Successfully queued reinitialization event after {} retries", eventProcessRetryCount + 1);
+ return ReconTaskController.ReInitializationResult.SUCCESS;
+ }
+ return null;
+ }
+
+ private Long validateRetryCountAndDelay() {
+ // Check if we should retry based on timing for iteration-based retries
+ long currentTime = System.currentTimeMillis();
+ if (eventProcessRetryCount > 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 retry attempt (need {}ms)",
+ timeSinceLastRetry, RETRY_DELAY_MS);
+ return null;
+ }
+ LOG.info("Attempting retry for checkpoint creation (retry attempt count: {}, delay: {}ms)",
+ eventProcessRetryCount + 1, timeSinceLastRetry);
+ }
+ return currentTime;
+ }
+
+ /**
+ * Handle iteration failure by updating retry counters.
+ */
+ private void handleEventFailure(long currentTime) {
+ eventProcessRetryCount++;
+ lastCheckpointRetryTimestamp = currentTime;
+ LOG.error("Event processing failed {} times.", eventProcessRetryCount);
+ }
+
+ /**
+ * Determine the appropriate retry result based on current event retry count.
+ */
+ private ReconTaskController.ReInitializationResult getEventRetryResult() {
+ if (eventProcessRetryCount >= MAX_EVENT_PROCESS_RETRIES) {
+ LOG.warn("Maximum iteration retries ({}) exceeded, resetting counters and signaling full snapshot fallback",
+ MAX_EVENT_PROCESS_RETRIES);
+ eventProcessRetryCount = 0;
+ lastCheckpointRetryTimestamp = 0;
+ return ReconTaskController.ReInitializationResult.MAX_RETRIES_EXCEEDED;
+ } else {
+ return ReconTaskController.ReInitializationResult.RETRY_LATER;
+ }
+ }
+
+ public void drainEventBufferAndCleanExistingCheckpoints() {
+ // First drain all events to check for any ReconTaskReInitializationEvent that need checkpoint cleanup
+ List<ReconEvent> drainedEvents = new ArrayList<>();
+ int drainedCount = eventBuffer.drainTo(drainedEvents);
+
+ if (drainedCount > 0) {
+ LOG.info("Drained {} events from buffer before clearing. Checking for checkpoint cleanup.", drainedCount);
+
+ // Check for any ReconTaskReInitializationEvent and cleanup their checkpoints
+ for (ReconEvent event : drainedEvents) {
+ if (event instanceof ReconTaskReInitializationEvent) {
+ ReconTaskReInitializationEvent reinitEvent = (ReconTaskReInitializationEvent) event;
+ ReconOMMetadataManager checkpointedManager = reinitEvent.getCheckpointedOMMetadataManager();
+ if (checkpointedManager != null) {
+ LOG.info("Cleaning up unprocessed checkpoint from drained ReconTaskReInitializationEvent");
+ cleanupCheckpoint(checkpointedManager);
+ }
+ }
+ }
+ }
+ }
+
+ @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);
+
+ return omMetaManager.createCheckpointReconMetadataManager(configuration, checkpoint);
+ }
+
+ /**
+ * 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" + "_" + System.currentTimeMillis()).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());
+ resetTasksFailureFlag();
+ 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());
+ boolean isRunSuccessful = reInitializeTasks(checkpointedOMMetadataManager, null);
+ if (!isRunSuccessful) {
+ // Setting this taskFailed flag as true here will block consuming delta events and stop buffering events
+ // in eventBuffer until we successfully queue a reinit event and also reprocess all the tasks of
+ // the reinit event succeed.
+ tasksFailed.compareAndSet(false, true);
+ LOG.error("Task reinitialization failed, tasksFailed flag set to true");
+ } else {
+ resetEventFlags();
+ resetRetryCounters();
+ LOG.info("Completed async task reinitialization");
+ }
+
+ // Clean up the checkpointed metadata manager and its files after use
+ cleanupCheckpoint(checkpointedOMMetadataManager);
Review Comment:
cleanup checkpoint in finally for all cases
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -423,12 +463,348 @@ public void resetEventBufferOverflowFlag() {
}
@Override
- public boolean hasDeltaTasksFailed() {
- return deltaTasksFailed.get();
+ public boolean hasTasksFailed() {
+ return tasksFailed.get();
+ }
+
+ @Override
+ public void resetTasksFailureFlag() {
+ tasksFailed.compareAndSet(true, false);
+ }
+
+ @Override
+ public synchronized ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (retry attempt count: {})", reason,
+ eventProcessRetryCount);
+
+ Long currentTime = validateRetryCountAndDelay();
+ if (currentTime == null) {
+ return ReInitializationResult.RETRY_LATER;
+ }
+
+ // Drain all events in buffer and cleanup any existing checkpoints before falling back to full snapshot.
+ // Events can be present in queue when reinit checkpoint creation fails multiple times because only after
+ // successful creation of checkpoint, we are clearing the event buffer.
+ drainEventBufferAndCleanExistingCheckpoints();
+
+ // Try checkpoint creation (single attempt per iteration)
+ ReconOMMetadataManager checkpointedOMMetadataManager = null;
+
+ try {
+ LOG.info("Attempting checkpoint creation (retry attempt: {})", eventProcessRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ LOG.info("Checkpoint creation succeeded");
+ } catch (IOException e) {
+ LOG.error("Checkpoint creation failed: {}", e.getMessage());
+ handleEventFailure(currentTime);
Review Comment:
I think failure count is handled for for checkpoint creation, not for continuous reprocess failure. Adding that need have this before create-checkpoint itself.
--
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
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
Posted by "sumitagrawl (via GitHub)" <gi...@apache.org>.
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
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
Posted by "ashishkumar50 (via GitHub)" <gi...@apache.org>.
ashishkumar50 commented on PR #8962:
URL: https://github.com/apache/ozone/pull/8962#issuecomment-3223289932
bugbot run
--
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
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
Posted by "devmadhuu (via GitHub)" <gi...@apache.org>.
devmadhuu commented on code in PR #8962:
URL: https://github.com/apache/ozone/pull/8962#discussion_r2315984925
##########
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:
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();
+ }
+
+ @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:
Ok
--
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
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
Posted by "sumitagrawl (via GitHub)" <gi...@apache.org>.
sumitagrawl commented on code in PR #8962:
URL: https://github.com/apache/ozone/pull/8962#discussion_r2297300769
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java:
##########
@@ -672,21 +676,30 @@ public boolean syncDataFromOM() {
boolean tasksFailed = reconTaskController.hasDeltaTasksFailed();
if (bufferOverflowed || tasksFailed) {
- String reason = bufferOverflowed ? "Event buffer overflow" : "Delta tasks failed after retry";
- LOG.warn("{}, triggering task reinitialization", reason);
+ ReconTaskReInitializationEvent.ReInitializationReason reason = bufferOverflowed ?
+ ReconTaskReInitializationEvent.ReInitializationReason.BUFFER_OVERFLOW :
+ ReconTaskReInitializationEvent.ReInitializationReason.TASK_FAILURES;
+
+ LOG.warn("Detected condition for task reinitialization: {}, queueing async reinitialization event",
+ reason);
metrics.incrNumDeltaRequestsFailed();
deltaReconTaskStatusUpdater.setLastTaskRunStatus(-1);
deltaReconTaskStatusUpdater.recordRunCompletion();
- reconTaskController.reInitializeTasks(omMetadataManager, null);
-
- // Reset appropriate flags after reinitialization
- if (bufferOverflowed) {
- reconTaskController.resetEventBufferOverflowFlag();
- }
- if (tasksFailed) {
- reconTaskController.resetDeltaTasksFailureFlag();
+ // Queue async reinitialization event instead of blocking call
+ boolean queued = reconTaskController.queueReInitializationEvent(reason);
Review Comment:
for full snapshot also, we can call, same flow of reinit Event and be non-blocking
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OMUpdateEventBuffer.java:
##########
@@ -43,41 +43,41 @@ public OMUpdateEventBuffer(int maxCapacity) {
}
/**
- * Add an event batch to the buffer.
+ * Add an event to the buffer.
*
- * @param eventBatch The event batch to buffer
+ * @param event The event to buffer
* @return true if successfully buffered, false if queue full
*/
- public boolean offer(OMUpdateEventBatch eventBatch) {
- boolean added = eventQueue.offer(eventBatch);
+ public boolean offer(ReconEvent event) {
+ boolean added = eventQueue.offer(event);
if (added) {
- totalBufferedEvents.addAndGet(eventBatch.getEvents().size());
- LOG.debug("Buffered event batch with {} events. Queue size: {}, Total buffered events: {}",
- eventBatch.getEvents().size(), eventQueue.size(), totalBufferedEvents.get());
+ totalBufferedEvents.addAndGet(event.getEventCount());
+ LOG.debug("Buffered event {} with {} events. Queue size: {}, Total buffered events: {}",
Review Comment:
bufferedEvents and queue size has no relation. 1 Entry in queue of EventType Update will have multiple updates. Check if this log may be misleading.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java:
##########
@@ -672,21 +676,30 @@ public boolean syncDataFromOM() {
boolean tasksFailed = reconTaskController.hasDeltaTasksFailed();
if (bufferOverflowed || tasksFailed) {
- String reason = bufferOverflowed ? "Event buffer overflow" : "Delta tasks failed after retry";
- LOG.warn("{}, triggering task reinitialization", reason);
+ ReconTaskReInitializationEvent.ReInitializationReason reason = bufferOverflowed ?
+ ReconTaskReInitializationEvent.ReInitializationReason.BUFFER_OVERFLOW :
+ ReconTaskReInitializationEvent.ReInitializationReason.TASK_FAILURES;
+
+ LOG.warn("Detected condition for task reinitialization: {}, queueing async reinitialization event",
+ reason);
metrics.incrNumDeltaRequestsFailed();
deltaReconTaskStatusUpdater.setLastTaskRunStatus(-1);
deltaReconTaskStatusUpdater.recordRunCompletion();
- reconTaskController.reInitializeTasks(omMetadataManager, null);
-
- // Reset appropriate flags after reinitialization
- if (bufferOverflowed) {
- reconTaskController.resetEventBufferOverflowFlag();
- }
- if (tasksFailed) {
- reconTaskController.resetDeltaTasksFailureFlag();
+ // Queue async reinitialization event instead of blocking call
+ boolean queued = reconTaskController.queueReInitializationEvent(reason);
+ if (!queued) {
Review Comment:
IMO, queue must not be a problem as before adding to queue, queue can be cleaned up.
and as part of this queueReInitializationEvent() in ReconTaskController:
1. cleanup queue
2. add to queue
3. reset buffer/failure flag
And not need expose this reset to OMServiceProviderImpl
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -431,4 +450,74 @@ 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);
+
+ // Clear the buffer to discard all pending events
+ eventBuffer.clear();
+
+ // Create and queue the reinitialization event
+ ReconTaskReInitializationEvent reinitEvent = new ReconTaskReInitializationEvent(reason);
+ boolean queued = eventBuffer.offer(reinitEvent);
+
+ if (queued) {
+ LOG.info("Successfully queued reinitialization event");
+ } else {
+ LOG.error("Failed to queue reinitialization event - buffer is full");
+ }
+
+ return queued;
+ }
+
+ @Override
+ public void updateOMMetadataManager(ReconOMMetadataManager omMetadataManager) {
+ this.currentOMMetadataManager = omMetadataManager;
+ }
+
+ /**
+ * 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 current OM metadata manager for reinitialization
+ if (currentOMMetadataManager != null) {
+ LOG.info("Starting async task reinitialization due to: {}", event.getReason());
+ reInitializeTasks(currentOMMetadataManager, null);
+ LOG.info("Completed async task reinitialization");
+ } else {
+ LOG.warn("Current OM metadata manager is null, cannot perform reinitialization");
+ deltaTasksFailed.set(true);
+ return;
+ }
+
+ // Reset appropriate flags based on the reason
+ if (event.getReason() == ReconTaskReInitializationEvent.ReInitializationReason.BUFFER_OVERFLOW) {
Review Comment:
reset should be done for all cases as cleanup after success.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskReInitializationEvent.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.recon.tasks;
+
+/**
+ * Custom event to trigger task reinitialization asynchronously.
+ * This event is generated when the event buffer overflows or when tasks fail
+ * and need reinitialization.
+ */
+public class ReconTaskReInitializationEvent implements ReconEvent {
+
+ private final ReInitializationReason reason;
Review Comment:
As this will hold OM checkpoint reference, to be provided to task, there can be multiple checkpoint
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -431,4 +450,74 @@ 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);
+
+ // Clear the buffer to discard all pending events
+ eventBuffer.clear();
+
+ // Create and queue the reinitialization event
+ ReconTaskReInitializationEvent reinitEvent = new ReconTaskReInitializationEvent(reason);
Review Comment:
should create a checkpoint of OM db before putting to queue, so that parallel db updates can happen. And need use same for all cases to be updated in task.
--
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
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
Posted by "devmadhuu (via GitHub)" <gi...@apache.org>.
devmadhuu commented on code in PR #8962:
URL: https://github.com/apache/ozone/pull/8962#discussion_r2318998626
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -423,12 +463,348 @@ public void resetEventBufferOverflowFlag() {
}
@Override
- public boolean hasDeltaTasksFailed() {
- return deltaTasksFailed.get();
+ public boolean hasTasksFailed() {
+ return tasksFailed.get();
+ }
+
+ @Override
+ public void resetTasksFailureFlag() {
+ tasksFailed.compareAndSet(true, false);
+ }
+
+ @Override
+ public synchronized ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (retry attempt count: {})", reason,
+ eventProcessRetryCount);
+
+ Long currentTime = validateRetryCountAndDelay();
+ if (currentTime == null) {
+ return ReInitializationResult.RETRY_LATER;
+ }
+
+ // Drain all events in buffer and cleanup any existing checkpoints before falling back to full snapshot.
+ // Events can be present in queue when reinit checkpoint creation fails multiple times because only after
+ // successful creation of checkpoint, we are clearing the event buffer.
+ drainEventBufferAndCleanExistingCheckpoints();
+
+ // Try checkpoint creation (single attempt per iteration)
+ ReconOMMetadataManager checkpointedOMMetadataManager = null;
+
+ try {
+ LOG.info("Attempting checkpoint creation (retry attempt: {})", eventProcessRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ LOG.info("Checkpoint creation succeeded");
+ } catch (IOException e) {
+ LOG.error("Checkpoint creation failed: {}", e.getMessage());
+ handleEventFailure(currentTime);
+ return getEventRetryResult();
+ }
+
+ // 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();
+ resetRetryCounters();
Review Comment:
Want to separate retry times for queuing and reprocess, that is why resetting here also.
--
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
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
Posted by "devmadhuu (via GitHub)" <gi...@apache.org>.
devmadhuu commented on code in PR #8962:
URL: https://github.com/apache/ozone/pull/8962#discussion_r2318998626
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -423,12 +463,348 @@ public void resetEventBufferOverflowFlag() {
}
@Override
- public boolean hasDeltaTasksFailed() {
- return deltaTasksFailed.get();
+ public boolean hasTasksFailed() {
+ return tasksFailed.get();
+ }
+
+ @Override
+ public void resetTasksFailureFlag() {
+ tasksFailed.compareAndSet(true, false);
+ }
+
+ @Override
+ public synchronized ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (retry attempt count: {})", reason,
+ eventProcessRetryCount);
+
+ Long currentTime = validateRetryCountAndDelay();
+ if (currentTime == null) {
+ return ReInitializationResult.RETRY_LATER;
+ }
+
+ // Drain all events in buffer and cleanup any existing checkpoints before falling back to full snapshot.
+ // Events can be present in queue when reinit checkpoint creation fails multiple times because only after
+ // successful creation of checkpoint, we are clearing the event buffer.
+ drainEventBufferAndCleanExistingCheckpoints();
+
+ // Try checkpoint creation (single attempt per iteration)
+ ReconOMMetadataManager checkpointedOMMetadataManager = null;
+
+ try {
+ LOG.info("Attempting checkpoint creation (retry attempt: {})", eventProcessRetryCount + 1);
+ checkpointedOMMetadataManager = createOMCheckpoint(currentOMMetadataManager);
+ LOG.info("Checkpoint creation succeeded");
+ } catch (IOException e) {
+ LOG.error("Checkpoint creation failed: {}", e.getMessage());
+ handleEventFailure(currentTime);
+ return getEventRetryResult();
+ }
+
+ // 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();
+ resetRetryCounters();
Review Comment:
Want to separate retry times for queuing and reprocess, that is why resetting here also.
--
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
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
Posted by "sumitagrawl (via GitHub)" <gi...@apache.org>.
sumitagrawl commented on code in PR #8962:
URL: https://github.com/apache/ozone/pull/8962#discussion_r2322180470
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/recovery/ReconOmMetadataManagerImpl.java:
##########
@@ -72,6 +74,31 @@ public ReconOmMetadataManagerImpl(OzoneConfiguration configuration,
this.ozoneConfiguration = configuration;
}
+ private ReconOmMetadataManagerImpl(OzoneConfiguration configuration, File dir, String name, ReconUtils reconUtils)
+ throws IOException {
+ super(configuration, dir, name);
+ this.reconUtils = reconUtils;
+ this.ozoneConfiguration = configuration;
+ }
+
+ @Override
+ public ReconOMMetadataManager createCheckpointReconMetadataManager(
+ OzoneConfiguration conf, DBCheckpoint checkpoint) throws IOException {
+ Path path = checkpoint.getCheckpointLocation();
+ Path parent = path.getParent();
+ if (parent == null) {
+ throw new IllegalStateException("DB checkpoint parent path should not "
Review Comment:
check if can use IOException instead of IllegalStateException
##########
hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/TestReconOmMetaManagerUtils.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.recon;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.hadoop.ozone.recon.tasks.OMUpdateEventBuffer;
+
+/**
+ * Test Recon Utility methods.
+ */
+public class TestReconOmMetaManagerUtils {
+
+ /**
+ * 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
+ */
+ public CompletableFuture<Void> waitForEventBufferEmpty(OMUpdateEventBuffer eventBuffer) {
Review Comment:
can use GenericTestUtils.waitFor() here
##########
hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/TestReconContainerEndpoint.java:
##########
@@ -107,6 +110,16 @@ public void testContainerEndpointForFSOLayout() throws Exception {
recon.getReconServer().getOzoneManagerServiceProvider();
impl.syncDataFromOM();
+ // Wait for async event processing to complete
+ // Events are processed asynchronously, so wait for processing to finish
+ ReconTaskControllerImpl reconTaskController =
+ (ReconTaskControllerImpl) recon.getReconServer().getReconTaskController();
+ CompletableFuture<Void> completableFuture =
+ omMetaManagerUtils.waitForEventBufferEmpty(reconTaskController.getEventBuffer());
+ while (!completableFuture.isDone()) {
Review Comment:
we can use GenericTestUtils.waitFor() for this case, instead of infinite loop. similar changes can be done other places
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -416,19 +459,363 @@ private void processEventBatchDirectly(OMUpdateEventBatch events) {
public boolean hasEventBufferOverflowed() {
return eventBuffer.getDroppedBatches() > 0;
}
-
- @Override
+
+ /**
+ * Reset the event buffer overflow flag after full snapshot is completed.
+ */
public void resetEventBufferOverflowFlag() {
eventBuffer.resetDroppedBatches();
}
@Override
- public boolean hasDeltaTasksFailed() {
- return deltaTasksFailed.get();
+ public boolean hasTasksFailed() {
+ return tasksFailed.get();
+ }
+
+ /**
+ * Reset the task(s) failure flag after reinitialization is completed.
+ */
+ public void resetTasksFailureFlag() {
+ tasksFailed.compareAndSet(true, false);
+ }
+
+ @Override
+ public synchronized ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (retry attempt count: {})", reason,
+ eventProcessRetryCount.get());
+
+ ReInitializationResult reInitializationResult = validateRetryCountAndDelay();
+ if (null != reInitializationResult) {
+ return reInitializationResult;
+ }
+
+ // Drain all events in buffer and cleanup any existing checkpoints before falling back to full snapshot.
+ // Events can be present in queue when reinit checkpoint creation fails multiple times because only after
+ // successful creation of checkpoint, we are clearing the event buffer.
+ drainEventBufferAndCleanExistingCheckpoints();
+
+ // Try checkpoint creation (single attempt per iteration)
+ ReconOMMetadataManager checkpointedOMMetadataManager = null;
+
+ 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());
+ handleEventFailure();
+ return ReInitializationResult.RETRY_LATER;
+ }
+
+ // 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() {
+ // Check if we should retry based on timing for iteration-based retries
+ long currentTime = System.currentTimeMillis();
+ if (eventProcessRetryCount.get() > 0) {
+ // Check if 2 seconds have passed since last iteration
+ long timeSinceLastRetry = currentTime - lastRetryTimestamp.get();
+ if (timeSinceLastRetry < RETRY_DELAY_MS) {
+ LOG.debug("Skipping retry, only {}ms since last retry attempt (need {}ms)",
+ timeSinceLastRetry, RETRY_DELAY_MS);
+ return ReInitializationResult.RETRY_LATER;
+ }
+ LOG.info("Attempting retry (retry attempt count: {}, delay: {}ms)",
+ eventProcessRetryCount.get() + 1, timeSinceLastRetry);
+ }
+ return getEventRetryResult();
+ }
+
+ /**
+ * Handle iteration failure by updating retry counters.
+ */
+ private void handleEventFailure() {
+ long currentTime = System.currentTimeMillis();
+ lastRetryTimestamp.set(currentTime);
+ eventProcessRetryCount.getAndIncrement();
+ tasksFailed.compareAndSet(false, true);
+ LOG.error("Event processing failed {} times.", eventProcessRetryCount);
+ }
+
+ /**
+ * Determine the appropriate retry result based on current event retry count.
+ */
+ private ReconTaskController.ReInitializationResult getEventRetryResult() {
+ if (eventProcessRetryCount.get() >= MAX_EVENT_PROCESS_RETRIES) {
+ LOG.warn("Maximum iteration retries ({}) exceeded, resetting counters and signaling full snapshot fallback",
+ MAX_EVENT_PROCESS_RETRIES);
+ resetRetryCounters();
+ return ReconTaskController.ReInitializationResult.MAX_RETRIES_EXCEEDED;
+ }
+ return null;
+ }
+
+ public void drainEventBufferAndCleanExistingCheckpoints() {
+ // First drain all events to check for any ReconTaskReInitializationEvent that need checkpoint cleanup
+ List<ReconEvent> drainedEvents = new ArrayList<>();
+ int drainedCount = eventBuffer.drainTo(drainedEvents);
+
+ if (drainedCount > 0) {
+ LOG.info("Drained {} events from buffer before clearing. Checking for checkpoint cleanup.", drainedCount);
+
+ // Check for any ReconTaskReInitializationEvent and cleanup their checkpoints
+ for (ReconEvent event : drainedEvents) {
+ if (event instanceof ReconTaskReInitializationEvent) {
+ ReconTaskReInitializationEvent reinitEvent = (ReconTaskReInitializationEvent) event;
+ ReconOMMetadataManager checkpointedManager = reinitEvent.getCheckpointedOMMetadataManager();
+ if (checkpointedManager != null) {
+ LOG.info("Cleaning up unprocessed checkpoint from drained ReconTaskReInitializationEvent");
+ cleanupCheckpoint(checkpointedManager);
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public void updateOMMetadataManager(ReconOMMetadataManager omMetadataManager) {
+ LOG.debug("Updating OM metadata manager");
+ this.currentOMMetadataManager = omMetadataManager;
+ }
+
+ /**
+ * Create a checkpoint of the current OM metadata manager.
+ * This method creates a snapshot of the current OM database state
+ * to prevent data inconsistency during reinitialization.
+ *
+ * @param omMetaManager the OM metadata manager to checkpoint
+ * @return a checkpointed ReconOMMetadataManager instance
+ * @throws IOException if checkpoint creation fails
+ */
+ 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);
+
+ return omMetaManager.createCheckpointReconMetadataManager(configuration, checkpoint);
+ }
+
+ /**
+ * 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");
Review Comment:
here NullPointerException, but signature tells IOException. May be we can use IOException for all cases
--
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
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
Posted by "devmadhuu (via GitHub)" <gi...@apache.org>.
devmadhuu commented on code in PR #8962:
URL: https://github.com/apache/ozone/pull/8962#discussion_r2299966706
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OMUpdateEventBuffer.java:
##########
@@ -43,41 +43,41 @@ public OMUpdateEventBuffer(int maxCapacity) {
}
/**
- * Add an event batch to the buffer.
+ * Add an event to the buffer.
*
- * @param eventBatch The event batch to buffer
+ * @param event The event to buffer
* @return true if successfully buffered, false if queue full
*/
- public boolean offer(OMUpdateEventBatch eventBatch) {
- boolean added = eventQueue.offer(eventBatch);
+ public boolean offer(ReconEvent event) {
+ boolean added = eventQueue.offer(event);
if (added) {
- totalBufferedEvents.addAndGet(eventBatch.getEvents().size());
- LOG.debug("Buffered event batch with {} events. Queue size: {}, Total buffered events: {}",
- eventBatch.getEvents().size(), eventQueue.size(), totalBufferedEvents.get());
+ totalBufferedEvents.addAndGet(event.getEventCount());
+ LOG.debug("Buffered event {} with {} events. Queue size: {}, Total buffered events: {}",
Review Comment:
No, we are counting the size of events here in `org.apache.hadoop.ozone.recon.tasks.OMUpdateEventBatch#getEventCount`
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java:
##########
@@ -672,21 +676,30 @@ public boolean syncDataFromOM() {
boolean tasksFailed = reconTaskController.hasDeltaTasksFailed();
if (bufferOverflowed || tasksFailed) {
- String reason = bufferOverflowed ? "Event buffer overflow" : "Delta tasks failed after retry";
- LOG.warn("{}, triggering task reinitialization", reason);
+ ReconTaskReInitializationEvent.ReInitializationReason reason = bufferOverflowed ?
+ ReconTaskReInitializationEvent.ReInitializationReason.BUFFER_OVERFLOW :
+ ReconTaskReInitializationEvent.ReInitializationReason.TASK_FAILURES;
+
+ LOG.warn("Detected condition for task reinitialization: {}, queueing async reinitialization event",
+ reason);
metrics.incrNumDeltaRequestsFailed();
deltaReconTaskStatusUpdater.setLastTaskRunStatus(-1);
deltaReconTaskStatusUpdater.recordRunCompletion();
- reconTaskController.reInitializeTasks(omMetadataManager, null);
-
- // Reset appropriate flags after reinitialization
- if (bufferOverflowed) {
- reconTaskController.resetEventBufferOverflowFlag();
- }
- if (tasksFailed) {
- reconTaskController.resetDeltaTasksFailureFlag();
+ // Queue async reinitialization event instead of blocking call
+ boolean queued = reconTaskController.queueReInitializationEvent(reason);
Review Comment:
done.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java:
##########
@@ -672,21 +676,30 @@ public boolean syncDataFromOM() {
boolean tasksFailed = reconTaskController.hasDeltaTasksFailed();
if (bufferOverflowed || tasksFailed) {
- String reason = bufferOverflowed ? "Event buffer overflow" : "Delta tasks failed after retry";
- LOG.warn("{}, triggering task reinitialization", reason);
+ ReconTaskReInitializationEvent.ReInitializationReason reason = bufferOverflowed ?
+ ReconTaskReInitializationEvent.ReInitializationReason.BUFFER_OVERFLOW :
+ ReconTaskReInitializationEvent.ReInitializationReason.TASK_FAILURES;
+
+ LOG.warn("Detected condition for task reinitialization: {}, queueing async reinitialization event",
+ reason);
metrics.incrNumDeltaRequestsFailed();
deltaReconTaskStatusUpdater.setLastTaskRunStatus(-1);
deltaReconTaskStatusUpdater.recordRunCompletion();
- reconTaskController.reInitializeTasks(omMetadataManager, null);
-
- // Reset appropriate flags after reinitialization
- if (bufferOverflowed) {
- reconTaskController.resetEventBufferOverflowFlag();
- }
- if (tasksFailed) {
- reconTaskController.resetDeltaTasksFailureFlag();
+ // Queue async reinitialization event instead of blocking call
+ boolean queued = reconTaskController.queueReInitializationEvent(reason);
+ if (!queued) {
Review Comment:
done.
--
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
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
Posted by "devmadhuu (via GitHub)" <gi...@apache.org>.
devmadhuu commented on code in PR #8962:
URL: https://github.com/apache/ozone/pull/8962#discussion_r2299967250
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -431,4 +450,74 @@ 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);
+
+ // Clear the buffer to discard all pending events
+ eventBuffer.clear();
+
+ // Create and queue the reinitialization event
+ ReconTaskReInitializationEvent reinitEvent = new ReconTaskReInitializationEvent(reason);
+ boolean queued = eventBuffer.offer(reinitEvent);
+
+ if (queued) {
+ LOG.info("Successfully queued reinitialization event");
+ } else {
+ LOG.error("Failed to queue reinitialization event - buffer is full");
+ }
+
+ return queued;
+ }
+
+ @Override
+ public void updateOMMetadataManager(ReconOMMetadataManager omMetadataManager) {
+ this.currentOMMetadataManager = omMetadataManager;
+ }
+
+ /**
+ * 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 current OM metadata manager for reinitialization
+ if (currentOMMetadataManager != null) {
+ LOG.info("Starting async task reinitialization due to: {}", event.getReason());
+ reInitializeTasks(currentOMMetadataManager, null);
+ LOG.info("Completed async task reinitialization");
+ } else {
+ LOG.warn("Current OM metadata manager is null, cannot perform reinitialization");
+ deltaTasksFailed.set(true);
+ return;
+ }
+
+ // Reset appropriate flags based on the reason
+ if (event.getReason() == ReconTaskReInitializationEvent.ReInitializationReason.BUFFER_OVERFLOW) {
Review Comment:
done.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskReInitializationEvent.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.recon.tasks;
+
+/**
+ * Custom event to trigger task reinitialization asynchronously.
+ * This event is generated when the event buffer overflows or when tasks fail
+ * and need reinitialization.
+ */
+public class ReconTaskReInitializationEvent implements ReconEvent {
+
+ private final ReInitializationReason reason;
Review Comment:
done.
--
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
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
Posted by "devmadhuu (via GitHub)" <gi...@apache.org>.
devmadhuu commented on code in PR #8962:
URL: https://github.com/apache/ozone/pull/8962#discussion_r2315815056
##########
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:
done.
--
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
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
Posted by "sumitagrawl (via GitHub)" <gi...@apache.org>.
sumitagrawl merged PR #8962:
URL: https://github.com/apache/ozone/pull/8962
--
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
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
Posted by "sumitagrawl (via GitHub)" <gi...@apache.org>.
sumitagrawl commented on code in PR #8962:
URL: https://github.com/apache/ozone/pull/8962#discussion_r2320913596
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -416,19 +459,358 @@ private void processEventBatchDirectly(OMUpdateEventBatch events) {
public boolean hasEventBufferOverflowed() {
return eventBuffer.getDroppedBatches() > 0;
}
-
- @Override
+
+ /**
+ * Reset the event buffer overflow flag after full snapshot is completed.
+ */
public void resetEventBufferOverflowFlag() {
eventBuffer.resetDroppedBatches();
}
@Override
- public boolean hasDeltaTasksFailed() {
- return deltaTasksFailed.get();
+ public boolean hasTasksFailed() {
+ return tasksFailed.get();
+ }
+
+ /**
+ * Reset the task(s) failure flag after reinitialization is completed.
+ */
+ public void resetTasksFailureFlag() {
+ tasksFailed.compareAndSet(true, false);
+ }
+
+ @Override
+ public synchronized ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (retry attempt count: {})", reason,
+ eventProcessRetryCount.get());
+
+ ReInitializationResult reInitializationResult = validateRetryCountAndDelay();
+ if (reInitializationResult == ReInitializationResult.MAX_RETRIES_EXCEEDED) {
+ return reInitializationResult;
+ }
+
+ // Drain all events in buffer and cleanup any existing checkpoints before falling back to full snapshot.
+ // Events can be present in queue when reinit checkpoint creation fails multiple times because only after
+ // successful creation of checkpoint, we are clearing the event buffer.
+ drainEventBufferAndCleanExistingCheckpoints();
+
+ // Try checkpoint creation (single attempt per iteration)
+ ReconOMMetadataManager checkpointedOMMetadataManager = null;
+
+ 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());
+ handleEventFailure();
+ return ReInitializationResult.RETRY_LATER;
+ }
+
+ // 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() {
+ // Check if we should retry based on timing for iteration-based retries
+ long currentTime = System.currentTimeMillis();
+ if (eventProcessRetryCount.get() > 0) {
+ // Check if 2 seconds have passed since last iteration
+ long timeSinceLastRetry = currentTime - lastRetryTimestamp.get();
+ if (timeSinceLastRetry < RETRY_DELAY_MS) {
+ LOG.debug("Skipping retry, only {}ms since last retry attempt (need {}ms)",
+ timeSinceLastRetry, RETRY_DELAY_MS);
+ return ReInitializationResult.RETRY_LATER;
+ }
+ LOG.info("Attempting retry (retry attempt count: {}, delay: {}ms)",
+ eventProcessRetryCount.get() + 1, timeSinceLastRetry);
+ }
+ return getEventRetryResult();
+ }
+
+ /**
+ * Handle iteration failure by updating retry counters.
+ */
+ private void handleEventFailure() {
+ long currentTime = System.currentTimeMillis();
+ lastRetryTimestamp.set(currentTime);
+ eventProcessRetryCount.getAndIncrement();
+ LOG.error("Event processing failed {} times.", eventProcessRetryCount);
+ }
+
+ /**
+ * Determine the appropriate retry result based on current event retry count.
+ */
+ private ReconTaskController.ReInitializationResult getEventRetryResult() {
+ if (eventProcessRetryCount.get() >= MAX_EVENT_PROCESS_RETRIES) {
+ LOG.warn("Maximum iteration retries ({}) exceeded, resetting counters and signaling full snapshot fallback",
+ MAX_EVENT_PROCESS_RETRIES);
+ resetRetryCounters();
+ return ReconTaskController.ReInitializationResult.MAX_RETRIES_EXCEEDED;
+ }
+ return null;
+ }
+
+ public void drainEventBufferAndCleanExistingCheckpoints() {
+ // First drain all events to check for any ReconTaskReInitializationEvent that need checkpoint cleanup
+ List<ReconEvent> drainedEvents = new ArrayList<>();
+ int drainedCount = eventBuffer.drainTo(drainedEvents);
+
+ if (drainedCount > 0) {
+ LOG.info("Drained {} events from buffer before clearing. Checking for checkpoint cleanup.", drainedCount);
+
+ // Check for any ReconTaskReInitializationEvent and cleanup their checkpoints
+ for (ReconEvent event : drainedEvents) {
+ if (event instanceof ReconTaskReInitializationEvent) {
+ ReconTaskReInitializationEvent reinitEvent = (ReconTaskReInitializationEvent) event;
+ ReconOMMetadataManager checkpointedManager = reinitEvent.getCheckpointedOMMetadataManager();
+ if (checkpointedManager != null) {
+ LOG.info("Cleaning up unprocessed checkpoint from drained ReconTaskReInitializationEvent");
+ cleanupCheckpoint(checkpointedManager);
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public void updateOMMetadataManager(ReconOMMetadataManager omMetadataManager) {
+ LOG.info("Updating OM metadata manager");
+ this.currentOMMetadataManager = omMetadataManager;
+ }
+
+ /**
+ * Create a checkpoint of the current OM metadata manager.
+ * This method creates a snapshot of the current OM database state
+ * to prevent data inconsistency during reinitialization.
+ *
+ * @param omMetaManager the OM metadata manager to checkpoint
+ * @return a checkpointed ReconOMMetadataManager instance
+ * @throws IOException if checkpoint creation fails
+ */
+ 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);
+
+ return omMetaManager.createCheckpointReconMetadataManager(configuration, checkpoint);
+ }
+
+ /**
+ * 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" + "_" + UUID.randomUUID()).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());
+ resetTasksFailureFlag();
+ // Use the checkpointed OM metadata manager for reinitialization to prevent data inconsistency
+ ReconOMMetadataManager checkpointedOMMetadataManager = event.getCheckpointedOMMetadataManager();
+ try {
+ if (checkpointedOMMetadataManager != null) {
+ LOG.info("Starting async task reinitialization with checkpointed OM metadata manager due to: {}",
+ event.getReason());
+ boolean isRunSuccessful = reInitializeTasks(checkpointedOMMetadataManager, null);
+ if (!isRunSuccessful) {
+ // Setting this taskFailed flag as true here will block consuming delta events and stop buffering events
+ // in eventBuffer until we successfully queue a new reinit event again.
+ tasksFailed.compareAndSet(false, true);
+ handleEventFailure();
+ LOG.error("Task reinitialization failed, tasksFailed flag set to true");
+ } else {
+ resetEventFlags();
Review Comment:
remove resetEventFlags() here, as there may be next failure waiting. Its reset when put to queue is success.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -416,19 +459,358 @@ private void processEventBatchDirectly(OMUpdateEventBatch events) {
public boolean hasEventBufferOverflowed() {
return eventBuffer.getDroppedBatches() > 0;
}
-
- @Override
+
+ /**
+ * Reset the event buffer overflow flag after full snapshot is completed.
+ */
public void resetEventBufferOverflowFlag() {
eventBuffer.resetDroppedBatches();
}
@Override
- public boolean hasDeltaTasksFailed() {
- return deltaTasksFailed.get();
+ public boolean hasTasksFailed() {
+ return tasksFailed.get();
+ }
+
+ /**
+ * Reset the task(s) failure flag after reinitialization is completed.
+ */
+ public void resetTasksFailureFlag() {
+ tasksFailed.compareAndSet(true, false);
+ }
+
+ @Override
+ public synchronized ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (retry attempt count: {})", reason,
+ eventProcessRetryCount.get());
+
+ ReInitializationResult reInitializationResult = validateRetryCountAndDelay();
+ if (reInitializationResult == ReInitializationResult.MAX_RETRIES_EXCEEDED) {
+ return reInitializationResult;
+ }
+
+ // Drain all events in buffer and cleanup any existing checkpoints before falling back to full snapshot.
+ // Events can be present in queue when reinit checkpoint creation fails multiple times because only after
+ // successful creation of checkpoint, we are clearing the event buffer.
+ drainEventBufferAndCleanExistingCheckpoints();
+
+ // Try checkpoint creation (single attempt per iteration)
+ ReconOMMetadataManager checkpointedOMMetadataManager = null;
+
+ 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());
+ handleEventFailure();
+ return ReInitializationResult.RETRY_LATER;
+ }
+
+ // 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() {
+ // Check if we should retry based on timing for iteration-based retries
+ long currentTime = System.currentTimeMillis();
+ if (eventProcessRetryCount.get() > 0) {
+ // Check if 2 seconds have passed since last iteration
+ long timeSinceLastRetry = currentTime - lastRetryTimestamp.get();
+ if (timeSinceLastRetry < RETRY_DELAY_MS) {
+ LOG.debug("Skipping retry, only {}ms since last retry attempt (need {}ms)",
+ timeSinceLastRetry, RETRY_DELAY_MS);
+ return ReInitializationResult.RETRY_LATER;
+ }
+ LOG.info("Attempting retry (retry attempt count: {}, delay: {}ms)",
+ eventProcessRetryCount.get() + 1, timeSinceLastRetry);
+ }
+ return getEventRetryResult();
+ }
+
+ /**
+ * Handle iteration failure by updating retry counters.
+ */
+ private void handleEventFailure() {
+ long currentTime = System.currentTimeMillis();
+ lastRetryTimestamp.set(currentTime);
+ eventProcessRetryCount.getAndIncrement();
+ LOG.error("Event processing failed {} times.", eventProcessRetryCount);
+ }
+
+ /**
+ * Determine the appropriate retry result based on current event retry count.
+ */
+ private ReconTaskController.ReInitializationResult getEventRetryResult() {
+ if (eventProcessRetryCount.get() >= MAX_EVENT_PROCESS_RETRIES) {
+ LOG.warn("Maximum iteration retries ({}) exceeded, resetting counters and signaling full snapshot fallback",
+ MAX_EVENT_PROCESS_RETRIES);
+ resetRetryCounters();
+ return ReconTaskController.ReInitializationResult.MAX_RETRIES_EXCEEDED;
+ }
+ return null;
+ }
+
+ public void drainEventBufferAndCleanExistingCheckpoints() {
+ // First drain all events to check for any ReconTaskReInitializationEvent that need checkpoint cleanup
+ List<ReconEvent> drainedEvents = new ArrayList<>();
+ int drainedCount = eventBuffer.drainTo(drainedEvents);
+
+ if (drainedCount > 0) {
+ LOG.info("Drained {} events from buffer before clearing. Checking for checkpoint cleanup.", drainedCount);
+
+ // Check for any ReconTaskReInitializationEvent and cleanup their checkpoints
+ for (ReconEvent event : drainedEvents) {
+ if (event instanceof ReconTaskReInitializationEvent) {
+ ReconTaskReInitializationEvent reinitEvent = (ReconTaskReInitializationEvent) event;
+ ReconOMMetadataManager checkpointedManager = reinitEvent.getCheckpointedOMMetadataManager();
+ if (checkpointedManager != null) {
+ LOG.info("Cleaning up unprocessed checkpoint from drained ReconTaskReInitializationEvent");
+ cleanupCheckpoint(checkpointedManager);
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public void updateOMMetadataManager(ReconOMMetadataManager omMetadataManager) {
+ LOG.info("Updating OM metadata manager");
Review Comment:
there will be frequent info log as its called every time in while loop for delta update
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -416,19 +459,358 @@ private void processEventBatchDirectly(OMUpdateEventBatch events) {
public boolean hasEventBufferOverflowed() {
return eventBuffer.getDroppedBatches() > 0;
}
-
- @Override
+
+ /**
+ * Reset the event buffer overflow flag after full snapshot is completed.
+ */
public void resetEventBufferOverflowFlag() {
eventBuffer.resetDroppedBatches();
}
@Override
- public boolean hasDeltaTasksFailed() {
- return deltaTasksFailed.get();
+ public boolean hasTasksFailed() {
+ return tasksFailed.get();
+ }
+
+ /**
+ * Reset the task(s) failure flag after reinitialization is completed.
+ */
+ public void resetTasksFailureFlag() {
+ tasksFailed.compareAndSet(true, false);
+ }
+
+ @Override
+ public synchronized ReconTaskController.ReInitializationResult queueReInitializationEvent(
+ ReconTaskReInitializationEvent.ReInitializationReason reason) {
+ LOG.info("Queueing task reinitialization event due to: {} (retry attempt count: {})", reason,
+ eventProcessRetryCount.get());
+
+ ReInitializationResult reInitializationResult = validateRetryCountAndDelay();
+ if (reInitializationResult == ReInitializationResult.MAX_RETRIES_EXCEEDED) {
Review Comment:
it can return retry_later also, if null!= result, then we can return reInitializtionResult
--
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
Re: [PR] HDDS-13576. Recon reprocess of all tasks should be non-blocking. [ozone]
Posted by "devmadhuu (via GitHub)" <gi...@apache.org>.
devmadhuu commented on code in PR #8962:
URL: https://github.com/apache/ozone/pull/8962#discussion_r2315815056
##########
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:
done.
##########
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:
ok done.
--
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