[BUG] Wait on a completion state in the Elasticsearch exporter

CHANGELOG.md

@@ -15,6 +15,9 @@ Increment the:
 
 ## [Unreleased]
 
+* [BUG] Wait on a completion state in the Elasticsearch exporter
+  [#4298](https://github.com/open-telemetry/opentelemetry-cpp/pull/4298)
+
 * [CODE HEALTH] Move metrics storage test fixtures into anonymous namespace
   [#4286](https://github.com/open-telemetry/opentelemetry-cpp/pull/4286)
 

exporters/elasticsearch/src/es_log_record_exporter.cc

@@ -1,10 +1,10 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-#include <stdint.h>
 #include <atomic>
 #include <chrono>
 #include <condition_variable>
+#include <cstdint>
 #include <map>
 #include <memory>  // IWYU pragma: keep
 #include <mutex>
@@ -100,21 +100,24 @@ class ResponseHandler : public http_client::EventHandler
                                 << log_message);
       }
 
-      // Set the response_received_ flag to true and notify any threads waiting on this result
-      response_received_ = true;
+      // Record the outcome and notify any threads waiting on this result
+      recordCompletionLocked(CompletionState::Success);
     }
     cv_.notify_all();
   }
 
   /**
-   * A method the user calls to block their thread until the response is received. The longest
-   * duration is the timeout of the request, set by SetTimeoutMs()
+   * A method the user calls to block their thread until the request has either produced a
+   * response or failed. The longest duration is the timeout of the request, set by
+   * SetTimeoutMs(), which arrives here as a TimedOut session event.
    */
   bool waitForResponse()
   {
     std::unique_lock<std::mutex> lk(mutex_);
-    cv_.wait(lk);
-    return response_received_;
+    // Waiting on a predicate rather than bare: the completion may already have been recorded
+    // before this thread got here, in which case there is no notification left to receive.
+    cv_.wait(lk, [this] { return completion_ != CompletionState::Pending; });
+    return completion_ == CompletionState::Success;
   }
 
   /**
@@ -135,20 +138,23 @@ class ResponseHandler : public http_client::EventHandler
     {
       case http_client::SessionState::CreateFailed:
         OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed to create session");
-        cv_.notify_all();
+        recordCompletion(CompletionState::Failure);
         break;
       case http_client::SessionState::Created:
         OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Session created");
         break;
       case http_client::SessionState::Destroyed:
         OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Session destroyed");
+        // Nothing else will arrive after this. If no outcome was recorded, the session ended
+        // without a response, so release the waiter rather than leaving it blocked forever.
+        recordCompletion(CompletionState::Failure);
         break;
       case http_client::SessionState::Connecting:
         OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Connecting to peer");
         break;
       case http_client::SessionState::ConnectFailed:
         OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed to connect to peer");
-        cv_.notify_all();
+        recordCompletion(CompletionState::Failure);
         break;
       case http_client::SessionState::Connected:
         OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Connected to peer");
@@ -158,22 +164,22 @@ class ResponseHandler : public http_client::EventHandler
         break;
       case http_client::SessionState::SendFailed:
         OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed to send request");
-        cv_.notify_all();
+        recordCompletion(CompletionState::Failure);
         break;
       case http_client::SessionState::Response:
         OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Received response");
         break;
       case http_client::SessionState::SSLHandshakeFailed:
         OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed SSL Handshake");
-        cv_.notify_all();
+        recordCompletion(CompletionState::Failure);
         break;
       case http_client::SessionState::TimedOut:
         OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Request timed out");
-        cv_.notify_all();
+        recordCompletion(CompletionState::Failure);
         break;
       case http_client::SessionState::NetworkError:
         OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Network error");
-        cv_.notify_all();
+        recordCompletion(CompletionState::Failure);
         break;
       case http_client::SessionState::ReadError:
         OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Read error");
@@ -183,18 +189,47 @@ class ResponseHandler : public http_client::EventHandler
         break;
       case http_client::SessionState::Cancelled:
         OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] (manually) cancelled");
-        cv_.notify_all();
+        recordCompletion(CompletionState::Failure);
         break;
     }
   }
 
 private:
+  enum class CompletionState : std::uint8_t
+  {
+    Pending,
+    Success,
+    Failure
+  };
+
+  /**
+   * Record the outcome of the request, first writer wins, then release any waiter. Keeping the
+   * first outcome means a session destroyed after a successful response does not overwrite it.
+   */
+  void recordCompletion(CompletionState state)
+  {
+    {
+      std::unique_lock<std::mutex> lk(mutex_);
+      recordCompletionLocked(state);
+    }
+    cv_.notify_all();
+  }
+
+  /// As recordCompletion(), for callers that already hold mutex_ and notify themselves.
+  void recordCompletionLocked(CompletionState state)
+  {
+    if (completion_ == CompletionState::Pending)
+    {
+      completion_ = state;
+    }
+  }
+
   // Define a condition variable and mutex
   std::condition_variable cv_;
   std::mutex mutex_;
 
-  // Whether the response from Elasticsearch has been received
-  bool response_received_ = false;
+  // Whether the request has completed, and how
+  CompletionState completion_ = CompletionState::Pending;
 
   // A string to store the response body
   std::string body_ = "";