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

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?