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 f9cf938e44aee..4e8712b6619f2 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 @@ -182,6 +182,8 @@ private void restoreWriteMetadata() throws Exception { boolean eventSent = false; for (WriteMetadataEvent event : this.writeMetadataState.get()) { if (Objects.equals(lastInflight, event.getInstantTime())) { + // Reset taskID for event + event.setTaskID(taskID); // The checkpoint succeed but the meta does not commit, // re-commit the inflight instant this.eventGateway.sendEventToCoordinator(event);