This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new b119273e3f0 [fix](load) Keep graceful BE stop bounded when an audit
stream load is in flight (#66797)
b119273e3f0 is described below
commit b119273e3f06b3425a09908fc0ac65742e6a1b96
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Sun Aug 16 14:35:42 2026 +0800
[fix](load) Keep graceful BE stop bounded when an audit stream load is in
flight (#66797)
### What problem does this PR solve?
Issue Number: close #66796
Problem Summary:
A graceful BE shutdown could hang for up to 10 minutes and fail the
pipeline's
"stop grace" check.
`doris_main()` stops the servers first (`Http service stopped` -> `Brpc
service
stopped` -> `Backend Service stopped`) and only then calls
`ExecEnv::destroy()`,
which reaches `SAFE_STOP(_stream_load_recorder_manager)` and joins the
recorder's
worker `std::thread`. If that worker is inside `_send_stream_load()` at
the time,
its request is a stream load against this BE's own http endpoint, which
has
already been torn down, so the load can never complete and the request
only
returns once the `DEFAULT_STREAM_LOAD_TIMEOUT_SEC = 600` curl timeout
expires.
The join blocks for that long. Deployment scripts give graceful stop a
10 minute
budget, so whenever the race fires the stop check is guaranteed to lose
it and
the BE gets `kill -ABRT`ed.
The window is narrow (an audit batch has to be in flight within about
one worker
iteration of the service teardown), which is why it fired only once in
the last
30 NonConcurrentRegression runs, but any BE with a pending audit-log
batch at
stop time can hang this way.
The PR applies three changes, each of which bounds the wait on its own:
1. **Ordering** — `doris_main()` stops the recorder manager *before*
tearing down
the http service, so on the normal path the last audit batch is flushed
against a live server and the worker is already gone by the time
`ExecEnv::destroy()` runs. `stop()` is idempotent, so the existing
`SAFE_STOP()` in `destroy()` becomes a no-op.
2. **Interruptible send** — new `HttpClient::set_abort_callback()`
registers a
functor that libcurl polls through `CURLOPT_XFERINFOFUNCTION`, roughly
once
per second while the connection is idle. Returning true aborts the
transfer
with `CURLE_ABORTED_BY_CALLBACK` instead of running to
`CURLOPT_TIMEOUT_MS`.
`StreamLoadRecorderManager` hooks its `_stop` flag up to it, which
covers both
a request that is already in flight when `stop()` is called and one that
races its way past the check in 3.
3. **No new work after stop** — the worker no longer starts an audit
load once
shutdown has begun, and waits on a condition variable instead of an
unconditional 1s sleep, so `stop()` wakes it immediately rather than
after up
to a second.
`stop()` now also logs on completion. The original hang was hard to
locate in
the log precisely because a raw `std::thread::join()` is silent, unlike
doris
`Thread::join`, which prints `Waited for ...ms trying to join`.
The 600s curl timeout itself is left alone: it is a sane bound for a
load that
is not racing shutdown, and the abort hook makes it irrelevant for
shutdown.
### Release note
Fix a graceful BE shutdown that could hang for up to 10 minutes when an
audit
log stream load was in flight while the BE's http service was being
stopped.
### Check List (For Author)
- Test
- [x] Unit Test
`HttpClientTest.abort_in_flight_request` points an `HttpClient` at a
listening
socket that is never accepted from — the kernel completes the handshake
and
buffers the request, but no response ever comes, which is exactly what
an audit
stream load looks like once the http service serving it is gone. With
`CURLOPT_TIMEOUT_MS` at 60s, the abort callback is raised 300ms in and
the
request has to end well before the timeout.
Verified separately against the libcurl the BE links (8.2.1) that the
progress
callback is polled while idle-waiting: the request ends in ~1.2s with
`CURLE_ABORTED_BY_CALLBACK` (callback invoked 13 times) rather than at
60s.
The shutdown ordering itself is not unit-testable — it needs a full BE
stop
with an audit batch in flight, which is the race described in the issue.
---
.../stream_load/stream_load_recorder_manager.cpp | 25 ++++++++-
.../stream_load/stream_load_recorder_manager.h | 14 +++++
be/src/runtime/exec_env.h | 3 ++
be/src/service/doris_main.cpp | 10 ++++
be/src/service/http/http_client.cpp | 22 ++++++++
be/src/service/http/http_client.h | 12 +++++
be/test/service/http/http_client_test.cpp | 59 ++++++++++++++++++++++
7 files changed, 143 insertions(+), 2 deletions(-)
diff --git a/be/src/load/stream_load/stream_load_recorder_manager.cpp
b/be/src/load/stream_load/stream_load_recorder_manager.cpp
index a8fe851efa6..76913bf8f65 100644
--- a/be/src/load/stream_load/stream_load_recorder_manager.cpp
+++ b/be/src/load/stream_load/stream_load_recorder_manager.cpp
@@ -80,9 +80,16 @@ void StreamLoadRecorderManager::_load_last_fetch_key() {
}
void StreamLoadRecorderManager::stop() {
- _stop = true;
+ {
+ std::lock_guard<std::mutex> lock(_stop_mutex);
+ _stop = true;
+ }
+ // Wakes the worker from its idle wait, and aborts the audit stream load
it may
+ // currently be blocked on, see the abort callback in _send_stream_load().
+ _stop_cv.notify_all();
if (_worker_thread.joinable()) {
_worker_thread.join();
+ LOG(INFO) << "StreamLoadRecorderManager is stopped";
}
}
@@ -90,11 +97,21 @@ void StreamLoadRecorderManager::_worker_thread_func() {
SCOPED_ATTACH_TASK(_mem_tracker);
while (!_stop) {
_fetch_and_buffer_records();
+ // Do not start a new audit stream load once shutdown has begun. The
load is served
+ // by this BE's own http service, which is about to go away.
+ if (_stop) {
+ break;
+ }
_load_if_necessary();
- std::this_thread::sleep_for(std::chrono::seconds(1));
+ _wait_for_stop(1000);
}
}
+void StreamLoadRecorderManager::_wait_for_stop(int64_t wait_ms) {
+ std::unique_lock<std::mutex> lock(_stop_mutex);
+ _stop_cv.wait_for(lock, std::chrono::milliseconds(wait_ms), [this]() {
return _stop.load(); });
+}
+
void StreamLoadRecorderManager::_fetch_and_buffer_records() {
if (!_recorder) {
LOG(WARNING) << "StreamLoadRecorder is not initialized";
@@ -262,6 +279,10 @@ Status StreamLoadRecorderManager::_send_stream_load(const
std::string& data) {
if (!st.ok()) {
return Status::InternalError("Failed to init http client: {}",
st.to_string());
}
+ // This load is served by this BE's own http service. Once shutdown starts
that service
+ // stops answering, and without an abort hook the request would sit here
for the full
+ // DEFAULT_STREAM_LOAD_TIMEOUT_SEC, blocking the join() in stop().
+ client.set_abort_callback([this]() { return _stop.load(); });
client.set_authorization("Basic YWRtaW46");
client.set_header("Expect", "100-continue");
client.set_content_type("text/plain; charset=UTF-8");
diff --git a/be/src/load/stream_load/stream_load_recorder_manager.h
b/be/src/load/stream_load/stream_load_recorder_manager.h
index b7908b14b86..acfe76dde1a 100644
--- a/be/src/load/stream_load/stream_load_recorder_manager.h
+++ b/be/src/load/stream_load/stream_load_recorder_manager.h
@@ -18,8 +18,10 @@
#pragma once
#include <atomic>
+#include <condition_variable>
#include <cstdint>
#include <memory>
+#include <mutex>
#include <string>
#include <thread>
@@ -57,6 +59,13 @@ public:
void start();
+ // Stops the worker thread and waits for it to exit. Aborts the audit
stream load that
+ // the worker may currently be running, so that this returns within about
a second even
+ // if the request would never be answered.
+ //
+ // The manager sends its records to this BE's own http service, so it must
be stopped
+ // before that service is torn down. doris_main() does that explicitly;
the SAFE_STOP()
+ // in ExecEnv::destroy() is then a no-op. Calling this more than once is
safe.
void stop();
private:
@@ -80,12 +89,17 @@ private:
void _reset_batch(int64_t current_time);
+ // Waits at most `wait_ms` for stop() to be called. The caller re-checks
_stop itself.
+ void _wait_for_stop(int64_t wait_ms);
+
private:
std::shared_ptr<StreamLoadRecorder> _recorder;
std::shared_ptr<MemTrackerLimiter> _mem_tracker;
std::thread _worker_thread;
std::atomic<bool> _stop;
+ std::mutex _stop_mutex;
+ std::condition_variable _stop_cv;
faststring _buffer;
diff --git a/be/src/runtime/exec_env.h b/be/src/runtime/exec_env.h
index 885828b8eff..8d30f60258a 100644
--- a/be/src/runtime/exec_env.h
+++ b/be/src/runtime/exec_env.h
@@ -301,6 +301,9 @@ public:
StreamLoadExecutor* stream_load_executor() { return
_stream_load_executor.get(); }
RoutineLoadTaskExecutor* routine_load_task_executor() { return
_routine_load_task_executor; }
+ StreamLoadRecorderManager* stream_load_recorder_manager() {
+ return _stream_load_recorder_manager;
+ }
HeartbeatFlags* heartbeat_flags() { return _heartbeat_flags; }
FileMetaCache* file_meta_cache() { return _file_meta_cache; }
MemTableMemoryLimiter* memtable_memory_limiter() { return
_memtable_memory_limiter.get(); }
diff --git a/be/src/service/doris_main.cpp b/be/src/service/doris_main.cpp
index 12f5af0e2a4..dc55d6ce142 100644
--- a/be/src/service/doris_main.cpp
+++ b/be/src/service/doris_main.cpp
@@ -73,6 +73,7 @@
#include "common/signal_handler.h"
#include "common/status.h"
#include "io/cache/block_file_cache_factory.h"
+#include "load/stream_load/stream_load_recorder_manager.h"
#include "runtime/exec_env.h"
#include "runtime/user_function_cache.h"
#include "service/arrow_flight/flight_sql_service.h"
@@ -731,6 +732,15 @@ int main(int argc, char** argv) {
heartbeat_thrift_starter->stop();
heartbeat_thrift_starter->join();
LOG(INFO) << "Heartbeat server stopped";
+ // The stream load recorder manager writes its audit records through this
BE's own http
+ // service, so it has to be stopped while that service is still up.
Otherwise an audit
+ // load that is in flight here can never be answered, and it blocks the
join() done by
+ // SAFE_STOP(_stream_load_recorder_manager) in ExecEnv::destroy() for up
to the stream
+ // load timeout, which is longer than the grace period of stop_be.sh
--grace.
+ if (auto* recorder_manager = exec_env->stream_load_recorder_manager();
+ recorder_manager != nullptr) {
+ recorder_manager->stop();
+ }
// TODO(zhiqiang): http_service
http_starter->stop();
http_starter->join();
diff --git a/be/src/service/http/http_client.cpp
b/be/src/service/http/http_client.cpp
index 7141e6f6012..7a82549f78e 100644
--- a/be/src/service/http/http_client.cpp
+++ b/be/src/service/http/http_client.cpp
@@ -295,6 +295,8 @@ Status HttpClient::init(const std::string& url, bool
set_fail_on_error) {
curl_slist_free_all(_header_list);
_header_list = nullptr;
}
+ // curl_easy_reset() dropped the progress callback options, so drop the
functor too.
+ _abort_callback = nullptr;
// set error_buf
_error_buf[0] = 0;
auto code = curl_easy_setopt(_curl, CURLOPT_ERRORBUFFER, _error_buf);
@@ -388,6 +390,26 @@ void HttpClient::set_method(HttpMethod method) {
}
}
+void HttpClient::set_abort_callback(std::function<bool()> callback) {
+ _abort_callback = std::move(callback);
+ if (!_abort_callback) {
+ curl_easy_setopt(_curl, CURLOPT_NOPROGRESS, 1L);
+ return;
+ }
+
+ curl_xferinfo_callback xferinfo = [](void* param, curl_off_t /*dltotal*/,
curl_off_t /*dlnow*/,
+ curl_off_t /*ultotal*/, curl_off_t
/*ulnow*/) -> int {
+ auto* client = (HttpClient*)param;
+ // A non-zero return value makes libcurl abort the transfer with
+ // CURLE_ABORTED_BY_CALLBACK.
+ return client->_abort_callback() ? 1 : 0;
+ };
+ curl_easy_setopt(_curl, CURLOPT_XFERINFOFUNCTION, xferinfo);
+ curl_easy_setopt(_curl, CURLOPT_XFERINFODATA, (void*)this);
+ // libcurl only calls the progress callback when the progress meter is
enabled.
+ curl_easy_setopt(_curl, CURLOPT_NOPROGRESS, 0L);
+}
+
void HttpClient::set_speed_limit() {
curl_easy_setopt(_curl, CURLOPT_LOW_SPEED_LIMIT,
config::download_low_speed_limit_kbps * 1024);
curl_easy_setopt(_curl, CURLOPT_LOW_SPEED_TIME,
config::download_low_speed_time);
diff --git a/be/src/service/http/http_client.h
b/be/src/service/http/http_client.h
index 8c57d8a2e6a..cdeaf9a4d34 100644
--- a/be/src/service/http/http_client.h
+++ b/be/src/service/http/http_client.h
@@ -117,6 +117,17 @@ public:
curl_easy_setopt(_curl, CURLOPT_TIMEOUT_MS, timeout_ms);
}
+ // Register a callback that libcurl polls while the request is in flight:
often while
+ // data is flowing, and about once per second when the connection is idle.
Returning
+ // true aborts the transfer right away, so `execute()` fails instead of
blocking until
+ // CURLOPT_TIMEOUT_MS expires.
+ //
+ // This is for callers that must be able to give up on a request which may
never be
+ // answered, e.g. a background worker that is being stopped while the http
service
+ // serving its request is going away. Must be called after init(), which
resets all
+ // curl options. Passing an empty callback clears a previously registered
one.
+ void set_abort_callback(std::function<bool()> callback);
+
// used to get content length
// return -1 as error
Status get_content_length(uint64_t* length) const {
@@ -198,6 +209,7 @@ private:
CURL* _curl = nullptr;
using HttpCallback = std::function<bool(const void* data, size_t length)>;
const HttpCallback* _callback = nullptr;
+ std::function<bool()> _abort_callback;
char _error_buf[CURL_ERROR_SIZE];
curl_slist* _header_list = nullptr;
HttpMethod _method = GET;
diff --git a/be/test/service/http/http_client_test.cpp
b/be/test/service/http/http_client_test.cpp
index 7b6cb2fd9f7..06d23a64185 100644
--- a/be/test/service/http/http_client_test.cpp
+++ b/be/test/service/http/http_client_test.cpp
@@ -21,12 +21,18 @@
#include <fcntl.h>
#include <gtest/gtest-message.h>
#include <gtest/gtest-test-part.h>
+#include <netinet/in.h>
#include <sys/mman.h>
+#include <sys/socket.h>
#include <sys/stat.h>
#include <unistd.h>
+#include <atomic>
#include <boost/algorithm/string/predicate.hpp>
+#include <chrono>
+#include <cstring>
#include <filesystem>
+#include <thread>
#include "gtest/gtest_pred_impl.h"
#include "io/fs/local_file_system.h"
@@ -669,4 +675,57 @@ TEST_F(HttpClientTest, batch_download) {
EXPECT_TRUE(st.ok());
}
+TEST_F(HttpClientTest, abort_in_flight_request) {
+ // A listening socket that is never accepted from. The kernel completes
the handshake
+ // and buffers the request, so the client believes it is connected, but no
response ever
+ // comes back. This is what an audit stream load looks like when the http
service that
+ // was supposed to serve it has been torn down.
+ int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
+ ASSERT_GE(listen_fd, 0);
+ struct sockaddr_in addr;
+ memset(&addr, 0, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+ addr.sin_port = 0; // let the kernel pick a free port
+ ASSERT_EQ(0, bind(listen_fd, reinterpret_cast<struct sockaddr*>(&addr),
sizeof(addr)));
+ ASSERT_EQ(0, listen(listen_fd, 8));
+ socklen_t addr_len = sizeof(addr);
+ ASSERT_EQ(0, getsockname(listen_fd, reinterpret_cast<struct
sockaddr*>(&addr), &addr_len));
+ std::string url = "http://127.0.0.1:"; +
std::to_string(ntohs(addr.sin_port)) + "/no_answer";
+
+ HttpClient client;
+ auto st = client.init(url);
+ EXPECT_TRUE(st.ok()) << st;
+ client.set_method(GET);
+ // Much longer than the abort is expected to take, so that finishing early
can only be
+ // the abort callback and not the timeout.
+ client.set_timeout_ms(60 * 1000);
+ std::atomic<bool> should_abort {false};
+ client.set_abort_callback([&should_abort]() { return should_abort.load();
});
+
+ // Ask for the abort only once the request is on the wire, like a shutdown
starting
+ // while a worker is already blocked inside execute().
+ std::thread aborter([&should_abort]() {
+ std::this_thread::sleep_for(std::chrono::milliseconds(300));
+ should_abort = true;
+ });
+
+ auto start = std::chrono::steady_clock::now();
+ std::string response;
+ st = client.execute(&response);
+ auto elapsed_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+ std::chrono::steady_clock::now() - start)
+ .count();
+ aborter.join();
+ close(listen_fd);
+
+ EXPECT_FALSE(st.ok());
+ // The request really was stuck waiting rather than failing outright, ...
+ EXPECT_GE(elapsed_ms, 300) << "request did not reach the server, it took "
<< elapsed_ms
+ << "ms";
+ // ... and the abort ended it instead of CURLOPT_TIMEOUT_MS. libcurl polls
the callback
+ // about once a second while the connection is idle.
+ EXPECT_LT(elapsed_ms, 15000) << "request was not aborted, it took " <<
elapsed_ms << "ms";
+}
+
} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
