cshuo commented on code in PR #19952:
URL: https://github.com/apache/hudi/pull/19952#discussion_r4012966070
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java:
##########
@@ -260,11 +271,93 @@ public void testReceiveInvalidEvent() {
"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);
Review Comment:
**[P1] COW / CDC scale-up still blocks**
After scaling from 2 to 4, two bootstrap events cannot complete the old
buffer. COW upsert and MOR upsert with CDC wait for that old commit before
creating a new instant, but deferred recommit requires the next checkpoint to
complete. Both configurations reproduce a timeout. Please resolve bootstrap
completion and cover these cases.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java:
##########
@@ -166,15 +160,13 @@ public void initializeState(FunctionInitializationContext
context) throws Except
"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);
Review Comment:
**[P1] Global RLI checkpoint mismatch**
`IndexBackendFactory` still initializes the cache at `-1`, while the
restored writer now uses e.g. `42`. `markAsEvictable(42)` can then evict
uncommitted index entries, causing updates to be treated as inserts. The
regression test fails with this change and passes with the previous
implementation. Please align both initialization paths.
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java:
##########
@@ -260,11 +271,93 @@ public void testReceiveInvalidEvent() {
"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")
Review Comment:
**Deferred metadata loss**
The coordinator’s next checkpoint snapshot excludes the incomplete bootstrap
buffer, while the writers’ snapshots replace the old batch metadata with the
new batch. If the job restarts after that checkpoint succeeds but before
deferred commit, neither snapshot contains the old batch metadata. Enabling the
disabled recovery test reproduces this loss. Please preserve the restored
pending metadata until commit completes.
--
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]
