This is an automated email from the ASF dual-hosted git repository. danny0405 pushed a commit to branch release-1.1.2 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 0bdeb30e84fda106656907e9b4d8454be33d82f2 Author: Danny Chan <[email protected]> AuthorDate: Thu Oct 15 12:45:52 2026 +0800 fix(flink): resend recovered write metadata on task retries (#20323) * fix(flink): resend recovered write metadata on task retries * fix(flink): ignore late bootstrap events for completed instants * fix(flink): mark checkpointed write metadata as the last batch * fix(flink): preserve received bootstrap metadata on retries (cherry picked from commit 0170359e59d30a5f09ccadd47e796763c653f395) --- .../hudi/sink/StreamWriteOperatorCoordinator.java | 36 +++++++++---- .../sink/common/AbstractStreamWriteFunction.java | 30 +++++------ .../sink/TestStreamWriteOperatorCoordinator.java | 63 ++++++++++++++++++++++ .../org/apache/hudi/sink/TestWriteCopyOnWrite.java | 13 +++-- .../common/TestAbstractStreamWriteFunction.java | 14 ++--- 5 files changed, 115 insertions(+), 41 deletions(-) diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java index 6b7efe3f5822..a43c98cf766c 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java @@ -95,16 +95,21 @@ import static org.apache.hudi.util.StreamerUtil.initTableIfNotExists; * * <p>Task failover workflow: * <pre> - * task failover - * | - * initially started ? - * /(yes) \(no) - * / \ - * has pending send an empty - * instant event ? bootstrap event - * /(yes) \(no) - * / \ - * resend no-op + * task failover + * | + * restored pending instant event ? + * /(yes) \(no) + * / \ + * resend metadata | + * on every attempt | + * \ / + * \ / + * attempt > 0 ? + * /(yes) \(no) + * / \ + * send an empty no-op + * bootstrap event + * to clear legacy events * </pre> * * @see StreamWriteFunction for the data inputs checkpointing and semantics @@ -534,8 +539,17 @@ public class StreamWriteOperatorCoordinator this.eventBuffers.cleanLegacyEvents(event); return; } + // The writer may have observed a pending instant before its normal commit finished. + // Do not recreate the committed buffer when that bootstrap event arrives later. + if (this.eventBuffers.getEventBuffer(event.getCheckpointId()) == null + && this.metaClient.reloadActiveTimeline().filterCompletedInstants().containsInstant(event.getInstantTime())) { + return; + } WriteMetadataEvent[] eventBuffer = this.eventBuffers.getOrCreateBootstrapBuffer(event, this.parallelism); - eventBuffer[event.getTaskID()] = event; + WriteMetadataEvent existing = eventBuffer[event.getTaskID()]; + if (existing == null || !existing.isBootstrap()) { + eventBuffer[event.getTaskID()] = event; + } if (Arrays.stream(eventBuffer).allMatch(evt -> evt != null && evt.isBootstrap())) { // start to recommit the instant. recommitInstant(event.getCheckpointId(), event.getInstantTime(), eventBuffer); diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java index cc72da9bd12b..af4c65682366 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java @@ -190,25 +190,20 @@ public abstract class AbstractStreamWriteFunction<I> // ------------------------------------------------------------------------- protected void sendBootstrapEvent(int attemptId, boolean isRestored) throws Exception { - if (attemptId <= 0) { - if (isRestored) { - HoodieTimeline pendingTimeline = this.metaClient.getActiveTimeline().filterPendingExcludingCompaction(); - // if the task is initially started, resend the pending event. - for (WriteMetadataEvent event : this.writeMetadataState.get()) { - // Must filter out the completed instants in case it is a partial failover, - // the write status should not be accumulated in such case. - if (pendingTimeline.containsInstant(event.getInstantTime())) { - // Reset taskID for event - event.setTaskID(taskID); - // The checkpoint succeed but the meta does not commit, - // re-commit the inflight instant - this.eventGateway.sendEventToCoordinator(event); - LOG.info("Send uncommitted write metadata event to coordinator, task[{}].", taskID); - } + if (isRestored) { + HoodieTimeline pendingTimeline = this.metaClient.getActiveTimeline().filterPendingExcludingCompaction(); + // Resend pending metadata on every attempt in case an earlier delivery failed. + for (WriteMetadataEvent event : this.writeMetadataState.get()) { + // A partial failover may restore metadata for an already completed instant. + if (pendingTimeline.containsInstant(event.getInstantTime())) { + event.setTaskID(taskID); + this.eventGateway.sendEventToCoordinator(event); + LOG.info("Send uncommitted write metadata event to coordinator, task[{}].", taskID); } } - } else { - // otherwise sends an empty bootstrap event instead. + } + if (attemptId > 0) { + // Clear events from the failed attempt at or after the restored checkpoint. this.eventGateway.sendEventToCoordinator(WriteMetadataEvent.emptyBootstrap(taskID, checkpointId)); LOG.info("Send bootstrap write metadata event to coordinator, task[{}].", taskID); } @@ -224,6 +219,7 @@ public abstract class AbstractStreamWriteFunction<I> .checkpointId(checkpointId) .instantTime(currentInstant) .writeStatus(new ArrayList<>(writeStatuses)) + .lastBatch(true) .bootstrap(true) .build(); this.writeMetadataState.add(event); diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java index 2ecf543859a4..bb30ca928f61 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java @@ -96,6 +96,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; /** @@ -451,6 +452,68 @@ public class TestStreamWriteOperatorCoordinator { assertThat("Recommits the instant with partial uncommitted events", lastCompleted, is(instant)); } + @ParameterizedTest + @ValueSource(strings = {"COPY_ON_WRITE", "MERGE_ON_READ"}) + void testLateBootstrapAfterNormalCommit(String tableType) throws Exception { + Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); + conf.set(FlinkOptions.TABLE_TYPE, tableType); + conf.set(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT, 1_000L); + coordinator.close(); + reset(); + coordinator = startCoordinator(conf, 2); + String instant = requestInstantTime(41); + coordinator.handleEventFromOperator(0, createOperatorEvent(0, 41, instant, "par1", false, true, 0.1)); + coordinator.handleEventFromOperator(1, createOperatorEvent(1, 41, instant, "par2", false, true, 0.1)); + + // The restored writer observes a pending instant, but its event is handled after the normal commit. + WriteMetadataEvent delayed = createBootstrapEvent(0, 41, instant, "par1"); + coordinator.notifyCheckpointComplete(42); + assertEquals(instant, TestUtils.getLastCompleteInstant(tempFile.getAbsolutePath())); + assertNull(coordinator.getEventBuffer(41)); + coordinator.handleEventFromOperator(0, delayed); + assertNull(coordinator.getEventBuffer(41), "A late bootstrap must not recreate a normally committed buffer"); + + // COW instant creation must not block on a buffer recreated by the late event. + String nextInstant = requestInstantTime(42); + coordinator.handleEventFromOperator(0, createOperatorEvent(0, 42, nextInstant, "par1", false, true, 0.1)); + coordinator.handleEventFromOperator(1, createOperatorEvent(1, 42, nextInstant, "par2", false, true, 0.1)); + coordinator.notifyCheckpointComplete(43); + assertEquals(nextInstant, TestUtils.getLastCompleteInstant(tempFile.getAbsolutePath())); + assertNull(((MockOperatorCoordinatorContext) coordinator.getContext()).getJobFailureReason()); + } + + @Test + void testDuplicateBootstrapEvents() { + String instant = requestInstantTime(-1); + WriteMetadataEvent partial = createOperatorEvent(0, -1, instant, "par1", false, false, 0.1); + partial.setLastBatch(false); + coordinator.handleEventFromOperator(0, partial); + WriteMetadataEvent first = createBootstrapEvent(0, -1, instant, "par1"); + coordinator.handleEventFromOperator(0, first); + assertSame(first, coordinator.getEventBuffer(-1)[0], + "Recovery must replace an incomplete ordinary event"); + coordinator.handleEventFromOperator(0, createBootstrapEvent(0, -1, instant, "par1")); + assertSame(first, coordinator.getEventBuffer(-1)[0], + "A retry must preserve the existing bootstrap event"); + assertEquals(1, coordinator.getEventBuffer(-1)[0].getWriteStatuses().size(), + "A retry must not accumulate write statuses"); + assertNull(TestUtils.getLastCompleteInstant(tempFile.getAbsolutePath())); + + // The other task's first delivery was lost; its retry completes the recovered batch. + coordinator.handleEventFromOperator(1, createBootstrapEvent(1, -1, instant, "par2")); + assertEquals(instant, TestUtils.getLastCompleteInstant(tempFile.getAbsolutePath())); + assertNull(coordinator.getEventBuffer(-1)); + coordinator.handleEventFromOperator(0, first); + assertNull(coordinator.getEventBuffer(-1), "A late retry must not recreate a committed buffer"); + + String nextInstant = requestInstantTime(1); + coordinator.handleEventFromOperator(0, createBootstrapEvent(0, 1, nextInstant, "par1")); + coordinator.handleEventFromOperator(1, createBootstrapEvent(1, 1, nextInstant, "par2")); + assertEquals(nextInstant, TestUtils.getLastCompleteInstant(tempFile.getAbsolutePath())); + coordinator.handleEventFromOperator(0, first); + assertNull(coordinator.getEventBuffer(-1), "Older retries must remain ignored after a newer checkpoint"); + } + @Test public void testStopHeartbeatForUncommittedEventWithLazyCleanPolicy() throws Exception { // reset diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java index 4cc8eae7a0bd..b32a7a3b7b82 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java @@ -128,13 +128,11 @@ public class TestWriteCopyOnWrite extends TestWriteBase { .assertEmptyEvent() .checkpoint(3) .assertEmptyEvent() - // 1. triggers a new checkpoint; - // 2. triggers task 0 failover, there is no pending instant that needs to recommit, - // the task sends an empty bootstrap event to trigger cleaning of legacy events; - // 3. triggers a checkpoint ack to simulate a following-up checkpoint success, - // the handling of the checkpoint ack event would recommit 3. + // Recover the pending checkpoint metadata on retry, then clean events from the failed attempt. + // A following checkpoint notification must not commit the recovered instant again. .subTaskFails(0, 2) - .assertEmptyEvent() + .assertEmptyEvent() // recovered metadata + .assertEmptyEvent() // cleanup of the failed attempt .checkpointComplete(4) .checkCompletedInstantCount(3) // rollback the last complete instant to inflight state, to simulate an instant commit failure @@ -160,7 +158,8 @@ public class TestWriteCopyOnWrite extends TestWriteBase { .assertEmptyEvent() // trigger a partial failure .subTaskFails(0, 1) - .assertNextEvent() + .assertNextEvent() // recovered metadata + .assertEmptyEvent() // cleanup of the failed attempt // make sure coordinator send an ack event to unblock the writers. .assertNextSubTaskEvent() // write a set of data and check the result. diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/common/TestAbstractStreamWriteFunction.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/common/TestAbstractStreamWriteFunction.java index bfd9e06184fd..5b31abf59d65 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/common/TestAbstractStreamWriteFunction.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/common/TestAbstractStreamWriteFunction.java @@ -41,7 +41,6 @@ import org.apache.flink.streaming.api.operators.collect.utils.MockFunctionSnapsh import org.apache.flink.table.data.RowData; import org.apache.flink.util.Collector; import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.ValueSource; @@ -116,30 +115,33 @@ class TestAbstractStreamWriteFunction { assertEquals(42L, snapshot.getCheckpointId(), "Saved metadata belongs to the batch before checkpoint 43"); assertEquals("002", snapshot.getInstantTime()); assertTrue(snapshot.isBootstrap()); + assertTrue(snapshot.isLastBatch(), "Checkpoint metadata contains the complete flushed batch"); function.instantToWrite(true); verify(correspondent).requestInstantTime(43L, instantRequestPollBudget()); } - @Test - void testRestoredMetadataKeepsOriginalCheckpointId() throws Exception { + @ParameterizedTest + @CsvSource({"0, false", "0, true", "1, false", "1, true", "2, false", "2, true"}) + void testRestoredMetadataKeepsOriginalCheckpointId(int attempt, boolean lastBatch) throws Exception { WriteMetadataEvent restored = WriteMetadataEvent.builder() .taskID(0) .checkpointId(41L) .instantTime("001") .writeStatus(Collections.emptyList()) .bootstrap(true) - .lastBatch(true) + .lastBatch(lastBatch) .build(); writeMetadataState().add(restored); when(pendingTimeline.containsInstant("001")).thenReturn(true); - initialize(42L, 0); + initialize(42L, attempt); - assertEquals(1, events.size()); + assertEquals(attempt == 0 ? 1 : 2, events.size()); WriteMetadataEvent bootstrap = (WriteMetadataEvent) events.get(0); assertEquals(2, bootstrap.getTaskID(), "Restore must use the new subtask ID"); assertEquals(41L, bootstrap.getCheckpointId()); assertEquals("001", bootstrap.getInstantTime()); + assertEquals(lastBatch, bootstrap.isLastBatch(), "Replay preserves the checkpointed completion flag"); function.instantToWrite(true); verify(correspondent).requestInstantTime(42L, instantRequestPollBudget()); }
