[server][test] Fix record-level metric transition timing and test thread leak (linkedin#2553)

[server][test] Fix record-level metric transition timing and test thread leak  

- StoreIngestionTask: invoke  
  `mayResumeRecordLevelMetricsForCurrentVersion()` inside  
  `processConsumerRecord()` when a partition transitions to complete.  
  Previously this logic ran only on the SIT `run()` thread, while  
  record processing occurs on the StoreBufferDrainer thread. After  
  EOP marked a partition complete, the drainer could process  
  subsequent records before the SIT loop updated  
  `recordLevelMetricEnabled`. The flag is now updated on the drainer  
  thread immediately after the ready-to-serve check so post-EOP  
  records in the same batch emit full metrics. Guarded by  
  `!recordLevelMetricEnabled.get() && !wasComplete &&  
  partitionConsumptionState.isComplete()` to trigger only on actual  
  transitions and avoid repeated partition-state scans.  
- KafkaConsumerServiceDelegatorTest: fix thread leak and misleading  
  assertion in `testKafkaConsumerServiceResubscriptionConcurrency`.  
  Resubscription threads are now always interrupted and joined before  
  assertions to prevent leaks and ensure the real race-condition  
  failure is not masked by thread-state checks.  
Root cause:  
- `testRecordLevelMetricForCurrentVersion`: resume logic executed only  
  on the SIT `run()` loop thread. Because record processing flows  
  through `ConsumptionTask → StorePartitionDataReceiver →  
  StoreBufferService → StoreBufferDrainer`, the drainer could process  
  post-EOP records before the SIT loop set  
  `recordLevelMetricEnabled = true`, causing  
  `recordTotalBytesConsumed` to be invoked fewer times than expected.  
- `testKafkaConsumerServiceResubscriptionConcurrency`: a  
  resubscription thread could terminate after throwing an exception  
  and count down the latch. The thread-state assertion then failed  
  (TERMINATED vs RUNNABLE/WAITING), obscuring the real race assertion  
  and leaking other running threads.

Author: sushantmane

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java

@@ -2909,7 +2909,15 @@ public void processConsumerRecord(
     long syncBytesInterval = getSyncBytesInterval(partitionConsumptionState);
     boolean recordsProcessedAboveSyncIntervalThreshold = (syncBytesInterval > 0
         && (partitionConsumptionState.getProcessedRecordSizeSinceLastSync() >= syncBytesInterval));
+    // Capture completion state before the ready-to-serve check so we can detect a transition.
+    boolean wasComplete = partitionConsumptionState.isComplete();
     getDefaultReadyToServeChecker().apply(partitionConsumptionState, recordsProcessedAboveSyncIntervalThreshold);
+    // Re-evaluate record-level metrics only when the current partition just transitioned to complete, so that records
+    // in the same batch as EOP get full metrics without waiting for the next loop cycle. The guard avoids repeatedly
+    // scanning all partition states on every record while metrics are still disabled.
+    if (!recordLevelMetricEnabled.get() && !wasComplete && partitionConsumptionState.isComplete()) {
+      mayResumeRecordLevelMetricsForCurrentVersion();
+    }
 
     /**
      * Syncing offset checking in syncOffset() should be the very last step for processing a record.

clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/KafkaConsumerServiceDelegatorTest.java

@@ -10,7 +10,6 @@
 import static org.mockito.Mockito.reset;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
-import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertTrue;
 
 import com.linkedin.davinci.config.VeniceServerConfig;
@@ -358,22 +357,17 @@ public void testKafkaConsumerServiceResubscriptionConcurrency() throws Exception
       });
     }
     long currentTime = System.currentTimeMillis();
-    Boolean raceConditionFound = countDownLatch.await(30, TimeUnit.SECONDS);
+    boolean raceConditionFound = countDownLatch.await(30, TimeUnit.SECONDS);
     long elapsedTime = System.currentTimeMillis() - currentTime;
+    // Always clean up threads before asserting so leaked threads don't affect subsequent tests.
     for (Thread infiniteSubUnSubThread: infiniteSubUnSubThreads) {
-      assertTrue(
-          infiniteSubUnSubThread.getState().equals(Thread.State.WAITING)
-              || infiniteSubUnSubThread.getState().equals(Thread.State.TIMED_WAITING)
-              || infiniteSubUnSubThread.getState().equals(Thread.State.BLOCKED)
-              || infiniteSubUnSubThread.getState().equals(Thread.State.RUNNABLE));
       infiniteSubUnSubThread.interrupt();
       infiniteSubUnSubThread.join();
-      assertEquals(Thread.State.TERMINATED, infiniteSubUnSubThread.getState());
     }
+    delegator.close();
     Assert.assertFalse(
         raceConditionFound,
         "Found race condition in KafkaConsumerService with time passed in milliseconds: " + elapsedTime);
-    delegator.close();
   }
 
   private Runnable getResubscriptionRunnableFor(

clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTaskTest.java

@@ -1623,13 +1623,7 @@ public void testRecordLevelMetricForCurrentVersion(boolean enableRecordLevelMetr
     StoreIngestionTaskTestConfig config = new StoreIngestionTaskTestConfig(Utils.setOf(PARTITION_FOO), () -> {
       verify(mockAbstractStorageEngine, timeout(TEST_TIMEOUT_MS))
           .put(PARTITION_FOO, putKeyFoo2, ByteBuffer.wrap(ValueRecord.create(SCHEMA_ID, putValue).serialize()));
-      /**
-       * Verify host-level metrics
-       *
-       * N.B.: the below verification for {@link HostLevelIngestionStats#recordTotalBytesConsumed(long)} is flaky, and
-       *       sometimes comes up with 1 fewer invocation than desired (in both branches of the if). The retries mask
-       *       the issue as the rate of flakiness is low. But there does seem to be something going on here...
-       */
+      // Verify host-level metrics
       if (enableRecordLevelMetricForCurrentVersionBootstrapping) {
         verify(mockStoreIngestionStats, timeout(TEST_TIMEOUT_MS).times(3)).recordTotalBytesConsumed(anyLong());
       } else {