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

オープン
  • 何か問題がありますか？
#4,199 コメント 1 件 リアクション 0 件 担当者 1 名 GitHub で見る

メンテナーはふだん 1 日以内に返信

@goutamadwant がすでに取り組んでいます。

2026年9月16日 から。

評価

この issue はまだ評価されていません。

説明

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
主要言語
Rust
スター
5.2k
フォーク
516
平均マージ
2日 21時間
マージ済み PR（30日）
159

環境構築

Codespaces で開く

このプロジェクトの開発コンテナを、あなたの GitHub アカウントでブラウザ上に起動します。

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

apache/iggy のほかの issue

apache/iggy の issue をすべて見る

似ている issue

Rust の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。