Copilot commented on code in PR #11262:
URL: https://github.com/apache/ozone/pull/11262#discussion_r4050087562
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -473,8 +475,23 @@ private void processTasks(
// Track task delta processing success
taskMetrics.incrTaskDeltaProcessingSuccess(taskName);
- taskStatusUpdater.setLastTaskRunStatus(0);
-
taskStatusUpdater.setLastUpdatedSeqNumber(events.getLastSequenceNumber());
+ // Make the derived-table RocksDB writes for this batch durable
before the task-status
+ // cursor is committed to Derby. The cursor row is fsync-durable
while the RocksDB writes
+ // are not synced by default, so without this barrier a power loss
can leave the durable
+ // cursor 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. If the sync fails, leave the
cursor unadvanced so the
+ // batch is reprocessed instead of recording an un-durable success.
+ if (syncReconDbLog()) {
+ taskStatusUpdater.setLastTaskRunStatus(0);
+
taskStatusUpdater.setLastUpdatedSeqNumber(events.getLastSequenceNumber());
+ } else {
+ failedTasks.add(new ReconOmTask.TaskResult.Builder()
+ .setTaskName(taskName)
+ .setSubTaskSeekPositions(result.getSubTaskSeekPositions())
+ .build());
Review Comment:
`failedTasks` is an `ArrayList`, but these completion callbacks run
concurrently on `executorService`. If multiple tasks hit this new sync-failure
branch, concurrent `add` calls can lose entries, so some failed tasks may not
be retried. Make the collection thread-safe (and apply the same fix to the
existing add at line 469).
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -503,6 +520,27 @@ private void processTasks(
LOG.error("Some tasks were cancelled with exception", ce);
}
}
+
+ /**
+ * 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;
+ }
+ try {
+ dbStore.flushLog(true);
+ return true;
+ } catch (RocksDatabaseException e) {
+ LOG.error("Failed to sync Recon DB WAL before advancing task status
cursor; "
+ + "leaving cursor unadvanced so the batch is reprocessed.", e);
+ return false;
Review Comment:
The failure behavior introduced here is not covered: no test makes
`flushLog(true)` throw and verifies that the task cursor stays at its previous
sequence, status becomes `-1`, and the retry path is exercised. This is the
critical failure mode of the durability barrier and should be protected by a
unit test.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java:
##########
@@ -473,8 +475,23 @@ private void processTasks(
// Track task delta processing success
taskMetrics.incrTaskDeltaProcessingSuccess(taskName);
- taskStatusUpdater.setLastTaskRunStatus(0);
-
taskStatusUpdater.setLastUpdatedSeqNumber(events.getLastSequenceNumber());
+ // Make the derived-table RocksDB writes for this batch durable
before the task-status
+ // cursor is committed to Derby. The cursor row is fsync-durable
while the RocksDB writes
+ // are not synced by default, so without this barrier a power loss
can leave the durable
+ // cursor 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. If the sync fails, leave the
cursor unadvanced so the
+ // batch is reprocessed instead of recording an un-durable success.
+ if (syncReconDbLog()) {
+ taskStatusUpdater.setLastTaskRunStatus(0);
+
taskStatusUpdater.setLastUpdatedSeqNumber(events.getLastSequenceNumber());
Review Comment:
This callback runs once per successful task, so a batch with the six
registered Recon tasks can issue multiple `flushLog(true)` calls, each
potentially forcing an fsync on the same WAL. That adds avoidable durable-sync
latency and blocks task executor threads; coalesce the barrier after all task
writes complete, then commit the successful task cursors, or otherwise share
one sync per batch.
##########
hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestReconTaskControllerImpl.java:
##########
@@ -932,6 +935,46 @@ public void
testCreateOMCheckpointThrowsWhenCheckpointNull() throws Exception {
assertThrows(IOException.class, () ->
controller.createOMCheckpoint(omMetadataManager));
}
+ /**
+ * A successful delta batch must sync the derived-data RocksDB WAL to stable
storage before the
+ * fsync-durable task-status cursor is advanced. Without that barrier a
power loss can lose the
+ * un-synced derived write while the durable cursor survives, leaving the
cursor ahead of the
+ * data; startup reconciliation then sees equal cursors and never
reprocesses, permanently
+ * dropping the applied update.
+ */
+ @Test
+ public void testDerivedDbSyncedBeforeCursorAdvanceOnSuccess() throws
Exception {
+ ReconOmTask reconOmTaskMock = getMockTask("SyncBarrierTask");
+ when(reconOmTaskMock.process(any(OMUpdateEventBatch.class), anyMap()))
+ .thenReturn(new ReconOmTask.TaskResult.Builder()
+ .setTaskName("SyncBarrierTask").setTaskSuccess(true).build());
+ reconTaskController.registerTask(reconOmTaskMock);
+
+ OMUpdateEventBatch batch = mock(OMUpdateEventBatch.class);
+ when(batch.getLastSequenceNumber()).thenReturn(100L);
+ when(batch.isEmpty()).thenReturn(false);
+ when(batch.getEvents()).thenReturn(new ArrayList<>());
+
when(batch.getEventType()).thenReturn(ReconEvent.EventType.OM_UPDATE_BATCH);
+ when(batch.getEventCount()).thenReturn(1);
+
+ reconTaskController.consumeOMEvents(batch, mock(OMMetadataManager.class));
+
+ GenericTestUtils.waitFor(() -> {
+ try {
+ ReconTaskStatus status =
reconTaskStatusDao.findById("SyncBarrierTask");
+ return status != null && status.getLastTaskRunStatus() == 0
+ && status.getLastUpdatedSeqNumber() == 100L;
+ } catch (Exception e) {
+ return false;
+ }
+ }, 100, 5000);
+
+ // The derived-data RocksDB WAL must be synced (sync == true) before the
durable cursor advances.
+ verify(reconDbStore, times(1)).flushLog(true);
+ // A non-syncing flush would leave the write in page cache and reintroduce
the durability gap.
+ verify(reconDbStore, never()).flushLog(false);
Review Comment:
The new test only verifies that `flushLog(true)` was eventually called after
observing the committed status. A regression that calls `recordRunCompletion()`
first and then flushes could still pass if the flush wins this race, so the
required ordering is not actually asserted. Use an ordered verification or
record the flush and cursor-update events explicitly.
--
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]
