cshuo commented on code in PR #19952:
URL: https://github.com/apache/hudi/pull/19952#discussion_r4012966060
##########
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:
**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);
Review Comment:
** 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.
--
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]
