This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 835e867a848 branch-4.1: [fix](s3) handle S3 file writer async task
submission failures #64779 (#65160)
835e867a848 is described below
commit 835e867a8480abd3fdd013c98a588b2cdc60ef33
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Jul 3 20:36:25 2026 +0800
branch-4.1: [fix](s3) handle S3 file writer async task submission failures
#64779 (#65160)
Cherry-picked from #64779
Co-authored-by: hui lai <[email protected]>
---
be/src/io/fs/s3_file_writer.cpp | 50 ++++++++++---
be/src/io/fs/s3_file_writer.h | 1 +
.../s3/test_s3_file_writer_submit_error.groovy | 84 ++++++++++++++++++++++
3 files changed, 127 insertions(+), 8 deletions(-)
diff --git a/be/src/io/fs/s3_file_writer.cpp b/be/src/io/fs/s3_file_writer.cpp
index d543c09328d..ec8e54849f6 100644
--- a/be/src/io/fs/s3_file_writer.cpp
+++ b/be/src/io/fs/s3_file_writer.cpp
@@ -153,13 +153,30 @@ Status S3FileWriter::close(bool non_block) {
_async_close_pack = std::make_unique<AsyncCloseStatusPack>();
_async_close_pack->future = _async_close_pack->promise.get_future();
s3_file_writer_async_close_queuing << 1;
- return
ExecEnv::GetInstance()->non_block_close_thread_pool()->submit_func([&]() {
+ Status submit_status = Status::OK();
+ DBUG_EXECUTE_IF("S3FileWriter.close.submit_async_close.inject_error", {
+ submit_status =
Status::IOError("S3FileWriter.close.submit_async_close.inject_error");
+ });
+ if (submit_status.ok()) {
+ submit_status =
+
ExecEnv::GetInstance()->non_block_close_thread_pool()->submit_func([&]() {
+ s3_file_writer_async_close_queuing << -1;
+ s3_file_writer_async_close_processing << 1;
+ _st = _close_impl();
+ _async_close_pack->promise.set_value(_st);
+ s3_file_writer_async_close_processing << -1;
+ });
+ }
+ if (!submit_status.ok()) {
s3_file_writer_async_close_queuing << -1;
- s3_file_writer_async_close_processing << 1;
+ LOG(WARNING) << "failed to submit async close for "
+ << _obj_storage_path_opts.path.native()
+ << ", fallback to sync close, status=" <<
submit_status;
_st = _close_impl();
_async_close_pack->promise.set_value(_st);
- s3_file_writer_async_close_processing << -1;
- });
+ return _st;
+ }
+ return Status::OK();
}
_st = _close_impl();
_state = State::CLOSED;
@@ -250,6 +267,20 @@ Status S3FileWriter::_build_upload_buffer() {
return Status::OK();
}
+Status S3FileWriter::_submit_upload_buffer(const std::shared_ptr<FileBuffer>&
buf) {
+ _countdown_event.add_count();
+ DBUG_EXECUTE_IF("S3FileWriter.submit_upload_buffer.inject_error", {
+ auto st =
Status::IOError("S3FileWriter.submit_upload_buffer.inject_error");
+ _complete_part_task_callback(st);
+ return st;
+ });
+ auto st = FileBuffer::submit(buf);
+ if (!st.ok()) [[unlikely]] {
+ _complete_part_task_callback(st);
+ }
+ return st;
+}
+
Status S3FileWriter::_close_impl() {
VLOG_DEBUG << "S3FileWriter::close, path: " <<
_obj_storage_path_opts.path.native();
@@ -276,9 +307,12 @@ Status S3FileWriter::_close_impl() {
}
if (_pending_buf != nullptr) { // there is remaining data in buffer need
to be uploaded
- _countdown_event.add_count();
- RETURN_IF_ERROR(FileBuffer::submit(std::move(_pending_buf)));
+ auto st = _submit_upload_buffer(_pending_buf);
_pending_buf = nullptr;
+ if (!st.ok()) {
+ _wait_until_finish("pending buffer submit failed");
+ return st;
+ }
}
RETURN_IF_ERROR(_complete());
@@ -330,9 +364,9 @@ Status S3FileWriter::appendv(const Slice* data, size_t
data_cnt) {
RETURN_IF_ERROR(_create_multi_upload_request());
}
_cur_part_num++;
- _countdown_event.add_count();
- RETURN_IF_ERROR(FileBuffer::submit(std::move(_pending_buf)));
+ auto st = _submit_upload_buffer(_pending_buf);
_pending_buf = nullptr;
+ RETURN_IF_ERROR(st);
}
_bytes_appended += data_size_to_append;
}
diff --git a/be/src/io/fs/s3_file_writer.h b/be/src/io/fs/s3_file_writer.h
index f31a9edef2c..5a8075e03cf 100644
--- a/be/src/io/fs/s3_file_writer.h
+++ b/be/src/io/fs/s3_file_writer.h
@@ -85,6 +85,7 @@ private:
void _upload_one_part(int part_num, UploadFileBuffer& buf);
bool _complete_part_task_callback(Status s);
Status _build_upload_buffer();
+ Status _submit_upload_buffer(const std::shared_ptr<FileBuffer>& buf);
void _record_close_latency();
ObjectStoragePathOptions _obj_storage_path_opts;
diff --git
a/regression-test/suites/cloud_p0/s3/test_s3_file_writer_submit_error.groovy
b/regression-test/suites/cloud_p0/s3/test_s3_file_writer_submit_error.groovy
new file mode 100644
index 00000000000..17518c968b5
--- /dev/null
+++ b/regression-test/suites/cloud_p0/s3/test_s3_file_writer_submit_error.groovy
@@ -0,0 +1,84 @@
+// 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.
+
+suite("test_s3_file_writer_submit_error", "p0, nonConcurrent") {
+ if (!isCloudMode()) {
+ return
+ }
+
+ def submitUploadBufferErrorPoint =
"S3FileWriter.submit_upload_buffer.inject_error"
+ def asyncCloseSubmitErrorPoint =
"S3FileWriter.close.submit_async_close.inject_error"
+ def debugPoints = [submitUploadBufferErrorPoint,
asyncCloseSubmitErrorPoint]
+
+ def disableDebugPoints = {
+ debugPoints.each { point ->
+ GetDebugPoint().disableDebugPointForAllBEs(point)
+ }
+ }
+
+ def insertAndCheck = { String label, boolean expectSuccess ->
+ def insertSql = "INSERT INTO test_s3_file_writer_submit_error " +
+ "VALUES (1, '${label}'), (2, '${label}')"
+ if (expectSuccess) {
+ sql insertSql
+ } else {
+ test {
+ sql insertSql
+ exception "S3FileWriter.submit_upload_buffer.inject_error"
+ }
+ }
+ }
+
+ def runWithDebugPoint = { String label, boolean expectSuccess, String
point ->
+ GetDebugPoint().enableDebugPointForAllBEs(point)
+ try {
+ insertAndCheck(label, expectSuccess)
+ } finally {
+ GetDebugPoint().disableDebugPointForAllBEs(point)
+ }
+ }
+
+ sql """ DROP TABLE IF EXISTS test_s3_file_writer_submit_error """
+ sql """
+ CREATE TABLE IF NOT EXISTS test_s3_file_writer_submit_error (
+ `k1` int NULL,
+ `v1` varchar(32) NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`k1`)
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 1
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1"
+ )
+ """
+
+ setBeConfigTemporary([
+ "enable_file_cache_adaptive_write": "false",
+ "enable_packed_file": "true",
+ "small_file_threshold_bytes": "1"
+ ]) {
+ try {
+ sql """ SET disable_file_cache = true """
+ disableDebugPoints()
+
+ runWithDebugPoint("submit_err", false,
submitUploadBufferErrorPoint)
+ runWithDebugPoint("close_submit_err", true,
asyncCloseSubmitErrorPoint)
+ } finally {
+ sql """ SET disable_file_cache = false """
+ disableDebugPoints()
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
