[BUG] Elasticsearch async ForceFlush reports success on timeout and ignores the caller deadline

[thc1006]

Describe your environment

main at 11fa0db0, Linux, GCC 14.2, configured with -DWITH_ELASTICSEARCH=ON -DWITH_ASYNC_EXPORT_PREVIEW=ON. The path is inside #ifdef ENABLE_ASYNC_EXPORT, so it only exists in an async build.

Steps to reproduce

An in-test fake HttpClient whose Session::SendRequest() accepts the handler and never calls it back, so the session Export() starts never finishes.

auto client = std::make_shared<FakeHttpClient>([](http_client::EventHandler &) {});
ElasticsearchExporterOptions options;
options.response_timeout_ = 2;  // seconds
ElasticsearchLogRecordExporter exporter(options, client);

auto record = exporter.MakeRecordable();
exporter.Export(nostd::span<std::unique_ptr<sdklogs::Recordable>>(&record, 1));

const auto start   = std::chrono::steady_clock::now();
const bool flushed = exporter.ForceFlush(std::chrono::milliseconds{1});
const auto elapsed = std::chrono::steady_clock::now() - start;

What is the expected behavior?

flushed == false, returning at roughly one millisecond. That is what the method's own documentation promises in es_log_record_exporter.h:

 * @param timeout an option timeout, default to max.
 * @return return true when all data are exported, and false when timeout

What is the actual behavior?

[probe] ForceFlush(1ms) returned true after 2000 ms

Both halves are wrong. It reports success while a session is still running, and it ignores the caller's deadline in favour of response_timeout_.

Additional context

exporters/elasticsearch/src/es_log_record_exporter.cc, in ForceFlush:

while (timeout_steady > std::chrono::steady_clock::duration::zero())
{
  if (finished_session_counter_.load(...) >= running_counter)
  {
    break;
  }

  std::chrono::steady_clock::time_point start_timepoint = std::chrono::steady_clock::now();
  if (std::cv_status::no_timeout != force_flush_cv.wait_for(
                                        lk_cv, std::chrono::seconds{options_.response_timeout_}))
  {
    break;
  }
  timeout_steady -= std::chrono::steady_clock::now() - start_timepoint;
}

return timeout_steady > std::chrono::steady_clock::duration::zero();

The timeout branch leaves the loop without subtracting the time it just spent, and the loop condition has already established that timeout_steady is positive, so the return is true on that path no matter what. It is not a corner case: every exit through the inner timeout reports success. The only way to get false is to be notified repeatedly without completing until the subtraction drains the budget.

The wait itself is seconds{options_.response_timeout_} rather than the caller's remaining time, which is where the 2000 ms comes from.

There is a spec angle beyond the header's own wording. LogRecordProcessor ForceFlush says "If a timeout is specified, the LogRecordProcessor MUST prioritize honoring the timeout over finishing all calls", and the exporter's Export "MUST NOT block indefinitely, there MUST be a reasonable upper limit after which the call must time out". A ForceFlush that overruns its deadline and then claims success works against both.

Two smaller things in the same area, mentioned so they are not lost rather than to widen this issue:

• The completion callback increments finished_session_counter_ and calls notify_all() without holding force_flush_cv_m, so a completion landing between the waiter's predicate check and its park is not seen until the next wakeup. OtlpHttpClient::ReleaseSession publishes its counter under a lock for this reason.
• Shutdown(timeout) has its parameter commented out and always returns true. Bounding it properly needs a deadline aware FinishAllSessions in ext/http, which is an API question rather than an exporter fix, so I would rather raise that separately if you agree it is worth doing.

I have a fix and a regression test for the ForceFlush part and am happy to open a PR. The shape is a single steady_clock deadline from the caller timeout, wait_until with a completion predicate, and the predicate as the return value, plus publishing the counter under force_flush_cv_m. One decision I would rather have your view on first: should ForceFlush wait only for the sessions that were in flight when it was entered, which is what the current counter snapshot intends, or should it keep waiting while new exports arrive?

[thc1006]

A third defect in the same function, found after filing and reproduced against the branch in #4337.

Both counters are monotonic totals with no session identity, so any completion satisfies any waiter:

const std::size_t running_counter = session_counter_.load(...);           // total ever started
const auto flushed = [running_counter] {
  return finished_session_counter_.load(...) >= running_counter;          // total ever finished
};

A flush that entered with sessions A and B outstanding waits for finished >= 2. If session C starts afterwards and completes, and B completes, the count reaches two and the flush reports success while A is still running.

Reproduced with a fake client, three exports and a flush on its own thread:

ForceFlush reported success while the first session was still outstanding
[  FAILED  ] ElasticsearchForceFlushTests.ALaterSessionCanSatisfyAnEarlierFlush (101 ms)

main compares the same way, so this is not new. It is also not what #4338 fixes: making each session publish exactly one completion does not make that completion attributable to the session a waiter is waiting for.

A fix has to track which sessions are outstanding rather than how many. OtlpHttpClient already carries that shape, in running_sessions_, which is why its loop can also break on running_sessions_.empty(). The Elasticsearch exporter's SynchronizationData holds only the two counters, so this is a change to that struct rather than to ForceFlush.

I have not folded it into #4337, which corrects the deadline and the predicate and leaves the counter semantics as they are on main. Happy to take it as a follow-up once you have a view on the shape, since it touches the same struct #4338 does.