@@ -117,6 +117,9 @@ Increment the:
   deprecated C headers (`stdint.h`, `stddef.h`, `stdlib.h`, `string.h`,
   `stdio.h`, `ctype.h`, `limits.h`, `assert.h`) with their C++ equivalents
   ([#4349](https://github.com/open-telemetry/opentelemetry-cpp/pull/4349))
+* [BUG] Stop the Elasticsearch async ForceFlush reporting success without
+  waiting for the sessions it was asked about
+  [#4337](https://github.com/open-telemetry/opentelemetry-cpp/pull/4337)
 
 * [CONFIGURATION] Add SDK component builder interfaces to the registry
   [#4358](https://github.com/open-telemetry/opentelemetry-cpp/issues/4358)
 
@@ -55,4 +55,12 @@ if(OTELCPP_BUILD_TESTING)
     TARGET es_log_record_exporter_test
     TEST_PREFIX exporter.
     TEST_LIST es_log_record_exporter_test)
+
+  # AnIndefiniteFlushParksUntilTheOutcomeArrives really does park on the
+  # condition variable, so a regression there does not fail, it stalls until the
+  # exporter's own bound expires. CTest's default is 25 minutes, which is a long
+  # time to spend learning that. This is deliberately well above the bounds the
+  # cases use themselves, so that one of them expiring reports which case failed
+  # rather than being killed from the outside.
+  set_tests_properties(${es_log_record_exporter_test} PROPERTIES TIMEOUT 60)
 endif() # OTELCPP_BUILD_TESTING
@@ -16,8 +16,9 @@
 
 #ifdef ENABLE_ASYNC_EXPORT
 #  include <condition_variable>
-#  include <cstddef>
+#  include <cstdint>
 #  include <mutex>
+#  include <set>
 #endif
 
 OPENTELEMETRY_BEGIN_NAMESPACE
@@ -122,8 +123,16 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo
 
   /**
    * Force flush the exporter.
+   *
+   * Waits for the asynchronous sessions already started when the call was made, bounded by the
+   * caller's timeout rather than by the exporter's response timeout.
+   *
    * @param timeout an option timeout, default to max.
-   * @return return true when all data are exported, and false when timeout
+   * @return true when each of those exports has reported a terminal outcome, false on timeout.
+   *         The outcome is published before the session is torn down, so a true return does not
+   *         mean FinishSession() has run. Nor does it mean the batch reached Elasticsearch: a
+   *         failed export reports too, through the internal log. Surfacing that here is
+   *         [#3075](https://github.com/open-telemetry/opentelemetry-cpp/issues/3075).
    */
   bool ForceFlush(
       std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept override;
@@ -149,13 +158,21 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo
 #ifdef ENABLE_ASYNC_EXPORT
   struct SynchronizationData
   {
-    std::atomic<std::size_t> session_counter_{0};
-    std::atomic<std::size_t> finished_session_counter_{0};
+    // next_session_id and running_sessions are guarded by force_flush_cv_m, so a start or a
+    // finish cannot interleave with a waiter's snapshot. The wait compares ids by order, so they
+    // must keep increasing: uint64_t rather than size_t.
+    std::uint64_t next_session_id{0};
+    std::set<std::uint64_t> running_sessions;
     std::condition_variable force_flush_cv;
     std::mutex force_flush_cv_m;
-    std::recursive_mutex force_flush_m;
+
+    // Test synchronization. Raised under force_flush_cv_m once per ForceFlush(), immediately
+    // after that call takes its watermark. No exporter behaviour reads it.
+    std::uint64_t watermarks_taken{0};
   };
   nostd::shared_ptr<SynchronizationData> synchronization_data_;
+
+  friend class ElasticsearchExporterTestPeer;
 #endif
 };
 }  // namespace logs
 
@@ -5,7 +5,7 @@
 #include <chrono>
 #include <condition_variable>
 #include <cstdint>
-#include <map>
+#include <map>     // IWYU pragma: keep
 #include <memory>  // IWYU pragma: keep
 #include <mutex>
 #include <nlohmann/json.hpp>
@@ -27,9 +27,13 @@
 #include "opentelemetry/sdk/logs/recordable.h"
 #include "opentelemetry/version.h"
 
+// Half this file only exists under ENABLE_ASYNC_EXPORT, and include-what-you-use asks for these
+// in the configurations that build it and asks for them to go in the ones that do not.
 #ifdef ENABLE_ASYNC_EXPORT
 #  include <cstddef>
 #  include <functional>
+#  include <set>
+
 #  include "opentelemetry/common/timestamp.h"
 #endif
 
@@ -423,12 +427,7 @@ ElasticsearchLogRecordExporter::ElasticsearchLogRecordExporter(
       ,
       synchronization_data_(new SynchronizationData())
 #endif
-{
-#ifdef ENABLE_ASYNC_EXPORT
-  synchronization_data_->finished_session_counter_.store(0);
-  synchronization_data_->session_counter_.store(0);
-#endif
-}
+{}
 
 std::unique_ptr<sdklogs::Recordable> ElasticsearchLogRecordExporter::MakeRecordable() noexcept
 {
@@ -438,9 +437,72 @@ std::unique_ptr<sdklogs::Recordable> ElasticsearchLogRecordExporter::MakeRecorda
 sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
     const nostd::span<std::unique_ptr<sdklogs::Recordable>> &records) noexcept
 {
+#ifdef ENABLE_ASYNC_EXPORT
+  // Registered before anything that can return, so a flush asked from the moment these records
+  // arrive waits for them, and every exit below reports through the guard.
+  const std::size_t span_count = records.size();
+  auto synchronization_data    = synchronization_data_;
+
+  std::uint64_t session_id = 0;
+  {
+    std::lock_guard<std::mutex> lock(synchronization_data_->force_flush_cv_m);
+    session_id = synchronization_data_->next_session_id++;
+    synchronization_data_->running_sessions.insert(session_id);
+  }
+
+  // Retiring the export is separate from describing what happened to it, because the refusal
+  // below has to retire without having an outcome to report through the completion.
+  auto retire = [session_id, synchronization_data]() noexcept {
+    bool retired = false;
+    {
+      // Published under the mutex ForceFlush() waits on. A waiter that has evaluated its
+      // predicate but not yet parked would otherwise not see this until the next wakeup.
+      std::lock_guard<std::mutex> lock(synchronization_data->force_flush_cv_m);
+      retired = synchronization_data->running_sessions.erase(session_id) == 1;
+    }
+    synchronization_data->force_flush_cv.notify_all();
+    return retired;
+  };
+
+  using Completion    = std::function<bool(opentelemetry::sdk::common::ExportResult)>;
+  Completion complete = [span_count,
+                         retire](opentelemetry::sdk::common::ExportResult result) noexcept {
+    retire();
+
+    // Logged after the session is retired. The log handler is replaceable, and one that calls
+    // ForceFlush() would otherwise wait for the very session this call has not let go of yet.
+    //
+    // That is the whole of what the ordering buys, and it is worth being exact about the rest.
+    // It does not make a log handler safe to re-enter the exporter from in general: one that
+    // flushes from any callback the HTTP client dispatches can still wait on work that only the
+    // client thread it is standing on can advance. That is
+    // https://github.com/open-telemetry/opentelemetry-cpp/issues/4435, and it is not fixed here.
+    if (result != opentelemetry::sdk::common::ExportResult::kSuccess)
+    {
+      OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] ERROR: Export "
+                              << span_count
+                              << " log record(s) error: " << static_cast<int>(result));
+    }
+    else
+    {
+      OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Export " << span_count
+                                                          << " log record(s) success");
+    }
+    return true;
+  };
+
+  // Exactly two exits below: the shutdown refusal retires by hand, and the handler takes the
+  // session over after SendRequest(). A return added between them must do one or the other.
+#endif
+
   // Return failure if this exporter has been shutdown
   if (isShutdown())
   {
+#ifdef ENABLE_ASYNC_EXPORT
+    // Retired before the log line below, so a handler that flushes does not wait on this
+    // Export(). The refusal is reported there, not through the completion.
+    retire();
+#endif
     OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting "
                             << records.size() << " log(s) failed, exporter is shutdown");
     return sdk::common::ExportResult::kFailure;
@@ -481,29 +543,10 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
 
 #ifdef ENABLE_ASYNC_EXPORT
   // Send the request
-  synchronization_data_->session_counter_.fetch_add(1, std::memory_order_release);
-  std::size_t span_count    = records.size();
-  auto synchronization_data = synchronization_data_;
-  auto handler              = std::make_shared<AsyncResponseHandler>(
-      session,
-      [span_count, synchronization_data](opentelemetry::sdk::common::ExportResult result) {
-        if (result != opentelemetry::sdk::common::ExportResult::kSuccess)
-        {
-          OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] ERROR: Export "
-                                               << span_count
-                                               << " log record(s) error: " << static_cast<int>(result));
-        }
-        else
-        {
-          OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Export " << span_count
-                                                                           << " log record(s) success");
-        }
-
-        synchronization_data->finished_session_counter_.fetch_add(1, std::memory_order_release);
-        synchronization_data->force_flush_cv.notify_all();
-        return true;
-      },
-      options_.console_debug_);
+  // The handler reports this session from here on, so the completion goes to it rather than
+  // being copied to it.
+  auto handler =
+      std::make_shared<AsyncResponseHandler>(session, std::move(complete), options_.console_debug_);
   session->SendRequest(handler);
   return sdk::common::ExportResult::kSuccess;
 #else
@@ -548,41 +591,38 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou
                                                 OPENTELEMETRY_MAYBE_UNUSED) noexcept
 {
 #ifdef ENABLE_ASYNC_EXPORT
-  std::lock_guard<std::recursive_mutex> lock_guard{synchronization_data_->force_flush_m};
-  std::size_t running_counter =
-      synchronization_data_->session_counter_.load(std::memory_order_acquire);
   // ASAN will report chrono: runtime error: signed integer overflow: A + B cannot be represented
   //   in type 'long int' here. So we reset timeout to meet signed long int limit here.
+  // Returns zero for a timeout not worth waiting against, which is also how a caller asks for
+  //   no deadline.
   timeout = opentelemetry::common::DurationUtil::AdjustWaitForTimeout(
       timeout, std::chrono::microseconds::zero());
 
-  std::chrono::steady_clock::duration timeout_steady =
-      std::chrono::duration_cast<std::chrono::steady_clock::duration>(timeout);
-  if (timeout_steady <= std::chrono::steady_clock::duration::zero())
+  // No deadline still gets one, since a client is not obliged to call back. response_timeout_
+  // is the value the request itself carries, so an outstanding session owes an event within it.
+  if (timeout <= std::chrono::microseconds::zero())
   {
-    timeout_steady = (std::chrono::steady_clock::duration::max)();
+    timeout = std::chrono::seconds{options_.response_timeout_};
   }
 
-  std::unique_lock<std::mutex> lk_cv(synchronization_data_->force_flush_cv_m);
-  // Wait for all the sessions to finish
-  while (timeout_steady > std::chrono::steady_clock::duration::zero())
-  {
-    if (synchronization_data_->finished_session_counter_.load(std::memory_order_acquire) >=
-        running_counter)
-    {
-      break;
-    }
+  // Before the lock, so acquiring the mutex spends the caller's budget and not a fresh one.
+  const std::chrono::steady_clock::time_point deadline =
+      std::chrono::steady_clock::now() +
+      std::chrono::duration_cast<std::chrono::steady_clock::duration>(timeout);
 
-    std::chrono::steady_clock::time_point start_timepoint = std::chrono::steady_clock::now();
-    if (std::cv_status::no_timeout != synchronization_data_->force_flush_cv.wait_for(
-                                          lk_cv, std::chrono::seconds{options_.response_timeout_}))
-    {
-      break;
-    }
-    timeout_steady -= std::chrono::steady_clock::now() - start_timepoint;
-  }
+  std::unique_lock<std::mutex> lock(synchronization_data_->force_flush_cv_m);
+
+  // The next id, not a count: a session started later takes a larger id and cannot stand in for
+  // one of these. Ids are issued in order, so the smallest one still running decides.
+  const std::uint64_t watermark = synchronization_data_->next_session_id;
+  ++synchronization_data_->watermarks_taken;
+  const auto flushed = [this, watermark]() {
+    const auto &running = synchronization_data_->running_sessions;
+    return running.empty() || *running.begin() >= watermark;
+  };
 
-  return timeout_steady > std::chrono::steady_clock::duration::zero();
+  // wait_until() returns the predicate, so running out of time cannot report success.
+  return synchronization_data_->force_flush_cv.wait_until(lock, deadline, flushed);
 #else
   return true;
 #endif