[da-vinci][changelog] Keep lag reporter alive when recordStats fails

[minhmo1620]

Problem Statement

The changelog consumer's background reporter thread emits two metrics on a fixed interval:
HeartBeatDelay and CurrentConsumingVersion. It is started at most once per consumer, and
startHeartbeatReporterIfNeeded() guards the start with a Thread.State.NEW check, so once the
thread terminates it is never restarted.

The run loop only caught InterruptedException:

while (!Thread.interrupted()) {
  try {
    recordStats(getLastHeartbeatPerPartition(), changeCaptureStats, getTopicAssignment());
    TimeUnit.SECONDS.sleep(...);
  } catch (InterruptedException e) {
    ...
  }
}

Any RuntimeException escaping recordStats therefore propagates out of run() and permanently
kills lag reporting for the rest of the consumer's lifetime. recordStats is not exception-free —
it calls Version.parseVersionFromKafkaTopicName, which does an unguarded Integer.parseInt on the
topic name and throws NumberFormatException for any assignment entry that is not a versioned topic.

Two things make this failure mode hard to notice:

1. The poll-path metrics (PollCount, RecordsConsumedCount, VersionSwapCount) are emitted from a
   different code path and keep reporting normally, so the consumer still looks healthy. Only the two
   reporter-thread metrics go silent.
2. An uncaught exception in a thread is routed to the default handler and written to stderr, so it
   does not go through the application logger and can be absent from log aggregation entirely.

Solution

Wrap the recordStats call so that a failed reporting cycle is logged and the loop continues on the
next interval, instead of terminating the thread. The InterruptedException handling is unchanged,
so shutdown behaviour on close() is preserved.

The same loop pattern exists in both VeniceChangelogConsumerImpl.HeartbeatReporterThread and
VeniceChangelogConsumerDaVinciRecordTransformerImpl.BackgroundReporterThread, so both are updated.

Exception is caught rather than Throwable, so Error conditions such as OutOfMemoryError still
propagate rather than being swallowed and retried every interval.

Code changes

• Added new code behind a config. If so list the config names and their default values in the PR description.
• Introduced new log lines.
  • Confirmed if logs need to be rate limited to avoid excessive logging.
    • The new LOGGER.error can fire at most once per reporter interval
      (backgroundReporterThreadSleepIntervalInSeconds, default 60s) per consumer, so it is
      inherently rate limited and does not need additional throttling.

Concurrency-Specific Checks

Both reviewer and PR author to verify

• Code has no race conditions or thread safety issues.
• Proper synchronization mechanisms (e.g., synchronized, RWLock) are used where needed.
• No blocking calls inside critical sections that could lead to deadlocks or performance degradation.
• Verified thread-safe collections are used (e.g., ConcurrentHashMap, CopyOnWriteArrayList).
• Validated proper exception handling in multi-threaded code to avoid silent thread termination.
  • This is precisely what the PR fixes.

How was this PR tested?

• New unit tests added.
• New integration tests added.
• Modified or extended existing tests.
• Verified backward compatibility (if applicable).

Added testMetricReportingThreadSurvivesRecordStatsFailure, which drives the failure through the
production path by making getTopicAssignment() throw, starts the real reporter thread, and asserts
the thread is still alive after a reporting cycle.

The test was verified to be a genuine regression test: it fails on the unmodified source (the
thread terminates) and passes with the fix applied. VeniceChangelogConsumerImplTest and
VeniceChangelogConsumerDaVinciRecordTransformerImplTest both pass, and root spotlessCheck is clean.

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

No. This only changes failure handling inside an internal reporter thread. In the failure case the
thread now survives and keeps emitting metrics on subsequent intervals instead of stopping silently;
the success path is unchanged.

---

🤖 Generated with GitHub Copilot CLI

[Copilot]

🟡 Changes recommended

Strengthen the existing regression test and add coverage for the transformer reporter thread.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Keeps changelog metric reporter threads alive when recordStats() throws.

Changes:

• Catches and logs per-cycle reporting failures in both reporter threads.
• Adds regression coverage for heartbeat reporter survival.
• Preserves existing interruption and shutdown behavior.

File summaries

File	Summary
clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImplTest.java	Tests reporter survival; review requests stronger synchronization with completed reporting cycles.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImpl.java	Keeps the heartbeat reporter alive after reporting failures.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerDaVinciRecordTransformerImpl.java	Keeps the background reporter alive; review requests equivalent regression coverage.

Review details

Suppressed comments (1)

clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImplTest.java:726

• atLeastOnce() only waits until Mockito records the call; on the unfixed implementation that invocation is recorded before the thrown NumberFormatException finishes unwinding, so the following isAlive() check can observe the thread during that brief window and pass even though it is about to terminate. Use a short reporter interval and require a second assignment call (or otherwise wait for the cycle to complete) so this regression test proves the loop continued.

          () -> Mockito.verify(mockPubSubConsumer, atLeastOnce()).getAssignment());

• Files reviewed: 3/3 changed files
• Comments generated: 1
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[minhmo1620]

Both review points addressed in 9118faa.

1. atLeastOnce() could pass on the unfixed implementation (suppressed comment, VeniceChangelogConsumerImplTest:726)

Agreed — the invocation is recorded before the NumberFormatException finishes unwinding, so a single-call assertion can catch the thread mid-unwind. Changed to set setBackgroundReporterThreadSleepIntervalInSeconds(1L) and require atLeast(2) calls to getAssignment(), so the assertion only passes if the loop genuinely resumed for a second cycle.

2. No coverage for BackgroundReporterThread

Added the equivalent test in VeniceChangelogConsumerDaVinciRecordTransformerImplTest, with reporter-thread cleanup in a finally block. Details in the inline reply.

Verification — ran both suites in each direction rather than just asserting the new tests pass:

	VeniceChangelogConsumerImplTest	...DaVinciRecordTransformerImplTest
With the try/catch guard	PASS (1.06s)	PASS (1.52s)
Guard reverted in main source	FAIL, every retry	FAIL, every retry

Full suites green with the guard restored: 22 tests / 0 failures and 28 tests / 0 failures. spotlessApply was a no-op.

[Copilot]

🟢 Approval recommended

No unresolved issues were identified that would block approval.

Review details

• Files reviewed: 4/4 changed files
• Comments generated: 0 new
• Review effort level: Lite