A note for the community
- Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
- If you are interested in working on this issue or have submitted a pull request, please leave a comment
Use Cases
We are using the kafka source to populate a memory enrichment table from a compacted Kafka topic. The topic is the source of truth for a small lookup table, and we want each Vector aggregator to rebuild its in-memory copy of that table from offset 0 on every restart.
The intended pattern, as documented in the kafka source reference, is:
- Use a unique group_id per consumer instance.
- Set auto_offset_reset: earliest.
- Disable offset commits so a clean shutdown does not leave a __consumer_offsets entry that would cause the next start to resume mid-stream.
We can satisfy (1) and (2). What we cannot satisfy is (3): there is no first-class config option that prevents offsets from being committed.
In our use case, the topic is replicated state, not a queue. we always want to read everything from offset 0 on each start so the enrichment table is complete. Resuming from a committed offset means the
enrichment table boots up with only the records that have arrived since the previous shutdown, which is incorrect for our use case.
Attempted Solutions
- librdkafka_options.enable.auto.commit: "false". This disables librdkafka's background auto-commit timer, but Vector also performs explicit synchronous commits via commit_consumer_state(CommitMode::Sync) at four lifecycle points in
src/sources/kafka.rs:
- end of kafka_source (clean shutdown),
- inside KafkaSourceContext::shutdown,
- inside revoke_partitions (rebalance drain),
- inside pre_rebalance for the Rebalance::Revoke arm.
These bypass the librdkafka setting and call the broker commit RPC directly.
- Rotating group_id per Vector restart. Our current workaround. We append a per-process identifier to group_id (e.g. ${HOSTNAME}-${VECTOR_RUN_ID} where VECTOR_RUN_ID is set by a wrapper script before Vector launches). Each new group has no committed
offsets, so auto_offset_reset: earliest triggers and replay works.
This is functional but undesirable:
- Requires every operator deploying this pattern to script their own group_id rotation.
- Old groups accumulate in __consumer_offsets until offsets.retention.minutes reaps them.
- Group churn is visible to anyone monitoring consumer groups on the broker.
Proposal
Add a boolean config option on the kafka source that disables both Vector's explicit commit_consumer_state calls and librdkafka's auto-commit. Suggested name: commit_offsets, default true (preserves current behaviour).
References
No response
Version
vector 0.51.1
A note for the community
Use Cases
We are using the kafka source to populate a memory enrichment table from a compacted Kafka topic. The topic is the source of truth for a small lookup table, and we want each Vector aggregator to rebuild its in-memory copy of that table from offset 0 on every restart.
The intended pattern, as documented in the kafka source reference, is:
We can satisfy (1) and (2). What we cannot satisfy is (3): there is no first-class config option that prevents offsets from being committed.
In our use case, the topic is replicated state, not a queue. we always want to read everything from offset 0 on each start so the enrichment table is complete. Resuming from a committed offset means the
enrichment table boots up with only the records that have arrived since the previous shutdown, which is incorrect for our use case.
Attempted Solutions
src/sources/kafka.rs:
These bypass the librdkafka setting and call the broker commit RPC directly.
offsets, so auto_offset_reset: earliest triggers and replay works.
This is functional but undesirable:
Proposal
Add a boolean config option on the kafka source that disables both Vector's explicit commit_consumer_state calls and librdkafka's auto-commit. Suggested name: commit_offsets, default true (preserves current behaviour).
References
No response
Version
vector 0.51.1