SS-455 - source: commit Kafka offsets on change and on a timer

[martykulma]

Each worker commits only when its offsets change, plus on a timer. The timer is needed because the consumer only ever assigns partitions, which results in the brokers treating the group as standalone. For standalone groups, brokers expire a partition's offset offsets.retention.minutes after its last commit, whether or not it is still current. The interval is the kafka_offset_commit_refresh_interval dyncfg: 10 minutes by default, 10 seconds in CI, zero disables it. A failed commit is retried on the next resume upper.

🤖 Generated with Claude Code
👨 Improved by human

[linear-code]

SS-455

[def-]

QA LLM Review

1. MEDIUM -- Idle partitions are never re-committed, so the broker expires their consumer-group offsets

src/storage/src/source/kafka.rs:1163

With the offsets != self.committed_offsets check, a worker whose partitions receive no new messages commits them once and never again, even while the rest of the source keeps advancing. Materialize consumes with assign(), which makes it a standalone consumer group, and brokers delete a standalone group's offset for a partition offsets.retention.minutes (7 days by default) after that partition's last commit. Lag-monitoring tools, which the Kafka source docs say these commits exist for, then lose those partitions.

Details

On main, reclock_committed_upper calls tx.send_replace every time it runs (source_reader_pipeline.rs:685 at the merge base), so every worker re-commits its offsets about once per tick, and that refreshes the commit timestamp. On #39461, any change to ResumeUppers still re-commits every worker's partitions. This PR takes away that last refresh for idle partitions. Two examples: a keyed topic where some partitions get no writes for a week, or a cluster with at least as many workers as partitions, where each worker owns at most one partition. After the retention window, kafka-consumer-groups --describe shows - for those partitions, and lag exporters either drop them or report them as missing. Data ingestion is not affected.

Suggested fix: keep the dedup, but re-commit unchanged offsets once the last successful commit is older than a refresh interval set well below typical broker retention (for example one hour, or a dyncfg). Because #39461 switched to send_if_modified, process_frontier is never called for a fully idle source. So the refresh needs a timer in resume_uppers_process_loop, for example a select! over the stream and a tokio::time::interval, rather than only a looser condition here. The same timer would also retry a failed commit on an idle source. Today the doc comment's "retried on the next resume upper" may never happen for such a source.

[martykulma]

1. MEDIUM -- Idle partitions are never re-committed, so the broker expires their consumer-group offsets

Confirmed and addressed