[fix](move-memtable) close stream when cancel load stream stub (#38912) (#39039)

kaijchen · web-flow · commit 0a3874f203f2 · 2024-08-07T23:24:00.000+08:00
backport #38912
diff --git a/be/src/vec/sink/load_stream_stub.cpp b/be/src/vec/sink/load_stream_stub.cpp
@@ -338,6 +338,9 @@ Status LoadStreamStub::close_wait(RuntimeState* state, int64_t timeout_ms) {
 
 void LoadStreamStub::cancel(Status reason) {
     LOG(WARNING) << *this << " is cancelled because of " << reason;
+    if (_is_init.load()) {
+        brpc::StreamClose(_stream_id);
+    }
     {
         std::lock_guard<bthread::Mutex> lock(_cancel_mutex);
         _cancel_reason = reason;

Original file line number	Diff line number	Diff line change
`@@ -338,6 +338,9 @@ Status LoadStreamStub::close_wait(RuntimeState* state, int64_t timeout_ms) {`
`338`	`338`
`339`	`339`	`void LoadStreamStub::cancel(Status reason) {`
`340`	`340`	`LOG(WARNING) << *this << " is cancelled because of " << reason;`
	`341`	`+ if (_is_init.load()) {`
	`342`	`+ brpc::StreamClose(_stream_id);`
	`343`	`+ }`
`341`	`344`	`{`
`342`	`345`	`std::lock_guard<bthread::Mutex> lock(_cancel_mutex);`
`343`	`346`	`_cancel_reason = reason;`