Native Kafka source: Flink semantic parity and remaining tails

[jordepic]

What

The remaining semantic-parity, feature, and productionization work for the native Kafka source. The source is built, default-on, and decisively faster than the shallow path on every format (divergences/19, docs/optimizations.md); per-partition watermarks/idleness and specific-offsets/topic-pattern startup have shipped.

Correctness parity comes before expanding coverage or publishing further exactly-once Kafka comparisons.

Flink checkpoint and consume semantics — highest priority

• Bounded stopping must use the Kafka consumer position and enforce the captured stop while building each partition batch. Today progress comes from the last delivered non-null payload and a whole poll batch can overshoot the enumerator's captured latest offset. Match Flink's KafkaPartitionSplitReader: emit only records whose offsets are below the stop, finish when the actual consumer position reaches it, and finish immediately when start >= stop. Cover empty partitions, records appended after stop capture, and multiple partitions independently.
• Make bounded read_committed completion control-record aware. Aborted transactional records and control markers advance Kafka's position without producing application records; they must still let the split reach its stop rather than hang at the tail.
• Propagate Kafka consumer failures. The native callback currently discards all librdkafka error events. Classify benign events such as partition EOF separately and fail the Flink source for the same fatal/non-retriable authentication, authorization, unknown-topic, and offset failures that stock Flink surfaces.
• Handle null Kafka values/tombstones without losing offset progress. Every consumed record must advance its split position. Preserve a nullable body for formats that can represent it, and otherwise invoke the format's normal failure/ignore-parse-errors behavior instead of silently dropping it in the consumer callback.
• Match group-offset startup and auto.offset.reset exactly. In particular, absent committed offsets with no explicit reset strategy must fail (none) like Flink rather than inherit librdkafka's latest default. Test explicit earliest/latest/none plus missing and out-of-range committed offsets.
• Commit offsets on completed checkpoints when Flink's commit.offsets.on.checkpoint is enabled. Recovery correctness continues to come exclusively from Flink checkpoint state; Kafka commits are external monitoring state and a commit failure must not fail or corrupt a completed checkpoint. Include finished splits and discard pending commits from older checkpoints exactly as KafkaSourceReader does.
• Implement partition pause/resume parity. Delegate FLIP-27 split pause/resume to librdkafka so watermark alignment remains valid if the planner later admits aligned Kafka watermarks. Keep the current planner fallback until this is covered.

Checkpoint-performance validation

• Measure Arrow-batch checkpoint granularity. A native source advances checkpoint state once per emitted Arrow batch rather than once per Kafka record. This is result-correct, but can enlarge the replay unit and delay a barrier behind one batch. Benchmark checkpoint duration and recovery replay at representative batch sizes; retain the columnar batch boundary unless evidence justifies a smaller native batch—not a rowwise hot path.
• Add a real FLIP-27 failover harness with a transactional verifier. 2ebf873 completes a checkpoint, fails and restarts the task, restores native split state, and proves 10,000 records remain unique through an exactly-once Kafka sink plus read_committed verifier.
• Add the ordinary checkpointed-sink failover variant. 3b2c22a restores the native source into Flink's checkpointed filesystem sink and proves 10,000 records remain gap-free and unique independently of Kafka transaction visibility.

Kafka observability and identity parity

• Register Kafka source metrics. At minimum expose current/committed offsets, records lag, bytes in, and librdkafka consumer metrics with stable Flink metric-group names where practical.
• Remove the unconditional synthetic group ID. Preserve Flink's group-ID requirements and client-ID behavior: require a group for group-offset startup/commits, but do not manufacture group membership for earliest/latest/specific/timestamp modes. Verify ACL behavior and client identity.
• Verify cancellation and split-change latency. wakeUp() currently relies on the 100 ms bounded poll timeout. Keep it only if cancellation, checkpoint shutdown, and partition discovery/reassignment tests show Flink-equivalent responsiveness; otherwise wire librdkafka queue/consumer wakeup.

Feature gaps

• key.format / multi-format tables — currently fall back.
• SASL/SSL at runtime — the bundled librdkafka is built without SSL/SASL. A SASL/SSL cluster needs the ssl + gssapi-vendored rdkafka features (heavier build); the config translator already emits the config, so this is a build-features + verification item.
• Bounded modes beyond latest-offset — still fall back.

Productionize the fast path

• Linux mimalloc full validation — the symbol-level link-alias check passed in a container build (aliases bind library-locally, mirroring the macOS result), but a full suite + benchmark-ladder run on a Linux host is still outstanding before recommending the feature there unreservedly.
• Multi-broker parallel fetch — unmeasured (single-broker Testcontainers only). librdkafka fetches per-broker on parallel threads vs the Java client's single network thread, so this can only widen the native win — but confirm on a 3-broker cluster before quoting it.
• Revisit mimalloc v3 when its dlopen'd lazy thread-init stabilizes (crashes in _mi_subproc today — divergences/19). v3's cross-thread free is markedly faster than v2's on the raw-consume pattern (5.3M/s vs 3.4M/s micro).
• Upstream librdkafka ARM hardware CRC32C (crc32c.c is x86 SSE4.2 only). Would let us re-pin check.crcs=true at ~no cost on ARM.

Acceptance gate

Before this issue's semantic-parity section is considered complete, run native-vs-stock Flink parity tests for:

• normal checkpoint/restore and rescale;
• completed-checkpoint Kafka offset visibility and failed asynchronous commits;
• empty bounded partitions and start == stop;
• appends after latest-offset capture, proving no overshoot;
• read_committed aborted transactions/control records at the bounded tail;
• null values at the middle and tail of a partition;
• broker/auth/authorization/invalid-offset failures;
• group-offset startup with each reset strategy;
• watermark-alignment pause/resume; and
• bounded and unbounded multi-partition consumption.

All performance measurements must use the release bench profile with the Kafka and mimalloc features, and must report checkpoint/recovery costs alongside steady-state throughput.

[jordepic]

Bounded empty/start-at-stop acceptance shipped in 98708fd. The full FLIP-27 SQL path now terminates both an empty earliest-to-latest partition and a non-empty latest-to-latest partition with zero rows. The fix enables librdkafka partition-EOF progress and keeps symbolic-start splits assigned until the fetcher reports completion; bounded-cap and read_committed control-record regressions remain green.