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 <[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
---
.../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());
}
