Fix: BatchLogProcessor to invoke shutdown on exporter (#2696)

Author: cijothomas

opentelemetry-sdk/CHANGELOG.md

@@ -33,17 +33,25 @@
     }
 }
 ```
+
 - **Breaking** The SpanExporter::export() method no longer requires a mutable reference to self.
-  Before: 
+  Before:
+
   ```rust
     async fn export(&mut self, batch: Vec<SpanData>) -> OTelSdkResult 
   ```
-  After: 
-  ```rust 
+
+  After:
+
+  ```rust
     async fn export(&self, batch: Vec<SpanData>) -> OTelSdkResult  
   ```
+
   Custom exporters will need to internally synchronize any mutable state, if applicable.
-  
+
+- Bug Fix: `BatchLogProcessor` now correctly calls `shutdown` on the exporter
+  when its `shutdown` is invoked.
+
 ## 0.28.0
 
 Released 2025-Feb-10

opentelemetry-sdk/src/logs/batch_log_processor.rs

@@ -436,6 +436,7 @@ impl BatchLogProcessor {
                                 &current_batch_size,
                                 &config,
                             );
+                            let _ = exporter.shutdown();
                             let _ = sender.send(result);
 
                             otel_debug!(
@@ -925,7 +926,8 @@ mod tests {
         processor.shutdown().unwrap();
         // todo: expect to see errors here. How should we assert this?
         processor.emit(&mut record, &instrumentation);
-        assert_eq!(1, exporter.get_emitted_logs().unwrap().len())
+        assert_eq!(1, exporter.get_emitted_logs().unwrap().len());
+        assert!(exporter.is_shutdown_called());
     }
 
     #[tokio::test(flavor = "current_thread")]

opentelemetry-sdk/src/logs/in_memory_exporter.rs

@@ -4,6 +4,7 @@ use crate::logs::{LogBatch, LogExporter};
 use crate::Resource;
 use opentelemetry::InstrumentationScope;
 use std::borrow::Cow;
+use std::sync::atomic::AtomicBool;
 use std::sync::{Arc, Mutex};
 
 type LogResult<T> = Result<T, OTelSdkError>;
@@ -42,6 +43,7 @@ pub struct InMemoryLogExporter {
     logs: Arc<Mutex<Vec<OwnedLogData>>>,
     resource: Arc<Mutex<Resource>>,
     should_reset_on_shutdown: bool,
+    shutdown_called: Arc<AtomicBool>,
 }
 
 impl Default for InMemoryLogExporter {
@@ -124,6 +126,7 @@ impl InMemoryLogExporterBuilder {
             logs: Arc::new(Mutex::new(Vec::new())),
             resource: Arc::new(Mutex::new(Resource::builder().build())),
             should_reset_on_shutdown: self.reset_on_shutdown,
+            shutdown_called: Arc::new(AtomicBool::new(false)),
         }
     }
 
@@ -137,6 +140,12 @@ impl InMemoryLogExporterBuilder {
 }
 
 impl InMemoryLogExporter {
+    /// Returns true if shutdown was called.
+    pub fn is_shutdown_called(&self) -> bool {
+        self.shutdown_called
+            .load(std::sync::atomic::Ordering::Relaxed)
+    }
+
     /// Returns the logs emitted via Logger as a vector of `LogDataWithResource`.
     ///
     /// # Example
@@ -203,6 +212,8 @@ impl LogExporter for InMemoryLogExporter {
     }
 
     fn shutdown(&mut self) -> OTelSdkResult {
+        self.shutdown_called
+            .store(true, std::sync::atomic::Ordering::Relaxed);
         if self.should_reset_on_shutdown {
             self.reset();
         }

opentelemetry-sdk/src/logs/simple_log_processor.rs

@@ -223,7 +223,8 @@ mod tests {
 
         processor.emit(&mut record, &instrumentation);
 
-        assert_eq!(1, exporter.get_emitted_logs().unwrap().len())
+        assert_eq!(1, exporter.get_emitted_logs().unwrap().len());
+        assert!(exporter.is_shutdown_called());
     }
 
     #[test]