Update TelemetryClient to direct enqueue/flush/export pipeline

Remove ActivityListener and MetricsAggregator dependencies. TelemetryClient
now directly enqueues TelemetryFrontendLog objects and flushes them via
the circuit breaker-protected exporter.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Jade Wang

csharp/src/Telemetry/TelemetryClient.cs

@@ -15,6 +15,7 @@
 */
 
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Threading;
@@ -30,7 +31,7 @@ namespace AdbcDrivers.Databricks.Telemetry
     /// <remarks>
     /// <para>
     /// This client orchestrates the complete telemetry pipeline:
-    /// DatabricksActivityListener → MetricsAggregator → CircuitBreakerTelemetryExporter → DatabricksTelemetryExporter
+    /// CircuitBreakerTelemetryExporter → DatabricksTelemetryExporter
     /// </para>
     /// <para>
     /// Key Behaviors:
@@ -52,8 +53,9 @@ internal sealed class TelemetryClient : ITelemetryClient
 
         private readonly DatabricksTelemetryExporter _databricksExporter;
         private readonly CircuitBreakerTelemetryExporter _circuitBreakerExporter;
-        private readonly MetricsAggregator _metricsAggregator;
-        private readonly DatabricksActivityListener _activityListener;
+        private readonly ITelemetryExporter _effectiveExporter;
+        private readonly ConcurrentQueue<TelemetryFrontendLog> _pendingLogs;
+        private readonly int _batchSize;
         private readonly CancellationTokenSource _cts;
         private volatile bool _disposed;
 
@@ -64,13 +66,15 @@ internal sealed class TelemetryClient : ITelemetryClient
         /// <param name="httpClient">The HTTP client to use for exporting telemetry.</param>
         /// <param name="isAuthenticated">Whether the connection is authenticated (determines telemetry endpoint).</param>
         /// <param name="configuration">The telemetry configuration.</param>
+        /// <param name="exporterOverride">Optional exporter override for testing.</param>
         /// <exception cref="ArgumentNullException">Thrown when host, httpClient, or configuration is null.</exception>
         /// <exception cref="ArgumentException">Thrown when host is empty or whitespace.</exception>
         public TelemetryClient(
             string host,
             System.Net.Http.HttpClient httpClient,
             bool isAuthenticated,
-            TelemetryConfiguration configuration)
+            TelemetryConfiguration configuration,
+            ITelemetryExporter? exporterOverride = null)
         {
             if (string.IsNullOrWhiteSpace(host))
             {
@@ -88,6 +92,8 @@ public TelemetryClient(
             }
 
             _cts = new CancellationTokenSource();
+            _pendingLogs = new ConcurrentQueue<TelemetryFrontendLog>();
+            _batchSize = configuration.BatchSize;
 
             try
             {
@@ -98,14 +104,8 @@ public TelemetryClient(
                 // 2. CircuitBreakerTelemetryExporter (wraps exporter with circuit breaker protection)
                 _circuitBreakerExporter = new CircuitBreakerTelemetryExporter(_databricksExporter, host);
 
-                // 3. MetricsAggregator (aggregates activities and exports via circuit breaker)
-                _metricsAggregator = new MetricsAggregator(
-                    _circuitBreakerExporter,
-                    batchSize: configuration.BatchSize,
-                    flushInterval: TimeSpan.FromMilliseconds(configuration.FlushIntervalMs));
-
-                // 4. DatabricksActivityListener (listens to activities and delegates to aggregator)
-                _activityListener = new DatabricksActivityListener(_metricsAggregator, configuration);
+                // Use override exporter if provided (for testing), otherwise use circuit breaker exporter
+                _effectiveExporter = exporterOverride ?? (ITelemetryExporter)_circuitBreakerExporter;
 
                 Activity.Current?.AddEvent(new ActivityEvent("telemetry.client.initialized",
                     tags: new ActivityTagsCollection
@@ -121,8 +121,6 @@ public TelemetryClient(
                 // Clean up any partially initialized resources
                 try
                 {
-                    _activityListener?.Dispose();
-                    _metricsAggregator?.DisposeAsync().AsTask().Wait(TimeSpan.FromSeconds(1));
                     _cts?.Dispose();
                 }
                 catch
@@ -148,10 +146,6 @@ public TelemetryClient(
         /// <param name="log">The telemetry frontend log to enqueue.</param>
         /// <remarks>
         /// <para>
-        /// This method delegates to the MetricsAggregator, which handles batching and export.
-        /// The aggregator processes activities from the ActivityListener automatically.
-        /// </para>
-        /// <para>
         /// This method never throws exceptions. If the client is closed or disposing,
         /// the event is silently dropped.
         /// </para>
@@ -165,16 +159,13 @@ public void Enqueue(TelemetryFrontendLog log)
 
             try
             {
-                // For the activity-based pipeline, this method is not typically used.
-                // Activities are automatically captured by the listener and processed by the aggregator.
-                // However, we provide this method for compatibility with ITelemetryClient interface
-                // in case manual enqueueing is needed.
+                _pendingLogs.Enqueue(log);
 
-                Activity.Current?.AddEvent(new ActivityEvent("telemetry.client.manual_enqueue",
-                    tags: new ActivityTagsCollection
-                    {
-                        { "frontend_log_id", log.FrontendLogEventId ?? "(none)" }
-                    }));
+                // Trigger flush if batch size reached
+                if (_pendingLogs.Count >= _batchSize)
+                {
+                    _ = Task.Run(() => FlushAsync(CancellationToken.None));
+                }
             }
             catch (Exception ex)
             {
@@ -213,8 +204,17 @@ public async Task FlushAsync(CancellationToken ct = default)
 
             try
             {
-                // Delegate to aggregator to flush all pending metrics
-                await _metricsAggregator.FlushAsync(ct).ConfigureAwait(false);
+                // Flush directly enqueued logs (V3 direct-object path)
+                List<TelemetryFrontendLog> logsToFlush = new List<TelemetryFrontendLog>();
+                while (_pendingLogs.TryDequeue(out TelemetryFrontendLog? log))
+                {
+                    logsToFlush.Add(log);
+                }
+
+                if (logsToFlush.Count > 0)
+                {
+                    await _effectiveExporter.ExportAsync(logsToFlush, ct).ConfigureAwait(false);
+                }
 
                 Activity.Current?.AddEvent(new ActivityEvent("telemetry.client.flush_completed"));
             }
@@ -241,10 +241,8 @@ public async Task FlushAsync(CancellationToken ct = default)
         /// </para>
         /// <para>
         /// The close operation performs the following steps:
-        /// 1. Stop the activity listener (no more activities will be processed)
-        /// 2. Flush all remaining queued metrics from the aggregator
-        /// 3. Cancel any pending background tasks
-        /// 4. Dispose all resources (listener, aggregator, cancellation token source)
+        /// 1. Cancel any pending background tasks
+        /// 2. Dispose all resources (cancellation token source)
         /// </para>
         /// <para>
         /// This method is idempotent - calling it multiple times is safe and has no effect
@@ -255,11 +253,11 @@ public async Task FlushAsync(CancellationToken ct = default)
         /// logged internally.
         /// </para>
         /// </remarks>
-        public async Task CloseAsync()
+        public Task CloseAsync()
         {
             if (_disposed)
             {
-                return;
+                return Task.CompletedTask;
             }
 
             _disposed = true;
@@ -268,37 +266,7 @@ public async Task CloseAsync()
             {
                 Activity.Current?.AddEvent(new ActivityEvent("telemetry.client.closing"));
 
-                // Step 1: Stop the activity listener (no more activities will be captured)
-                try
-                {
-                    await _activityListener.StopAsync(_cts.Token).ConfigureAwait(false);
-                }
-                catch (Exception ex)
-                {
-                    Activity.Current?.AddEvent(new ActivityEvent("telemetry.client.listener_stop_error",
-                        tags: new ActivityTagsCollection
-                        {
-                            { "error.message", ex.Message },
-                            { "error.type", ex.GetType().Name }
-                        }));
-                }
-
-                // Step 2: Flush all remaining metrics from the aggregator
-                try
-                {
-                    await _metricsAggregator.FlushAsync(CancellationToken.None).ConfigureAwait(false);
-                }
-                catch (Exception ex)
-                {
-                    Activity.Current?.AddEvent(new ActivityEvent("telemetry.client.final_flush_error",
-                        tags: new ActivityTagsCollection
-                        {
-                            { "error.message", ex.Message },
-                            { "error.type", ex.GetType().Name }
-                        }));
-                }
-
-                // Step 3: Cancel any pending background tasks
+                // Cancel any pending background tasks
                 try
                 {
                     _cts.Cancel();
@@ -313,21 +281,6 @@ public async Task CloseAsync()
                         }));
                 }
 
-                // Step 4: Dispose all resources
-                try
-                {
-                    await _metricsAggregator.DisposeAsync().ConfigureAwait(false);
-                }
-                catch (Exception ex)
-                {
-                    Activity.Current?.AddEvent(new ActivityEvent("telemetry.client.aggregator_dispose_error",
-                        tags: new ActivityTagsCollection
-                        {
-                            { "error.message", ex.Message },
-                            { "error.type", ex.GetType().Name }
-                        }));
-                }
-
                 Activity.Current?.AddEvent(new ActivityEvent("telemetry.client.closed"));
             }
             catch (Exception ex)
@@ -352,6 +305,8 @@ public async Task CloseAsync()
                     // Swallow CTS dispose exceptions
                 }
             }
+
+            return Task.CompletedTask;
         }
 
         /// <summary>