BugFix: fatal error when the delta writer is busy (#3878)

This PR makes retry when the AsyncDeltaWriter is busy to avoid the bthread's fatal error.

Author: rickif

be/src/storage/async_delta_writer_executor.h

@@ -3,32 +3,40 @@
 #include "common/compiler_util.h"
 DIAGNOSTIC_PUSH
 DIAGNOSTIC_IGNORE("-Wclass-memaccess")
+#include <bthread/bthread.h>
 #include <bthread/execution_queue.h>
 DIAGNOSTIC_POP
 
 #include "common/config.h"
 #include "util/threadpool.h"
 
 namespace starrocks {
+const int64_t kRetryIntervalMs = 50;
 
 // Used to run bthread::ExecutionQueue task in pthread instead of bthread.
 // Reference: https://github.com/apache/incubator-brpc/blob/master/docs/cn/execution_queue.md
 class AsyncDeltaWriterExecutor : public bthread::Executor {
 public:
-    Status init() {
+    Status init(int max_queue_size = 40960) {
         if (_thread_pool != nullptr) {
             return Status::InternalError("already initialized");
         }
         return ThreadPoolBuilder("delta_writer")
                 .set_min_threads(config::number_tablet_writer_threads / 2)
                 .set_max_threads(std::max<int>(1, config::number_tablet_writer_threads))
-                .set_max_queue_size(40960)
+                .set_max_queue_size(max_queue_size)
                 .set_idle_timeout(MonoDelta::FromMilliseconds(5 * 60 * 1000))
                 .build(&_thread_pool);
     }
 
     int submit(void* (*fn)(void*), void* args) override {
-        auto st = _thread_pool->submit_func([=]() { fn(args); });
+        Status st;
+        while (true) {
+            st = _thread_pool->submit_func([=]() { fn(args); });
+            if (!st.is_service_unavailable()) break;
+            LOG(INFO) << "async_delta_writer is busy, retry after " << kRetryIntervalMs << "ms";
+            bthread_usleep(kRetryIntervalMs * 1000);
+        }
         LOG_IF(WARNING, !st.ok()) << st;
         return st.ok() ? 0 : -1;
     }

be/test/CMakeLists.txt

@@ -152,6 +152,7 @@ set(EXEC_FILES
         ./storage/rowset/zone_map_index_test.cpp
         ./storage/rowset/unique_rowset_id_generator_test.cpp
         ./storage/rowset/index_page_test.cpp
+        ./storage/async_delta_writer_executor_test.cpp
         ./storage/selection_vector_test.cpp
         ./storage/snapshot_meta_test.cpp
         ./storage/short_key_index_test.cpp

be/test/storage/async_delta_writer_executor_test.cpp

@@ -0,0 +1,30 @@
+// This file is licensed under the Elastic License 2.0. Copyright 2021-present, StarRocks Limited.
+
+#include "storage/async_delta_writer_executor.h"
+
+#include <gtest/gtest.h>
+
+#include <vector>
+
+#include "common/config.h"
+#include "testutil/parallel_test.h"
+#include "util/monotime.h"
+
+namespace starrocks {
+
+PARALLEL_TEST(AsyncDeltaWriterExecutorTest, test_overload) {
+    AsyncDeltaWriterExecutor executor;
+    ASSERT_TRUE(executor.init(1).ok());
+
+    // submit overloading tasks.
+    for (size_t i = 0; i < 4 * config::number_tablet_writer_threads; ++i) {
+        ASSERT_EQ(0, executor.submit(
+                             [](void*) -> void* {
+                                 SleepFor(MonoDelta::FromSeconds(1));
+                                 return nullptr;
+                             },
+                             nullptr));
+    }
+}
+
+} // namespace starrocks