[cc] Unblock shared drainer on CDC shutdown

[kvargha]

Problem Statement

A version-specific changelog consumer can stop being polled while its bounded output queue is full. A shared StoreBuffer drainer then blocks in ArrayBlockingQueue.put, preventing it from processing records for other stores assigned to the same drainer.

Consumer shutdown previously called daVinciClient.close() without first releasing that blocked enqueue. The consumer could close while the shared drainer remained parked, leaving unrelated CDC consumers unable to make progress.

Solution

• Make consumer close terminal and reject attempts to restart the same closed object.
• Fence output before DaVinci teardown by marking the consumer closed/stopped and clearing its served-version state.
• Clear the output queue before close to wake producers already blocked in put.
• Remove messages inserted by producers that crossed the shutdown boundary.
• Prevent poll() from emitting output after close begins.
• Release a pending asynchronous start and prevent its reporter thread from starting after close.
• Shut down the consumer's dedicated asynchronous-start executor.
• Keep the shared StoreBuffer drainer alive; the fix does not interrupt or modify shared drainer behavior.

Code changes

• Added new code behind a config.
• Introduced new log lines.
  • Confirmed if logs need to be rate limited to avoid excessive logging.

Concurrency-Specific Checks

• Code has no race conditions or thread safety issues.
• Proper synchronization mechanisms are used where needed.
• No blocking calls exist inside critical sections. stop() intentionally holds the consumer monitor during daVinciClient.close() to serialize terminal close against start/seek; the shared drainer path does not acquire this monitor.
• Existing thread-safe queues, maps, and atomic state are preserved.
• Interrupted shared drainer threads are not used as a cancellation mechanism.

How was this PR tested?

• Local code review completed.
• New unit tests added.
• New integration tests added.
• Modified or extended existing tests.
• Verified backward compatibility.

Red-before-green integration reproduction:

VersionSpecificCDCShutdownTest.testVersionSpecificCDCConsumerRestartWithinFlinkTimeout

• Before the production fix: Store B could not observe the first queued record because the shared drainer remained blocked in Store A's full output queue.
• After the production fix: Store B resumed, and the restarted Store A consumer replayed every expected record.
• After refactoring the regression, temporarily disabling the stop-side queue clears and post-put removal reproduced the same Store B starvation. Restoring the fix passed the same test once with retries disabled.

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

• No. You can skip the rest of this section.
• Yes. stop()/close() are terminal for a consumer instance. Callers that need to resume consumption must request a fresh consumer from VeniceChangelogConsumerClientFactory. The consumer interfaces now document this lifecycle contract, and restart integration tests use fresh instances. However, no one is depending on this behavior today.

[Copilot]

Pull request overview

This PR hardens DaVinci-backed version-specific CDC consumer shutdown so a blocked enqueue into a full per-consumer output queue can’t park a shared StoreBuffer drainer and starve other stores during Flink-style task restarts.

Changes:

• Make consumer shutdown terminal (isClosed), rejecting start/seek after close and preventing poll() from emitting output once close begins.
• Fence and drain output on shutdown (clear buffer, drop messages crossing the shutdown boundary) to unblock producers/drainers and avoid post-close output.
• Add deterministic unit and integration regressions reproducing the shared-drainer starvation and validating restart correctness.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.

File	Description
clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerDaVinciRecordTransformerImpl.java	Implements terminal close semantics and shutdown fencing/queue clearing to unblock shared drainer paths.
clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerDaVinciRecordTransformerImplTest.java	Adds unit coverage for terminal close, blocked-producer release, late-put removal, and poll suppression during close.
internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/consumer/VersionSpecificCDCShutdownTest.java	Adds deterministic integration reproduction of shared-drainer blockage and verifies store B resumes + store A restart replay.

---

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

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (1)

clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerDaVinciRecordTransformerImpl.java:353

• stop() now makes the consumer terminal (isClosed stays true) but never shuts down the per-consumer completableFutureThreadPool. Since these consumers can be created/closed repeatedly (e.g., Flink restarts), leaving an executor alive per closed consumer can accumulate threads and other resources over time. Consider shutting down the executor in the stop() finally block so resources are released deterministically.

    } finally {
      isStarted.set(false);
      pubSubMessages.clear();
      veniceChangelogConsumerClientFactory.deregisterClient(changelogClientConfig.getConsumerName());
      clearPartitionState(Collections.emptySet());

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated no new comments.

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

Copilot reviewed 7 out of 7 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (1)

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/consumer/VersionSpecificCDCShutdownTest.java:360

• getOutputQueue() hard-codes VeniceChangelogConsumerDaVinciRecordTransformerImpl.class when reflecting the output queue field. This will fail if the factory ever returns a subclass/proxy wrapper, even if it still has the same field. Since this test already has a readField(...) helper that walks the class hierarchy, use it to make the test less brittle.

  private BlockingQueue<?> getOutputQueue(VeniceChangelogConsumer<?, ?> consumer) throws ReflectiveOperationException {
    Field outputQueueField =
        VeniceChangelogConsumerDaVinciRecordTransformerImpl.class.getDeclaredField("pubSubMessages");
    outputQueueField.setAccessible(true);
    return (BlockingQueue<?>) outputQueueField.get(consumer);

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

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

Comments suppressed due to low confidence (1)

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/consumer/VersionSpecificCDCShutdownTest.java:307

• getOutputQueue() downcasts the consumer to VeniceChangelogConsumerDaVinciRecordTransformerImpl. Today the factory returns that type, but if it ever returns a wrapper/proxy this will fail with an unhelpful ClassCastException. Adding an instanceof assertion (with the actual class name in the message) will make failures diagnosable and keep the test intent clear.

  private BlockingQueue<?> getOutputQueue(VeniceChangelogConsumer<?, ?> consumer) {
    return ((VeniceChangelogConsumerDaVinciRecordTransformerImpl<?, ?>) consumer).getPubSubMessages();
  }

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

Copilot reviewed 7 out of 7 changed files in this pull request and generated 1 comment.