This is an automated email from the ASF dual-hosted git repository.
dataroaring 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 09939fd69d5 [improve](binlog) Download binlogs with persistent
connection (#48467)
09939fd69d5 is described below
commit 09939fd69d50f525d6f29158b41b7cb9c456ac00
Author: walter <[email protected]>
AuthorDate: Mon Mar 3 21:58:23 2025 +0800
[improve](binlog) Download binlogs with persistent connection (#48467)
---
be/src/http/http_client.cpp | 29 ++++++++++++++++++++++++++---
be/src/http/http_client.h | 4 ++++
be/src/service/backend_service.cpp | 19 ++++++++++---------
3 files changed, 40 insertions(+), 12 deletions(-)
diff --git a/be/src/http/http_client.cpp b/be/src/http/http_client.cpp
index 767377cea3f..d41193c912a 100644
--- a/be/src/http/http_client.cpp
+++ b/be/src/http/http_client.cpp
@@ -188,7 +188,7 @@ private:
_written_size += write_size;
if (_written_size == _file_size) {
// This file has been downloaded, switch to the next one.
- switchToNextFile();
+ switch_to_next_file();
}
return write_size;
@@ -196,7 +196,7 @@ private:
Status finish_inner() {
if (!_is_reading_header && _written_size == _file_size) {
- switchToNextFile();
+ switch_to_next_file();
}
if (_fd >= 0) {
@@ -219,7 +219,7 @@ private:
return Status::OK();
}
- void switchToNextFile() {
+ void switch_to_next_file() {
DCHECK(_fd >= 0);
DCHECK(_written_size == _file_size);
@@ -513,6 +513,29 @@ const char* HttpClient::_get_url() const {
return url;
}
+// execute remote call action with retry
+Status HttpClient::execute(int retry_times, int sleep_time,
+ const std::function<Status(HttpClient*)>& callback)
{
+ Status status;
+ for (int i = 0; i < retry_times; ++i) {
+ status = callback(this);
+ if (status.ok()) {
+ auto http_status = get_http_status();
+ if (http_status == 200) {
+ return status;
+ } else {
+ std::string url = mask_token(_get_url());
+ auto error_msg = fmt::format("http status code is not 200,
code={}, url={}",
+ http_status, url);
+ LOG(WARNING) << error_msg;
+ return Status::HttpError(error_msg);
+ }
+ }
+ sleep(sleep_time);
+ }
+ return status;
+}
+
Status HttpClient::execute_with_retry(int retry_times, int sleep_time,
const
std::function<Status(HttpClient*)>& callback) {
Status status;
diff --git a/be/src/http/http_client.h b/be/src/http/http_client.h
index a6f2f4fdff5..be7674fb90a 100644
--- a/be/src/http/http_client.h
+++ b/be/src/http/http_client.h
@@ -157,6 +157,10 @@ public:
// execute remote call action
Status execute(const std::function<bool(const void* data, size_t length)>&
callback = {});
+ // execute remote call action with retry, like execute_with_retry but keep
the http client instance
+ Status execute(int retry_times, int sleep_time,
+ const std::function<Status(HttpClient*)>& callback);
+
size_t on_response_data(const void* data, size_t length);
// The file name of the variant column with the inverted index contains %
diff --git a/be/src/service/backend_service.cpp
b/be/src/service/backend_service.cpp
index 815e651b875..07aa1afc43c 100644
--- a/be/src/service/backend_service.cpp
+++ b/be/src/service/backend_service.cpp
@@ -101,6 +101,9 @@ struct IngestBinlogArg {
};
void _ingest_binlog(StorageEngine& engine, IngestBinlogArg* arg) {
+ // Save the http client instance for persistent connection
+ thread_local HttpClient client;
+
auto txn_id = arg->txn_id;
auto partition_id = arg->partition_id;
auto local_tablet_id = arg->local_tablet_id;
@@ -178,7 +181,7 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg*
arg) {
client->set_timeout_ms(config::download_binlog_meta_timeout_ms);
return client->execute(&binlog_info);
};
- auto status = HttpClient::execute_with_retry(max_retry, 1,
get_binlog_info_cb);
+ auto status = client.execute(max_retry, 1, get_binlog_info_cb);
if (!status.ok()) {
LOG(WARNING) << "failed to get binlog info from " <<
get_binlog_info_url
<< ", status=" << status.to_string();
@@ -217,7 +220,7 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg*
arg) {
client->set_timeout_ms(config::download_binlog_meta_timeout_ms);
return client->execute(&rowset_meta_str);
};
- status = HttpClient::execute_with_retry(max_retry, 1, get_rowset_meta_cb);
+ status = client.execute(max_retry, 1, get_rowset_meta_cb);
if (!status.ok()) {
LOG(WARNING) << "failed to get rowset meta from " <<
get_rowset_meta_url
<< ", status=" << status.to_string();
@@ -268,7 +271,7 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg*
arg) {
return client->get_content_length(&segment_file_size);
};
- status = HttpClient::execute_with_retry(max_retry, 1,
get_segment_file_size_cb);
+ status = client.execute(max_retry, 1, get_segment_file_size_cb);
if (!status.ok()) {
LOG(WARNING) << "failed to get segment file size from " <<
get_segment_file_size_url
<< ", status=" << status.to_string();
@@ -357,7 +360,7 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg*
arg) {
io::LocalFileSystem::PERMS_OWNER_RW);
};
- auto status = HttpClient::execute_with_retry(max_retry, 1,
get_segment_file_cb);
+ auto status = client.execute(max_retry, 1, get_segment_file_cb);
if (!status.ok()) {
LOG(WARNING) << "failed to get segment file from " <<
get_segment_file_url
<< ", status=" << status.to_string();
@@ -398,8 +401,7 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg*
arg) {
InvertedIndexDescriptor::get_index_file_path_prefix(segment_path), index_id,
index->get_index_suffix()));
- status = HttpClient::execute_with_retry(max_retry, 1,
-
get_segment_index_file_size_cb);
+ status = client.execute(max_retry, 1,
get_segment_index_file_size_cb);
if (!status.ok()) {
LOG(WARNING) << "failed to get segment file size from "
<< get_segment_index_file_size_url
@@ -435,8 +437,7 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg*
arg) {
segment_index_file_names.push_back(InvertedIndexDescriptor::get_index_file_path_v2(
InvertedIndexDescriptor::get_index_file_path_prefix(segment_path)));
- status = HttpClient::execute_with_retry(max_retry, 1,
-
get_segment_index_file_size_cb);
+ status = client.execute(max_retry, 1,
get_segment_index_file_size_cb);
if (!status.ok()) {
LOG(WARNING) << "failed to get segment file size from "
<< get_segment_index_file_size_url
@@ -528,7 +529,7 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg*
arg) {
io::LocalFileSystem::PERMS_OWNER_RW);
};
- status = HttpClient::execute_with_retry(max_retry, 1,
get_segment_index_file_cb);
+ status = client.execute(max_retry, 1, get_segment_index_file_cb);
if (!status.ok()) {
LOG(WARNING) << "failed to get segment index file from " <<
get_segment_index_file_url
<< ", status=" << status.to_string();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
