[test] Add test logic for verifying thread pool metric emitted before server shutdown during integration test. (#1505)

In production, we previously observed no state transition metrics emitted, here we add test logic to verify these metrics will be emitted for every Venice server instances.

Co-authored-by: Hao Xu <xhao@xhao-mn3.linkedin.biz>

Author: haoxu07

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/VeniceServerWrapper.java

@@ -51,13 +51,15 @@
 import com.linkedin.venice.server.VeniceServerContext;
 import com.linkedin.venice.servicediscovery.ServiceDiscoveryAnnouncer;
 import com.linkedin.venice.tehuti.MetricsAware;
+import com.linkedin.venice.tehuti.MockTehutiReporter;
 import com.linkedin.venice.utils.ForkedJavaProcess;
 import com.linkedin.venice.utils.PropertyBuilder;
 import com.linkedin.venice.utils.SslUtils;
 import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.Utils;
 import com.linkedin.venice.utils.VeniceProperties;
 import com.linkedin.venice.utils.metrics.MetricsRepositoryUtils;
+import io.tehuti.Metric;
 import io.tehuti.metrics.MetricsRepository;
 import java.io.File;
 import java.io.IOException;
@@ -78,6 +80,7 @@
 import org.apache.commons.io.FileUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.testng.Assert;
 
 
 /**
@@ -454,12 +457,28 @@ protected void internalStart() throws Exception {
   @Override
   protected void internalStop() throws Exception {
     if (!forkServer) {
+      verifyHelixParticipantServicePoolMetricsReporting(veniceServer);
       veniceServer.shutdown();
     } else {
       serverProcess.destroy();
     }
   }
 
+  private void verifyHelixParticipantServicePoolMetricsReporting(VeniceServer veniceServer) {
+    MetricsRepository metricsRepository = veniceServer.getMetricsRepository();
+    MockTehutiReporter reporter = new MockTehutiReporter();
+    metricsRepository.addReporter(reporter);
+    Metric activeThreadNumber = reporter.query(".Venice_L/F_ST_thread_pool--active_thread_number.LambdaStat");
+    Assert.assertNotNull(activeThreadNumber);
+    Assert.assertTrue(activeThreadNumber.value() >= 0);
+    Metric maxThreadNumber = reporter.query(".Venice_L/F_ST_thread_pool--max_thread_number.LambdaStat");
+    Assert.assertNotNull(maxThreadNumber);
+    Assert.assertTrue(maxThreadNumber.value() > 0);
+    Metric queuedTaskNumber = reporter.query(".Venice_L/F_ST_thread_pool--queued_task_number.LambdaStat");
+    Assert.assertNotNull(queuedTaskNumber);
+    Assert.assertTrue(queuedTaskNumber.value() >= 0);
+  }
+
   @Override
   protected void newProcess() throws Exception {
     if (forkServer) {

internal/venice-test-common/src/main/java/com/linkedin/venice/tehuti/MockTehutiReporter.java

@@ -18,6 +18,10 @@ public MockTehutiReporter() {
 
   @Override
   public void init(List<TehutiMetric> metrics) {
+    // Ensure lately added reporter has all metrics.
+    for (TehutiMetric metric: metrics) {
+      this.metrics.put(metric.name(), metric);
+    }
   }
 
   @Override