[PECOBLR-703][PECOBLR-735] Improve telemetry performance + Force telemetry to use daemon threads (#931)

## Description

- Performance improvements : 
  -  Centralised thread pool, to avoid OOM and improve performance
  - Force executors on every export event
 - There are 2 types of thread creation : 
    - One for in-general flush 
- One for scheduling flush every 300 seconds. (Daemon threads were not
being in this - this prevented JVM from exiting. This PR changes that
too)


## Testing
- Manually tested E2E using `ForceEnableTelemetry=1`


NO_CHANGELOG=true

Author: samikshya-db

src/main/java/com/databricks/jdbc/common/util/ProcessNameUtil.java

@@ -2,15 +2,13 @@
 
 import static com.databricks.jdbc.common.util.WildcardUtil.isNullOrEmpty;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 /**
  * Utility class for determining the current process name as it would appear in Activity Monitor.
+ * Note : removing logging as it methods are called on static INIT and logging might not be fully
+ * configured.
  */
 public class ProcessNameUtil {
   private static final String FALL_BACK_PROCESS_NAME = "UnknownJavaProcess";
-  private static final Logger LOGGER = LoggerFactory.getLogger(ProcessNameUtil.class);
 
   /**
    * Gets the current process name as it would appear in Activity Monitor.
@@ -22,14 +20,12 @@ public static String getProcessName() {
       // Step 1: Try ProcessHandle API (Java 9+)
       String processName = getProcessNameFromHandle();
       if (!isNullOrEmpty(processName)) {
-        LOGGER.trace("getProcessNameFromHandle: {}", processName);
         return processName;
       }
 
       // Fallback
       return FALL_BACK_PROCESS_NAME;
     } catch (Exception e) {
-      LOGGER.trace("Error getting process name {}, returning fallback", e);
       return FALL_BACK_PROCESS_NAME;
     }
   }
@@ -62,7 +58,6 @@ public static String getProcessNameFromHandle() {
       }
       return null;
     } catch (Exception e) {
-      LOGGER.trace("Error getting process name from handle {}, returning null", e);
       return null;
     }
   }

src/main/java/com/databricks/jdbc/model/telemetry/StatementTelemetryDetails.java

@@ -10,7 +10,10 @@
 public class StatementTelemetryDetails {
   private boolean isInternalCall;
   private final ChunkDetails chunkDetails;
-  private final ResultLatency resultLatency;
+  private final ResultLatency resultLatency =
+      new ResultLatency()
+          .setResultSetReadyLatencyMillis(null)
+          .setResultSetConsumptionLatencyMillis(null);
   private final OperationDetail operationDetail;
   private Long operationLatencyMillis;
   private final String statementId;
@@ -25,10 +28,6 @@ public class StatementTelemetryDetails {
 
   public StatementTelemetryDetails(String statementId) {
     this.chunkDetails = new ChunkDetails();
-    this.resultLatency =
-        new ResultLatency()
-            .setResultSetReadyLatencyMillis(null)
-            .setResultSetConsumptionLatencyMillis(null);
     this.isInternalCall = false;
     this.operationDetail = new OperationDetail(isInternalCall);
     this.operationLatencyMillis = null;

src/main/java/com/databricks/jdbc/telemetry/TelemetryClient.java

@@ -7,9 +7,12 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 public class TelemetryClient implements ITelemetryClient {
 
@@ -24,6 +27,19 @@ public class TelemetryClient implements ITelemetryClient {
   private ScheduledFuture<?> flushTask;
   private final int flushIntervalMillis;
 
+  private static ThreadFactory createSchedulerThreadFactory() {
+    return new ThreadFactory() {
+      private final AtomicInteger threadNumber = new AtomicInteger(1);
+
+      @Override
+      public Thread newThread(Runnable r) {
+        Thread thread = new Thread(r, "Telemetry-Scheduler-" + threadNumber.getAndIncrement());
+        thread.setDaemon(true);
+        return thread;
+      }
+    };
+  }
+
   public TelemetryClient(
       IDatabricksConnectionContext connectionContext,
       ExecutorService executorService,
@@ -35,7 +51,7 @@ public TelemetryClient(
     this.databricksConfig = config;
     this.executorService = executorService;
     this.scheduledExecutorService =
-        java.util.concurrent.Executors.newSingleThreadScheduledExecutor();
+        Executors.newSingleThreadScheduledExecutor(createSchedulerThreadFactory());
     this.flushIntervalMillis = context.getTelemetryFlushIntervalInMilliseconds();
     this.lastFlushedTime = System.currentTimeMillis();
     schedulePeriodicFlush();
@@ -50,7 +66,7 @@ public TelemetryClient(
     this.databricksConfig = null;
     this.executorService = executorService;
     this.scheduledExecutorService =
-        java.util.concurrent.Executors.newSingleThreadScheduledExecutor();
+        Executors.newSingleThreadScheduledExecutor(createSchedulerThreadFactory());
     this.flushIntervalMillis = context.getTelemetryFlushIntervalInMilliseconds();
     this.lastFlushedTime = System.currentTimeMillis();
     schedulePeriodicFlush();

src/main/java/com/databricks/jdbc/telemetry/TelemetryClientFactory.java

@@ -10,6 +10,8 @@
 import java.util.LinkedHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
 
 public class TelemetryClientFactory {
 
@@ -26,8 +28,21 @@ public class TelemetryClientFactory {
 
   private final ExecutorService telemetryExecutorService;
 
+  private static ThreadFactory createThreadFactory() {
+    return new ThreadFactory() {
+      private final AtomicInteger threadNumber = new AtomicInteger(1);
+
+      @Override
+      public Thread newThread(Runnable r) {
+        Thread thread = new Thread(r, "Telemetry-Thread-" + threadNumber.getAndIncrement());
+        thread.setDaemon(false);
+        return thread;
+      }
+    };
+  }
+
   private TelemetryClientFactory() {
-    telemetryExecutorService = Executors.newFixedThreadPool(10);
+    telemetryExecutorService = Executors.newFixedThreadPool(10, createThreadFactory());
   }
 
   public static TelemetryClientFactory getInstance() {

src/main/java/com/databricks/jdbc/telemetry/TelemetryPushTask.java

@@ -52,7 +52,7 @@ class TelemetryPushTask implements Runnable {
 
   @Override
   public void run() {
-    LOGGER.debug("Pushing Telemetry logs of size {}", queueToBePushed.size());
+    LOGGER.trace("Pushing Telemetry logs of size {}", queueToBePushed.size());
     TelemetryRequest request = new TelemetryRequest();
     if (queueToBePushed.isEmpty()) {
       return;
@@ -67,7 +67,7 @@ public void run() {
                         try {
                           return objectMapper.writeValueAsString(event);
                         } catch (JsonProcessingException e) {
-                          LOGGER.error(
+                          LOGGER.trace(
                               "Failed to serialize Telemetry event {} with error: {}", event, e);
                           return null; // Return null for failed serialization
                         }
@@ -98,13 +98,13 @@ public void run() {
         TelemetryResponse telResponse =
             objectMapper.readValue(
                 EntityUtils.toString(response.getEntity()), TelemetryResponse.class);
-        LOGGER.debug(
+        LOGGER.trace(
             "Pushed Telemetry logs with request-Id {} with events {} with error count {}",
             response.getFirstHeader(REQUEST_ID_HEADER),
             telResponse.getNumProtoSuccess(),
             telResponse.getErrors().size());
         if (!telResponse.getErrors().isEmpty()) {
-          LOGGER.debug("Failed to push telemetry logs with error: {}", telResponse.getErrors());
+          LOGGER.trace("Failed to push telemetry logs with error: {}", telResponse.getErrors());
         }
         if (queueToBePushed.size() != telResponse.getNumProtoSuccess()) {
           LOGGER.debug(

src/test/java/com/databricks/jdbc/telemetry/latency/TelemetryCollectorTest.java

@@ -4,17 +4,14 @@
 import static org.mockito.Mockito.*;
 
 import com.databricks.jdbc.common.util.DatabricksThreadContextHolder;
-import com.databricks.jdbc.model.telemetry.StatementTelemetryDetails;
 import com.databricks.jdbc.model.telemetry.latency.ChunkDetails;
 import com.databricks.jdbc.model.telemetry.latency.OperationType;
-import com.databricks.jdbc.telemetry.TelemetryHelper;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.EnumSource;
-import org.mockito.MockedStatic;
 
 public class TelemetryCollectorTest {
   private final TelemetryCollector handler = TelemetryCollector.getInstance();
@@ -74,12 +71,16 @@ void testRecordOperationLatency_WithCloseOperation() {
     String methodName = "closeStatement";
     long latency = 100L;
 
-    try (MockedStatic<TelemetryHelper> mockedStatic = mockStatic(TelemetryHelper.class)) {
-      handler.recordOperationLatency(latency, methodName);
-
-      mockedStatic.verify(
-          () -> TelemetryHelper.exportTelemetryLog(any(StatementTelemetryDetails.class)));
+    // Ensure telemetry is collected for the statement id prior to close
+    handler.recordGetOperationStatus(TEST_STATEMENT_ID, 1L);
+    handler.recordOperationLatency(latency, methodName);
+    // Allow asynchronous export to complete and internal map to be cleared
+    try {
+      Thread.sleep(200);
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
     }
+    assertNull(handler.getTelemetryDetails(TEST_STATEMENT_ID));
   }
 
   @ParameterizedTest