[changelog-consumer] Opt-in periodic offset commit to enable Kafka consumer-group lag monitoring

[minhmo1620]

Problem Statement

We want to leverage Kafka's existing consumer-group lag mechanism to detect laggy VeniceChangelogConsumer (DVRT) jobs. Kafka already computes lag for any consumer that commits to __consumer_offsets — the broker has the latest committed offset, the high-watermark, and exposes the difference through every standard consumer-group tool (kafka-consumer-groups.sh, Burrow, lag dashboards, regex-based alerts). Reusing this means we don't have to stand up a parallel lag-monitoring stack just for DVRT.

Today DVRT can't plug into this. It never writes to __consumer_offsets, so from the broker's point of view the consumer group doesn't exist, and every tool that reads broker-side lag returns empty.

Solution

Let the changelog consumer optionally publish its current position to __consumer_offsets purely so external tools can see it. Source-of-truth for offsets stays with the caller — Kafka-side commits are never read back on restart and a failed commit is swallowed.

• Add commitOffsets() to VeniceChangelogConsumer (default no-op; Kafka-backed impl delegates to the underlying consumer's commitSync()).
• Call it periodically from poll() at most once per consumerOffsetCommitIntervalMs. The interval is measured after each commit returns so a slow commit can't trigger an immediate re-commit on the next poll.

Opting in is fully explicit and requires two knobs. Commits actually happen only when both:

1. ChangelogClientConfig.consumerOffsetCommitIntervalMs > 0 (default 0 = off), and
2. A Kafka group.id is set on the consumer properties (caller configures this on the existing ChangelogClientConfig.consumerProperties bag).

If either is missing the path short-circuits silently — no Kafka call, no log noise. Existing callers see zero behavior change. No new factory overload or Properties parameter — callers wire both knobs through the existing ChangelogClientConfig API.

Code changes

• Added new code behind a config. ChangelogClientConfig.consumerOffsetCommitIntervalMs (default 0 = off). To enable, the caller must additionally set group.id on the consumer properties.
• Introduced new log lines. One LOGGER.warn (with stack trace) for actual broker-side commit failures. No log noise when commits are disabled or when group.id is unset — the path short-circuits silently.

Concurrency-Specific Checks

• Uses existing locks (consumerLock, subscriptionLock) — no new synchronization primitives.
• No new blocking calls inside critical sections.
• No new shared collections.
• Commit failures are caught and logged in two layers; never propagate to the caller.

How was this PR tested?

• New unit tests added across the adapter and the changelog impl:
  • commitSync() delegates to the underlying Kafka consumer when group.id is configured.
  • commitSync() short-circuits silently when group.id is missing (no Kafka call, no log).
  • commitSync() swallows real broker exceptions (logged once with stack).
  • commitOffsets() delegates to the underlying pub-sub consumer.
  • Periodic commit cadence is respected when the interval is positive; the default-zero / explicit-zero path doesn't commit at all.
• Modified or extended existing tests.
• Verified backward compatibility — interface additions are default no-ops; the commit path is fully off by default.

Does this PR introduce any user-facing or breaking changes?

• No. All additions are backwards-compatible and the new commit path is off by default.

[Copilot]

Pull request overview

This PR adds an optional “monitoring-only” offset commit path for Venice changelog consumers so external Kafka consumer-group monitoring tools can observe lag, and adds factory overloads to allow per-call Kafka consumer property overrides (e.g., group.id) without mutating global config.

Changes:

• Add commitSync() to PubSubConsumerAdapter (default no-op) and implement it in ApacheKafkaConsumerAdapter by delegating to Kafka commitSync() while swallowing failures.
• Add commitOffsets() to VeniceChangelogConsumer (default no-op) and periodically invoke it from poll() in the changelog consumer implementation, gated by consumerOffsetCommitIntervalMs.
• Add ChangelogClientConfig.consumerOffsetCommitIntervalMs (default 30s; 0 disables) and add Properties override overloads in VeniceChangelogConsumerClientFactory with tests.

Reviewed changes

Copilot reviewed 9 out of 9 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
internal/venice-common/src/main/java/com/linkedin/venice/pubsub/api/PubSubConsumerAdapter.java	Adds a monitoring-only commitSync() default method to the pub-sub consumer abstraction.
internal/venice-common/src/main/java/com/linkedin/venice/pubsub/adapter/kafka/consumer/ApacheKafkaConsumerAdapter.java	Implements monitoring-only commits by delegating to Kafka commitSync() and swallowing errors.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/VeniceChangelogConsumer.java	Adds a monitoring-only commitOffsets() default method to the changelog consumer API.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImpl.java	Periodically calls commitOffsets() from poll() based on the new interval config.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerClientFactory.java	Adds overloads to merge per-call consumer Properties into resolved per-store configs without leaking globally.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/ChangelogClientConfig.java	Introduces consumerOffsetCommitIntervalMs config with clone support.
internal/venice-common/src/test/java/com/linkedin/venice/pubsub/adapter/kafka/consumer/ApacheKafkaConsumerAdapterTest.java	Adds unit tests for commitSync() delegation and exception swallowing.
clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImplTest.java	Adds unit tests for commit interval config, delegation, cadence, and interval=0 behavior.
clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerClientFactoryTest.java	Adds test verifying per-call consumer overrides merge without mutating global config.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Copilot encountered an error and was unable to review this pull request. You can try again by re-requesting a review.

[Copilot]

Pull request overview

Copilot reviewed 7 out of 7 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (1)

clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImpl.java:789

• maybeCommitOffsets() uses System.currentTimeMillis() even though this class already has an injectable Time time abstraction (used elsewhere via time.getMilliseconds()). Using the Time field here would keep time handling consistent and make commit-cadence behavior easier to test deterministically (e.g., with TestMockTime).

  private void maybeCommitOffsets() {
    if (consumerOffsetCommitIntervalMs <= 0) {
      return;
    }
    long now = System.currentTimeMillis();
    if (now - lastCommitTimeMs >= consumerOffsetCommitIntervalMs) {
      commitOffsets();
      lastCommitTimeMs = System.currentTimeMillis();
    }