Address PR #325 review feedback

- Fix CircuitBreaker HALF_OPEN: any failure now immediately reopens the
  circuit instead of accumulating to failureThreshold, aligning with
  Resilience4j behavior used in the JDBC driver
- Add maxPendingMetrics bound to MetricsAggregator (default 500) to
  prevent unbounded buffer growth when exports keep failing
- Inline buildUrl into DatabricksTelemetryExporter and FeatureFlagCache;
  remove urlUtils.ts (single-use utility)
- Add missing unit tests for DatabricksTelemetryExporter, MetricsAggregator,
  and TelemetryEventEmitter (HTTP calls, retry logic, batching, flush
  timers, event routing)

Co-authored-by: Isaac

Author: samikshya-db

lib/contracts/IClientContext.ts

@@ -32,6 +32,7 @@ export interface ClientConfig {
   telemetryAuthenticatedExport?: boolean;
   telemetryCircuitBreakerThreshold?: number;
   telemetryCircuitBreakerTimeout?: number;
+  telemetryMaxPendingMetrics?: number;
 }
 
 export default interface IClientContext {

lib/telemetry/CircuitBreaker.ts

@@ -166,8 +166,9 @@ export class CircuitBreaker {
 
     logger.log(LogLevel.debug, `Circuit breaker failure (${this.failureCount}/${this.config.failureThreshold})`);
 
-    if (this.failureCount >= this.config.failureThreshold) {
-      // Transition to OPEN
+    // In HALF_OPEN state, any failure immediately reopens the circuit.
+    // In CLOSED state, reopen only after failureThreshold consecutive failures.
+    if (this.state === CircuitBreakerState.HALF_OPEN || this.failureCount >= this.config.failureThreshold) {
       this.state = CircuitBreakerState.OPEN;
       this.nextAttempt = new Date(Date.now() + this.config.timeout);
       logger.log(LogLevel.debug, `Circuit breaker transitioned to OPEN (will retry after ${this.config.timeout}ms)`);

lib/telemetry/DatabricksTelemetryExporter.ts

@@ -20,7 +20,6 @@ import { LogLevel } from '../contracts/IDBSQLLogger';
 import { TelemetryMetric, DEFAULT_TELEMETRY_CONFIG } from './types';
 import { CircuitBreakerRegistry } from './CircuitBreaker';
 import ExceptionClassifier from './ExceptionClassifier';
-import { buildUrl } from './urlUtils';
 
 /**
  * Databricks telemetry log format for export.
@@ -210,8 +209,8 @@ export default class DatabricksTelemetryExporter {
     // Determine endpoint based on authentication mode
     const authenticatedExport = config.telemetryAuthenticatedExport ?? DEFAULT_TELEMETRY_CONFIG.authenticatedExport;
     const endpoint = authenticatedExport
-      ? buildUrl(this.host, '/telemetry-ext')
-      : buildUrl(this.host, '/telemetry-unauth');
+      ? this.buildUrl(this.host, '/telemetry-ext')
+      : this.buildUrl(this.host, '/telemetry-unauth');
 
     // Format payload - each log is JSON-stringified to match JDBC format
     const telemetryLogs = metrics.map((m) => this.toTelemetryLog(m));
@@ -319,6 +318,16 @@ export default class DatabricksTelemetryExporter {
     return log;
   }
 
+  /**
+   * Build full URL from host and path, handling protocol correctly.
+   */
+  private buildUrl(host: string, path: string): string {
+    if (host.startsWith('http://') || host.startsWith('https://')) {
+      return `${host}${path}`;
+    }
+    return `https://${host}${path}`;
+  }
+
   /**
    * Generate a UUID v4.
    */

lib/telemetry/FeatureFlagCache.ts

@@ -18,7 +18,6 @@ import fetch from 'node-fetch';
 import IClientContext from '../contracts/IClientContext';
 import { LogLevel } from '../contracts/IDBSQLLogger';
 import driverVersion from '../version';
-import { buildUrl } from './urlUtils';
 
 /**
  * Context holding feature flag state for a specific host.
@@ -120,7 +119,7 @@ export default class FeatureFlagCache {
       const version = this.getDriverVersion();
 
       // Build feature flags endpoint for Node.js driver
-      const endpoint = buildUrl(host, `/api/2.0/connector-service/feature-flags/NODEJS/${version}`);
+      const endpoint = this.buildUrl(host, `/api/2.0/connector-service/feature-flags/NODEJS/${version}`);
 
       // Get authentication headers
       const authHeaders = await this.context.getAuthHeaders();
@@ -181,6 +180,16 @@ export default class FeatureFlagCache {
     }
   }
 
+  /**
+   * Build full URL from host and path, handling protocol correctly.
+   */
+  private buildUrl(host: string, path: string): string {
+    if (host.startsWith('http://') || host.startsWith('https://')) {
+      return `${host}${path}`;
+    }
+    return `https://${host}${path}`;
+  }
+
   /**
    * Gets the driver version without -oss suffix for API calls.
    * Format: "1.12.0" from "1.12.0-oss"

lib/telemetry/MetricsAggregator.ts

@@ -64,11 +64,14 @@ export default class MetricsAggregator {
 
   private flushIntervalMs: number;
 
+  private maxPendingMetrics: number;
+
   constructor(private context: IClientContext, private exporter: DatabricksTelemetryExporter) {
     try {
       const config = context.getConfig();
       this.batchSize = config.telemetryBatchSize ?? DEFAULT_TELEMETRY_CONFIG.batchSize;
       this.flushIntervalMs = config.telemetryFlushIntervalMs ?? DEFAULT_TELEMETRY_CONFIG.flushIntervalMs;
+      this.maxPendingMetrics = config.telemetryMaxPendingMetrics ?? DEFAULT_TELEMETRY_CONFIG.maxPendingMetrics;
 
       // Start periodic flush timer
       this.startFlushTimer();
@@ -80,6 +83,7 @@ export default class MetricsAggregator {
       // Initialize with default values
       this.batchSize = DEFAULT_TELEMETRY_CONFIG.batchSize;
       this.flushIntervalMs = DEFAULT_TELEMETRY_CONFIG.flushIntervalMs;
+      this.maxPendingMetrics = DEFAULT_TELEMETRY_CONFIG.maxPendingMetrics;
     }
   }
 
@@ -283,11 +287,21 @@ export default class MetricsAggregator {
   }
 
   /**
-   * Add a metric to pending batch and flush if batch size reached
+   * Add a metric to pending batch and flush if batch size reached.
+   * Drops oldest metrics if the buffer exceeds maxPendingMetrics to prevent
+   * unbounded growth when the circuit breaker keeps failing exports.
    */
   private addPendingMetric(metric: TelemetryMetric): void {
     this.pendingMetrics.push(metric);
 
+    // Cap the buffer to avoid unbounded memory growth when exports keep failing
+    if (this.pendingMetrics.length > this.maxPendingMetrics) {
+      const dropped = this.pendingMetrics.length - this.maxPendingMetrics;
+      this.pendingMetrics = this.pendingMetrics.slice(dropped);
+      const logger = this.context.getLogger();
+      logger.log(LogLevel.debug, `Dropped ${dropped} oldest telemetry metrics (buffer full at ${this.maxPendingMetrics})`);
+    }
+
     // Check if batch size reached
     if (this.pendingMetrics.length >= this.batchSize) {
       this.flush();

lib/telemetry/types.ts

@@ -54,6 +54,9 @@ export interface TelemetryConfiguration {
 
   /** Circuit breaker timeout in milliseconds */
   circuitBreakerTimeout?: number;
+
+  /** Maximum number of pending metrics buffered before dropping oldest (prevents unbounded growth when export keeps failing) */
+  maxPendingMetrics?: number;
 }
 
 /**
@@ -67,6 +70,7 @@ export const DEFAULT_TELEMETRY_CONFIG: Required<TelemetryConfiguration> = {
   authenticatedExport: true,
   circuitBreakerThreshold: 5,
   circuitBreakerTimeout: 60000, // 1 minute
+  maxPendingMetrics: 500,
 };
 
 /**

lib/telemetry/urlUtils.ts

@@ -367,42 +367,39 @@ describe('CircuitBreaker', () => {
       expect(breaker.getState()).to.equal(CircuitBreakerState.CLOSED);
     });
 
-    it('should reopen if operation fails in HALF_OPEN state', async () => {
+    it('should reopen immediately if operation fails in HALF_OPEN state', async () => {
       const context = new ClientContextStub();
       const breaker = new CircuitBreaker(context);
 
       await openAndWaitForHalfOpen(breaker);
 
-      // First success
+      // First success moves to HALF_OPEN
       const successOp = sinon.stub().resolves('success');
       await breaker.execute(successOp);
       expect(breaker.getState()).to.equal(CircuitBreakerState.HALF_OPEN);
       expect(breaker.getSuccessCount()).to.equal(1);
 
-      // Failure should reset success count but not immediately open
+      // Any failure in HALF_OPEN immediately reopens the circuit
       const failOp = sinon.stub().rejects(new Error('Failed'));
       try {
         await breaker.execute(failOp);
       } catch {}
 
       expect(breaker.getSuccessCount()).to.equal(0); // Reset
-      expect(breaker.getFailureCount()).to.equal(1);
-      expect(breaker.getState()).to.equal(CircuitBreakerState.HALF_OPEN);
+      expect(breaker.getState()).to.equal(CircuitBreakerState.OPEN);
     });
 
-    it('should track failures and eventually reopen circuit', async () => {
+    it('should reopen immediately on first failure in HALF_OPEN state', async () => {
       const context = new ClientContextStub();
       const breaker = new CircuitBreaker(context);
 
       await openAndWaitForHalfOpen(breaker);
 
-      // Now in HALF_OPEN, fail 5 times to reopen
+      // A single failure in HALF_OPEN reopens immediately (not after threshold)
       const failOp = sinon.stub().rejects(new Error('Failed'));
-      for (let i = 0; i < 5; i++) {
-        try {
-          await breaker.execute(failOp);
-        } catch {}
-      }
+      try {
+        await breaker.execute(failOp);
+      } catch {}
 
       expect(breaker.getState()).to.equal(CircuitBreakerState.OPEN);
     });