This is an automated email from the ASF dual-hosted git repository.
sollhui 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 cdabe6c5f48 [fix](s3) handle S3 file writer async task submission
failures (#64779)
cdabe6c5f48 is described below
commit cdabe6c5f48fdc25eff17d6ca3ede53fb649c020
Author: hui lai <[email protected]>
AuthorDate: Thu Jul 2 18:04:37 2026 +0800
[fix](s3) handle S3 file writer async task submission failures (#64779)
### What problem does this PR solve?
`S3FileWriter` did not handle two async submit failure paths correctly.
1. Upload buffer submit failure:
The writer added the upload task to `_countdown_event` before submitting
it to the async upload thread pool. If `FileBuffer::submit()` failed,
the task would never run, so `_complete_part_task_callback()` would not
be called. This could hide the real failure and leave later
close/destructor paths waiting for a countdown that would never be
released.
2. Async close submit failure:
`close(true)` could leave the writer in `ASYNC_CLOSING` after failing to
submit the close task, without a valid future completion path. Later
`try_finish_close()` or destructor logic could then observe an
inconsistent async-close state.
This PR centralizes S3 upload-buffer submission so submit failures go
through the normal completion callback and release the countdown. It
also falls back to synchronous close when async close task submission
fails, returning the real close result.
---
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 8b9a9f2dfd6..f8b836607a1 100644
--- a/be/src/io/fs/s3_file_writer.cpp
+++ b/be/src/io/fs/s3_file_writer.cpp
@@ -152,13 +152,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;
@@ -249,6 +266,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();
@@ -275,9 +306,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());
@@ -329,9 +363,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]
