K1taIkuyo0421 opened a new issue, #19922:
URL: https://github.com/apache/hudi/issues/19922
### Bug Description
**What happened:**
To reproduce a scenario where a checkpoint has completed but the
corresponding Hudi commit is still pending, I added a wait in
`StreamWriteOperatorCoordinator.notifyCheckpointComplete()`, before
`commitInstants()`. I stopped the job after **the first Flink checkpoint had
completed but its Hudi commit had not yet been submitted**, increased the
parallelism, and restored from that checkpoint. The job then failed with
`Receive an unexpected event for instant` and restarted.
The investigation identified the following sequence:
1. The Writer's initial `checkpointId` is `-1`, so the write metadata saved
in the first checkpoint also carries `checkpointId = -1`. During recovery,
`sendBootstrapEvent()` resends this uncommitted metadata.
2. In this scale-up recovery scenario, the bootstrap events do not fill all
slots in the `eventBuffer` array corresponding to the new parallelism.
Consequently, `recommit` is not triggered, and the old `-1 → instant` mapping
remains in the coordinator.
3. Due to the existing initialization checks, the restored Writer keeps
`checkpointId = -1` instead of using `restoredCheckpointId`. Instant requests
for new records therefore also use `-1`, sharing the same key as the restored,
uncommitted batch. The coordinator finds the old mapping and directly returns
the existing instant, bypassing the normal instant creation and initialization
flow. As a result, the coordinator's `this.instant` field on the JobManager
remains an empty string.
4. Subsequently, `handleWriteMetaEvent()` receives a write event carrying a
non-empty instant and compares it against the empty `this.instant`. The
validation fails and triggers a job restart.
After changing the restored Writer's `checkpointId` to use the valid
`restoredCheckpointId`, new writes no longer collide with the old `-1` mapping.
**The exception no longer occurs in the same test scenario.**
**What you expected:**
After increasing parallelism and restoring from a checkpoint, the job should
correctly handle a completed checkpoint whose Hudi commit is still pending and
continue running normally.
However, the original code explicitly contains this comment:
```java
// do not set up the checkpoint id if the state comes from the old job.
```
I would therefore appreciate clarification from the maintainers: **What
scenario was this check originally intended to protect?** Is it valid to
directly use a valid `restoredCheckpointId` during recovery, and could doing so
affect other recovery scenarios?
**Steps to reproduce:**
1. Start a Flink MOR upsert job, use the test modification described above
to delay the Hudi commit associated with the first checkpoint, and wait for
that Flink checkpoint to complete.
2. Stop the job while the Hudi commit is still pending, and increase the
parallelism from 2 to 4.
3. Restore from that checkpoint under a new JobID. The original
initialization logic produces the exception described above. After switching to
the valid `restoredCheckpointId`, the exception no longer occurs in the same
scenario.
@
### Environment
**Hudi version:**1.1
**Query engine:**Apache Flink 1.20
**Relevant configs:**
- Table type: `MERGE_ON_READ`.
- Write operation: `upsert`.
- Checkpointing: enabled.
- Parallelism change on recovery: `2 → 4`.
- The test modification delays the Hudi commit associated with the first
checkpoint.
### Logs and Stack Trace
Relevant log excerpts are shown below. The temporary table name and operator
ID have been anonymized. Other log entries and some stack frames have been
omitted.
```text
2026-09-08 21:26:04.814 ERROR [meta-event-handle:ent-handle]
org.apache.hudi.sink.StreamWriteOperatorCoordinator - Executor executes action
[handle write metadata event for instant ] error
java.lang.IllegalStateException: Receive an unexpected event for instant
20260908212052922 from task 2
at
org.apache.hudi.common.util.ValidationUtils.checkState(ValidationUtils.java:82)
at
org.apache.hudi.sink.StreamWriteOperatorCoordinator.handleWriteMetaEvent(StreamWriteOperatorCoordinator.java:1630)
at
org.apache.hudi.sink.StreamWriteOperatorCoordinator.lambda$handleEventFromOperator$5(StreamWriteOperatorCoordinator.java:592)
at
org.apache.hudi.sink.utils.NonThrownExecutor.lambda$wrapAction$0(NonThrownExecutor.java:131)
...
2026-09-08 21:26:04.820 INFO [flink-pekko.actor.de:patcher-16]
org.apache.flink.runtime.jobmaster.JobMaster - Trying to recover from a global
failure.
org.apache.flink.util.FlinkException: Global failure triggered by
OperatorCoordinator for 'Sink: bucket_write(table=checkpoint_restore_test)'
(operator <operator-id>).
at
org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder$LazyInitializedCoordinatorContext.failJob(OperatorCoordinatorHolder.java:651)
at
org.apache.hudi.sink.StreamWriteOperatorCoordinator.lambda$start$0(StreamWriteOperatorCoordinator.java:373)
...
```
--
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]
