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

Abierto
  • ¿Algo no está bien?
#4,199 1 comentario 0 reacciones 1 asignado Ver en GitHub

Los mantenedores suelen responder en 1 día

@goutamadwant ya está trabajando en esto.

Desde el 16/9/2026.

Evaluación

Este issue todavía no se ha evaluado.

Descripción

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
Lenguaje dominante
Rust
Estrellas
5.2k
Forks
502
Merge medio
3 d 8 h
PR fusionados (30 d)
146

Preparar el entorno

Abrir en Codespaces

Inicia el contenedor de desarrollo del proyecto en tu navegador, con tu propia cuenta de GitHub.

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de apache/iggy

Todos los issues de apache/iggy

Issues similares

Más issues de Rust

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.