[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.

Author: morningman

be/src/load/stream_load/stream_load_recorder_manager.cpp

@@ -80,21 +80,38 @@ 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";
     }
 }
 
 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");

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 @@ class StreamLoadRecorderManager {
 
     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 @@ class StreamLoadRecorderManager {
 
     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;
 

be/src/runtime/exec_env.h

@@ -301,6 +301,9 @@ class ExecEnv {
 
     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(); }

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();

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);

be/src/service/http/http_client.h

@@ -117,6 +117,17 @@ class HttpClient {
         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 @@ class HttpClient {
     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;

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