chore: Adjust after review

- Adjusts null propagation, naming, and comments.
- Adjusts BatchExportTaskWorker and PeriodicExportingMetricReaderTaskWorker to start the Task with a known cancellation token

Author: jeromelaban

src/OpenTelemetry/BatchExportProcessor.cs

@@ -1,7 +1,6 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-using System.Diagnostics;
 using System.Runtime.CompilerServices;
 using OpenTelemetry.Internal;
 
@@ -20,7 +19,6 @@ public abstract class BatchExportProcessor<T> : BaseExportProcessor<T>
     internal const int DefaultMaxExportBatchSize = 512;
 
     internal readonly int MaxExportBatchSize;
-
     internal readonly int ScheduledDelayMilliseconds;
     internal readonly int ExporterTimeoutMilliseconds;
 
@@ -65,18 +63,19 @@ protected BatchExportProcessor(
         : base(exporter)
     {
         Guard.ThrowIfNull(options);
-#pragma warning disable CA1062 // Validate arguments of public methods - needed for netstandard2.1
-        Guard.ThrowIfOutOfRange(options.MaxQueueSize, min: 1);
-#pragma warning restore CA1062 // Validate arguments of public methods - needed for netstandard2.1
-        Guard.ThrowIfOutOfRange(options.MaxExportBatchSize, min: 1, max: options.MaxQueueSize, maxName: nameof(options.MaxQueueSize));
-        Guard.ThrowIfOutOfRange(options.ScheduledDelayMilliseconds, min: 1);
-        Guard.ThrowIfOutOfRange(options.ExporterTimeoutMilliseconds, min: 0);
-
-        this.circularBuffer = new CircularBuffer<T>(options.MaxQueueSize);
-        this.ScheduledDelayMilliseconds = options.ScheduledDelayMilliseconds;
-        this.ExporterTimeoutMilliseconds = options.ExporterTimeoutMilliseconds;
-        this.MaxExportBatchSize = options.MaxExportBatchSize;
-        this.useThreads = options.UseThreads;
+
+        var maxQueueSize = options?.MaxQueueSize ?? 0;
+        Guard.ThrowIfOutOfRange(maxQueueSize, min: 1);
+
+        this.circularBuffer = new CircularBuffer<T>(maxQueueSize);
+        this.ScheduledDelayMilliseconds = options?.ScheduledDelayMilliseconds ?? 0;
+        this.ExporterTimeoutMilliseconds = options?.ExporterTimeoutMilliseconds ?? -1;
+        this.MaxExportBatchSize = options?.MaxExportBatchSize ?? 0;
+        this.useThreads = options?.UseThreads ?? true;
+
+        Guard.ThrowIfOutOfRange(this.MaxExportBatchSize, min: 1, max: maxQueueSize, maxName: nameof(options.MaxQueueSize));
+        Guard.ThrowIfOutOfRange(this.ScheduledDelayMilliseconds, min: 1);
+        Guard.ThrowIfOutOfRange(this.ExporterTimeoutMilliseconds, min: 0);
 
         this.worker = this.CreateWorker();
         this.worker.Start();
@@ -145,9 +144,7 @@ protected override bool OnShutdown(int timeoutMilliseconds)
             return this.exporter.Shutdown(0) && result;
         }
 
-        var sw = Stopwatch.StartNew();
-        var timeout = timeoutMilliseconds - sw.ElapsedMilliseconds;
-        return this.exporter.Shutdown((int)Math.Max(timeout, 0)) && result;
+        return this.exporter.Shutdown(timeoutMilliseconds) && result;
     }
 
     /// <inheritdoc/>

src/OpenTelemetry/BatchExportProcessorOptions.cs

@@ -31,7 +31,7 @@ public class BatchExportProcessorOptions<T>
     public int MaxExportBatchSize { get; set; } = BatchExportProcessor<T>.DefaultMaxExportBatchSize;
 
     /// <summary>
-    /// Gets or sets a value indicating whether to use threads. Enables the use of <see cref="Thread" /> when true, <see cref="Task"/> when false. The default value is true.
+    /// Gets or sets a value indicating whether to use threads. Enables the use of <see cref="Thread" /> when <see langword="true"/>; otherwise <see cref="Task"/> is used. The default value is <see langword="true"/>.
     /// </summary>
     public bool UseThreads { get; set; } = true;
 }

src/OpenTelemetry/Internal/BatchExportTaskWorker.cs

@@ -42,7 +42,7 @@ public override void Start()
     {
         this.workerTask = Task.Factory.StartNew(
             this.ExporterProcAsync,
-            CancellationToken.None,
+            this.cancellationTokenSource.Token,
             TaskCreationOptions.LongRunning,
             TaskScheduler.Default).Unwrap();
     }
@@ -74,8 +74,8 @@ public override bool TriggerExport()
     /// <inheritdoc/>
     public override bool WaitForExport(int timeoutMilliseconds)
     {
-        var tail = this.circularBuffer.RemovedCount;
-        var head = this.circularBuffer.AddedCount;
+        var tail = this.CircularBuffer.RemovedCount;
+        var head = this.CircularBuffer.AddedCount;
 
         if (head == tail)
         {
@@ -98,7 +98,7 @@ public override bool WaitForExport(int timeoutMilliseconds)
     /// <inheritdoc/>
     public override bool Shutdown(int timeoutMilliseconds)
     {
-        this.SetShutdownDrainTarget(this.circularBuffer.AddedCount);
+        this.SetShutdownDrainTarget(this.CircularBuffer.AddedCount);
 
         try
         {
@@ -160,7 +160,7 @@ private async Task<bool> WaitForExportAsync(int timeoutMilliseconds, long target
                 var remaining = timeoutMilliseconds - sw.ElapsedMilliseconds;
                 if (remaining <= 0)
                 {
-                    return this.circularBuffer.RemovedCount >= targetHead;
+                    return this.CircularBuffer.RemovedCount >= targetHead;
                 }
 
                 timeout = Math.Min((int)remaining, pollingMilliseconds);
@@ -187,7 +187,7 @@ await Task.WhenAny(
                 // Expected when timeout or shutdown occurs
             }
 
-            if (this.circularBuffer.RemovedCount >= targetHead)
+            if (this.CircularBuffer.RemovedCount >= targetHead)
             {
                 return true;
             }
@@ -208,13 +208,13 @@ private async Task ExporterProcAsync()
             while (true)
             {
                 // only wait when the queue doesn't have enough items, otherwise keep busy and send data continuously
-                if (this.circularBuffer.Count < this.maxExportBatchSize)
+                if (this.CircularBuffer.Count < this.MaxExportBatchSize)
                 {
                     try
                     {
                         await Task.WhenAny(
                             this.exportTrigger.WaitAsync(cancellationToken),
-                            Task.Delay(this.scheduledDelayMilliseconds, cancellationToken)).ConfigureAwait(false);
+                            Task.Delay(this.ScheduledDelayMilliseconds, cancellationToken)).ConfigureAwait(false);
                     }
                     catch (OperationCanceledException)
                     {

src/OpenTelemetry/Internal/BatchExportThreadWorker.cs

@@ -66,8 +66,8 @@ public override bool TriggerExport()
     /// <inheritdoc/>
     public override bool WaitForExport(int timeoutMilliseconds)
     {
-        var tail = this.circularBuffer.RemovedCount;
-        var head = this.circularBuffer.AddedCount;
+        var tail = this.CircularBuffer.RemovedCount;
+        var head = this.CircularBuffer.AddedCount;
 
         if (head == tail)
         {
@@ -113,7 +113,7 @@ public override bool WaitForExport(int timeoutMilliseconds)
 
                 if (timeout <= 0)
                 {
-                    return this.circularBuffer.RemovedCount >= head;
+                    return this.CircularBuffer.RemovedCount >= head;
                 }
 
                 try
@@ -126,7 +126,7 @@ public override bool WaitForExport(int timeoutMilliseconds)
                 }
             }
 
-            if (this.circularBuffer.RemovedCount >= head)
+            if (this.CircularBuffer.RemovedCount >= head)
             {
                 return true;
             }
@@ -141,7 +141,7 @@ public override bool WaitForExport(int timeoutMilliseconds)
     /// <inheritdoc/>
     public override bool Shutdown(int timeoutMilliseconds)
     {
-        this.SetShutdownDrainTarget(this.circularBuffer.AddedCount);
+        this.SetShutdownDrainTarget(this.CircularBuffer.AddedCount);
 
         try
         {
@@ -188,11 +188,11 @@ private void ExporterProc()
         while (true)
         {
             // only wait when the queue doesn't have enough items, otherwise keep busy and send data continuously
-            if (this.circularBuffer.Count < this.maxExportBatchSize)
+            if (this.CircularBuffer.Count < this.MaxExportBatchSize)
             {
                 try
                 {
-                    WaitHandle.WaitAny(triggers, this.scheduledDelayMilliseconds);
+                    WaitHandle.WaitAny(triggers, this.ScheduledDelayMilliseconds);
                 }
                 catch (ObjectDisposedException)
                 {

src/OpenTelemetry/Internal/BatchExportWorker.cs

@@ -10,12 +10,6 @@ namespace OpenTelemetry.Internal;
 internal abstract class BatchExportWorker<T> : IDisposable
     where T : class
 {
-    protected readonly CircularBuffer<T> circularBuffer;
-    protected readonly BaseExporter<T> exporter;
-    protected readonly int maxExportBatchSize;
-    protected readonly int scheduledDelayMilliseconds;
-    protected readonly int exporterTimeoutMilliseconds;
-
     private long shutdownDrainTarget = long.MaxValue;
     private long droppedCount;
 
@@ -34,18 +28,49 @@ protected BatchExportWorker(
         int scheduledDelayMilliseconds,
         int exporterTimeoutMilliseconds)
     {
-        this.circularBuffer = circularBuffer;
-        this.exporter = exporter;
-        this.maxExportBatchSize = maxExportBatchSize;
-        this.scheduledDelayMilliseconds = scheduledDelayMilliseconds;
-        this.exporterTimeoutMilliseconds = exporterTimeoutMilliseconds;
+        this.CircularBuffer = circularBuffer;
+        this.Exporter = exporter;
+        this.MaxExportBatchSize = maxExportBatchSize;
+        this.ScheduledDelayMilliseconds = scheduledDelayMilliseconds;
+        this.ExporterTimeoutMilliseconds = exporterTimeoutMilliseconds;
+    }
+
+    ~BatchExportWorker()
+    {
+        // Finalizer to ensure resources are cleaned up if Dispose is not called
+        this.Dispose(false);
     }
 
     /// <summary>
     /// Gets the number of telemetry objects dropped by the processor.
     /// </summary>
     internal long DroppedCount => Volatile.Read(ref this.droppedCount);
 
+    /// <summary>
+    /// Gets the circular buffer for storing telemetry objects.
+    /// </summary>
+    protected CircularBuffer<T> CircularBuffer { get; }
+
+    /// <summary>
+    /// Gets the exporter instance.
+    /// </summary>
+    protected BaseExporter<T> Exporter { get; }
+
+    /// <summary>
+    /// Gets the maximum batch size for exports.
+    /// </summary>
+    protected int MaxExportBatchSize { get; }
+
+    /// <summary>
+    /// Gets the delay between exports in milliseconds.
+    /// </summary>
+    protected int ScheduledDelayMilliseconds { get; }
+
+    /// <summary>
+    /// Gets the timeout for export operations in milliseconds.
+    /// </summary>
+    protected int ExporterTimeoutMilliseconds { get; }
+
     /// <summary>
     /// Gets the shutdown drain target.
     /// </summary>
@@ -59,7 +84,7 @@ protected BatchExportWorker(
     /// <summary>
     /// Triggers an export operation.
     /// </summary>
-    /// <returns>True if the export was triggered successfully; otherwise, false.</returns>
+    /// <returns><see langword="true"/> if the export was triggered successfully; otherwise, <see langword="false"/>.</returns>
     public abstract bool TriggerExport();
 
     /// <summary>
@@ -105,11 +130,11 @@ protected void SetShutdownDrainTarget(long target)
     /// </summary>
     protected void PerformExport()
     {
-        if (this.circularBuffer.Count > 0)
+        if (this.CircularBuffer.Count > 0)
         {
-            using (var batch = new Batch<T>(this.circularBuffer, this.maxExportBatchSize))
+            using (var batch = new Batch<T>(this.CircularBuffer, this.MaxExportBatchSize))
             {
-                this.exporter.Export(batch);
+                this.Exporter.Export(batch);
             }
         }
     }
@@ -120,7 +145,7 @@ protected void PerformExport()
     /// <returns>True if shutdown should occur; otherwise, false.</returns>
     protected bool ShouldShutdown()
     {
-        return this.circularBuffer.RemovedCount >= this.ShutdownDrainTarget;
+        return this.CircularBuffer.RemovedCount >= this.ShutdownDrainTarget;
     }
 
     /// <summary>

src/OpenTelemetry/Internal/PeriodicExportingMetricReaderTaskWorker.cs

@@ -36,7 +36,7 @@ public override void Start()
     {
         this.workerTask = Task.Factory.StartNew(
             this.ExporterProcAsync,
-            CancellationToken.None,
+            this.cancellationTokenSource.Token,
             TaskCreationOptions.LongRunning,
             TaskScheduler.Default).Unwrap();
     }
@@ -122,7 +122,7 @@ private async Task ExporterProcAsync()
         {
             while (!cancellationToken.IsCancellationRequested)
             {
-                var timeout = (int)(this.exportIntervalMilliseconds - (sw.ElapsedMilliseconds % this.exportIntervalMilliseconds));
+                var timeout = (int)(this.ExportIntervalMilliseconds - (sw.ElapsedMilliseconds % this.ExportIntervalMilliseconds));
 
                 var exportTriggerTask = this.exportTrigger.WaitAsync(cancellationToken);
                 Task? triggeredTask = null;
@@ -141,7 +141,7 @@ private async Task ExporterProcAsync()
                 if (cancellationToken.IsCancellationRequested)
                 {
                     OpenTelemetrySdkEventSource.Log.MetricReaderEvent("PeriodicExportingMetricReader calling MetricReader.Collect because Shutdown was triggered.");
-                    this.metricReader.Collect(this.exportTimeoutMilliseconds);
+                    this.MetricReader.Collect(this.ExportTimeoutMilliseconds);
                     break;
                 }
 
@@ -157,7 +157,7 @@ private async Task ExporterProcAsync()
                     OpenTelemetrySdkEventSource.Log.MetricReaderEvent("PeriodicExportingMetricReader calling MetricReader.Collect because the export interval has elapsed.");
                 }
 
-                this.metricReader.Collect(this.exportTimeoutMilliseconds);
+                this.MetricReader.Collect(this.ExportTimeoutMilliseconds);
             }
         }
         catch (Exception ex)

src/OpenTelemetry/Internal/PeriodicExportingMetricReaderThreadWorker.cs

@@ -109,7 +109,7 @@ private void ExporterProc()
 
         while (true)
         {
-            timeout = (int)(this.exportIntervalMilliseconds - (sw.ElapsedMilliseconds % this.exportIntervalMilliseconds));
+            timeout = (int)(this.ExportIntervalMilliseconds - (sw.ElapsedMilliseconds % this.ExportIntervalMilliseconds));
 
             try
             {
@@ -124,15 +124,15 @@ private void ExporterProc()
             {
                 case 0: // export
                     OpenTelemetrySdkEventSource.Log.MetricReaderEvent("PeriodicExportingMetricReader calling MetricReader.Collect because Export was triggered.");
-                    this.metricReader.Collect(this.exportTimeoutMilliseconds);
+                    this.MetricReader.Collect(this.ExportTimeoutMilliseconds);
                     break;
                 case 1: // shutdown
                     OpenTelemetrySdkEventSource.Log.MetricReaderEvent("PeriodicExportingMetricReader calling MetricReader.Collect because Shutdown was triggered.");
-                    this.metricReader.Collect(this.exportTimeoutMilliseconds);
+                    this.MetricReader.Collect(this.ExportTimeoutMilliseconds);
                     return;
                 case WaitHandle.WaitTimeout: // timer
                     OpenTelemetrySdkEventSource.Log.MetricReaderEvent("PeriodicExportingMetricReader calling MetricReader.Collect because the export interval has elapsed.");
-                    this.metricReader.Collect(this.exportTimeoutMilliseconds);
+                    this.MetricReader.Collect(this.ExportTimeoutMilliseconds);
                     break;
             }
         }