This is an automated email from the ASF dual-hosted git repository.
liaoxin01 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 9a259ed4748 [fix](load_stream) fix receiver-side stream leak race on
close_load (#65584)
9a259ed4748 is described below
commit 9a259ed4748bb5de640c52e783d8324a1d687e20
Author: lizhuoyu5 <[email protected]>
AuthorDate: Mon Aug 24 23:05:00 2026 +0800
[fix](load_stream) fix receiver-side stream leak race on close_load (#65584)
Loads that open incremental destinations at runtime, including `INSERT
OVERWRITE ... PARTITION(*)`, auto-partition creation, and auto-detect
overwrite, could hang forever.
On the receiver, `LoadStream` counted `CLOSE_LOAD` messages in `close()`
and registered stream IDs for deferred close in a later lock scope, with
`_report_result()` network I/O between them. Under concurrent
`CLOSE_LOAD`s, the thread that observed `all_closed=true` could drain
and clear `_closing_stream_ids` before another already-counted stream
registered its ID. The late stream was then never `StreamClose`d: its
`on_closed()` callback never fired, the sink remained in `close_wait`,
the fragment never reported completion, and the FE latch was never
released.
This PR introduces `mark_eos_sent_and_collect()`. Deferred registration,
the all-received check, and collection of streams to close now run under
the same lock. The function is called after the current stream's EOS
response write returns, while `brpc::StreamClose()` remains outside the
lock. A write-once `_all_close_load_received` latch ensures that a
counted-but-late thread observes that all `CLOSE_LOAD`s have arrived and
drains its own deferred stream instead of leaving it orphaned. The
existing close fence for loads whose sender opened incremental
destinations is preserved.
The PR adds a BE unit test and a docker regression suite. The inert
debug point `LoadStream.close_load.delay_incremental_register`
deterministically opens the count-before-register race window: the buggy
implementation leaks streams and hangs, while the fixed implementation
closes every stream and completes.
### What problem does this PR solve?
Issue Number: close #65582
Related PR: #56120
Problem Summary:
`INSERT OVERWRITE ... PARTITION(*)` and other loads that open
incremental destinations at runtime could hang until
`insert_load_default_timeout_second`. The receiver split `CLOSE_LOAD`
counting and deferred-close registration across two lock scopes. A
thread could therefore contribute to `_close_load_cnt`, pause before
registering its stream ID, and resume only after the last-counting
thread had already drained and cleared `_closing_stream_ids`. Because no
later event would drain the newly registered ID, that BRPC stream
remained open permanently and prevented the load from completing.
`_closing_stream_ids` was introduced in #56120 to preserve the close
fence required by dynamic destinations. This PR keeps that fence but
makes registration and the all-received decision atomic under `_lock`,
and performs the resulting `StreamClose()` calls outside the critical
section.
### Release note
Fix a permanent hang of `INSERT OVERWRITE ... PARTITION(*)` and other
loads that open incremental destinations at runtime, caused by a
receiver-side load-stream close race.
---------
Co-authored-by: lizhuoyu5 <[email protected]>
---
be/src/load/channel/load_stream.cpp | 71 ++++++++---
be/src/load/channel/load_stream.h | 19 ++-
be/test/runtime/load_stream_test.cpp | 63 +++++++++-
.../test_iot_overwrite_partition_star_hang.out | 6 +
.../test_iot_overwrite_partition_star_hang.groovy | 139 +++++++++++++++++++++
5 files changed, 280 insertions(+), 18 deletions(-)
diff --git a/be/src/load/channel/load_stream.cpp
b/be/src/load/channel/load_stream.cpp
index 6cae3c7852e..64c8beb0a2a 100644
--- a/be/src/load/channel/load_stream.cpp
+++ b/be/src/load/channel/load_stream.cpp
@@ -525,6 +525,39 @@ bool LoadStream::close(int64_t src_id, const
std::vector<PTabletID>& tablets_to_
return true;
}
+std::vector<int64_t> LoadStream::mark_eos_sent_and_collect(int64_t stream_id,
bool is_incremental) {
+ std::lock_guard<bthread::Mutex> lock_guard(_lock);
+ std::vector<int64_t> to_close;
+ // A non-incremental stream is closed as soon as its own CLOSE_LOAD (and
EOS)
+ // is handled -- this is the first batch of streams, known up front, not
subject
+ // to fencing. Closing it promptly also means a duplicate/late CLOSE_LOAD
lands
+ // on an already-closed stream and is dropped, instead of being counted
again.
+ // An incremental stream must be deferred (fencing #56120: it may only
close once
+ // every non-incremental stream is closed), so it is parked in
_eos_sent_stream_ids
+ // until all CLOSE_LOADs have been received.
+ if (is_incremental) {
+ // Parked only after the caller sent this stream's EOS via
_report_result,
+ // so every parked id is safe to close.
+ _eos_sent_stream_ids.push_back(stream_id);
+ } else {
+ to_close.push_back(stream_id);
+ }
+ // `_close_load_cnt == _total_streams` means every CLOSE_LOAD has been
counted by
+ // close(). Latch it so that any thread reaching here afterwards also
drains the
+ // parked incremental streams, guaranteeing none is left un-closed
regardless of
+ // thread interleaving (fixes the split-lock leak race).
+ if (_close_load_cnt >= _total_streams) {
+ _all_close_load_received = true;
+ }
+ if (_all_close_load_received) {
+ for (const auto& parked_id : _eos_sent_stream_ids) {
+ to_close.push_back(parked_id);
+ }
+ _eos_sent_stream_ids.clear();
+ }
+ return to_close;
+}
+
void LoadStream::_report_result(StreamId stream, const Status& status,
const std::vector<int64_t>& success_tablet_ids,
const FailedTablets& failed_tablets, bool eos)
{
@@ -762,24 +795,32 @@ void LoadStream::_dispatch(StreamId id, const
PStreamHeader& hdr, butil::IOBuf*
std::vector<int64_t> success_tablet_ids;
FailedTablets failed_tablets;
std::vector<PTabletID> tablets_to_commit(hdr.tablets().begin(),
hdr.tablets().end());
- bool all_closed =
+ // Step 1: count this CLOSE_LOAD and, if this is the last one, commit.
Under _lock.
+ bool all_received =
close(hdr.src_id(), tablets_to_commit, &success_tablet_ids,
&failed_tablets);
+ // Step 2: send THIS stream's EOS (network IO, must be outside _lock).
A stream
+ // must not be StreamClose'd before its own EOS is delivered,
otherwise the
+ // sender sees on_closed without EOS and reports "Stream closed
without EOS".
_report_result(id, Status::OK(), success_tablet_ids, failed_tablets,
true);
- std::lock_guard<bthread::Mutex> lock_guard(_lock);
- // if incremental stream, we need to wait for all non-incremental
streams to be closed
- // before closing incremental streams. We need a fencing mechanism to
avoid use after closing
- // across different be.
- if (hdr.has_num_incremental_streams() && hdr.num_incremental_streams()
> 0) {
- _closing_stream_ids.push_back(id);
- } else {
- brpc::StreamClose(id);
+ bool is_incremental =
+ hdr.has_num_incremental_streams() &&
hdr.num_incremental_streams() > 0;
+ // Test-only: delay every incremental stream except the one that made
+ // all_received, so a non-last incremental stream parks after the last
+ // stream drained the list. On the buggy code this orphans it and the
+ // load hangs; on the fix the latch drains the late registration under
+ // the same lock. Inert unless enable_debug_points=true.
+ if (is_incremental && !all_received) {
+ DBUG_EXECUTE_IF("LoadStream.close_load.delay_incremental_register",
+ { bthread_usleep(3000000); });
}
-
- if (all_closed) {
- for (auto& closing_id : _closing_stream_ids) {
- brpc::StreamClose(closing_id);
- }
- _closing_stream_ids.clear();
+ // Step 3: close this stream (non-incremental) or park it for deferred
close
+ // (incremental, fencing), then collect everything that is now safe to
close.
+ // Registration happens only after step 2, so a collected stream
already had
+ // its EOS delivered (fixes the close-before-EOS race); the
all-received latch
+ // inside makes any late thread drain the parked streams (fixes the
leak race).
+ auto streams_to_close = mark_eos_sent_and_collect(id, is_incremental);
+ for (auto& closing_id : streams_to_close) {
+ brpc::StreamClose(closing_id);
}
} break;
case PStreamHeader::GET_SCHEMA: {
diff --git a/be/src/load/channel/load_stream.h
b/be/src/load/channel/load_stream.h
index ff5c7d1d957..4c8865e1603 100644
--- a/be/src/load/channel/load_stream.h
+++ b/be/src/load/channel/load_stream.h
@@ -136,10 +136,19 @@ public:
}
}
- // return true if all streams are closed, otherwise return false
+ // Count this CLOSE_LOAD. Returns true once CLOSE_LOAD from all streams
has been
+ // received (i.e. the load is ready to commit). Only counts and commits;
stream
+ // closing is handled by the caller in _dispatch to keep EOS-before-close
ordering.
bool close(int64_t src_id, const std::vector<PTabletID>& tablets_to_commit,
std::vector<int64_t>* success_tablet_ids, FailedTablets*
failed_tablet_ids);
+ // Close/park `stream_id` after its EOS was sent, then return stream ids
that are
+ // now safe to close. A non-incremental stream is returned right away; an
incremental
+ // stream is parked (fencing) until all CLOSE_LOADs arrive, after which
any thread
+ // reaching here drains the parked streams. Only a stream whose EOS was
sent is ever
+ // returned -- this closes both the split-lock leak race and the
close-before-EOS race.
+ std::vector<int64_t> mark_eos_sent_and_collect(int64_t stream_id, bool
is_incremental);
+
// callbacks called by brpc
int on_received_messages(StreamId id, butil::IOBuf* const messages[],
size_t size) override;
void on_idle_timeout(StreamId id) override;
@@ -189,7 +198,13 @@ private:
RuntimeProfile::Counter* _close_wait_timer = nullptr;
LoadStreamMgr* _load_stream_mgr = nullptr;
std::shared_ptr<ResourceContext> _resource_ctx;
- std::vector<int64_t> _closing_stream_ids;
+ // Streams whose EOS has been sent and are waiting to be closed. A stream
is added
+ // here (under _lock) only after its _report_result() finished, and
drained once
+ // _all_close_load_received becomes true. Replaces the old
_closing_stream_ids whose
+ // registration happened in a separate lock scope from the all-received
check,
+ // causing a race that leaked (never-closed) streams.
+ std::vector<int64_t> _eos_sent_stream_ids;
+ bool _all_close_load_received = false;
bool _is_incremental = false;
};
diff --git a/be/test/runtime/load_stream_test.cpp
b/be/test/runtime/load_stream_test.cpp
index baf1f10e967..b17f302d586 100644
--- a/be/test/runtime/load_stream_test.cpp
+++ b/be/test/runtime/load_stream_test.cpp
@@ -496,13 +496,16 @@ public:
_light_work_pool(4, 32, "load_stream_test_light") {}
void close_load(MockSinkClient& client, const std::vector<PTabletID>&
tablets_to_commit = {},
- uint32_t sender_id = NORMAL_SENDER_ID) {
+ uint32_t sender_id = NORMAL_SENDER_ID, int
num_incremental_streams = 0) {
butil::IOBuf append_buf;
PStreamHeader header;
header.mutable_load_id()->set_hi(1);
header.mutable_load_id()->set_lo(1);
header.set_opcode(PStreamHeader::CLOSE_LOAD);
header.set_src_id(sender_id);
+ if (num_incremental_streams > 0) {
+ header.set_num_incremental_streams(num_incremental_streams);
+ }
for (const auto& tablet : tablets_to_commit) {
*header.add_tablets() = tablet;
}
@@ -1357,4 +1360,62 @@ TEST_F(LoadStreamMgrTest,
two_client_one_close_before_the_other_open) {
}
}
+// Verify the latch drain prevents stream leaks: 3 distinct senders each
+// send one incremental CLOSE_LOAD. The debug point delays the first 2
+// (non-last) streams while the 3rd sets the _all_close_load_received
+// latch and drains _eos_sent_stream_ids. On the FIXED code the delayed
+// streams also drain themselves when they resume, so all 3 streams are
+// properly closed and LoadStream is cleaned up.
+//
+// Three distinct MockSinkClient connections are required because a single
+// connection shares one source entry; after the first CLOSE_LOAD consumes
+// that entry the remaining messages are rejected as "no open stream".
+TEST_F(LoadStreamMgrTest, incremental_close_race_orhpans_streams) {
+ constexpr int kStreams = 3;
+ MockSinkClient clients[kStreams];
+
+ // Open three independent senders; each advertises kStreams so that
+ // _total_streams == kStreams and add_source tracks every sender.
+ for (int i = 0; i < kStreams; i++) {
+ auto st = clients[i].connect_stream(NORMAL_SENDER_ID + i, kStreams);
+ ASSERT_TRUE(st.ok());
+ }
+ reset_response_stat();
+
+ PTabletID tablet;
+ tablet.set_partition_id(NORMAL_PARTITION_ID);
+ tablet.set_index_id(NORMAL_INDEX_ID);
+ tablet.set_tablet_id(NORMAL_TABLET_ID);
+
+ // Enable the debug point: non-last incremental streams sleep 3s.
+ auto debug_point_name =
std::string("LoadStream.close_load.delay_incremental_register");
+ bool saved_debug_points = config::enable_debug_points;
+ config::enable_debug_points = true;
+ auto* debug_points = DebugPoints::instance();
+ debug_points->add(debug_point_name);
+ ASSERT_TRUE(debug_points->is_enable(debug_point_name)) << "debug point was
not registered!";
+
+ // Send one incremental CLOSE_LOAD from each sender. The last one
+ // makes _close_load_cnt == _total_streams (all_received).
+ for (int i = 0; i < kStreams; i++) {
+ close_load(clients[i], {tablet}, NORMAL_SENDER_ID + i, 1);
+ }
+
+ // At least one EOS response confirms the load ran.
+ wait_for_ack(1);
+ EXPECT_GE(g_response_stat.num, 1);
+
+ // Let the delayed streams wake up (3s sleep + small margin).
+ bthread_usleep(4 * 1000 * 1000);
+
+ // On the fixed code the latch drains all delayed streams, so no
+ // orphan remains and LoadStream is properly destroyed.
+ auto remaining = _load_stream_mgr->get_load_stream_num();
+ EXPECT_EQ(remaining, 0) << "stream leak detected! LoadStream should be
cleaned up. "
+ << "remaining=" << remaining;
+
+ debug_points->remove(debug_point_name);
+ config::enable_debug_points = saved_debug_points;
+}
+
} // namespace doris
diff --git
a/regression-test/data/insert_overwrite_p0/test_iot_overwrite_partition_star_hang.out
b/regression-test/data/insert_overwrite_p0/test_iot_overwrite_partition_star_hang.out
new file mode 100644
index 00000000000..abace22a8d0
--- /dev/null
+++
b/regression-test/data/insert_overwrite_p0/test_iot_overwrite_partition_star_hang.out
@@ -0,0 +1,6 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !sql --
+50
+
+-- !sql --
+50 0 49
diff --git
a/regression-test/suites/insert_overwrite_p0/test_iot_overwrite_partition_star_hang.groovy
b/regression-test/suites/insert_overwrite_p0/test_iot_overwrite_partition_star_hang.groovy
new file mode 100644
index 00000000000..0f4f9e2b252
--- /dev/null
+++
b/regression-test/suites/insert_overwrite_p0/test_iot_overwrite_partition_star_hang.groovy
@@ -0,0 +1,139 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.apache.doris.regression.suite.ClusterOptions
+
+// Reproduces the INSERT OVERWRITE ... PARTITION(*) hang on a multi-BE cluster.
+//
+// The race is WITHIN A SINGLE LOAD, on the receiver's per-LoadStream close
path.
+// A PARTITION(*) auto-detect overwrite that creates NEW partitions mid-load
opens
+// both non-incremental (base) and incremental (new partition) streams. On
+// CLOSE_LOAD the buggy _dispatch does, per stream:
+// close(); _report_result(); // count + send this stream's EOS
+// lock; if incremental: _closing_stream_ids.push_back(id);
+// else: StreamClose(id);
+// if (all_closed) { StreamClose all in _closing_stream_ids;
clear(); }
+// The push_back of an incremental stream happens in a SEPARATE lock scope from
+// close()'s counting. If the last (non-incremental) stream reaches all_closed
and
+// drains+clears the list before a counted-but-delayed incremental stream does
its
+// push_back, that incremental stream is never StreamClose'd -> the load's brpc
+// streams never finish -> the load hangs forever.
+//
+// In a fast single-host docker cluster that window is microseconds, so it will
+// not open organically. The debug point
LoadStream.close_load.delay_incremental_register
+// (BE-only, test build) sleeps an incremental stream between close() and
push_back,
+// deterministically opening the window. On the FIXED binary registration and
the
+// all-received check are under one lock, so no stream is orphaned and the load
+// completes even with the delay.
+//
+// Expectation: buggy binary -> load hangs -> this suite throws (RED).
+// fixed binary -> load completes -> suite passes (GREEN).
+suite("test_iot_overwrite_partition_star_hang", "docker") {
+ def options = new ClusterOptions()
+ options.feNum = 1
+ options.beNum = 3
+ options.cloudMode = false
+ options.beConfigs += [
+ 'enable_debug_points=true'
+ ]
+
+ docker(options) {
+ sql "set enable_auto_create_when_overwrite = true;"
+ sql " drop table if exists iot_star_hang; "
+ sql """
+ create table iot_star_hang(
+ k0 int null
+ )
+ auto partition by list (k0)
+ (
+ PARTITION p1 values in ((0))
+ )
+ DISTRIBUTED BY HASH(`k0`) BUCKETS 1
+ properties("replication_num" = "1");
+ """
+ // Seed ONLY the base partition. With replication_num=1 + BUCKETS 1 it
lands on
+ // a single BE, so the base load opens a non-incremental stream to
just one
+ // backend. New partitions created mid-overwrite land on the OTHER BEs
as fresh
+ // backends -> those are incremental streams (num_incremental_streams
> 0), which
+ // is exactly the condition the receiver's deferred-close race needs.
+ sql """ insert into iot_star_hang values (0); """
+
+ def deadline = 60000
+
+ try {
+ // Delay incremental-stream registration on all BEs to force the
orphan.
+
GetDebugPoint().enableDebugPointForAllBEs("LoadStream.close_load.delay_incremental_register")
+ log.info("debug point enabled: delay incremental register")
+
+ // Expose the JDBC statement so the watchdog can cancel the blocked
+ // socket I/O before tearing the thread down (Thread.interrupt does
+ // not reliably unblock Connector/J). Also hold the connection
+ // reference so it can be closed if a cancel arrives before the
+ // worker finishes.
+ def stmtRef = null
+ def connRef = null
+ def loadEx = null
+
+ def t = Thread.start {
+ def conn = null
+ try {
+ conn = context.getConnection()
+ connRef = conn
+ def stmt = conn.createStatement()
+ stmtRef = stmt
+ stmt.execute("set enable_auto_create_when_overwrite =
true;")
+ // key 0 -> existing base partition (non-incremental
stream);
+ // keys 1..49 -> new partitions (incremental streams). One
load, both kinds.
+ stmt.execute("""
+ insert overwrite table iot_star_hang partition(*)
+ select number from numbers("number" = "50");
+ """)
+ } catch (Throwable ex) {
+ loadEx = ex
+ } finally {
+ if (conn != null) conn.close()
+ }
+ }
+
+ t.join(deadline)
+ if (t.isAlive()) {
+ try {
+ if (stmtRef != null) stmtRef.cancel()
+ if (connRef != null) connRef.close()
+ } catch (Throwable ignore) {}
+ t.join(5000)
+ if (t.isAlive()) {
+ t.interrupt()
+ }
+ throw new Exception("INSERT OVERWRITE PARTITION(*) hung: load
did not finish within ${deadline}ms (orphaned incremental stream never closed)")
+ }
+ if (loadEx != null) {
+ throw new Exception("INSERT OVERWRITE PARTITION(*) failed:
${loadEx.message}")
+ }
+ } finally {
+ try {
+
GetDebugPoint().disableDebugPointForAllBEs("LoadStream.close_load.delay_incremental_register")
+ } catch (Throwable ignore) {}
+ }
+
+ // Assert the load both completed and published the correct
+ // deterministic result (keys 0..49) so the suite proves bounded
+ // completion AND correct commit visibility, not just non-hang.
+ qt_sql """select count(*) from iot_star_hang;"""
+ qt_sql """select count(distinct k0), min(k0), max(k0) from
iot_star_hang;"""
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
