fix: route telemetry HTTP through shared connection stack; fix close() flush race

- DatabricksTelemetryExporter.sendRequest and FeatureFlagCache.fetchWithRetry
  now use connectionProvider.getRetryPolicy().invokeWithRetry(), matching the
  CloudFetchResultHandler pattern instead of bespoke fetch/retry logic
- MetricsAggregator: add closing flag so batch-triggered fire-and-forget flushes
  are suppressed during close(), ensuring a single awaited flushForClose() drains
  all remaining metrics without racing past process.exit()

Co-authored-by: samikshya-chand_data

Author: samikshya-db

lib/telemetry/DatabricksTelemetryExporter.ts

@@ -15,7 +15,7 @@
  */
 
 import { v4 as uuidv4 } from 'uuid';
-import fetch, { RequestInit, Response } from 'node-fetch';
+import fetch, { RequestInit, Response, Request } from 'node-fetch';
 import IClientContext from '../contracts/IClientContext';
 import { LogLevel } from '../contracts/IDBSQLLogger';
 import IAuthentication from '../connection/contracts/IAuthentication';
@@ -304,7 +304,13 @@ export default class DatabricksTelemetryExporter {
   private async sendRequest(url: string, init: RequestInit): Promise<Response> {
     const connectionProvider = await this.context.getConnectionProvider();
     const agent = await connectionProvider.getAgent();
-    return fetch(url, { ...init, agent });
+    const retryPolicy = await connectionProvider.getRetryPolicy();
+    const requestConfig: RequestInit = { agent, ...init };
+    const result = await retryPolicy.invokeWithRetry(() => {
+      const request = new Request(url, requestConfig);
+      return fetch(request).then((response) => ({ request, response }));
+    });
+    return result.response;
   }
 
   private toTelemetryLog(

lib/telemetry/FeatureFlagCache.ts

@@ -14,12 +14,11 @@
  * limitations under the License.
  */
 
-import fetch, { RequestInit, Response } from 'node-fetch';
+import fetch, { RequestInit, Response, Request } from 'node-fetch';
 import IClientContext from '../contracts/IClientContext';
 import { LogLevel } from '../contracts/IDBSQLLogger';
 import IAuthentication from '../connection/contracts/IAuthentication';
 import { buildTelemetryUrl, normalizeHeaders } from './telemetryUtils';
-import ExceptionClassifier from './ExceptionClassifier';
 import buildUserAgentString from '../utils/buildUserAgentString';
 import driverVersion from '../version';
 
@@ -193,20 +192,13 @@ export default class FeatureFlagCache {
   private async fetchWithRetry(url: string, init: RequestInit): Promise<Response> {
     const connectionProvider = await this.context.getConnectionProvider();
     const agent = await connectionProvider.getAgent();
-    const logger = this.context.getLogger();
-
-    try {
-      return await fetch(url, { ...init, agent });
-    } catch (err: any) {
-      if (!ExceptionClassifier.isRetryable(err)) {
-        throw err;
-      }
-      logger.log(LogLevel.debug, `Feature flag fetch retry after transient: ${err?.code ?? err?.message ?? err}`);
-      await new Promise((resolve) => {
-        setTimeout(resolve, 100 + Math.random() * 100);
-      });
-      return fetch(url, { ...init, agent });
-    }
+    const retryPolicy = await connectionProvider.getRetryPolicy();
+    const requestConfig: RequestInit = { agent, ...init };
+    const result = await retryPolicy.invokeWithRetry(() => {
+      const request = new Request(url, requestConfig);
+      return fetch(request).then((response) => ({ request, response }));
+    });
+    return result.response;
   }
 
   private async getAuthHeaders(): Promise<Record<string, string>> {

lib/telemetry/MetricsAggregator.ts

@@ -51,6 +51,8 @@ export default class MetricsAggregator {
 
   private closed = false;
 
+  private closing = false;
+
   private batchSize: number;
 
   private flushIntervalMs: number;
@@ -317,9 +319,11 @@ export default class MetricsAggregator {
       );
     }
 
-    if (this.pendingMetrics.length >= this.batchSize) {
+    if (this.pendingMetrics.length >= this.batchSize && !this.closing) {
       // resetTimer=false so the periodic tail-drain keeps its cadence even
       // under sustained batch-size bursts.
+      // Suppressed during close() so fire-and-forget promises don't race past
+      // the single awaited flushForClose().
       const logger = this.context.getLogger();
       Promise.resolve(this.flush(false)).catch((err: any) => {
         logger.log(LogLevel.debug, `Batch-trigger flush failed: ${err?.message ?? err}`);
@@ -406,6 +410,7 @@ export default class MetricsAggregator {
 
   async close(): Promise<void> {
     const logger = this.context.getLogger();
+    this.closing = true;
     this.closed = true;
 
     try {