diff --git a/CHANGELOG.md b/CHANGELOG.md
index 765eb43998..43afc03877 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,6 +15,17 @@ Increment the:
 
 ## [Unreleased]
 
+* [EXPORTER] Fix the Elasticsearch log exporter's `Shutdown()` ignoring its
+  timeout and always reporting success. It now flushes pending exports against
+  the caller's deadline before cancelling sessions, and returns whether that
+  flush actually completed in time. Also closes a race where a session could
+  register for export after `Shutdown()` had already taken its snapshot of
+  in-flight sessions, so `ForceFlush()` could return without ever waiting for
+  it. Also fixes that wait turning into a CPU spin when `response_timeout_` is
+  not positive, which (like `SetTimeoutMs()`/`CURLOPT_TIMEOUT_MS`) means no
+  timeout rather than zero.
+  [#4359](https://github.com/open-telemetry/opentelemetry-cpp/issues/4359)
+
 * [SDK] Clamp `BatchSpanProcessor` max_export_batch_size to max_queue_size,
   so a larger value no longer stops spans from being exported until the
   schedule delay expires.
diff --git a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h
index 6234df59a2..4d32d0a70f 100644
--- a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h
+++ b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h
@@ -129,8 +129,10 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo
       std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept override;
 
   /**
-   * Shutdown this exporter.
-   * @param timeout The maximum time to wait for the shutdown method to return
+   * Shutdown this exporter. Flushes any pending export within the given timeout before
+   * cancelling remaining sessions, then reports whether the flush completed in time.
+   * @param timeout The maximum time to wait for pending exports to flush before shutting down
+   * @return true if all pending exports flushed before the timeout, false otherwise
    */
   bool Shutdown(
       std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept override;
diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc
index cd3b1bdbfd..637d8b8324 100644
--- a/exporters/elasticsearch/src/es_log_record_exporter.cc
+++ b/exporters/elasticsearch/src/es_log_record_exporter.cc
@@ -1,6 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
+#include <algorithm>  // IWYU pragma: keep
 #include <atomic>
 #include <chrono>
 #include <condition_variable>
@@ -441,8 +442,29 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
   request->SetBody(body_vec);
 
 #ifdef ENABLE_ASYNC_EXPORT
-  // Send the request
-  synchronization_data_->session_counter_.fetch_add(1, std::memory_order_release);
+  bool rejected = false;
+  // Send the request. Registration has to happen under the same lock Shutdown() takes to
+  // flip is_shutdown_ and snapshot session_counter_ (see ForceFlush()) - otherwise a session
+  // that passes the isShutdown() check above can still register after Shutdown() has already
+  // taken its snapshot, and ForceFlush() would return without ever having waited for it.
+  {
+    std::lock_guard<std::recursive_mutex> lock_guard{synchronization_data_->force_flush_m};
+    rejected = isShutdown();
+    if (!rejected)
+    {
+      synchronization_data_->session_counter_.fetch_add(1, std::memory_order_release);
+    }
+  }
+
+  // Outside the lock: the client owns the session until somebody hands it back, and this is the
+  // only path that can, since the handler that would do it later is never built.
+  if (rejected)
+  {
+    session->FinishSession();
+    OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting "
+                            << records.size() << " log(s) failed, exporter is shutdown");
+    return sdk::common::ExportResult::kFailure;
+  }
   std::size_t span_count    = records.size();
   auto synchronization_data = synchronization_data_;
   auto handler              = std::make_shared<AsyncResponseHandler>(
@@ -525,7 +547,10 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou
   }
 
   std::unique_lock<std::mutex> lk_cv(synchronization_data_->force_flush_cv_m);
-  // Wait for all the sessions to finish
+  // Wait for all the sessions to finish. The condition variable's return value is not trusted
+  // on its own: the standard permits wait_for() to report cv_status::no_timeout on a spurious
+  // wakeup, indistinguishable from a real notification, so completion is always verified
+  // against finished_session_counter_ directly instead of inferred from the wait's return.
   while (timeout_steady > std::chrono::steady_clock::duration::zero())
   {
     if (synchronization_data_->finished_session_counter_.load(std::memory_order_acquire) >=
@@ -534,30 +559,55 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou
       break;
     }
 
+    // Clamp the wait to whatever is left of the caller's deadline: waiting the full
+    // response_timeout_ regardless of timeout_steady would let Shutdown(timeout) block far
+    // longer than the timeout it was given whenever nothing ever notifies this condition
+    // variable (e.g. an export that never completes). A non-positive response_timeout_ means
+    // "no timeout" (see SetTimeoutMs()/CURLOPT_TIMEOUT_MS), not zero, so it must not be allowed
+    // to collapse the wait into a spin; fall back to whatever is left of the caller's deadline.
+    const std::chrono::steady_clock::duration configured =
+        options_.response_timeout_ > 0
+            ? std::chrono::duration_cast<std::chrono::steady_clock::duration>(
+                  std::chrono::seconds{options_.response_timeout_})
+            : timeout_steady;
+    const std::chrono::steady_clock::duration wait_interval =
+        (std::min)(configured, timeout_steady);
+
     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;
-    }
+    synchronization_data_->force_flush_cv.wait_for(lk_cv, wait_interval);
     timeout_steady -= std::chrono::steady_clock::now() - start_timepoint;
   }
 
-  return timeout_steady > std::chrono::steady_clock::duration::zero();
+  return synchronization_data_->finished_session_counter_.load(std::memory_order_acquire) >=
+         running_counter;
 #else
   return true;
 #endif
 }
 
-bool ElasticsearchLogRecordExporter::Shutdown(std::chrono::microseconds /* timeout */) noexcept
+bool ElasticsearchLogRecordExporter::Shutdown(std::chrono::microseconds timeout) noexcept
 {
+#ifdef ENABLE_ASYNC_EXPORT
+  {
+    // Same lock Export() takes around its isShutdown() check and registration, so that by the
+    // time ForceFlush() below takes its session_counter_ snapshot, every session that is
+    // going to register for this shutdown already has.
+    std::lock_guard<std::recursive_mutex> lock_guard{synchronization_data_->force_flush_m};
+    is_shutdown_ = true;
+  }
+#else
   is_shutdown_ = true;
+#endif
+
+  // Flush with the caller's deadline before cancelling anything, so the wait below has
+  // something to wait for. Cancelling first would leave nothing pending to flush.
+  const bool flushed = ForceFlush(timeout);
 
   // Shutdown the session manager
   http_client_->CancelAllSessions();
   http_client_->FinishAllSessions();
 
-  return true;
+  return flushed;
 }
 
 bool ElasticsearchLogRecordExporter::isShutdown() const noexcept
diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc
index 52a5d9b6ef..bbfff96629 100644
--- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc
+++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc
@@ -19,10 +19,14 @@
 #include <gtest/gtest.h>
 #include <array>
 #include <chrono>
+#include <condition_variable>
 #include <cstddef>
 #include <cstdint>
+#include <mutex>
 #include <string>
+#include <thread>  // IWYU pragma: keep
 #include <utility>
+#include <vector>
 #include "nlohmann/json.hpp"
 
 namespace
@@ -106,6 +110,114 @@ class FakeSession final : public http_client::Session
   bool FinishSession() noexcept override { return true; }
 };
 
+// A client that owns what it creates, the way the curl client does: CreateSession() keeps a
+// reference in the client and only FinishSession() gives it back. CreateSession() also waits on
+// a gate, so a case can run Shutdown() while an export sits between its two shutdown checks.
+class RetainingSession final : public http_client::Session
+{
+public:
+  std::shared_ptr<http_client::Request> CreateRequest() noexcept override
+  {
+    return std::make_shared<FakeRequest>();
+  }
+
+  void SendRequest(std::shared_ptr<http_client::EventHandler>) noexcept override { sent_ = true; }
+
+  bool IsSessionActive() noexcept override { return finish_calls_ == 0; }
+  bool CancelSession() noexcept override { return true; }
+  // Counted rather than flagged: handing a session back twice is as wrong as not at all, and a
+  // flag reads the same either way.
+  bool FinishSession() noexcept override
+  {
+    ++finish_calls_;
+    return true;
+  }
+
+  bool sent_                = false;
+  std::size_t finish_calls_ = 0;
+};
+
+class RetainingHttpClient final : public http_client::HttpClient
+{
+public:
+  std::shared_ptr<http_client::Session> CreateSession(
+      opentelemetry::nostd::string_view) noexcept override
+  {
+    {
+      std::unique_lock<std::mutex> lock{gate_m_};
+      entered_ = true;
+      gate_cv_.notify_all();
+      gate_cv_.wait(lock, [this] { return released_; });
+    }
+    auto session = std::make_shared<RetainingSession>();
+    std::lock_guard<std::mutex> lock{sessions_m_};
+    sessions_.push_back(session);
+    return session;
+  }
+
+  bool CancelAllSessions() noexcept override { return true; }
+  bool FinishAllSessions() noexcept override { return true; }
+  void SetMaxSessionsPerConnection(std::size_t) noexcept override {}
+
+  void WaitUntilCreating()
+  {
+    std::unique_lock<std::mutex> lock{gate_m_};
+    gate_cv_.wait(lock, [this] { return entered_; });
+  }
+
+  void Release()
+  {
+    {
+      std::lock_guard<std::mutex> lock{gate_m_};
+      released_ = true;
+    }
+    gate_cv_.notify_all();
+  }
+
+  std::size_t Sent()
+  {
+    std::lock_guard<std::mutex> lock{sessions_m_};
+    std::size_t n = 0;
+    for (const auto &s : sessions_)
+    {
+      n += s->sent_ ? 1 : 0;
+    }
+    return n;
+  }
+
+  std::size_t FinishCalls()
+  {
+    std::lock_guard<std::mutex> lock{sessions_m_};
+    std::size_t n = 0;
+    for (const auto &s : sessions_)
+    {
+      n += s->finish_calls_;
+    }
+    return n;
+  }
+
+  // What the client is still holding that nobody handed back.
+  std::size_t Retained()
+  {
+    std::lock_guard<std::mutex> lock{sessions_m_};
+    std::size_t n = 0;
+    for (const auto &s : sessions_)
+    {
+      n += s->finish_calls_ == 0 ? 1 : 0;
+    }
+    return n;
+  }
+
+private:
+  std::mutex gate_m_;
+  std::condition_variable gate_cv_;
+  bool entered_  = false;
+  bool released_ = false;
+
+  std::mutex sessions_m_;
+  std::vector<std::shared_ptr<RetainingSession>> sessions_;
+};
+
 class FakeHttpClient final : public http_client::HttpClient
 {
 public:
@@ -120,6 +232,62 @@ class FakeHttpClient final : public http_client::HttpClient
   void SetMaxSessionsPerConnection(std::size_t) noexcept override {}
 };
 
+// A session that parks the handler it is given, in the test case rather than in the session
+// itself, without ever calling back into it. The handler stays alive (so the export it
+// represents keeps counting as outstanding) without creating a Session/AsyncResponseHandler
+// reference cycle: AsyncResponseHandler holds a shared_ptr to its session, so a session that
+// held the handler back would keep both alive for the exporter's own lifetime, which a leak
+// sanitizer would flag.
+//
+// Only meaningful under ENABLE_ASYNC_EXPORT: that is the only build where ForceFlush() waits on
+// anything at all (see ElasticsearchLogRecordExporter::ForceFlush).
+#ifdef ENABLE_ASYNC_EXPORT
+class HoldingSession final : public http_client::Session
+{
+public:
+  explicit HoldingSession(std::shared_ptr<http_client::EventHandler> *parked) : parked_(parked) {}
+
+  std::shared_ptr<http_client::Request> CreateRequest() noexcept override
+  {
+    return std::make_shared<FakeRequest>();
+  }
+
+  // Parked where the case can see it, not in this session: the handler owns its session, so a
+  // session that owned the handler back would keep the pair alive.
+  void SendRequest(std::shared_ptr<http_client::EventHandler> handler) noexcept override
+  {
+    *parked_ = std::move(handler);
+  }
+
+  bool IsSessionActive() noexcept override { return true; }
+  bool CancelSession() noexcept override { return true; }
+  bool FinishSession() noexcept override { return true; }
+
+private:
+  std::shared_ptr<http_client::EventHandler> *parked_;
+};
+
+class HoldingHttpClient final : public http_client::HttpClient
+{
+public:
+  explicit HoldingHttpClient(std::shared_ptr<http_client::EventHandler> *parked) : parked_(parked)
+  {}
+
+  std::shared_ptr<http_client::Session> CreateSession(
+      opentelemetry::nostd::string_view) noexcept override
+  {
+    return std::make_shared<HoldingSession>(parked_);
+  }
+
+  bool CancelAllSessions() noexcept override { return true; }
+  bool FinishAllSessions() noexcept override { return true; }
+  void SetMaxSessionsPerConnection(std::size_t) noexcept override {}
+
+private:
+  std::shared_ptr<http_client::EventHandler> *parked_;
+};
+#endif  // ENABLE_ASYNC_EXPORT
+
 }  // namespace
 
 namespace sdklogs       = opentelemetry::sdk::logs;
@@ -157,6 +325,151 @@ TEST(ElasticsearchLogsExporterTests, CustomClientConstructionSucceeds)
   ASSERT_NE(exporter, nullptr);
 }
 
+// Regression test: Shutdown() used to ignore its timeout parameter entirely and always return
+// true, whether or not anything had actually flushed. It should now report what flushing (via
+// ForceFlush) actually found. With FakeHttpClient, every export completes synchronously inside
+// Export() itself, so there is nothing left pending by the time Shutdown() runs.
+TEST(ElasticsearchLogsExporterTests, ShutdownReportsFlushCompletion)
+{
+  logs_exporter::ElasticsearchExporterOptions options;
+  auto http_client = std::make_shared<FakeHttpClient>();
+  auto exporter    = std::unique_ptr<sdklogs::LogRecordExporter>(
+      new logs_exporter::ElasticsearchLogRecordExporter(options, http_client));
+
+  auto record = exporter->MakeRecordable();
+  record->SetBody("shutdown regression test");
+  auto export_result =
+      exporter->Export(nostd::span<std::unique_ptr<sdklogs::Recordable>>(&record, 1));
+  ASSERT_EQ(export_result, opentelemetry::sdk::common::ExportResult::kSuccess);
+
+  EXPECT_TRUE(exporter->Shutdown(std::chrono::seconds(1)));
+}
+
+// Regression test: ForceFlush()'s wait loop always waited the full response_timeout_ on its
+// condition variable, regardless of how much of the caller's own timeout was left. With an
+// export still outstanding and nothing to notify the condition variable, Shutdown(1us) waited
+// the full response_timeout_ (30s by default) instead of returning after about 1us. The wait
+// is now clamped to whatever remains of the caller's deadline, matching the pattern already
+// used by the OTLP HTTP client's ForceFlush().
+#ifdef ENABLE_ASYNC_EXPORT
+TEST(ElasticsearchLogsExporterTests, ShutdownClampsWaitToCallerTimeoutWhenExportIsOutstanding)
+{
+  // Declared first so it outlives the client and the session that point at it.
+  std::shared_ptr<http_client::EventHandler> parked;
+  logs_exporter::ElasticsearchExporterOptions options;
+  auto http_client = std::make_shared<HoldingHttpClient>(&parked);
+  auto exporter    = std::unique_ptr<sdklogs::LogRecordExporter>(
+      new logs_exporter::ElasticsearchLogRecordExporter(options, http_client));
+
+  auto record = exporter->MakeRecordable();
+  record->SetBody("this export never completes");
+  auto export_result =
+      exporter->Export(nostd::span<std::unique_ptr<sdklogs::Recordable>>(&record, 1));
+  ASSERT_EQ(export_result, opentelemetry::sdk::common::ExportResult::kSuccess);
+
+  auto start   = std::chrono::steady_clock::now();
+  bool result  = exporter->Shutdown(std::chrono::microseconds(1));
+  auto elapsed = std::chrono::steady_clock::now() - start;
+
+  EXPECT_FALSE(result);
+  EXPECT_LT(elapsed, std::chrono::seconds(1));
+
+  // Let the export finish now that the assertions are done.
+  parked.reset();
+}
+
+// Regression test: a non-positive response_timeout_ made wait_interval collapse to zero
+// (duration_cast<steady_clock::duration>(seconds{0})), turning the wait loop into a spin that
+// re-armed on every near-instant wait_for() instead of blocking. response_timeout_ = 0 means
+// "no timeout" (see SetTimeoutMs()/CURLOPT_TIMEOUT_MS), matching the convention used by
+// ElasticsearchLogRecordExporter::Export()'s own deadline, not an instruction to poll
+// continuously. The result is still correct either way (bounded by timeout_steady), so this
+// mainly documents the fix; the absence of the spin itself was confirmed manually by reverting
+// configured's `response_timeout_ > 0 ? ... : timeout_steady` to the unconditional
+// response_timeout_ cast and observing the wait loop busy-spin instead of blocking.
+TEST(ElasticsearchLogsExporterTests, ShutdownDoesNotSpinWhenResponseTimeoutIsZero)
+{
+  // Declared first so it outlives the client and the session that point at it.
+  std::shared_ptr<http_client::EventHandler> parked;
+  logs_exporter::ElasticsearchExporterOptions options;
+  options.response_timeout_ = 0;
+  auto http_client          = std::make_shared<HoldingHttpClient>(&parked);
+  auto exporter             = std::unique_ptr<sdklogs::LogRecordExporter>(
+      new logs_exporter::ElasticsearchLogRecordExporter(options, http_client));
+
+  auto record = exporter->MakeRecordable();
+  record->SetBody("this export never completes");
+  auto export_result =
+      exporter->Export(nostd::span<std::unique_ptr<sdklogs::Recordable>>(&record, 1));
+  ASSERT_EQ(export_result, opentelemetry::sdk::common::ExportResult::kSuccess);
+
+  auto start   = std::chrono::steady_clock::now();
+  bool result  = exporter->Shutdown(std::chrono::milliseconds(50));
+  auto elapsed = std::chrono::steady_clock::now() - start;
+
+  EXPECT_FALSE(result);
+  EXPECT_LT(elapsed, std::chrono::seconds(1));
+
+  // Let the export finish now that the assertions are done.
+  parked.reset();
+}
+#endif  // ENABLE_ASYNC_EXPORT
+
+// Regression test: once Shutdown() has been called, any later Export() must fail rather than
+// silently trying to register a session against an exporter that is already tearing down.
+// The case is only meaningful where the second shutdown check exists, but it is registered in
+// both builds: gtest_add_tests reads the source, so a case behind #ifdef is still handed to CTest
+// in the build that does not compile it and reports a pass it never ran.
+TEST(ElasticsearchLogsExporterTests, ARejectedExportHandsItsSessionBack)
+{
+#ifndef ENABLE_ASYNC_EXPORT
+  GTEST_SKIP() << "the shutdown re-check this covers is compiled only with async export";
+#else
+  auto client = std::make_shared<RetainingHttpClient>();
+  logs_exporter::ElasticsearchExporterOptions options;
+  auto exporter = std::unique_ptr<sdklogs::LogRecordExporter>(
+      new logs_exporter::ElasticsearchLogRecordExporter(options, client));
+
+  auto record = exporter->MakeRecordable();
+  record->SetBody("a record the exporter will refuse");
+  std::array<std::unique_ptr<sdklogs::Recordable>, 1> batch = {std::move(record)};
+
+  auto result = opentelemetry::sdk::common::ExportResult::kSuccess;
+  std::thread exporting([&] {
+    result = exporter->Export(
+        nostd::span<std::unique_ptr<sdklogs::Recordable>>(batch.data(), batch.size()));
+  });
+
+  // Shutdown lands while the export is inside CreateSession, so it sees no registered session
+  // and returns, and the export then meets the second check on its way back.
+  client->WaitUntilCreating();
+  exporter->Shutdown();
+  client->Release();
+  exporting.join();
+
+  EXPECT_EQ(opentelemetry::sdk::common::ExportResult::kFailure, result);
+  EXPECT_EQ(0u, client->Sent()) << "a rejected export must not send";
+  EXPECT_EQ(0u, client->Retained())
+      << "the rejected session is still held by the client, so nothing will call FinishSession";
+  EXPECT_EQ(1u, client->FinishCalls()) << "handed back exactly once, not twice";
+#endif
+}
+
+TEST(ElasticsearchLogsExporterTests, ExportAfterShutdownFails)
+{
+  logs_exporter::ElasticsearchExporterOptions options;
+  auto http_client = std::make_shared<FakeHttpClient>();
+  auto exporter    = std::unique_ptr<sdklogs::LogRecordExporter>(
+      new logs_exporter::ElasticsearchLogRecordExporter(options, http_client));
+
+  ASSERT_TRUE(exporter->Shutdown(std::chrono::seconds(1)));
+
+  auto record = exporter->MakeRecordable();
+  auto result = exporter->Export(nostd::span<std::unique_ptr<sdklogs::Recordable>>(&record, 1));
+
+  EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure);
+}
+
 // Attempt to write a log to an invalid host/port, test that the Export() returns failure
 TEST(DISABLED_ElasticsearchLogsExporterTests, InvalidEndpoint)
 {