[SDK] Preserve unlimited batch processor shutdown timeout

CHANGELOG.md

@@ -15,6 +15,10 @@ Increment the:
 
 ## [Unreleased]
 
+* [SDK] Preserve the default unlimited timeout when batch span and log record
+  processors shut down their exporters.
+  [#4686](https://github.com/open-telemetry/opentelemetry-cpp/pull/4686)
+
 * [CODE HEALTH] Prevent `std::out_of_range` escaping `noexcept` substr callers.
   [#4562](https://github.com/open-telemetry/opentelemetry-cpp/pull/4562)
 

sdk/src/logs/batch_log_record_processor.cc

@@ -359,6 +359,11 @@ void BatchLogRecordProcessor::GetWaitAdjustedTime(
     std::chrono::microseconds &timeout,
     std::chrono::time_point<std::chrono::system_clock> &start_time)
 {
+  if (timeout == (std::chrono::microseconds::max)())
+  {
+    return;
+  }
+
   auto end_time = std::chrono::system_clock::now();
   auto offset   = std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time);
   start_time    = end_time;

sdk/src/logs/multi_log_record_processor.cc

@@ -169,9 +169,10 @@ bool MultiLogRecordProcessor::InternalForceFlush(std::chrono::microseconds timeo
 
 bool MultiLogRecordProcessor::InternalShutdown(std::chrono::microseconds timeout) noexcept
 {
-  bool result           = true;
-  auto start_time       = std::chrono::system_clock::now();
-  auto overflow_checker = std::chrono::system_clock::time_point::max();
+  bool result                = true;
+  bool has_unlimited_timeout = timeout == (std::chrono::microseconds::max)();
+  auto start_time            = std::chrono::system_clock::now();
+  auto overflow_checker      = std::chrono::system_clock::time_point::max();
   std::chrono::system_clock::time_point expire_time;
   if (std::chrono::duration_cast<std::chrono::microseconds>(overflow_checker - start_time) <=
       timeout)
@@ -189,6 +190,10 @@ bool MultiLogRecordProcessor::InternalShutdown(std::chrono::microseconds timeout
     {
       result = false;
     }
+    if (has_unlimited_timeout)
+    {
+      continue;
+    }
     start_time = std::chrono::system_clock::now();
     if (expire_time > start_time)
     {

sdk/src/trace/batch_span_processor.cc

@@ -348,6 +348,11 @@ void BatchSpanProcessor::GetWaitAdjustedTime(
     std::chrono::microseconds &timeout,
     std::chrono::time_point<std::chrono::system_clock> &start_time)
 {
+  if (timeout == (std::chrono::microseconds::max)())
+  {
+    return;
+  }
+
   auto end_time = std::chrono::system_clock::now();
   auto offset   = std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time);
   start_time    = end_time;

sdk/test/logs/batch_log_record_processor_test.cc

@@ -142,18 +142,22 @@ class MockLogExporter final : public LogRecordExporter
   }
 
   // toggles the boolean flag marking this exporter as shut down
-  bool Shutdown(std::chrono::microseconds /* timeout */) noexcept override
+  bool Shutdown(std::chrono::microseconds timeout) noexcept override
   {
-    *is_shutdown_ = true;
+    shutdown_timeout_ = timeout;
+    *is_shutdown_     = true;
     return true;
   }
 
+  std::chrono::microseconds GetShutdownTimeout() const noexcept { return shutdown_timeout_; }
+
 private:
   std::shared_ptr<std::vector<std::unique_ptr<MockLogRecordable>>> logs_received_;
   std::shared_ptr<std::atomic<std::size_t>> force_flush_counter_;
   std::shared_ptr<std::atomic<bool>> is_shutdown_;
   std::shared_ptr<std::atomic<bool>> is_export_completed_;
   std::chrono::milliseconds export_delay_;
+  std::chrono::microseconds shutdown_timeout_{std::chrono::microseconds::zero()};
 };
 
 /**
@@ -191,8 +195,12 @@ TEST_F(BatchLogRecordProcessorTest, TestShutdown)
       new std::vector<std::unique_ptr<MockLogRecordable>>);
   std::shared_ptr<std::atomic<std::size_t>> force_flush_counter(new std::atomic<std::size_t>(0));
   std::shared_ptr<std::atomic<bool>> is_shutdown(new std::atomic<bool>(false));
+  std::shared_ptr<std::atomic<bool>> is_export_completed(new std::atomic<bool>(false));
 
-  auto batch_processor = GetMockProcessor(logs_received, force_flush_counter, is_shutdown);
+  auto exporter =
+      new MockLogExporter(logs_received, force_flush_counter, is_shutdown, is_export_completed);
+  auto batch_processor = std::shared_ptr<LogRecordProcessor>(
+      new BatchLogRecordProcessor(std::unique_ptr<LogRecordExporter>(exporter)));
 
   // Create a few test log records and send them to the processor
   const int num_logs = 3;
@@ -221,6 +229,7 @@ TEST_F(BatchLogRecordProcessorTest, TestShutdown)
 
   // Also check that the processor is shut down at the end
   EXPECT_TRUE(is_shutdown->load());
+  EXPECT_EQ((std::chrono::microseconds::max)(), exporter->GetShutdownTimeout());
 }
 
 TEST_F(BatchLogRecordProcessorTest, TestForceFlush)

sdk/test/logs/multi_log_record_processor_test.cc

@@ -7,6 +7,7 @@
 #include <cstdint>
 #include <memory>
 #include <string>
+#include <thread>
 #include <utility>
 #include <vector>
 
@@ -172,7 +173,14 @@ TEST(MultiLogRecordProcessorTest, EmptyProcessorIsDisabled)
 class FixedResultProcessor final : public LogRecordProcessor
 {
 public:
-  explicit FixedResultProcessor(bool result) : result_(result) {}
+  explicit FixedResultProcessor(
+      bool result,
+      std::shared_ptr<std::chrono::microseconds> shutdown_timeout = nullptr,
+      std::chrono::microseconds shutdown_delay = std::chrono::microseconds::zero())
+      : result_(result),
+        shutdown_timeout_(std::move(shutdown_timeout)),
+        shutdown_delay_(shutdown_delay)
+  {}
 
   std::unique_ptr<Recordable> MakeRecordable() noexcept override
   {
@@ -187,10 +195,20 @@ class FixedResultProcessor final : public LogRecordProcessor
 
   bool ForceFlush(std::chrono::microseconds /* timeout */) noexcept override { return result_; }
 
-  bool Shutdown(std::chrono::microseconds /* timeout */) noexcept override { return result_; }
+  bool Shutdown(std::chrono::microseconds timeout) noexcept override
+  {
+    if (shutdown_timeout_ != nullptr)
+    {
+      *shutdown_timeout_ = timeout;
+    }
+    std::this_thread::sleep_for(shutdown_delay_);
+    return result_;
+  }
 
 private:
   bool result_;
+  std::shared_ptr<std::chrono::microseconds> shutdown_timeout_;
+  std::chrono::microseconds shutdown_delay_;
 };
 
 std::unique_ptr<LogRecordProcessor> MakeMultiProcessor(bool first, bool second)
@@ -217,4 +235,36 @@ TEST(MultiLogRecordProcessorTest, ShutdownFailsWhenAnyChildFails)
   EXPECT_FALSE(MakeMultiProcessor(false, true)->Shutdown());
 }
 
+TEST(MultiLogRecordProcessorTest, DefaultShutdownPassesUnlimitedTimeoutToAllChildren)
+{
+  auto first_timeout  = std::make_shared<std::chrono::microseconds>();
+  auto second_timeout = std::make_shared<std::chrono::microseconds>();
+
+  std::vector<std::unique_ptr<LogRecordProcessor>> processors;
+  processors.emplace_back(new FixedResultProcessor(true, first_timeout));
+  processors.emplace_back(new FixedResultProcessor(true, second_timeout));
+  MultiLogRecordProcessor processor(std::move(processors));
+
+  EXPECT_TRUE(processor.Shutdown());
+  EXPECT_EQ((std::chrono::microseconds::max)(), *first_timeout);
+  EXPECT_EQ((std::chrono::microseconds::max)(), *second_timeout);
+}
+
+TEST(MultiLogRecordProcessorTest, FiniteShutdownSharesTimeoutBudgetAcrossChildren)
+{
+  auto first_timeout  = std::make_shared<std::chrono::microseconds>();
+  auto second_timeout = std::make_shared<std::chrono::microseconds>();
+  auto timeout        = std::chrono::milliseconds(100);
+
+  std::vector<std::unique_ptr<LogRecordProcessor>> processors;
+  processors.emplace_back(
+      new FixedResultProcessor(true, first_timeout, std::chrono::milliseconds(2)));
+  processors.emplace_back(new FixedResultProcessor(true, second_timeout));
+  MultiLogRecordProcessor processor(std::move(processors));
+
+  EXPECT_TRUE(processor.Shutdown(timeout));
+  EXPECT_EQ(timeout, *first_timeout);
+  EXPECT_LT(*second_timeout, timeout);
+}
+
 }  // namespace

sdk/test/trace/batch_span_processor_test.cc

@@ -88,21 +88,25 @@ class MockSpanExporter final : public sdk::trace::SpanExporter
     return true;
   }
 
-  bool Shutdown(std::chrono::microseconds /* timeout */) noexcept override
+  bool Shutdown(std::chrono::microseconds timeout) noexcept override
   {
-    *is_shutdown_ = true;
+    shutdown_timeout_ = timeout;
+    *is_shutdown_     = true;
     return true;
   }
 
   bool IsExportCompleted() { return is_export_completed_->load(); }
 
+  std::chrono::microseconds GetShutdownTimeout() const noexcept { return shutdown_timeout_; }
+
 private:
   std::shared_ptr<std::vector<std::unique_ptr<sdk::trace::SpanData>>> spans_received_;
   std::shared_ptr<std::atomic<std::size_t>> shut_down_counter_;
   std::shared_ptr<std::atomic<bool>> is_shutdown_;
   std::shared_ptr<std::atomic<bool>> is_export_completed_;
   // Meant exclusively to test force flush timeout
   std::chrono::milliseconds export_delay_;
+  std::chrono::microseconds shutdown_timeout_{std::chrono::microseconds::zero()};
 };
 
 /**
@@ -138,11 +142,10 @@ TEST_F(BatchSpanProcessorTestPeer, TestShutdown)
   std::shared_ptr<std::vector<std::unique_ptr<sdk::trace::SpanData>>> spans_received(
       new std::vector<std::unique_ptr<sdk::trace::SpanData>>);
 
-  auto batch_processor =
-      std::shared_ptr<sdk::trace::BatchSpanProcessor>(new sdk::trace::BatchSpanProcessor(
-          std::unique_ptr<sdk::trace::SpanExporter>(
-              new MockSpanExporter(spans_received, shut_down_counter, is_shutdown)),
-          sdk::trace::BatchSpanProcessorOptions()));
+  auto exporter        = new MockSpanExporter(spans_received, shut_down_counter, is_shutdown);
+  auto batch_processor = std::shared_ptr<sdk::trace::BatchSpanProcessor>(
+      new sdk::trace::BatchSpanProcessor(std::unique_ptr<sdk::trace::SpanExporter>(exporter),
+                                         sdk::trace::BatchSpanProcessorOptions()));
   const int num_spans = 3;
 
   auto test_spans = GetTestSpans(batch_processor, num_spans);
@@ -163,6 +166,7 @@ TEST_F(BatchSpanProcessorTestPeer, TestShutdown)
   }
 
   EXPECT_TRUE(is_shutdown->load());
+  EXPECT_EQ((std::chrono::microseconds::max)(), exporter->GetShutdownTimeout());
 }
 
 TEST_F(BatchSpanProcessorTestPeer, TestForceFlush)