goutamadwant opened a new issue, #12849: URL: https://github.com/apache/seatunnel/issues/12849
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found no similar issues. ### What happened `PendingCheckpoint.acknowledgeTask` is not synchronized, and task acknowledgements run concurrently on Hazelcast generic operation threads (50 in `config/hazelcast.yaml`). Two acks for the same checkpoint can interleave: 1. Ack A removes its task from the pending set. 2. Ack B removes its task from the pending set. 3. Ack A reports its state, sees the checkpoint fully acknowledged, and completes it. 4. Ack B reports its state. `CompletedCheckpoint` shares the `actionStates` map, and `CheckpointCoordinator` serializes and stores it on its own executor. If that runs before step 4, the stored checkpoint has a `null` state for subtask B but is still reported as COMPLETED. On restore, `restoreTaskState` skips `null` slots, so that subtask starts with no state. In the same method, the checkpoint monitor numbers are wrong for any action with parallelism > 1: - `stateSize` counts state arrays (`map(s -> s.length).count()`), not bytes. - `TaskStatistics` is created per subtask with one slot, but `reportSubtaskStatistics` uses the subtask index as the slot. Every subtask with index >= 1 is rejected, so `acknowledged` never reaches `total`. Code on `dev`: - `PendingCheckpoint.java:121-153` - `CheckpointCoordinator.java:966-987, 1377, 422-424` - `CheckpointCoordinator.java:1153-1165` - `TaskStatistics.java:54-56` ### SeaTunnel Version dev ### SeaTunnel Config ```conf # Any Zeta job with checkpointing enabled and an action with parallelism > 1 env { parallelism = 2 job.mode = "STREAMING" checkpoint.interval = 5000 } ``` ### Running Command ```shell # Unit-level reproduction: a PendingCheckpoint built the way CheckpointCoordinator builds it, # for one action with parallelism 2, with both subtasks acked from two threads at once ./mvnw -pl seatunnel-engine/seatunnel-engine-server -DskipUT=false -DskipIT=true -Dtest=PendingCheckpointTest test ``` ### Error Exception ```log No exception. The checkpoint is stored as COMPLETED with a null subtask state. 20 000 concurrent runs: 16 (JDK 11) and 134 (JDK 17) checkpoints completed with a missing subtask state. Two serial acks of 100 bytes each: acknowledgedSubtasks=1 (expected 2), stateSize=1 (expected 200). ``` ### Zeta or Flink or Spark Version Zeta ### Java or Scala Version Java 11, Java 17 ### Screenshots _No response_ ### Are you willing to submit PR? - [x] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- 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]
