[regression](move-memtable) test LoadStream on_idle_timeout (apache#29354)

Signed-off-by: freemandealer <[email protected]>

Author: freemandealer

be/src/common/config.cpp

@@ -777,7 +777,7 @@ DEFINE_Int64(open_load_stream_timeout_ms, "60000"); // 60s
 DEFINE_Int64(close_load_stream_timeout_ms, "600000"); // 10 min
 
 // idle timeout for load stream in ms
-DEFINE_Int64(load_stream_idle_timeout_ms, "600000");
+DEFINE_mInt64(load_stream_idle_timeout_ms, "600000");
 // brpc streaming max_buf_size in bytes
 DEFINE_Int64(load_stream_max_buf_size, "20971520"); // 20MB
 // brpc streaming messages_in_batch

be/src/runtime/load_stream.cpp

@@ -454,7 +454,7 @@ Status LoadStream::_append_data(const PStreamHeader& header, butil::IOBuf* data)
     IndexStreamSharedPtr index_stream;
 
     int64_t index_id = header.index_id();
-    DBUG_EXECUTE_IF("TabletStream.add_segment.unknown_indexid",
+    DBUG_EXECUTE_IF("TabletStream._append_data.unknown_indexid",
                     { index_id = UNKNOWN_ID_FOR_TEST; });
     auto it = _index_streams_map.find(index_id);
     if (it == _index_streams_map.end()) {

be/src/service/internal_service.cpp

@@ -391,8 +391,7 @@ void PInternalServiceImpl::open_load_stream(google::protobuf::RpcController* con
         }
 
         stream_options.handler = load_stream.get();
-        // TODO : set idle timeout
-        // stream_options.idle_timeout_ms =
+        stream_options.idle_timeout_ms = config::load_stream_idle_timeout_ms;
 
         StreamId streamid;
         if (brpc::StreamAccept(&streamid, *cntl, &stream_options) != 0) {

be/src/vec/sink/load_stream_stub.cpp

@@ -21,6 +21,7 @@
 
 #include "olap/rowset/rowset_writer.h"
 #include "util/brpc_client_cache.h"
+#include "util/debug_points.h"
 #include "util/network_util.h"
 #include "util/thrift_util.h"
 #include "util/uid_util.h"
@@ -330,6 +331,10 @@ Status LoadStreamStub::_send_with_retry(butil::IOBuf& buf) {
         int ret;
         {
             SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->orphan_mem_tracker());
+            DBUG_EXECUTE_IF("LoadStreamStub._send_with_retry.delay_before_send", {
+                int64_t delay_ms = dp->param<int64>("delay_ms", 1000);
+                bthread_usleep(delay_ms * 1000);
+            });
             ret = brpc::StreamWrite(_stream_id, buf);
         }
         DBUG_EXECUTE_IF("LoadStreamStub._send_with_retry.stream_write_failed", { ret = EPIPE; });

regression-test/suites/fault_injection_p0/test_load_stream_fault_injection.groovy

@@ -67,6 +67,52 @@ suite("load_stream_fault_injection", "nonConcurrent") {
         file "baseall.txt"
     }
 
+    def backendId_to_backendIP = [:]
+    def backendId_to_backendHttpPort = [:]
+    def backendId_to_params = [string:[:]]
+
+    getBackendIpHttpPort(backendId_to_backendIP, backendId_to_backendHttpPort);
+
+    def set_be_param = { paramName, paramValue ->
+        // for eache be node, set paramName=paramValue
+        for (String id in backendId_to_backendIP.keySet()) {
+            def beIp = backendId_to_backendIP.get(id)
+            def bePort = backendId_to_backendHttpPort.get(id)
+            def (code, out, err) = curl("POST", String.format("http://%s:%s/api/update_config?%s=%s", beIp, bePort, paramName, paramValue))
+            assertTrue(out.contains("OK"))
+        }
+    }
+
+    def reset_be_param = { paramName ->
+        // for eache be node, reset paramName to default
+        for (String id in backendId_to_backendIP.keySet()) {
+            def beIp = backendId_to_backendIP.get(id)
+            def bePort = backendId_to_backendHttpPort.get(id)
+            def original_value = backendId_to_params.get(id).get(paramName)
+            def (code, out, err) = curl("POST", String.format("http://%s:%s/api/update_config?%s=%s", beIp, bePort, paramName, original_value))
+            assertTrue(out.contains("OK"))
+        }
+    }
+
+    def get_be_param = { paramName ->
+        // for eache be node, get param value by default
+        def paramValue = ""
+        for (String id in backendId_to_backendIP.keySet()) {
+            def beIp = backendId_to_backendIP.get(id)
+            def bePort = backendId_to_backendHttpPort.get(id)
+            // get the config value from be
+            def (code, out, err) = curl("GET", String.format("http://%s:%s/api/show_config?conf_item=%s", beIp, bePort, paramName))
+            assertTrue(code == 0)
+            assertTrue(out.contains(paramName))
+            // parsing
+            def resultList = parseJson(out)[0]
+            assertTrue(resultList.size() == 4)
+            // get original value
+            paramValue = resultList[2]
+            backendId_to_params.get(id, [:]).put(paramName, paramValue)
+        }
+    }
+
     def load_with_injection = { injection, expect_errmsg ->
         try {
             GetDebugPoint().enableDebugPointForAllBEs(injection)
@@ -110,10 +156,21 @@ suite("load_stream_fault_injection", "nonConcurrent") {
     // LoadStream add_segment meet unknown segid in request header
     load_with_injection("TabletStream.add_segment.unknown_segid", "")
     // LoadStream append_data meet unknown index id in request header
-    load_with_injection("abletStream.add_segment.unknown_indexid", "")
+    load_with_injection("TabletStream._append_data.unknown_indexid", "")
     // LoadStream dispatch meet unknown load id
     load_with_injection("LoadStream._dispatch.unknown_loadid", "")
     // LoadStream dispatch meet unknown src id
     load_with_injection("LoadStream._dispatch.unknown_srcid", "")
+
+    // LoadStream meets StreamRPC idle timeout
+    get_be_param("load_stream_idle_timeout_ms")
+    set_be_param("load_stream_idle_timeout_ms", 500)
+    try {
+        load_with_injection("LoadStreamStub._send_with_retry.delay_before_send", "")
+    } catch(Exception e) {
+        logger.info(e.getMessage())
+    } finally {
+        reset_be_param("load_stream_idle_timeout_ms")
+    }
 }