[BUG] Stop the Elasticsearch async ForceFlush reporting success without waiting

ForceFlush() returned true straight away when async export was enabled, so a
caller had no way to know whether anything had been delivered.

It now waits on the sessions that were in flight when it was called. Sessions
are identified rather than counted: each export takes the next id and joins a
set, and the wait ends when no id below the entry watermark is left, so a
completion cannot satisfy a flush that started after it. Counting alone lets a
later export stand in for an earlier one. The wait uses one deadline for the
call, so a wakeup that is not a completion resumes against what is left instead
of restarting the wait, and the result is the predicate rather than the leftover
duration.

The ids are uint64_t rather than size_t. The wait compares them by order, which
only holds while they keep increasing, and a 32 bit counter reaches its end in
days at a rate this exporter is meant to sustain; past that the next watermark
is small enough for a still running session to satisfy it.

AsyncResponseHandler reports at most once, through a compare and exchange. The
HTTP client can deliver both a response and a terminal session event for one
request, and the exporter counts one finished session per export.

An export registers its session before anything that can return, so a flush
asked from the moment the records arrive waits for them, and a guard reports
through the same completion on every early exit. Without it a return added later
would strand a waiter on a session that can never finish.

What a true return means is documented on the declaration, and it is weaker than
"the session ended". Both completion paths publish the outcome before the
session is torn down: OnResponse calls CompleteOnce() ahead of its logging, and
the handler destructor calls it ahead of FinishSession(). So a true return means
every snapshotted export has reported an outcome, and transport cleanup may
still be running.

Reported in #4336

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>

Author: thc1006

CHANGELOG.md

@@ -56,6 +56,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)

exporters/elasticsearch/CMakeLists.txt

@@ -55,4 +55,10 @@ if(BUILD_TESTING)
     TARGET es_log_record_exporter_test
     TEST_PREFIX exporter.
     TEST_LIST es_log_record_exporter_test)
+
+  # AnIndefiniteFlushReturnsOnceEverythingIsFinished exercises the branch that
+  # waits without a deadline, so a regression there does not fail, it stops.
+  # CTest's default bound is 25 minutes, which is a long time to spend learning
+  # that.
+  set_tests_properties(${es_log_record_exporter_test} PROPERTIES TIMEOUT 30)
 endif() # BUILD_TESTING

exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h

@@ -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,11 +158,17 @@ 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};
+    // Identified rather than counted, so a completion can only satisfy its own waiter. Guarded by
+    // force_flush_cv_m, the mutex the wait uses, so starting and finishing cannot interleave with a
+    // waiter's snapshot or predicate.
+    //
+    // Sized independently of the platform's size_t. The wait compares ids by order, which only
+    // holds while they keep increasing, and a 32 bit counter reaches its end in days at a rate
+    // this exporter is meant to sustain.
+    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;
   };
   nostd::shared_ptr<SynchronizationData> synchronization_data_;
 #endif