[feature](debug point) Add handler to debug point (#33350)

be/src/util/debug_points.h

@@ -17,6 +17,7 @@
 
 #pragma once
 
+#include <any>
 #include <atomic>
 #include <boost/lexical_cast.hpp>
 #include <chrono>
@@ -42,6 +43,7 @@
 
 // define some common debug actions
 // usage example: DBUG_EXECUTE_IF("xxx", DBUG_BLOCK);
+
 #define DBUG_BLOCK                                                      \
     {                                                                   \
         LOG(INFO) << "start debug block " << DP_NAME;                   \
@@ -51,6 +53,42 @@
         LOG(INFO) << "end debug block " << DP_NAME;                     \
     }
 
+// example of debug point with handler.
+//
+// base code:
+//
+// void demo_handler() {
+//    int a = 0;
+//
+//    DBUG_EXECUTE_IF("set_a", {
+//        auto handler = std::any_cast<std::function<void(int&)>>(dp->handler);
+//        handler(a);
+//    });
+//
+//    DBUG_EXECUTE_IF("get_a", {
+//        auto handler = std::any_cast<std::function<void(int)>>(dp->handler);
+//        handler(a);
+//    });
+//}
+//
+// test code:
+//
+//TEST(DebugPointsTest, Handler) {
+//    config::enable_debug_points = true;
+//    DebugPoints::instance()->clear();
+//
+//    int got_a = 0;
+//
+//    std::function<void(int&)> set_handler = [](int& a) { a = 1000; };
+//    std::function<void(int)> get_handler = [&got_a](int a) { got_a = a; };
+//    DebugPoints::instance()->add_with_handler("set_a", set_handler);
+//    DebugPoints::instance()->add_with_handler("get_a", get_handler);
+//
+//    demo_handler();
+//
+//    EXPECT_EQ(1000, got_a);
+//}
+
 namespace doris {
 
 struct DebugPoint {
@@ -60,6 +98,11 @@ struct DebugPoint {
 
     std::map<std::string, std::string> params;
 
+    // Usually `handler` use in be ut, to exchange local variable between base code and injected code,
+    // or change with different injected handlers.
+    // test/util/debug_points_test.cpp#Handler give a example.
+    std::any handler;
+
     template <typename T>
     T param(const std::string& key, T default_value = T()) {
         auto it = params.find(key);
@@ -120,6 +163,10 @@ public:
         add_with_params(name, {{"value", fmt::format("{}", value)}});
     }
 
+    void add_with_handler(const std::string& name, std::any handler) {
+        add(name, std::shared_ptr<DebugPoint>(new DebugPoint {.handler = handler}));
+    }
+
     static DebugPoints* instance();
 
 private:

be/src/vec/sink/load_stream_stub.cpp

@@ -291,10 +291,7 @@ Status LoadStreamStub::wait_for_schema(int64_t partition_id, int64_t index_id, i
 }
 
 Status LoadStreamStub::close_wait(RuntimeState* state, int64_t timeout_ms) {
-    DBUG_EXECUTE_IF("LoadStreamStub::close_wait.long_wait", {
-        while (true) {
-        };
-    });
+    DBUG_EXECUTE_IF("LoadStreamStub::close_wait.long_wait", DBUG_BLOCK);
     if (!_is_init.load()) {
         return Status::InternalError("stream {} is not opened, load_id={}", _stream_id,
                                      print_id(_load_id));

be/test/util/debug_points_test.cpp

@@ -96,4 +96,34 @@ TEST(DebugPointsTest, AddTest) {
               DebugPoints::instance()->get_debug_param_or_default<std::string>("dbug4", ""));
 }
 
+void demo_handler() {
+    int a = 0;
+
+    DBUG_EXECUTE_IF("set_a", {
+        auto handler = std::any_cast<std::function<void(int&)>>(dp->handler);
+        handler(a);
+    });
+
+    DBUG_EXECUTE_IF("get_a", {
+        auto handler = std::any_cast<std::function<void(int)>>(dp->handler);
+        handler(a);
+    });
+}
+
+TEST(DebugPointsTest, Handler) {
+    config::enable_debug_points = true;
+    DebugPoints::instance()->clear();
+
+    int got_a = 0;
+
+    std::function<void(int&)> set_handler = [](int& a) { a = 1000; };
+    std::function<void(int)> get_handler = [&got_a](int a) { got_a = a; };
+    DebugPoints::instance()->add_with_handler("set_a", set_handler);
+    DebugPoints::instance()->add_with_handler("get_a", get_handler);
+
+    demo_handler();
+
+    EXPECT_EQ(1000, got_a);
+}
+
 } // namespace doris