[fix](load) Remove deprecated auth_code handling

be/src/common/utils.h

@@ -41,8 +41,6 @@ struct AuthInfo {
     std::string passwd;
     std::string cluster;
     std::string user_ip;
-    // -1 as unset
-    int64_t auth_code = -1; // deprecated
     std::string token;
 
     std::string cert_pem;
@@ -59,14 +57,8 @@ struct AuthInfo {
 template <class T>
 void set_request_auth(T* req, const AuthInfo& auth) {
     req->user = auth.user; // always set user, because it may be used by FE
-    // auth code is deprecated and should be removed in 3.1
-    if (auth.auth_code != -1) {
-        // if auth_code is set, no need to set other info
-        req->__set_auth_code(auth.auth_code);
-        // user name and passwd is unused, but they are required field.
-        // so they have to be set.
-        req->passwd = "";
-    } else if (auth.token != "") {
+    // Internal loads, including Routine Load and WAL replay, use the FE token.
+    if (!auth.token.empty()) {
         req->__isset.token = true;
         req->token = auth.token;
     } else {

be/src/load/group_commit/group_commit_mgr.cpp

@@ -494,7 +494,7 @@ Status GroupCommitTable::_create_group_commit_load(
         request.__set_load_sql(ss.str());
         request.__set_loadId(tload_id);
         request.__set_label(label);
-        request.__set_token("group_commit"); // this is a fake, fe not check it now
+        request.__set_token(_exec_env->cluster_info()->curr_auth_token);
         request.__set_max_filter_ratio(1.0);
         request.__set_strictMode(false);
         request.__set_partial_update(false);
@@ -608,8 +608,6 @@ Status GroupCommitTable::_finish_group_commit_load(int64_t db_id, int64_t table_
         });
         // commit txn
         TLoadTxnCommitRequest request;
-        // deprecated and should be removed in 3.1, use token instead
-        request.__set_auth_code(0);
         request.__set_token(_exec_env->cluster_info()->curr_auth_token);
         request.__set_db_id(db_id);
         request.__set_table_id(table_id);
@@ -661,8 +659,6 @@ Status GroupCommitTable::_finish_group_commit_load(int64_t db_id, int64_t table_
     } else {
         // abort txn
         TLoadTxnRollbackRequest request;
-        // deprecated and should be removed in 3.1, use token instead
-        request.__set_auth_code(0);
         request.__set_token(_exec_env->cluster_info()->curr_auth_token);
         request.__set_db_id(db_id);
         request.__set_txnId(txn_id);

be/src/load/group_commit/wal/wal_table.cpp

@@ -22,6 +22,7 @@
 
 #include "io/fs/local_file_system.h"
 #include "io/fs/stream_load_pipe.h"
+#include "load/group_commit/wal/wal_file_reader.h"
 #include "load/group_commit/wal/wal_manager.h"
 #include "runtime/cluster_info.h"
 #include "runtime/fragment_mgr.h"
@@ -179,9 +180,6 @@ bool WalTable::_need_replay(std::shared_ptr<WalInfo> wal_info) {
 
 Status WalTable::_try_abort_txn(int64_t db_id, std::string& label) {
     TLoadTxnRollbackRequest request;
-    // this is a fake, fe not check it now
-    // should be removed in 3.1, use token instead
-    request.__set_auth_code(0);
     request.__set_token(_exec_env->cluster_info()->curr_auth_token);
     request.__set_db_id(db_id);
     request.__set_label(label);
@@ -276,8 +274,6 @@ Status WalTable::_handle_stream_load(int64_t wal_id, const std::string& wal,
         // wait stream load finish
         RETURN_IF_ERROR(ctx->load_status_future.get());
         if (ctx->status.ok()) {
-            // deprecated and should be removed in 3.1, use token instead.
-            ctx->auth.auth_code = wal_id;
             st = _exec_env->stream_load_executor()->commit_txn(ctx.get());
         } else {
             st = ctx->status;

be/src/load/routine_load/routine_load_task_executor.cpp

@@ -291,8 +291,6 @@ Status RoutineLoadTaskExecutor::submit_task(const TRoutineLoadTask& task) {
     ctx->db = task.db;
     ctx->table = task.tbl;
     ctx->label = task.label;
-    // deprecated, removed in 3.1, use auth token instead.
-    ctx->auth.auth_code = task.auth_code;
     ctx->auth.token = _exec_env->cluster_info()->curr_auth_token;
 
     if (task.__isset.max_interval_s) {

be/src/service/http/action/http_stream.cpp

@@ -356,9 +356,8 @@ Status HttpStreamAction::process_put(HttpRequest* http_req,
             request.__set_memtable_on_sink_node(value);
         }
     } else {
-        request.__set_token(ctx->auth.token);
         request.__set_load_sql(ctx->sql_str);
-        ctx->auth.token = "";
+        // WAL replay reuses this context for commit or rollback, so keep its token.
     }
     set_request_auth(&request, ctx->auth);
     request.__set_loadId(ctx->id.to_thrift());

be/src/service/http/http_common.h

@@ -68,7 +68,6 @@ static const std::string HTTP_TXN_OPERATION_KEY = "txn_operation";
 static const std::string HTTP_MEMTABLE_ON_SINKNODE = "memtable_on_sink_node";
 static const std::string HTTP_LOAD_STREAM_PER_NODE = "load_stream_per_node";
 static const std::string HTTP_WAL_ID_KY = "wal_id";
-static const std::string HTTP_AUTH_CODE = "auth_code"; // deprecated
 static const std::string HTTP_GROUP_COMMIT = "group_commit";
 static const std::string HTTP_CLOUD_CLUSTER = "cloud_cluster";
 static const std::string HTTP_EMPTY_FIELD_AS_NULL = "empty_field_as_null";

be/src/service/http/utils.cpp

@@ -43,6 +43,7 @@
 #include "service/http/http_method.h"
 #include "service/http/http_request.h"
 #include "service/http/http_status.h"
+#include "util/debug_points.h"
 #include "util/md5.h"
 #include "util/path_util.h"
 #include "util/security.h"
@@ -89,8 +90,6 @@ bool parse_basic_auth(const HttpRequest& req, std::string* user, std::string* pa
 bool parse_basic_auth(const HttpRequest& req, AuthInfo* auth) {
     // deprecated, removed in 3.1, use AUTH_TOKEN
     const auto& token = req.header("token");
-    // deprecated, removed in 3.1, use AUTH_TOKEN
-    const auto& auth_code = req.header(HTTP_AUTH_CODE);
     const auto& auth_token = req.header(HttpHeaders::AUTH_TOKEN);
 
     std::tuple<std::string, std::string, std::string> tmp;
@@ -111,16 +110,6 @@ bool parse_basic_auth(const HttpRequest& req, AuthInfo* auth) {
         auth->token = token; // deprecated
     } else if (!auth_token.empty()) {
         auth->token = auth_token;
-    } else if (!auth_code.empty()) {
-        // auth_code comes straight from a request header, a malformed one is just an
-        // invalid credential and must not throw out of the http callback. The value is a
-        // credential that is_sensitive_header() masks elsewhere, so it is not logged.
-        auto parsed_auth_code = safe_stoll(auth_code, HTTP_AUTH_CODE);
-        if (!parsed_auth_code.has_value()) {
-            LOG(WARNING) << "parse auth code failed, malformed " << HTTP_AUTH_CODE << " header";
-            return false;
-        }
-        auth->auth_code = parsed_auth_code.value(); // deprecated
     } else if (!valid_basic_auth) {
         return false;
     }

be/test/runtime/routine_load_task_executor_test.cpp

@@ -26,15 +26,20 @@
 #include <librdkafka/rdkafkacpp.h>
 #include <unistd.h>
 
+#include <future>
 #include <map>
 
 #include "common/config.h"
 #include "common/status.h"
+#include "common/utils.h"
 #include "gtest/gtest_pred_impl.h"
 #include "load/stream_load/new_load_stream_mgr.h"
+#include "load/stream_load/stream_load_context.h"
 #include "load/stream_load/stream_load_executor.h"
 #include "runtime/cluster_info.h"
 #include "runtime/exec_env.h"
+#include "util/defer_op.h"
+#include "util/threadpool.h"
 
 namespace doris {
 
@@ -74,6 +79,57 @@ class RoutineLoadTaskExecutorTest : public testing::Test {
     }
 };
 
+TEST_F(RoutineLoadTaskExecutorTest, LegacyTaskUsesHeartbeatToken) {
+    const auto original_pool_size = config::max_routine_load_thread_pool_size;
+    config::max_routine_load_thread_pool_size = 1;
+    Defer restore_config {[&] { config::max_routine_load_thread_pool_size = original_pool_size; }};
+    _env->cluster_info()->curr_auth_token = "routine-load-token";
+    RoutineLoadTaskExecutor executor(_env);
+    ASSERT_TRUE(executor.init(1024 * 1024).ok());
+
+    // Hold the worker so the submitted task can be inspected before execution.
+    std::promise<void> started;
+    std::promise<void> resume;
+    auto resumed = resume.get_future();
+    Defer stop_executor {[&] {
+        resume.set_value();
+        executor.stop();
+    }};
+    ASSERT_TRUE(executor.get_thread_pool()
+                        .submit_func([&] {
+                            started.set_value();
+                            resumed.wait();
+                        })
+                        .ok());
+    started.get_future().wait();
+
+    TRoutineLoadTask task;
+    task.type = TLoadSourceType::KAFKA;
+    task.__set_auth_code(12345); // Required field sent by older FEs.
+    task.__set_id(TUniqueId());
+    task.__set_txn_id(4);
+    task.__set_db("db1");
+    task.__set_tbl("tbl1");
+    ASSERT_TRUE(executor.submit_task(task).ok());
+    std::lock_guard<std::mutex> lock(executor._lock);
+    auto ctx = executor._task_map.at(UniqueId(task.id));
+    TStreamLoadPutRequest plan;
+    set_request_auth(&plan, ctx->auth);
+    TLoadTxnCommitRequest commit;
+    _env->stream_load_executor()->get_commit_request(ctx.get(), commit);
+    TLoadTxnRollbackRequest rollback;
+    set_request_auth(&rollback, ctx->auth);
+    EXPECT_EQ("routine-load-token", plan.token);
+    EXPECT_EQ(plan.token, commit.token);
+    EXPECT_EQ(plan.token, rollback.token);
+    EXPECT_TRUE(plan.__isset.token);
+    EXPECT_TRUE(commit.__isset.token);
+    EXPECT_TRUE(rollback.__isset.token);
+    EXPECT_FALSE(plan.__isset.auth_code);
+    EXPECT_FALSE(commit.__isset.auth_code);
+    EXPECT_FALSE(rollback.__isset.auth_code);
+}
+
 TEST_F(RoutineLoadTaskExecutorTest, exec_task) {
     TRoutineLoadTask task;
     task.type = TLoadSourceType::KAFKA;

be/test/service/http/http_utils_test.cpp

@@ -16,18 +16,17 @@
 // under the License.
 
 #include <event2/http.h>
+#include <gen_cpp/FrontendService_types.h>
 #include <gtest/gtest-message.h>
 #include <gtest/gtest-test-part.h>
 
 #include <string>
 
 #include "common/utils.h"
 #include "gtest/gtest_pred_impl.h"
-#include "service/http/http_common.h"
 #include "service/http/http_headers.h"
 #include "service/http/http_request.h"
 #include "service/http/utils.h"
-#include "util/string_util.h"
 #include "util/url_coding.h"
 
 struct evhttp_request;
@@ -46,6 +45,16 @@ class HttpUtilsTest : public testing::Test {
     }
 
 private:
+    template <typename Request>
+    void check_token_request(const AuthInfo& auth) {
+        Request request;
+        set_request_auth(&request, auth);
+        EXPECT_TRUE(request.__isset.token);
+        EXPECT_EQ("internal-token", request.token);
+        EXPECT_EQ("admin", request.user);
+        EXPECT_FALSE(request.__isset.auth_code);
+    }
+
     evhttp_request* _evhttp_req = nullptr;
 };
 
@@ -98,23 +107,36 @@ TEST_F(HttpUtilsTest, parse_basic_auth) {
     }
 }
 
-TEST_F(HttpUtilsTest, parse_auth_code) {
-    {
+TEST_F(HttpUtilsTest, deprecated_auth_code_does_not_replace_credentials) {
+    for (const auto* value : {"0", "1234567890123", "-1", "not_a_number"}) {
         HttpRequest req(_evhttp_req);
+        req._headers.emplace("auth_code", value);
+        AuthInfo missing_auth;
+        EXPECT_FALSE(parse_basic_auth(req, &missing_auth));
+
         req._headers.emplace(HttpHeaders::AUTHORIZATION, encode_basic_auth("doris", "passwd"));
-        req._headers.emplace(HTTP_AUTH_CODE, "1234567890123");
         AuthInfo auth;
         EXPECT_TRUE(parse_basic_auth(req, &auth));
-        EXPECT_EQ(1234567890123, auth.auth_code);
+        TLoadTxnBeginRequest request;
+        set_request_auth(&request, auth);
+        EXPECT_EQ("doris", request.user);
+        EXPECT_EQ("passwd", request.passwd);
+        EXPECT_FALSE(request.__isset.auth_code);
+        EXPECT_FALSE(request.__isset.token);
     }
-    {
-        // a malformed auth code is rejected instead of throwing out of the http callback
+}
+
+TEST_F(HttpUtilsTest, internal_token_is_forwarded_with_user) {
+    for (const auto& header : {std::string("token"), HttpHeaders::AUTH_TOKEN}) {
         HttpRequest req(_evhttp_req);
-        req._headers.emplace(HttpHeaders::AUTHORIZATION, encode_basic_auth("doris", "passwd"));
-        req._headers.emplace(HTTP_AUTH_CODE, "not_a_number");
+        req._headers.emplace(header, "internal-token");
+        req._headers.emplace("auth_code", "0");
+        req._headers.emplace(HttpHeaders::AUTHORIZATION, encode_basic_auth("admin", "unused"));
         AuthInfo auth;
-        EXPECT_FALSE(parse_basic_auth(req, &auth));
-        EXPECT_EQ(-1, auth.auth_code);
+        ASSERT_TRUE(parse_basic_auth(req, &auth));
+        check_token_request<TStreamLoadPutRequest>(auth);
+        check_token_request<TLoadTxnCommitRequest>(auth);
+        check_token_request<TLoadTxnRollbackRequest>(auth);
     }
 }
 

be/test/service/http/stream_load_test.cpp

@@ -22,13 +22,16 @@
 #include <memory>
 
 #include "common/config.h"
+#include "common/utils.h"
 #include "event2/buffer.h"
 #include "event2/event.h"
 #include "event2/http.h"
 #include "event2/http_struct.h"
 #include "evhttp.h"
 #include "load/group_commit/wal/wal_manager.h"
 #include "load/stream_load/stream_load_context.h"
+#include "load/stream_load/stream_load_executor.h"
+#include "runtime/cluster_info.h"
 #include "runtime/exec_env.h"
 #include "service/http/action/http_stream.h"
 #include "service/http/ev_http_server.h"
@@ -55,6 +58,38 @@ void http_request_done_cb(struct evhttp_request* req, void* arg) {
     event_base_loopbreak((struct event_base*)arg);
 }
 
+TEST_F(StreamLoadTest, WalPlanningFailurePreservesTokenForRetryAndRollback) {
+    auto* env = ExecEnv::GetInstance();
+    auto* original_cluster_info = env->cluster_info();
+    ClusterInfo cluster_info;
+    env->set_cluster_info(&cluster_info);
+    Defer restore_cluster_info {[&] { env->set_cluster_info(original_cluster_info); }};
+    auto ctx = std::make_shared<StreamLoadContext>(env);
+    ctx->auth.user = "admin";
+    ctx->auth.token = "internal-wal-token";
+    ctx->sql_str = "insert into wal_replay select * from http_stream()";
+    HttpStreamAction action(env);
+
+    // No FE address: planning fails after constructing the request, before any RPC.
+    EXPECT_FALSE(action.process_put(nullptr, ctx).ok());
+    EXPECT_EQ("internal-wal-token", ctx->auth.token);
+    EXPECT_FALSE(action.process_put(nullptr, ctx).ok());
+    EXPECT_EQ("internal-wal-token", ctx->auth.token);
+
+    TLoadTxnCommitRequest commit;
+    StreamLoadExecutor executor(env);
+    executor.get_commit_request(ctx.get(), commit);
+    EXPECT_TRUE(commit.__isset.token);
+    EXPECT_EQ("internal-wal-token", commit.token);
+    EXPECT_FALSE(commit.__isset.auth_code);
+
+    TLoadTxnRollbackRequest rollback;
+    set_request_auth(&rollback, ctx->auth);
+    EXPECT_TRUE(rollback.__isset.token);
+    EXPECT_EQ(commit.token, rollback.token);
+    EXPECT_FALSE(rollback.__isset.auth_code);
+}
+
 TEST_F(StreamLoadTest, TestHeader) {
     // 1G
     auto wal_mgr = WalManager::create_unique(ExecEnv::GetInstance(), config::group_commit_wal_path);