cshuo commented on code in PR #19952:
URL: https://github.com/apache/hudi/pull/19952#discussion_r4012966077
##########
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.
Ensuring bootstrap recommit completes before the next checkpoint succeeds,
as discussed in previous comment, should close this recovery gap.
--
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]
