make shutdown_with_timeout required on LogProcessor

Author: davidhewitt

examples/tracing-http-propagator/src/server.rs

@@ -121,6 +121,10 @@ impl LogProcessor for EnrichWithBaggageLogProcessor {
     fn force_flush(&self) -> OTelSdkResult {
         Ok(())
     }
+
+    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
+        Ok(())
+    }
 }
 
 /// A custom span processor that enriches spans with baggage attributes. Baggage

opentelemetry-appender-tracing/src/layer.rs

@@ -931,6 +931,10 @@ mod tests {
         fn force_flush(&self) -> OTelSdkResult {
             Ok(())
         }
+
+        fn shutdown_with_timeout(&self, _timeout: std::time::Duration) -> OTelSdkResult {
+            Ok(())
+        }
     }
 
     #[cfg(feature = "spec_unstable_logs_enabled")]

opentelemetry-proto/src/transform/logs.rs

@@ -241,6 +241,10 @@ mod tests {
         fn force_flush(&self) -> OTelSdkResult {
             Ok(())
         }
+
+        fn shutdown_with_timeout(&self, _timeout: std::time::Duration) -> OTelSdkResult {
+            Ok(())
+        }
     }
 
     fn create_test_log_data(

opentelemetry-sdk/CHANGELOG.md

@@ -14,6 +14,8 @@
   between logs and traces continues to work when the "trace" feature is
   explicitly enabled.
 
+- *Breaking* default implementation of `LogProcessor::shutdown_with_timeout()` method is removed. Implementors must now provide this to guarantee shutdown is performed properly.
+
 ## 0.30.0
 
 Released 2025-May-23
@@ -35,7 +37,7 @@ also modified to suppress telemetry before invoking exporters.
 
 - **Feature**: Implemented and enabled cardinality capping for Metrics by
   default. [#2901](https://github.com/open-telemetry/opentelemetry-rust/pull/2901)
-  - The default cardinality limit is 2000 and can be customized using Views.  
+  - The default cardinality limit is 2000 and can be customized using Views.
   - This feature was previously removed in version 0.28 due to the lack of
     configurability but has now been reintroduced with the ability to configure
     the limit.
@@ -176,7 +178,7 @@ Released 2025-Mar-21
   ```
 
   After:
-  
+
   ```rust
     async fn export(&self, batch: Vec<SpanData>) -> OTelSdkResult
   ```

opentelemetry-sdk/src/logs/log_processor.rs

@@ -57,9 +57,7 @@ pub trait LogProcessor: Send + Sync + Debug {
     /// Shuts down the processor.
     /// After shutdown returns the log processor should stop processing any logs.
     /// It's up to the implementation on when to drop the LogProcessor.
-    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
-        Ok(())
-    }
+    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult;
     /// Shuts down the processor with default timeout.
     fn shutdown(&self) -> OTelSdkResult {
         self.shutdown_with_timeout(Duration::from_secs(5))
@@ -140,6 +138,10 @@ pub(crate) mod tests {
         fn force_flush(&self) -> OTelSdkResult {
             Ok(())
         }
+
+        fn shutdown_with_timeout(&self, _timeout: std::time::Duration) -> OTelSdkResult {
+            Ok(())
+        }
     }
 
     #[derive(Debug)]
@@ -166,6 +168,10 @@ pub(crate) mod tests {
         fn force_flush(&self) -> OTelSdkResult {
             Ok(())
         }
+
+        fn shutdown_with_timeout(&self, _timeout: std::time::Duration) -> OTelSdkResult {
+            Ok(())
+        }
     }
 
     #[test]

opentelemetry-sdk/src/logs/log_processor_with_async_runtime.rs

@@ -87,7 +87,7 @@ impl<R: RuntimeChannel> LogProcessor for BatchLogProcessor<R> {
             .and_then(std::convert::identity)
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
         let dropped_logs = self.dropped_logs_count.load(Ordering::Relaxed);
         let max_queue_size = self.max_queue_size;
         if dropped_logs > 0 {

opentelemetry-sdk/src/logs/logger_provider.rs

@@ -402,6 +402,10 @@ mod tests {
             *res = resource.clone();
             self.exporter.set_resource(resource);
         }
+
+        fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
+            Ok(())
+        }
     }
     impl TestProcessorForResource {
         fn new(exporter: TestExporterForResource) -> Self {

opentelemetry-sdk/src/logs/mod.rs

@@ -171,7 +171,7 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> crate::error::OTelSdkResult {
+        fn shutdown_with_timeout(&self, _timeout: std::time::Duration) -> OTelSdkResult {
             Ok(())
         }
     }
@@ -277,7 +277,7 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> OTelSdkResult {
+        fn shutdown_with_timeout(&self, _timeout: std::time::Duration) -> OTelSdkResult {
             Ok(())
         }
     }

opentelemetry-sdk/src/logs/simple_log_processor.rs

@@ -28,6 +28,7 @@ use opentelemetry::{otel_debug, otel_error, otel_warn, Context, InstrumentationS
 use std::fmt::Debug;
 use std::sync::atomic::AtomicBool;
 use std::sync::Mutex;
+use std::time::Duration;
 
 /// A [`LogProcessor`] designed for testing and debugging purpose, that immediately
 /// exports log records as they are emitted. Log records are exported synchronously
@@ -116,11 +117,11 @@ impl<T: LogExporter> LogProcessor for SimpleLogProcessor<T> {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
         self.is_shutdown
             .store(true, std::sync::atomic::Ordering::Relaxed);
         if let Ok(exporter) = self.exporter.lock() {
-            exporter.shutdown()
+            exporter.shutdown_with_timeout(timeout)
         } else {
             Err(OTelSdkError::InternalFailure(
                 "SimpleLogProcessor mutex poison at shutdown".into(),