test(sdk): add shutdown regression and timeout tests for processor test suite

Add coverage for processor test suite gaps identified in #3381:

- Add shutdown() regression tests with TokioSpawn*Exporter mocks for
  BatchSpanProcessor, BatchLogProcessor, and PeriodicReader on
  multi_thread(1) runtime, verifying the same BlockingStrategy fix
  that resolved force_flush() deadlocks also works for shutdown().

- Add timeout behavior tests with HangingExporter mocks for
  BatchSpanProcessor and BatchLogProcessor, verifying that
  shutdown_with_timeout returns Err(Timeout) when exporters hang.

- Improve documentation on SimpleLogProcessor's ignored deadlock tests,
  explaining they demonstrate inherent design limitations (not bugs)
  and linking to relevant issues (#2802, #3381).

- Document current_thread runtime limitation across all processors
  and explain why PeriodicReader timeout test is omitted (hardcoded
  5s timeout makes it too slow for regular test suite).

Closes #3381

Author: bryantbiggs

opentelemetry-sdk/src/logs/batch_log_processor.rs

@@ -1075,6 +1075,56 @@ mod tests {
         processor.shutdown().unwrap();
     }
 
+    // Regression test: shutdown() goes through the same export path as force_flush()
+    // (get_logs_and_export via BlockingStrategy) and then calls exporter.shutdown().
+    // Without BlockingStrategy, this would deadlock on multi_thread(1) just like
+    // force_flush() did (#2802, #3356).
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn test_batch_log_processor_shutdown_with_tokio_spawn_exporter() {
+        let exporter = TokioSpawnLogExporter::new();
+        let exported_count = exporter.exported_count.clone();
+        let processor = BatchLogProcessor::new(exporter, BatchConfig::default());
+
+        let mut record = SdkLogRecord::new();
+        let instrumentation = InstrumentationScope::default();
+        processor.emit(&mut record, &instrumentation);
+
+        processor.shutdown().unwrap();
+
+        assert_eq!(exported_count.load(Ordering::Relaxed), 1);
+    }
+
+    // Test that shutdown() returns a timeout error when the exporter hangs.
+    // The BatchLogProcessor's shutdown_with_timeout uses recv_timeout (default 5s),
+    // so a hanging exporter should result in Err(Timeout).
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn test_batch_log_processor_shutdown_timeout_with_hanging_exporter() {
+        #[derive(Debug)]
+        struct HangingLogExporter;
+
+        impl LogExporter for HangingLogExporter {
+            async fn export(&self, _batch: LogBatch<'_>) -> OTelSdkResult {
+                // Block forever, simulating an exporter that cannot complete
+                futures_util::future::pending::<()>().await;
+                Ok(())
+            }
+
+            fn shutdown(&self) -> OTelSdkResult {
+                Ok(())
+            }
+        }
+
+        let processor = BatchLogProcessor::new(HangingLogExporter, BatchConfig::default());
+
+        let mut record = SdkLogRecord::new();
+        let instrumentation = InstrumentationScope::default();
+        processor.emit(&mut record, &instrumentation);
+
+        // Use a short timeout to avoid slow tests
+        let result = processor.shutdown_with_timeout(Duration::from_millis(500));
+        assert!(result.is_err(), "Expected timeout error from hanging exporter");
+    }
+
     /// A slow exporter that counts the number of logs received.
     /// Used for stress testing the BatchLogProcessor.
     #[derive(Debug, Clone)]

opentelemetry-sdk/src/logs/simple_log_processor.rs

@@ -346,24 +346,25 @@ mod tests {
         );
     }
 
-    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-    #[ignore]
-    // This test demonstrates a potential deadlock scenario in a multi-threaded Tokio runtime.
-    // It spawns Tokio tasks equal to the number of runtime worker threads (4) to emit log events.
-    // Each task attempts to acquire a mutex on the exporter in `SimpleLogProcessor::emit`.
-    // Only one task obtains the lock, while the others are blocked, waiting for its release.
+    // KNOWN DEADLOCK: This test is intentionally #[ignore]d because it demonstrates
+    // an inherent limitation of SimpleLogProcessor with async exporters.
     //
-    // The task holding the lock invokes the LogExporterThatRequiresTokio, which performs an
-    // asynchronous operation (e.g., network I/O simulated by `tokio::sleep`). This operation
-    // requires yielding control back to the Tokio runtime to make progress.
+    // SimpleLogProcessor uses futures_executor::block_on() to synchronously execute
+    // async export operations. When all tokio worker threads are occupied (one holding
+    // the exporter mutex in block_on, the rest waiting for the mutex), no thread is
+    // available to drive the tokio::spawn-ed tasks the exporter depends on.
     //
-    // However, all worker threads are occupied:
-    // - One thread is executing the async exporter operation
-    // - Three threads are blocked waiting for the mutex
+    // This is NOT a bug — it's a fundamental design trade-off of SimpleLogProcessor:
+    // - SimpleLogProcessor is designed for debugging/testing, not production use
+    // - For production with async exporters, use BatchLogProcessor which runs exports
+    //   on a dedicated background thread with BlockingStrategy
+    // - Unlike BatchLogProcessor's deadlock (#2802) which was fixable with BlockingStrategy,
+    //   this deadlock cannot be fixed without changing SimpleLogProcessor's synchronous design
     //
-    // This leads to a deadlock as there are no available threads to drive the async operation
-    // to completion, preventing the mutex from being released. Consequently, neither the blocked
-    // tasks nor the exporter can proceed.
+    // See: https://github.com/open-telemetry/opentelemetry-rust/issues/3381
+    //      https://github.com/open-telemetry/opentelemetry-rust/issues/2802
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[ignore]
     async fn test_simple_processor_async_exporter_with_all_runtime_worker_threads_blocked() {
         let exporter = LogExporterThatRequiresTokio::new();
         let processor = Arc::new(SimpleLogProcessor::new(exporter.clone()));
@@ -426,13 +427,22 @@ mod tests {
         assert_eq!(exporter.len(), 1);
     }
 
+    // KNOWN DEADLOCK: This test is intentionally #[ignore]d because it demonstrates
+    // a fundamental limitation of SimpleLogProcessor on current_thread runtimes.
+    //
+    // On a current_thread runtime, SimpleLogProcessor's futures_executor::block_on()
+    // blocks the only available thread. Any exporter that depends on tokio (e.g.,
+    // tokio::spawn, tokio::time::sleep, tonic gRPC) cannot make progress because
+    // there is no thread available to drive the tokio runtime.
+    //
+    // This applies to ALL processors on current_thread runtimes with tokio-dependent
+    // exporters — it's a fundamental limitation, not a bug. The multi_thread(1) scenario
+    // (common in 1-vCPU k8s pods) IS supported via BlockingStrategy in batch processors.
+    //
+    // See: https://github.com/open-telemetry/opentelemetry-rust/issues/3381
+    //      https://github.com/open-telemetry/opentelemetry-rust/issues/2802
     #[tokio::test(flavor = "current_thread")]
     #[ignore]
-    // This test uses a current-thread runtime, where all operations run on the main thread.
-    // The processor emits a log event while the runtime is blocked using `futures::block_on`
-    // to complete the export operation. The exporter, which performs an async operation and
-    // requires the runtime, cannot progress because the main thread is already blocked.
-    // This results in a deadlock, as the runtime cannot move forward.
     async fn test_simple_processor_async_exporter_with_current_thread_runtime() {
         let exporter = LogExporterThatRequiresTokio::new();
 

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -675,6 +675,39 @@ mod tests {
         assert!(exported_count.load(Ordering::Relaxed) >= 1);
     }
 
+    // Regression test: shutdown() goes through the same export path as force_flush()
+    // (collect_and_export via BlockingStrategy) and then calls exporter.shutdown().
+    // Without BlockingStrategy, this would deadlock on multi_thread(1) just like
+    // force_flush() did (#2802, #3356).
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn test_periodic_reader_shutdown_with_tokio_spawn_exporter() {
+        let exporter = TokioSpawnMetricExporter::default();
+        let exported_count = exporter.exported_count.clone();
+        let is_shutdown = exporter.is_shutdown.clone();
+
+        let reader = PeriodicReader::builder(exporter)
+            .with_interval(Duration::from_secs(120))
+            .build();
+
+        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
+        let meter = meter_provider.meter("test");
+        let counter = meter.u64_counter("test.counter").build();
+        counter.add(1, &[]);
+
+        meter_provider.shutdown().unwrap();
+
+        assert!(exported_count.load(Ordering::Relaxed) >= 1);
+        assert!(is_shutdown.load(Ordering::Relaxed));
+    }
+
+    // Note: A timeout test for PeriodicReader with a hanging exporter is not
+    // included here because PeriodicReader::shutdown() has a hardcoded 5-second
+    // timeout (see TODO in shutdown()) that is not configurable, making the test
+    // too slow for the regular test suite. The timeout behavior is architecturally
+    // identical to BatchSpanProcessor and BatchLogProcessor (recv_timeout on the
+    // response channel), so those tests provide sufficient coverage.
+    // See: https://github.com/open-telemetry/opentelemetry-rust/issues/3381
+
     #[test]
     fn collection_triggered_by_interval_multiple() {
         // Arrange

opentelemetry-sdk/src/trace/span_processor.rs

@@ -1607,6 +1607,68 @@ mod tests {
         assert_eq!(exported_spans.len(), 4);
     }
 
+    // Regression test: shutdown() goes through the same export path as force_flush()
+    // (get_spans_and_export via BlockingStrategy) and then calls exporter.shutdown().
+    // Without BlockingStrategy, this would deadlock on multi_thread(1) just like
+    // force_flush() did (#2802, #3356).
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn test_batch_processor_shutdown_with_tokio_spawn_exporter() {
+        let exporter = TokioSpawnSpanExporter::new();
+        let exporter_shared = exporter.exported_spans.clone();
+
+        let config = BatchConfigBuilder::default()
+            .with_max_queue_size(5)
+            .with_max_export_batch_size(3)
+            .build();
+
+        let processor = BatchSpanProcessor::new(exporter, config);
+
+        for _ in 0..4 {
+            let span = new_test_export_span_data();
+            processor.on_end(span);
+        }
+
+        processor.shutdown().unwrap();
+
+        let exported_spans = exporter_shared.lock().unwrap();
+        assert_eq!(exported_spans.len(), 4);
+    }
+
+    // Test that shutdown() returns a timeout error when the exporter hangs.
+    // The BatchSpanProcessor's shutdown_with_timeout uses recv_timeout (default 5s),
+    // so a hanging exporter should result in Err(Timeout).
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn test_batch_processor_shutdown_timeout_with_hanging_exporter() {
+        #[derive(Debug)]
+        struct HangingSpanExporter;
+
+        impl SpanExporter for HangingSpanExporter {
+            async fn export(&self, _batch: Vec<SpanData>) -> OTelSdkResult {
+                // Block forever, simulating an exporter that cannot complete
+                futures_util::future::pending::<()>().await;
+                Ok(())
+            }
+
+            fn shutdown(&self) -> OTelSdkResult {
+                Ok(())
+            }
+        }
+
+        let config = BatchConfigBuilder::default()
+            .with_max_queue_size(5)
+            .with_max_export_batch_size(3)
+            .build();
+
+        let processor = BatchSpanProcessor::new(HangingSpanExporter, config);
+
+        let span = new_test_export_span_data();
+        processor.on_end(span);
+
+        // Use a short timeout to avoid slow tests
+        let result = processor.shutdown_with_timeout(Duration::from_millis(500));
+        assert!(result.is_err(), "Expected timeout error from hanging exporter");
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     async fn test_batch_processor_multi_thread() {
         let exporter = MockSpanExporter::new();