Unexpected instant event during Flink scale-up recovery after the first checkpoint completes but before Hudi commit

### 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 ).
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)
...
```

Start in StreamWriteOperatorCoordinator, especially sendBootstrapEvent(), notifyCheckpointComplete(), commitInstants(), and handleWriteMetaEvent() near the reported failure. Reproduce the Flink 1.20 MERGE_ON_READ upsert recovery with parallelism changing from 2 to 4 and the delayed first Hudi commit. Done means the restored job continues without the unexpected-instant failure while preserving the intended old-job recovery behavior.

Written by the indexing model from the issue text.