[fix](brokerload) fix be core dump casued by broker load (apache#15874)

924060929 · Jan 17, 2023 · 9755358 · 9755358
1 parent b6d9e73
commit 9755358
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 7 deletions.
diff --git a/be/src/io/fs/broker_file_reader.cpp b/be/src/io/fs/broker_file_reader.cpp
@@ -51,12 +51,14 @@ Status BrokerFileReader::close() {
 
         TBrokerOperationStatus response;
         try {
+            std::shared_ptr<BrokerServiceConnection> client;
+            RETURN_IF_ERROR(_fs->get_client(&client));
             try {
-                (*_client)->closeReader(response, request);
+                (*client)->closeReader(response, request);
             } catch (apache::thrift::transport::TTransportException& e) {
                 std::this_thread::sleep_for(std::chrono::seconds(1));
-                RETURN_IF_ERROR((*_client).reopen());
-                (*_client)->closeReader(response, request);
+                RETURN_IF_ERROR((*client).reopen());
+                (*client)->closeReader(response, request);
             }
         } catch (apache::thrift::TException& e) {
             std::stringstream ss;
@@ -93,16 +95,18 @@ Status BrokerFileReader::read_at(size_t offset, Slice result, const IOContext& /
     request.__set_length(bytes_req);
 
     TBrokerReadResponse response;
+    std::shared_ptr<BrokerServiceConnection> client;
+    RETURN_IF_ERROR(_fs->get_client(&client));
     try {
         VLOG_RPC << "send pread request to broker:" << _broker_addr << " position:" << offset
                  << ", read bytes length:" << bytes_req;
         try {
-            (*_client)->pread(response, request);
+            (*client)->pread(response, request);
         } catch (apache::thrift::transport::TTransportException& e) {
             std::this_thread::sleep_for(std::chrono::seconds(1));
-            RETURN_IF_ERROR((*_client).reopen());
+            RETURN_IF_ERROR((*client).reopen());
             LOG(INFO) << "retry reading from broker: " << _broker_addr << ". reason: " << e.what();
-            (*_client)->pread(response, request);
+            (*client)->pread(response, request);
         }
     } catch (apache::thrift::TException& e) {
         std::stringstream ss;

diff --git a/be/src/io/fs/broker_file_reader.h b/be/src/io/fs/broker_file_reader.h
@@ -59,7 +59,6 @@ class BrokerFileReader : public FileReader {
 
     std::shared_ptr<BrokerFileSystem> _fs;
     std::atomic<bool> _closed = false;
-    std::shared_ptr<BrokerServiceConnection> _client;
 };
 } // namespace io
 } // namespace doris