goutamadwant opened a new issue, #4199: URL: https://github.com/apache/iggy/issues/4199
### Bug description The Flink source skips records after restoring a completed checkpoint when those records were polled before a task failure but were not included in the checkpoint. The job resumes processing newer records without replaying the skipped batch. Expected: records after the restored checkpoint are replayed, preserving at-least-once delivery. This does not require an exactly-once external sink. Actual: in a real Flink checkpoint/restart test, records 10–19 were missing from recovered downstream state, while records 0–9 and a newly published record 20 were present. An independent explicit-offset read confirmed all 21 records remained in Iggy. This is skipped delivery, not deletion from server storage. ### Affected area / component Connectors, Java SDK The Java Flink source under `foreign/java/external-processors/iggy-connector-flink/iggy-connector-library`. ### Deployment Docker (DockerHub image) ### Versions - Iggy server 0.9.0 and Java SDK/connector built from `71f29618ba04917fded0ef5cd085c206fef0ee49`. - Server image: `apache/iggy@sha256:4524170018b8d3b7c24753dfd63bfa6437055e3e4ff3e49027d300fd9dea9c14`; its revision label and startup log match the checkout. - Flink 2.3.0, Java 21.0.5. - The two source files discussed below are unchanged on current master `5c571ceddf532c250871cb138d58e0b5a43e0339`. ### Hardware / environment Local ARM64 environment with a Linux Iggy container and a Flink local MiniCluster. One Iggy server, one topic partition, one consumer group, and Flink parallelism 1. No multi-reader or rescaling claim. ### Reproduction Use the existing `IggySource<String>` with `StringDeserializationSchema`, poll batch size 10, earliest starting offset, checkpointing every 200 ms, and one fixed-delay restart. The downstream operator stores processed record IDs in checkpointed `ListState<String>`. 1. Create an isolated stream, a single-partition topic and a consumer group. Publish `record-0` through `record-9` to partition 0. 2. Start the Flink job and wait for a completed checkpoint containing those ten IDs in downstream state. 3. Publish `record-10` through `record-19` in one batch. 4. In the downstream operator, throw once when receiving `record-10`, before adding it to state. Keep the failure-injected flag outside checkpointed state so recovery does not fail repeatedly. 5. Wait for Flink to restore its completed checkpoint. The downstream operator restores records 0–9. 6. Publish `record-20` and wait for it to reach the recovered operator. 7. Compare the recovered IDs with records 0–20. Records 10–19 are missing. 8. Read the topic independently with explicit offset 0 and auto-commit disabled: all 21 input records are still present. The job-level reproduction produced the same result in a focused run and a subsequent full connector-module run. All 109 existing module tests and two uninterrupted-consumption controls passed; the added recovery assertions failed. Production source and build configuration were not modified. ### Logs Condensed from the reproduction output: ```text Completed checkpoint: processed = [record-0 ... record-9] Injected failure: before processing record-10 Restored operator state: [record-0 ... record-9] Expected recovered state: [record-0 ... record-20] Actual recovered state: [record-0 ... record-9, record-20] Independent Iggy read: [record-0 ... record-20] ``` ### Iggy server config Default server settings apart from test credentials and networking overrides: ```text IGGY_TCP_ADDRESS=0.0.0.0:18090 IGGY_HTTP_ADDRESS=0.0.0.0:18000 IGGY_NODE_ADVERTISED_ADDRESS=127.0.0.1 IGGY_SHARDING_CPU_ALLOCATION=all ``` Both ports were published only on loopback. The container was restricted to two CPUs with the usual integration-test seccomp/memlock settings. ### Suspected cause and scope [`IggyPartitionSplitReader.poll()`](https://github.com/apache/iggy/blob/71f29618ba04917fded0ef5cd085c206fef0ee49/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggyPartitionSplitReader.java#L128-L147) always polls with `PollingStrategy.next()` and `autoCommit=true`. The broker offset advances before Flink processing completes, and the offset carried by the restored split is not used to select the polling position. Checkpoint bookkeeping also needs consideration: [`IggySourceReader.snapshotState()`](https://github.com/apache/iggy/blob/71f29618ba04917fded0ef5cd085c206fef0ee49/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceReader.java#L125-L132) captures fetched split offsets even while records remain queued. A separate real-server reader test emitted only record 0 before snapshotting offset 10; the broker already stored offset 9. Restoring that serialized split did not emit records 1–9. The scope here is the Flink source's checkpoint/replay behavior. Related: #2205 tracks broader Java offset APIs; #2928 concerns the separate Rust sink runtime. Simply disabling auto-commit is not a complete proposed fix; polling position and checkpoint state need a consistent ownership model. ### Contribution - [x] I'm willing to submit a pull request to fix this bug Could this be assigned to me? I'd like to agree on the Flink checkpoint/offset approach here before implementing the fix, following the connector design-discussion requirement. -- 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]
