chihsuan commented on code in PR #11262:
URL: https://github.com/apache/ozone/pull/11262#discussion_r4052391717
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -502,6 +502,65 @@ private void processTasks(
} catch (CancellationException ce) {
LOG.error("Some tasks were cancelled with exception", ce);
}
+
+ failedTasks.addAll(executionFailedTasks);
+
+ if (!successfulTasks.isEmpty()) {
+ // Make the derived-table RocksDB writes for this batch durable before
the task-status
+ // cursors are committed to Derby. The cursor rows are fsync-durable
while the RocksDB writes
+ // are not synced by default, so without this barrier a power loss can
leave the durable
+ // cursors ahead of the (lost) derived data. Startup reconciliation then
sees the derived
+ // and delta cursors at the same sequence number and never reprocesses,
permanently
+ // dropping the applied update. Coalescing the barrier to once per batch
syncs all successful
+ // task writes together before any cursor advances. If the sync fails,
leave the cursors
+ // unadvanced so the batch is reprocessed instead of recording an
un-durable success.
+ 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();
+ failedTasks.add(new ReconOmTask.TaskResult.Builder()
Review Comment:
Should we signal reinit here instead of re-running process()? I noticed the
task already committed this batch, so the retry applies the same events twice.
With a transient sync failure, a counting task ends up double counted and the
cursor still advances.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -502,6 +502,65 @@ private void processTasks(
} catch (CancellationException ce) {
LOG.error("Some tasks were cancelled with exception", ce);
}
+
+ failedTasks.addAll(executionFailedTasks);
+
+ if (!successfulTasks.isEmpty()) {
+ // Make the derived-table RocksDB writes for this batch durable before
the task-status
+ // cursors are committed to Derby. The cursor rows are fsync-durable
while the RocksDB writes
+ // are not synced by default, so without this barrier a power loss can
leave the durable
+ // cursors ahead of the (lost) derived data. Startup reconciliation then
sees the derived
+ // and delta cursors at the same sequence number and never reprocesses,
permanently
+ // dropping the applied update. Coalescing the barrier to once per batch
syncs all successful
+ // task writes together before any cursor advances. If the sync fails,
leave the cursors
+ // unadvanced so the batch is reprocessed instead of recording an
un-durable success.
+ 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();
+ failedTasks.add(new ReconOmTask.TaskResult.Builder()
+ .setTaskName(taskName)
+ .setSubTaskSeekPositions(result.getSubTaskSeekPositions())
+ .build());
+ ReconTaskStatusUpdater taskStatusUpdater =
+ taskStatusUpdaterManager.getTaskStatusUpdater(taskName);
+ taskStatusUpdater.setLastTaskRunStatus(-1);
+ taskStatusUpdater.recordRunCompletion();
+ }
+ }
+ }
+ }
+
+ /**
+ * Flushes and syncs the Recon derived-data RocksDB write-ahead log to
stable storage so the
+ * derived writes for the processed batch are durable before the task-status
cursor advances.
+ * Returns {@code false} if the sync fails, in which case the caller must
not advance the cursor,
+ * so that the cursor is never persisted ahead of the derived data.
+ */
+ private boolean syncReconDbLog() {
+ DBStore dbStore = reconDBProvider.getDbStore();
+ if (dbStore == null) {
+ return true;
Review Comment:
nit: Should a null dbStore return false here? Treating it as synced advances
the cursor without any durability guarantee, though I realize this is mostly
unreachable in practice.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -502,6 +502,65 @@ private void processTasks(
} catch (CancellationException ce) {
LOG.error("Some tasks were cancelled with exception", ce);
}
+
+ failedTasks.addAll(executionFailedTasks);
+
+ if (!successfulTasks.isEmpty()) {
+ // Make the derived-table RocksDB writes for this batch durable before
the task-status
+ // cursors are committed to Derby. The cursor rows are fsync-durable
while the RocksDB writes
+ // are not synced by default, so without this barrier a power loss can
leave the durable
+ // cursors ahead of the (lost) derived data. Startup reconciliation then
sees the derived
+ // and delta cursors at the same sequence number and never reprocesses,
permanently
+ // dropping the applied update. Coalescing the barrier to once per batch
syncs all successful
+ // task writes together before any cursor advances. If the sync fails,
leave the cursors
+ // unadvanced so the batch is reprocessed instead of recording an
un-durable success.
+ 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();
+ failedTasks.add(new ReconOmTask.TaskResult.Builder()
+ .setTaskName(taskName)
+ .setSubTaskSeekPositions(result.getSubTaskSeekPositions())
+ .build());
+ ReconTaskStatusUpdater taskStatusUpdater =
+ taskStatusUpdaterManager.getTaskStatusUpdater(taskName);
+ taskStatusUpdater.setLastTaskRunStatus(-1);
Review Comment:
nit: Would it be worth counting this as a failure in taskMetrics too? Right
now a sync failure is invisible to anyone alerting on the delta metrics.
--
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]
