bug(connectors): Flink source skips records after checkpoint recovery

Aperta
  • Qualcosa non va?
#4,199 1 commento 0 reazioni 1 assegnatario Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

@goutamadwant ci sta già lavorando.

Dal 16/9/2026.

Valutazione

Questa issue non è ancora stata valutata.

Descrizione

bug connectors
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:

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:

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() 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() 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
  • I'm willing to submit a pull request to fix this bug
Lingua principale
Rust
Stelle
5.2k
Fork
502
Merge medio
3g 8h
PR unite (30g)
146

Preparare l'ambiente

Apri in Codespaces

Avvia il container di sviluppo del progetto nel browser, con il tuo account GitHub.

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di apache/iggy

Tutte le issue di apache/iggy

Issue simili

Altre issue su Rust

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.