You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2026/10/15 04:45:59 UTC
(hudi) branch master updated: fix(flink): resend recovered write metadata on task retries (#20323)
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 0170359e59d3 fix(flink): resend recovered write metadata on task retries (#20323)
0170359e59d3 is described below
commit 0170359e59d30a5f09ccadd47e796763c653f395
Author: Danny Chan <yu...@gmail.com>
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
---
.../hudi/sink/StreamWriteOperatorCoordinator.java | 39 ++++++----
.../sink/common/AbstractStreamWriteFunction.java | 37 ++++------
.../org/apache/hudi/sink/utils/EventBuffers.java | 9 ++-
.../sink/TestStreamWriteOperatorCoordinator.java | 85 ++++++++++++++++++++++
.../org/apache/hudi/sink/TestWriteCopyOnWrite.java | 13 ++--
.../common/TestAbstractStreamWriteFunction.java | 14 ++--
6 files changed, 145 insertions(+), 52 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 8a0e2a16ae99..e8c5ded34d70 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
@@ -98,16 +98,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
@@ -587,11 +592,17 @@ public class StreamWriteOperatorCoordinator
this.eventBuffers.cleanLegacyEvents(event);
return;
}
- EventBuffer eventBuffer = this.eventBuffers.getOrCreateBootstrapBuffer(event);
- eventBuffer.addBootstrapEvent(event);
- if (eventBuffer.allBootstrapEventsReceived()) {
+ // 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;
+ }
+ EventBuffer bootstrapBuffer = this.eventBuffers.getOrCreateBootstrapBuffer(event);
+ bootstrapBuffer.addBootstrapEvent(event);
+ if (bootstrapBuffer.allBootstrapEventsReceived()) {
// start to recommit the instant.
- boolean committed = recommitInstant(event.getCheckpointId(), event.getInstantTime(), eventBuffer);
+ boolean committed = recommitInstant(event.getCheckpointId(), event.getInstantTime(), bootstrapBuffer);
if (committed && tableState.isRecordLevelIndex) {
context.failJob(new HoodieException("There are pending instants recommitted on job restart,"
+ "triggering a global failover so that RLI bootstrap operator can load the record level index completely."));
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 19b72a4020d8..d7e050079c95 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
@@ -204,27 +204,21 @@ 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.
- StreamSupport.stream(this.writeMetadataState.get().spliterator(), false)
- .reduce(WriteMetadataEvent::mergeWithRescale)
- .ifPresent(metadataEvent -> {
- // 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(metadataEvent.getInstantTime())) {
- // Reset taskID for event
- metadataEvent.setTaskID(taskID);
- // The checkpoint succeed but the meta does not commit,
- // re-commit the inflight instant
- sendWriteMetadataEvent(metadataEvent);
- log.info("Send uncommitted write metadata event to coordinator, task[{}].", taskID);
- }
- });
- }
- } else {
- // otherwise sends an empty bootstrap event instead.
+ if (isRestored) {
+ HoodieTimeline pendingTimeline = this.metaClient.getActiveTimeline().filterPendingExcludingCompaction();
+ // Resend pending metadata on every attempt in case an earlier delivery failed.
+ StreamSupport.stream(this.writeMetadataState.get().spliterator(), false)
+ .reduce(WriteMetadataEvent::mergeWithRescale)
+ // A partial failover may restore metadata for an already completed instant.
+ .filter(metadataEvent -> pendingTimeline.containsInstant(metadataEvent.getInstantTime()))
+ .ifPresent(metadataEvent -> {
+ metadataEvent.setTaskID(taskID);
+ sendWriteMetadataEvent(metadataEvent);
+ log.info("Send uncommitted write metadata event to coordinator, task[{}].", taskID);
+ });
+ }
+ if (attemptId > 0) {
+ // Clear events from the failed attempt at or after the restored checkpoint.
sendWriteMetadataEvent(WriteMetadataEvent.emptyBootstrap(taskID, checkpointId, isMetadataTable));
log.info("Send bootstrap write metadata event to coordinator, task[{}].", taskID);
}
@@ -247,6 +241,7 @@ public abstract class AbstractStreamWriteFunction<I>
.checkpointId(checkpointId)
.instantTime(currentInstant)
.writeStatus(new ArrayList<>(writeStatuses))
+ .lastBatch(true)
.bootstrap(true)
.metadataTable(isMetadataTable)
.build();
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java
index 10f89637e342..8b0fc3cdbf9f 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java
@@ -206,11 +206,12 @@ public class EventBuffers implements Serializable {
}
public void addBootstrapEvent(WriteMetadataEvent event) {
- if (event.isMetadataTable()) {
- indexWriteEventBuffer[event.getTaskID()] = event;
- } else {
- dataWriteEventBuffer[event.getTaskID()] = event;
+ WriteMetadataEvent[] eventBuffer = event.isMetadataTable() ? indexWriteEventBuffer : dataWriteEventBuffer;
+ WriteMetadataEvent existing = eventBuffer[event.getTaskID()];
+ if (existing != null && existing.isBootstrap()) {
+ return;
}
+ eventBuffer[event.getTaskID()] = 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 5a06be17b40a..493282c2b139 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
@@ -118,6 +118,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.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -537,6 +538,90 @@ 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).getDataWriteEventBuffer()[0],
+ "Recovery must replace an incomplete ordinary event");
+ coordinator.handleEventFromOperator(0, createBootstrapEvent(0, -1, instant, "par1"));
+ assertSame(first, coordinator.getEventBuffer(-1).getDataWriteEventBuffer()[0],
+ "A retry must preserve the existing bootstrap event");
+ assertEquals(1, coordinator.getEventBuffer(-1).collectDataWriteStatuses().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
+ void testBootstrapDeduplicationSeparatesDataAndIndexWriters() throws Exception {
+ Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ conf.set(FlinkOptions.INDEX_TYPE, GLOBAL_RECORD_LEVEL_INDEX.name());
+ conf.set(FlinkOptions.INDEX_WRITE_TASKS, 2);
+ coordinator.close();
+ coordinator = startCoordinator(conf, 2);
+ String instant = requestInstantTime(-1);
+ WriteMetadataEvent dataEvent = createBootstrapEvent(0, -1, instant, "par1");
+ WriteMetadataEvent indexEvent = createOperatorEvent(0, -1, instant, "record_index", true, true, 0.1, true);
+ coordinator.handleEventFromOperator(0, dataEvent);
+ coordinator.handleEventFromOperator(0, indexEvent);
+ EventBuffers.EventBuffer buffer = coordinator.getEventBuffer(-1);
+ assertEquals(dataEvent, buffer.getDataWriteEventBuffer()[0]);
+ assertEquals(indexEvent, buffer.getIndexWriteEventBuffer()[0]);
+ coordinator.handleEventFromOperator(0, createOperatorEvent(0, -1, instant, "record_index", true, true, 0.1, true));
+ assertSame(indexEvent, buffer.getIndexWriteEventBuffer()[0], "An index retry must preserve the existing bootstrap event");
+ assertEquals(1, buffer.collectIndexWriteStatuses().size(), "An index retry must not accumulate write statuses");
+ assertEquals(dataEvent, buffer.getDataWriteEventBuffer()[0], "An index retry must not replace the data writer's slot");
+ assertFalse(buffer.allBootstrapEventsReceived());
+ }
+
@ParameterizedTest
@ValueSource(strings = {"GLOBAL_RECORD_LEVEL_INDEX", "RECORD_LEVEL_INDEX"})
public void testRecordLevelIndexFlag(String indexType) throws Exception {
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 8c263f38c9c4..b3ea17d255fe 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
@@ -166,13 +166,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
@@ -218,7 +216,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());
}