chihsuan commented on code in PR #11262:
URL: https://github.com/apache/ozone/pull/11262#discussion_r4056047408
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -554,7 +614,7 @@ private void processOMUpdateBatch(OMUpdateEventBatch
events) {
}
Collection<NamedCallableTask<ReconOmTask.TaskResult>> tasks = new
ArrayList<>();
- List<ReconOmTask.TaskResult> failedTasks = new ArrayList<>();
+ List<ReconOmTask.TaskResult> failedTasks =
Collections.synchronizedList(new ArrayList<>());
Review Comment:
nit: Could these two go back to plain `ArrayList`? I noticed `failedTasks`
and `retryFailedTasks` are now only touched on the calling thread after the
join. The synchronized wrapper may suggest a concurrent writer that no longer
exists.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -502,6 +502,66 @@ private void processTasks(
} catch (CancellationException ce) {
LOG.error("Some tasks were cancelled with exception", ce);
}
+
+ failedTasks.addAll(executionFailedTasks);
+
+ if (!successfulTasks.isEmpty()) {
+ // Sync derived-data RocksDB WAL before committing task-status cursors
to avoid durability gaps.
+ if (syncReconDbLog()) {
+ for (ReconOmTask.TaskResult result : successfulTasks) {
+ String taskName = result.getTaskName();
+ // Track task delta processing success
+ taskMetrics.incrTaskDeltaProcessingSuccess(taskName);
+
+ ReconTaskStatusUpdater taskStatusUpdater =
+ taskStatusUpdaterManager.getTaskStatusUpdater(taskName);
+ taskStatusUpdater.setLastTaskRunStatus(0);
+
taskStatusUpdater.setLastUpdatedSeqNumber(events.getLastSequenceNumber());
+ taskStatusUpdater.recordRunCompletion();
+ }
+ } else {
+ for (ReconOmTask.TaskResult result : successfulTasks) {
+ String taskName = result.getTaskName();
+ // Track task delta processing failure
+ taskMetrics.incrTaskDeltaProcessingFailures(taskName);
+
+ ReconTaskStatusUpdater taskStatusUpdater =
+ taskStatusUpdaterManager.getTaskStatusUpdater(taskName);
+ taskStatusUpdater.setLastTaskRunStatus(-1);
+ taskStatusUpdater.recordRunCompletion();
+ }
+ // Signal task reinitialization directly instead of retrying process(),
+ // because task writes were already applied and retrying would
re-apply non-idempotent events.
+ tasksFailed.compareAndSet(false, true);
Review Comment:
nit: Would it be worth folding these two loops into one? They only differ in
the status code and the metric bumped, so one loop keyed on the sync result
might read easier. I don't feel strongly either way.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
