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 43632ec5161f fix(flink): use restored checkpoint IDs for writers and
RLI caches (#19952)
43632ec5161f is described below
commit 43632ec5161fa672bbeb4e5d7e4688793792e80c
Author: Danny Chan <[email protected]>
AuthorDate: Wed Sep 16 12:30:09 2026 +0800
fix(flink): use restored checkpoint IDs for writers and RLI caches (#19952)
* fix(flink): use restored checkpoint IDs when initializing writers
---
.../sink/common/AbstractStreamWriteFunction.java | 43 +----
.../sink/partitioner/BucketAssignFunction.java | 2 +-
.../partitioner/index/IndexBackendFactory.java | 41 +----
.../sink/TestStreamWriteOperatorCoordinator.java | 171 +++++++++++++----
.../common/TestAbstractStreamWriteFunction.java | 204 +++++++++++++++++++++
.../index/TestGlobalRecordLevelIndexBackend.java | 35 ++++
6 files changed, 384 insertions(+), 112 deletions(-)
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 69c27dc7504a..567c312a156e 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
@@ -35,7 +35,6 @@ import org.apache.hudi.utils.RuntimeContextUtils;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
-import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -114,11 +113,6 @@ public abstract class AbstractStreamWriteFunction<I>
*/
private transient ListState<WriteMetadataEvent> writeMetadataState;
- /**
- * List state of the JobID.
- */
- private transient ListState<JobID> jobIdState;
-
/**
* Write status list for the current checkpoint.
*/
@@ -166,15 +160,13 @@ public abstract class AbstractStreamWriteFunction<I>
"write-metadata-state",
TypeInformation.of(WriteMetadataEvent.class)
));
- this.jobIdState = context.getOperatorStateStore().getListState(
- new ListStateDescriptor<>(
- "job-id-state",
- TypeInformation.of(JobID.class)
- ));
int attemptId = RuntimeContextUtils.getAttemptNumber(getRuntimeContext());
if (context.isRestored()) {
- initCheckpointId(attemptId,
context.getRestoredCheckpointId().orElse(-1L));
+ // sets up the known checkpoint id as the last successful checkpoint id
for purposes of:
+ // 1). old events cleaning;
+ // 2). instant time request for current checkpoint.
+ this.checkpointId = context.getRestoredCheckpointId().orElse(-1L);
}
sendBootstrapEvent(attemptId, context.isRestored());
}
@@ -187,8 +179,6 @@ public abstract class AbstractStreamWriteFunction<I>
snapshotState();
// Reload the snapshot state as the current state.
reloadWriteMetaState();
- // Reload the job ID state
- reloadJobIdState();
// Update checkpoint id
this.checkpointId = functionSnapshotContext.getCheckpointId();
}
@@ -212,23 +202,6 @@ public abstract class AbstractStreamWriteFunction<I>
// Utilities
// -------------------------------------------------------------------------
- private void initCheckpointId(int attemptId, long restoredCheckpointId)
throws Exception {
- if (attemptId <= 0) {
- // returns early if the job/task is initially started.
- return;
- }
- JobID currentJobId = RuntimeContextUtils.getJobId(getRuntimeContext());
- if (StreamSupport.stream(this.jobIdState.get().spliterator(), false)
- .noneMatch(currentJobId::equals)) {
- // do not set up the checkpoint id if the state comes from the old job.
- return;
- }
- // sets up the known checkpoint id as the last successful checkpoint id
for purposes of:
- // 1). old events cleaning;
- // 2). instant time request for current checkpoint.
- this.checkpointId = restoredCheckpointId;
- }
-
protected void sendBootstrapEvent(int attemptId, boolean isRestored) throws
Exception {
if (attemptId <= 0) {
if (isRestored) {
@@ -280,14 +253,6 @@ public abstract class AbstractStreamWriteFunction<I>
writeStatuses.clear();
}
- /**
- * Reload job id state as current job id.
- */
- private void reloadJobIdState() throws Exception {
- this.jobIdState.clear();
- this.jobIdState.add(RuntimeContextUtils.getJobId(getRuntimeContext()));
- }
-
public void handleOperatorEvent(OperatorEvent event) {
ValidationUtils.checkArgument(event instanceof CommitAckEvent,
"The write function can only handle CommitAckEvent");
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java
index a77b9cdf13e4..88478dce9443 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java
@@ -171,7 +171,7 @@ public class BucketAssignFunction
@Override
public void initializeState(FunctionInitializationContext context) throws
Exception {
- this.indexBackend = IndexBackendFactory.create(conf, context,
getRuntimeContext());
+ this.indexBackend = IndexBackendFactory.create(conf, context);
this.indexBackend.registerMetrics(getRuntimeContext().getMetricGroup());
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/IndexBackendFactory.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/IndexBackendFactory.java
index 9c61dbe0acb6..b53fea89090f 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/IndexBackendFactory.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/IndexBackendFactory.java
@@ -23,21 +23,14 @@ import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.configuration.OptionsResolver;
import org.apache.hudi.index.HoodieIndex;
-import org.apache.hudi.utils.RuntimeContextUtils;
import org.apache.hudi.utils.StateTtlConfigUtils;
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.functions.RuntimeContext;
-import org.apache.flink.api.common.state.ListState;
-import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
-import java.util.stream.StreamSupport;
-
/**
* Factory to create a {@link GlobalIndexBackend} based on the configured
index type.
*/
@@ -50,13 +43,11 @@ public class IndexBackendFactory {
*
* @param conf Flink write configuration
* @param context Flink function initialization context
- * @param runtimeContext Flink runtime context for job and attempt metadata
* @return global index backend for record-key lookups
*/
public static GlobalIndexBackend create(
Configuration conf,
- FunctionInitializationContext context,
- RuntimeContext runtimeContext) throws Exception {
+ FunctionInitializationContext context) throws Exception {
HoodieIndex.IndexType indexType = OptionsResolver.getIndexType(conf);
switch (indexType) {
case FLINK_STATE:
@@ -75,38 +66,12 @@ public class IndexBackendFactory {
if (conf.get(FlinkOptions.INDEX_BOOTSTRAP_ENABLED)) {
return new
RocksDBIndexBackend(conf.get(FlinkOptions.INDEX_BOOTSTRAP_ROCKSDB_PATH),
OptionsResolver.isPartitionedTable(conf));
} else {
- ListState<JobID> jobIdState =
context.getOperatorStateStore().getListState(
- new ListStateDescriptor<>(
- "bucket-assign-job-id-state",
- TypeInformation.of(JobID.class)
- ));
- long initCheckpointId = -1;
- if (context.isRestored()) {
- int attemptId =
RuntimeContextUtils.getAttemptNumber(runtimeContext);
- initCheckpointId = initCheckpointId(attemptId, jobIdState,
context.getRestoredCheckpointId().orElse(-1L), runtimeContext);
- }
- // set the jobId state with current job id.
- jobIdState.clear();
- jobIdState.add(RuntimeContextUtils.getJobId(runtimeContext));
+ // Match the writer's checkpoint ID so uncommitted index entries
remain protected from eviction.
+ long initCheckpointId = context.isRestored() ?
context.getRestoredCheckpointId().orElse(-1L) : -1L;
return new GlobalRecordLevelIndexBackend(conf, initCheckpointId);
}
default:
throw new UnsupportedOperationException("Index type " + indexType + "
is not supported for bucket assigning yet.");
}
}
-
- private static long initCheckpointId(int attemptId, ListState<JobID>
jobIdState, long restoredCheckpointId, RuntimeContext runtimeContext) throws
Exception {
- if (attemptId <= 0) {
- // returns early if the job/task is initially started.
- return -1;
- }
- JobID currentJobId = RuntimeContextUtils.getJobId(runtimeContext);
- if (StreamSupport.stream(jobIdState.get().spliterator(), false)
- .noneMatch(currentJobId::equals)) {
- // do not set up the checkpoint id if the state comes from the old job.
- return -1;
- }
- // sets up the known checkpoint id as the last successful checkpoint id
for purposes of cache cleaning.
- return restoredCheckpointId;
- }
}
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 4ac60852998c..352229beff9c 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
@@ -68,9 +68,11 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mockito;
import org.slf4j.Logger;
@@ -155,20 +157,22 @@ public class TestStreamWriteOperatorCoordinator {
* started coordinator, so it recommits its live event buffers directly.
*/
@ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testCheckpointAndRestore(boolean isStreamingIndexWriteEnabled)
throws Exception {
+ @CsvSource({"false, false", "false, true", "true, false", "true, true"})
+ public void testCheckpointAndRestore(boolean isStreamingIndexWriteEnabled,
boolean restartCoordinator) throws Exception {
Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
- conf.set(FlinkOptions.TABLE_TYPE, HoodieTableType.MERGE_ON_READ.name());
if (isStreamingIndexWriteEnabled) {
conf.set(FlinkOptions.INDEX_TYPE, GLOBAL_RECORD_LEVEL_INDEX.name());
conf.set(FlinkOptions.INDEX_WRITE_TASKS, 2);
+ coordinator.close();
+ coordinator = startCoordinator(conf, 2);
}
- coordinator = startCoordinator(conf, 2);
requestInstantTime(-1);
String instant = coordinator.getInstant();
assertNotEquals("", instant);
+ // Writer checkpoint 1 finishes after the coordinator has taken its
snapshot.
+ coordinator.checkpointCoordinator(1, new CompletableFuture<>());
OperatorEvent event0 = createOperatorEvent(0, instant, "par1", true, 0.1);
OperatorEvent event1 = createOperatorEvent(1, instant, "par2", true, 0.2);
coordinator.handleEventFromOperator(0, event0);
@@ -182,44 +186,48 @@ public class TestStreamWriteOperatorCoordinator {
}
CompletableFuture<byte[]> future = new CompletableFuture<>();
- coordinator.checkpointCoordinator(1, future);
-
- // Case 1: job restart restores checkpoint data before the coordinator
starts.
- try (StreamWriteOperatorCoordinator restoredCoordinator =
createCoordinator(conf, 2)) {
- restoredCoordinator.resetToCheckpoint(1, future.get());
+ coordinator.checkpointCoordinator(2, future);
- EventBuffers.EventBuffer eventBuffer =
restoredCoordinator.getEventBuffer(-1);
+ if (restartCoordinator) {
+ coordinator.close();
+ coordinator = createCoordinator(conf, 4);
+ coordinator.resetToCheckpoint(2, future.get());
+ EventBuffers.EventBuffer eventBuffer = coordinator.getEventBuffer(-1);
assertEquals(2, eventBuffer.getDataWriteEventBuffer().length);
assertEquals(isStreamingIndexWriteEnabled ? 2 : 0,
eventBuffer.getIndexWriteEventBuffer().length);
+ // Exercise recommit during start, not only checkpoint deserialization.
+ coordinator.start();
+ } else {
+ // Global failover recommits the live buffers of the already started
coordinator.
+ coordinator.resetToCheckpoint(2, future.get());
}
- // Case 2: global failover recommits the live buffers of the already
started coordinator.
- coordinator.resetToCheckpoint(1, future.get());
-
- assertNull(coordinator.getEventBuffer());
+ assertNull(coordinator.getEventBuffer(-1));
+ assertNull(((MockOperatorCoordinatorContext)
coordinator.getContext()).getJobFailureReason());
assertTrue(StreamerUtil.createMetaClient(conf).reloadActiveTimeline()
.filterCompletedInstants().containsInstant(instant));
}
/**
- * Verifies legacy checkpoint compatibility for both restore paths. Case 1
deserializes the legacy
- * checkpoint into a newly constructed coordinator. Case 2 intentionally
does not deserialize the
- * checkpoint because a coordinator surviving global failover recommits its
live buffers directly.
+ * Verifies both restore paths with the legacy checkpoint format.
*/
- @Test
- public void testRestoreFromLegacyState() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testRestoreFromLegacyState(boolean restartCoordinator) throws
Exception {
Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
requestInstantTime(-1);
String instant = coordinator.getInstant();
assertNotEquals("", instant);
+ // Writer checkpoint 1 finishes after the coordinator has taken its
snapshot.
+ coordinator.checkpointCoordinator(1, new CompletableFuture<>());
OperatorEvent event0 = createOperatorEvent(0, instant, "par1", true, 0.1);
OperatorEvent event1 = createOperatorEvent(1, instant, "par2", true, 0.2);
coordinator.handleEventFromOperator(0, event0);
coordinator.handleEventFromOperator(1, event1);
CompletableFuture<byte[]> future = new CompletableFuture<>();
- coordinator.checkpointCoordinator(1, future);
+ coordinator.checkpointCoordinator(2, future);
Map<Long, Pair<String, EventBuffers.EventBuffer>> eventBuffers =
SerializationUtils.deserialize(future.get());
// convert to legacy event buffers
@@ -228,19 +236,22 @@ public class TestStreamWriteOperatorCoordinator {
legacyEventBuffers.put(ckpId, Pair.of(eventBuffer.getLeft(),
eventBuffer.getRight().getDataWriteEventBuffer()));
});
- // Case 1: job restart restores legacy checkpoint data before the
coordinator starts.
- try (StreamWriteOperatorCoordinator restoredCoordinator =
createCoordinator(conf, 2)) {
- restoredCoordinator.resetToCheckpoint(1,
SerializationUtils.serialize(legacyEventBuffers));
-
- EventBuffers.EventBuffer eventBuffer =
restoredCoordinator.getEventBuffer(-1);
+ byte[] legacyState = SerializationUtils.serialize(legacyEventBuffers);
+ if (restartCoordinator) {
+ coordinator.close();
+ coordinator = createCoordinator(conf, 4);
+ coordinator.resetToCheckpoint(2, legacyState);
+ EventBuffers.EventBuffer eventBuffer = coordinator.getEventBuffer(-1);
assertEquals(2, eventBuffer.getDataWriteEventBuffer().length);
assertEquals(0, eventBuffer.getIndexWriteEventBuffer().length);
+ coordinator.start();
+ } else {
+ // A coordinator surviving global failover recommits its live buffers
directly.
+ coordinator.resetToCheckpoint(2, legacyState);
}
- // Case 2: global failover ignores checkpoint bytes and recommits the live
buffers.
- coordinator.resetToCheckpoint(1,
SerializationUtils.serialize(legacyEventBuffers));
-
- assertNull(coordinator.getEventBuffer());
+ assertNull(coordinator.getEventBuffer(-1));
+ assertNull(((MockOperatorCoordinatorContext)
coordinator.getContext()).getJobFailureReason());
assertTrue(StreamerUtil.createMetaClient(conf).reloadActiveTimeline()
.filterCompletedInstants().containsInstant(instant));
}
@@ -260,11 +271,93 @@ public class TestStreamWriteOperatorCoordinator {
"Receive an unexpected event for instant abc from task 0");
}
+ @Test
+ void testDeferredRecommitAfterScaleUp() throws Exception {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ String restoredInstant = restoreFirstCheckpointAfterScaleUp(conf);
+ String nextInstant = requestInstantTime(1);
+ assertNotEquals(restoredInstant, nextInstant);
+ assertEquals(nextInstant, coordinator.getInstant());
+ coordinator.checkpointCoordinator(2, new CompletableFuture<>());
+ sendCheckpointEvents(1, nextInstant, 4);
+
+ coordinator.notifyCheckpointComplete(2);
+
+ HoodieTimeline completed = StreamerUtil.createMetaClient(conf)
+ .reloadActiveTimeline().filterCompletedInstants();
+ assertTrue(completed.containsInstant(restoredInstant));
+ assertTrue(completed.containsInstant(nextInstant));
+ assertEquals(2,
completed.readCommitMetadata(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION,
restoredInstant))
+ .getPartitionToWriteStats().size(), "Both original writers must be
included in the restored commit");
+ assertEquals(4,
completed.readCommitMetadata(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION,
nextInstant))
+ .getPartitionToWriteStats().size(), "All scaled-up writers must be
included in the next commit");
+ assertNull(coordinator.getEventBuffer(-1));
+ assertNull(coordinator.getEventBuffer(1));
+ assertNull(((MockOperatorCoordinatorContext)
coordinator.getContext()).getJobFailureReason());
+ }
+
+ @Disabled("https://github.com/apache/hudi/issues/19922: deferred bootstrap
metadata is omitted from the next checkpoint")
+ @Test
+ void testDeferredRecommitSurvivesAnotherRestart() throws Exception {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ String restoredInstant = restoreFirstCheckpointAfterScaleUp(conf);
+ String nextInstant = requestInstantTime(1);
+ CompletableFuture<byte[]> secondCheckpoint = new CompletableFuture<>();
+ coordinator.checkpointCoordinator(2, secondCheckpoint);
+ sendCheckpointEvents(1, nextInstant, 4);
+ coordinator.close();
+
+ // Checkpoint 2 succeeded, but the job stopped before
notifyCheckpointComplete could commit it.
+ // The writers now restore only checkpoint 2's batch; the coordinator must
preserve the older batch.
+ coordinator = createCoordinator(conf, 4);
+ coordinator.resetToCheckpoint(2, secondCheckpoint.get());
+ coordinator.start();
+ setSynchronousExecutors(coordinator);
+ for (int task = 0; task < 4; task++) {
+ coordinator.handleEventFromOperator(task, createBootstrapEvent(task, 1,
nextInstant, "new" + task));
+ }
+ coordinator.notifyCheckpointComplete(3);
+
+ HoodieTimeline completed =
StreamerUtil.createMetaClient(conf).reloadActiveTimeline().filterCompletedInstants();
+ assertTrue(completed.containsInstant(nextInstant));
+ assertTrue(completed.containsInstant(restoredInstant), "The first batch
must survive a second restart before its deferred commit");
+ }
+
+ private String restoreFirstCheckpointAfterScaleUp(Configuration conf) throws
Exception {
+ resetToMergeOnRead(conf);
+ String restoredInstant = requestInstantTime(-1);
+ CompletableFuture<byte[]> firstCheckpoint = new CompletableFuture<>();
+ // The first coordinator snapshot precedes the writer snapshots, so only
writer state has this batch.
+ coordinator.checkpointCoordinator(1, firstCheckpoint);
+ coordinator.close();
+
+ coordinator = createCoordinator(conf, 4);
+ coordinator.resetToCheckpoint(1, firstCheckpoint.get());
+ coordinator.start();
+ setSynchronousExecutors(coordinator);
+ coordinator.handleEventFromOperator(0, createBootstrapEvent(0, -1,
restoredInstant, "par1"));
+ coordinator.handleEventFromOperator(1, createBootstrapEvent(1, -1,
restoredInstant, "par2"));
+ EventBuffers.EventBuffer buffer = coordinator.getEventBuffer(-1);
+ assertFalse(buffer.allBootstrapEventsReceived());
+ assertEquals("", coordinator.getInstant());
+
assertFalse(StreamerUtil.createMetaClient(conf).reloadActiveTimeline().filterCompletedInstants().containsInstant(restoredInstant));
+ return restoredInstant;
+ }
+
+ private void sendCheckpointEvents(long checkpointId, String instant, int
parallelism) {
+ for (int task = 0; task < parallelism; task++) {
+ coordinator.handleEventFromOperator(task,
+ createOperatorEvent(task, checkpointId, instant, "new" + task,
false, true, 0.1));
+ }
+ assertNull(((MockOperatorCoordinatorContext)
coordinator.getContext()).getJobFailureReason());
+ }
+
@Test
public void testEventReset() throws Exception {
Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
- conf.set(FlinkOptions.TABLE_TYPE, HoodieTableType.MERGE_ON_READ.name());
- coordinator = startCoordinator(conf, 2);
+ resetToMergeOnRead(conf);
CompletableFuture<byte[]> future = new CompletableFuture<>();
coordinator.checkpointCoordinator(1, future);
String instant = requestInstantTime(0);
@@ -693,8 +786,7 @@ public class TestStreamWriteOperatorCoordinator {
@Test
void testHandleInFlightInstantsRequest() throws Exception {
Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
- conf.set(FlinkOptions.TABLE_TYPE, HoodieTableType.MERGE_ON_READ.name());
- coordinator = startCoordinator(conf, 2);
+ resetToMergeOnRead(conf);
// Request an instant time to create an initial instant
String instant1 = requestInstantTime(1);
@@ -746,13 +838,24 @@ public class TestStreamWriteOperatorCoordinator {
}
}
+ private void resetToMergeOnRead(Configuration conf) throws Exception {
+ coordinator.close();
+ reset();
+ conf.set(FlinkOptions.TABLE_TYPE, HoodieTableType.MERGE_ON_READ.name());
+ coordinator = startCoordinator(conf, 2);
+ }
+
private static StreamWriteOperatorCoordinator startCoordinator(Configuration
conf, int subTasks) throws Exception {
StreamWriteOperatorCoordinator coordinator = createCoordinator(conf,
subTasks);
coordinator.start();
+ setSynchronousExecutors(coordinator);
+ return coordinator;
+ }
+
+ private static void setSynchronousExecutors(StreamWriteOperatorCoordinator
coordinator) throws Exception {
MockOperatorCoordinatorContext coordinatorContext =
(MockOperatorCoordinatorContext) coordinator.getContext();
coordinator.setExecutor(new MockCoordinatorExecutor(coordinatorContext));
coordinator.setInstantRequestExecutor(new
MockCoordinatorExecutor(coordinatorContext));
- return coordinator;
}
private static StreamWriteOperatorCoordinator
createCoordinator(Configuration conf, int subTasks) {
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
new file mode 100644
index 000000000000..b87ee0aa0a95
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/common/TestAbstractStreamWriteFunction.java
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.common;
+
+import org.apache.hudi.client.HoodieFlinkWriteClient;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.sink.event.Correspondent;
+import org.apache.hudi.sink.event.WriteMetadataEvent;
+import org.apache.hudi.sink.utils.MockOperatorStateStore;
+import org.apache.hudi.util.FlinkWriteClients;
+import org.apache.hudi.util.StreamerUtil;
+import org.apache.hudi.utils.RuntimeContextUtils;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.operators.coordination.OperatorEvent;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import
org.apache.flink.streaming.api.operators.collect.utils.MockFunctionSnapshotContext;
+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;
+import org.mockito.MockedStatic;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.OptionalLong;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests checkpoint identity and bootstrap events independently of the writer
implementation.
+ */
+class TestAbstractStreamWriteFunction {
+ private final Configuration conf = new Configuration();
+ private final JobID jobId = new JobID();
+ private final List<OperatorEvent> events = new ArrayList<>();
+
+ private final MockOperatorStateStore stateStore = new
MockOperatorStateStore();
+ private final HoodieTimeline pendingTimeline = mock(HoodieTimeline.class);
+ private final Correspondent correspondent = mock(Correspondent.class);
+ private final TestWriteFunction function = new TestWriteFunction(conf);
+
+ @AfterEach
+ void tearDown() throws Exception {
+ function.close();
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, 1})
+ void testFreshStartUsesInitialCheckpointId(int attempt) throws Exception {
+ initialize(-1L, attempt);
+
+ assertEquals("002", function.instantToWrite(true));
+ verify(correspondent).requestInstantTime(-1L);
+ if (attempt == 0) {
+ assertTrue(events.isEmpty());
+ } else {
+ assertCleanupEvent(-1L);
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"0, SAME", "0, DIFFERENT", "0, MISSING", "1, SAME", "1,
DIFFERENT", "1, MISSING"})
+ void testRestoredCheckpointId(int attempt, SavedJobState savedJob) throws
Exception {
+ // Old savepoints may contain this state; new scale-up subtasks may have
no operator state at all.
+ if (savedJob != SavedJobState.MISSING) {
+ stateStore.getListState(new ListStateDescriptor<>("job-id-state",
TypeInformation.of(JobID.class)))
+ .add(savedJob == SavedJobState.SAME ? jobId : new JobID());
+ }
+ initialize(42L, attempt);
+
+ assertEquals("002", function.instantToWrite(true));
+ verify(correspondent).requestInstantTime(42L);
+ if (attempt == 0) {
+ assertTrue(events.isEmpty());
+ } else {
+ assertCleanupEvent(42L);
+ }
+
+ function.snapshotState(new MockFunctionSnapshotContext(43L));
+ WriteMetadataEvent snapshot = writeMetadataState().get().iterator().next();
+ assertEquals(42L, snapshot.getCheckpointId(), "Saved metadata belongs to
the batch before checkpoint 43");
+ assertEquals("002", snapshot.getInstantTime());
+ assertTrue(snapshot.isBootstrap());
+ function.instantToWrite(true);
+ verify(correspondent).requestInstantTime(43L);
+ }
+
+ @Test
+ void testRestoredMetadataKeepsOriginalCheckpointId() throws Exception {
+ WriteMetadataEvent restored = WriteMetadataEvent.builder()
+ .taskID(0)
+ .checkpointId(41L)
+ .instantTime("001")
+ .writeStatus(Collections.emptyList())
+ .bootstrap(true)
+ .lastBatch(true)
+ .build();
+ writeMetadataState().add(restored);
+ when(pendingTimeline.containsInstant("001")).thenReturn(true);
+
+ initialize(42L, 0);
+
+ assertEquals(1, 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());
+ function.instantToWrite(true);
+ verify(correspondent).requestInstantTime(42L);
+ }
+
+ private void initialize(long checkpointId, int attempt) throws Exception {
+ RuntimeContext runtimeContext = mock(RuntimeContext.class);
+ FunctionInitializationContext context =
mock(FunctionInitializationContext.class);
+ when(context.getOperatorStateStore()).thenReturn(stateStore);
+ when(context.isRestored()).thenReturn(checkpointId >= 0);
+ when(context.getRestoredCheckpointId()).thenReturn(checkpointId >= 0 ?
OptionalLong.of(checkpointId) : OptionalLong.empty());
+ when(correspondent.requestInstantTime(anyLong())).thenReturn("002");
+ function.setRuntimeContext(runtimeContext);
+ function.setCorrespondent(correspondent);
+ function.setOperatorEventGateway(events::add);
+
+ try (MockedStatic<StreamerUtil> streamerUtil =
mockStatic(StreamerUtil.class);
+ MockedStatic<FlinkWriteClients> writeClients =
mockStatic(FlinkWriteClients.class);
+ MockedStatic<RuntimeContextUtils> runtimeContextUtils =
mockStatic(RuntimeContextUtils.class)) {
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class,
RETURNS_DEEP_STUBS);
+
when(metaClient.getActiveTimeline().filterPendingExcludingCompaction()).thenReturn(pendingTimeline);
+ streamerUtil.when(() ->
StreamerUtil.createMetaClient(conf)).thenReturn(metaClient);
+ writeClients.when(() -> FlinkWriteClients.createWriteClient(conf,
runtimeContext)).thenReturn(mock(HoodieFlinkWriteClient.class));
+ runtimeContextUtils.when(() ->
RuntimeContextUtils.getJobId(runtimeContext)).thenReturn(jobId);
+ runtimeContextUtils.when(() ->
RuntimeContextUtils.getIndexOfThisSubtask(runtimeContext)).thenReturn(2);
+ runtimeContextUtils.when(() ->
RuntimeContextUtils.getAttemptNumber(runtimeContext)).thenReturn(attempt);
+ function.initializeState(context);
+ }
+ }
+
+ private void assertCleanupEvent(long checkpointId) {
+ assertEquals(1, events.size());
+ WriteMetadataEvent bootstrap = (WriteMetadataEvent) events.get(0);
+ assertEquals(2, bootstrap.getTaskID());
+ assertEquals(checkpointId, bootstrap.getCheckpointId());
+ assertEquals(WriteMetadataEvent.BOOTSTRAP_INSTANT,
bootstrap.getInstantTime());
+ assertTrue(bootstrap.isBootstrap());
+ assertTrue(bootstrap.getWriteStatuses().isEmpty());
+ }
+
+ private ListState<WriteMetadataEvent> writeMetadataState() throws Exception {
+ return stateStore.getListState(new
ListStateDescriptor<>("write-metadata-state",
TypeInformation.of(WriteMetadataEvent.class)));
+ }
+
+ private enum SavedJobState {
+ SAME, DIFFERENT, MISSING
+ }
+
+ private static class TestWriteFunction extends
AbstractStreamWriteFunction<RowData> {
+ TestWriteFunction(Configuration conf) {
+ super(conf);
+ }
+
+ @Override
+ public void snapshotState() {
+ currentInstant = instantToWrite(true);
+ }
+
+ @Override
+ public void processElement(RowData value, Context ctx, Collector<RowData>
out) {
+ // No record processing is needed to exercise the base checkpoint
lifecycle.
+ }
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestGlobalRecordLevelIndexBackend.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestGlobalRecordLevelIndexBackend.java
index bff98cfe49d3..d267df7478d0 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestGlobalRecordLevelIndexBackend.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestGlobalRecordLevelIndexBackend.java
@@ -34,10 +34,13 @@ import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.metrics.NoOpMetricRegistry;
import org.apache.flink.runtime.metrics.groups.TaskManagerMetricGroup;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
import java.io.IOException;
@@ -45,6 +48,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.OptionalLong;
import java.util.UUID;
import static org.apache.hudi.common.model.HoodieTableType.COPY_ON_WRITE;
@@ -77,6 +81,37 @@ public class TestGlobalRecordLevelIndexBackend {
StreamerUtil.initTableIfNotExists(conf);
}
+ @ParameterizedTest
+ @ValueSource(longs = {-1L, 42L})
+ void testFactoryRetainsUncommittedLocations(long checkpointId) throws
Exception {
+ conf.set(FlinkOptions.INDEX_TYPE, "GLOBAL_RECORD_LEVEL_INDEX");
+ conf.set(FlinkOptions.INDEX_BOOTSTRAP_ENABLED, false);
+ conf.set(FlinkOptions.INDEX_RLI_CACHE_SIZE, 1L);
+ FunctionInitializationContext context =
mock(FunctionInitializationContext.class);
+ when(context.isRestored()).thenReturn(checkpointId >= 0);
+ when(context.getRestoredCheckpointId()).thenReturn(checkpointId >= 0 ?
OptionalLong.of(checkpointId) : OptionalLong.empty());
+
+ try (GlobalRecordLevelIndexBackend backend =
(GlobalRecordLevelIndexBackend) IndexBackendFactory.create(conf, context)) {
+ HoodieRecordGlobalLocation location = new
HoodieRecordGlobalLocation("par1", "001", "file1");
+ backend.update("uncommitted", location);
+ for (int i = 0; i < 1500; i++) {
+ backend.update("first_" + i, new HoodieRecordGlobalLocation("par1",
"001", UUID.randomUUID().toString()));
+ }
+ backend.onCheckpoint(checkpointId + 1);
+ Correspondent correspondent = mock(Correspondent.class);
+ // The writer's first batch after restore is still awaiting its Hudi
commit.
+
when(correspondent.requestInflightInstants()).thenReturn(Collections.singletonMap(checkpointId,
"001"));
+ backend.onCheckpointComplete(correspondent, checkpointId + 1);
+
+ // Force cache eviction pressure while the first batch remains
uncommitted.
+ for (int i = 0; i < 4000; i++) {
+ backend.update("next_" + i, new HoodieRecordGlobalLocation("par1",
"002", UUID.randomUUID().toString()));
+ }
+ backend.onCheckpoint(checkpointId + 2);
+ assertEquals(location, backend.get("uncommitted"), "Uncommitted
locations must survive cache cleaning");
+ }
+ }
+
@Test
void testRecordLevelIndexBackend() throws Exception {
TestData.writeData(TestData.DATA_SET_INSERT, conf);
