[fix](load) Validate Routine Load Kafka client properties

be/src/load/routine_load/data_consumer.cpp

@@ -18,6 +18,7 @@
 #include "load/routine_load/data_consumer.h"
 
 #include <absl/strings/str_split.h>
+#include <gen_cpp/BackendService_constants.h>
 #include <gen_cpp/Types_types.h>
 #include <gen_cpp/internal_service.pb.h>
 #include <librdkafka/rdkafkacpp.h>
@@ -47,6 +48,7 @@
 #include "common/metrics/doris_metrics.h"
 #include "common/status.h"
 #include "load/routine_load/consumer_helpers.h"
+#include "load/routine_load/kafka_properties.h"
 #include "load/routine_load/kinesis_conf.h"
 #include "runtime/aws_msk_iam_auth.h"
 #include "runtime/exec_env.h"
@@ -77,6 +79,10 @@ Status KafkaDataConsumer::init(std::shared_ptr<StreamLoadContext> ctx) {
     // conf has to be deleted finally
     Defer delete_conf {[conf]() { delete conf; }};
 
+    // Metadata probes and scheduled tasks share this path, including jobs from
+    // older FEs. Validate before any user-controlled Conf::set() can have effects.
+    RETURN_IF_ERROR(configure_kafka_security(*conf, ctx->kafka_info->properties));
+
     std::string errstr;
     auto set_conf = [&conf, &errstr](const std::string& conf_key, const std::string& conf_val) {
         RdKafka::Conf::ConfResult res = conf->set(conf_key, conf_val, errstr);
@@ -118,10 +124,10 @@ Status KafkaDataConsumer::init(std::shared_ptr<StreamLoadContext> ctx) {
     for (auto& item : ctx->kafka_info->properties) {
         _custom_properties.emplace(item.first, item.second);
 
-        // AWS properties (aws.*) are Doris-specific for MSK IAM authentication
-        // and should not be passed to librdkafka
-        if (starts_with(item.first, "aws.")) {
-            LOG(INFO) << "Skipping AWS property for librdkafka: " << item.first;
+        // Doris handles AWS authentication itself. Older FEs can also send offset
+        // bookkeeping properties; neither kind should reach librdkafka.
+        if (g_BackendService_constants.KAFKA_DORIS_PROPERTIES.contains(item.first)) {
+            LOG(INFO) << "Skipping Doris property for librdkafka: " << item.first;
             continue;
         }
 

be/src/load/routine_load/kafka_properties.cpp

@@ -0,0 +1,69 @@
+// 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.
+
+#include "load/routine_load/kafka_properties.h"
+
+#include <gen_cpp/BackendService_constants.h>
+#include <librdkafka/rdkafkacpp.h>
+
+#include "common/status.h"
+#include "util/string_util.h"
+
+namespace doris {
+
+Status configure_kafka_security(RdKafka::Conf& conf,
+                                const std::map<std::string, std::string>& properties) {
+    for (const auto& [key, value] : properties) {
+        // A new FE also sends zero explicitly to protect rolling upgrades to old BEs.
+        // Never allow a user to re-enable the default kinit command with a nonzero interval.
+        if (key == "sasl.kerberos.min.time.before.relogin" && value != "0") {
+            return Status::InvalidArgument(
+                    "PAUSE: Kafka property '{}' is not allowed in Routine Load with a nonzero "
+                    "value; "
+                    "Kerberos tickets must be managed externally",
+                    key);
+        }
+        // Certificate paths supplied by SQL users must be Doris-managed files.
+        if (g_BackendService_constants.KAFKA_FILE_PROPERTIES.contains(key)) {
+            if (!starts_with(value, "FILE:") || value.size() == 5) {
+                return Status::InvalidArgument(
+                        "PAUSE: Kafka property '{}' must reference a file registered with CREATE "
+                        "FILE",
+                        key);
+            }
+        } else if (!g_BackendService_constants.KAFKA_ALLOWED_PROPERTIES.contains(key) &&
+                   !g_BackendService_constants.KAFKA_DORIS_PROPERTIES.contains(key)) {
+            // For example, plugin.library.paths can load code during Conf::set(),
+            // before a consumer is even created. Unknown properties fail closed.
+            return Status::InvalidArgument(
+                    "PAUSE: Kafka property '{}' is not allowed in Routine Load", key);
+        }
+    }
+
+    // Rejecting kinit.cmd alone is insufficient: the default command interpolates
+    // the principal/keytab. Setting the interval to zero disables both the initial
+    // command and the refresh timer. Operators must maintain the ticket cache.
+    // This is mandatory, so do not use init's best-effort set_conf wrapper here.
+    std::string error;
+    if (conf.set("sasl.kerberos.min.time.before.relogin", "0", error) != RdKafka::Conf::CONF_OK) {
+        return Status::InternalError(
+                "PAUSE: failed to disable Kafka Kerberos command execution: {}", error);
+    }
+    return Status::OK();
+}
+
+} // namespace doris

be/src/load/routine_load/kafka_properties.h

@@ -0,0 +1,34 @@
+// 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.
+
+#pragma once
+
+#include <map>
+#include <string>
+
+namespace RdKafka {
+class Conf;
+}
+
+namespace doris {
+class Status;
+
+// Validate the entire map before applying any user properties, then disable
+// librdkafka's shell-based Kerberos ticket acquisition and renewal.
+Status configure_kafka_security(RdKafka::Conf& conf,
+                                const std::map<std::string, std::string>& properties);
+} // namespace doris

be/test/load/routine_load/kafka_properties_test.cpp

@@ -0,0 +1,105 @@
+// 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.
+
+#include "load/routine_load/kafka_properties.h"
+
+#include <gen_cpp/BackendService_constants.h>
+#include <gtest/gtest.h>
+#include <librdkafka/rdkafkacpp.h>
+
+#include <memory>
+
+#include "common/status.h"
+#include "load/routine_load/data_consumer.h"
+
+namespace doris {
+
+TEST(KafkaPropertiesTest, RejectExecutableAndUnknownProperties) {
+    std::unique_ptr<RdKafka::Conf> conf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
+    for (const auto* key :
+         {"sasl.kerberos.kinit.cmd", "sasl.kerberos.keytab",
+          "sasl.kerberos.min.time.before.relogin", "plugin.library.paths", "ssl.engine.location",
+          "ssl.engine.id", "ssl.providers", "unknown.future.property", "sasl.jaas.config"}) {
+        auto status = configure_kafka_security(*conf, {{key, "sensitive-value"}});
+        EXPECT_FALSE(status.ok()) << key;
+        EXPECT_NE(status.to_string().find("is not allowed in Routine Load"), std::string::npos);
+        EXPECT_EQ(status.to_string().find("sensitive-value"), std::string::npos);
+    }
+}
+
+TEST(KafkaPropertiesTest, RejectKerberosRefreshReenable) {
+    std::unique_ptr<RdKafka::Conf> conf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
+    for (const auto* value : {"1", "60000", "00", "FILE:123:checksum"}) {
+        EXPECT_FALSE(
+                configure_kafka_security(*conf, {{"sasl.kerberos.min.time.before.relogin", value}})
+                        .ok());
+    }
+}
+
+TEST(KafkaPropertiesTest, RequireManagedCertificateFiles) {
+    std::unique_ptr<RdKafka::Conf> conf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
+    for (const auto& key : g_BackendService_constants.KAFKA_FILE_PROPERTIES) {
+        for (const auto* value : {"/etc/ssl/private/key.pem", "relative.pem", "", "FILE:"}) {
+            EXPECT_FALSE(configure_kafka_security(*conf, {{key, value}}).ok()) << key;
+        }
+        EXPECT_TRUE(configure_kafka_security(*conf, {{key, "FILE:123:checksum"}}).ok()) << key;
+    }
+}
+
+TEST(KafkaPropertiesTest, DisableDefaultKerberosCommand) {
+    std::unique_ptr<RdKafka::Conf> conf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
+    ASSERT_TRUE(configure_kafka_security(*conf, {}).ok());
+    std::string value;
+    ASSERT_EQ(RdKafka::Conf::CONF_OK, conf->get("sasl.kerberos.min.time.before.relogin", value));
+    EXPECT_EQ("0", value);
+    // Principal values can contain shell metacharacters. They remain data because
+    // neither the initial kinit command nor its renewal timer can run.
+    ASSERT_TRUE(configure_kafka_security(*conf, {{"sasl.kerberos.principal", "user;unused"},
+                                                 {"security.protocol", "SASL_PLAINTEXT"},
+                                                 {"sasl.mechanisms", "GSSAPI"}})
+                        .ok());
+    ASSERT_EQ(RdKafka::Conf::CONF_OK, conf->get("sasl.kerberos.min.time.before.relogin", value));
+    EXPECT_EQ("0", value);
+}
+
+TEST(KafkaPropertiesTest, AcceptConsumerAndAwsProperties) {
+    std::unique_ptr<RdKafka::Conf> conf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
+    EXPECT_TRUE(configure_kafka_security(*conf, {{"group.id", "test-group"},
+                                                 {"sasl.kerberos.min.time.before.relogin", "0"},
+                                                 {"isolation.level", "read_committed"},
+                                                 {"fetch.max.bytes", "1048576"},
+                                                 {"security.protocol", "SASL_SSL"},
+                                                 {"sasl.mechanism", "OAUTHBEARER"},
+                                                 {"aws.region", "us-east-1"}})
+                        .ok());
+}
+
+TEST(KafkaPropertiesTest, ConsumerRejectsBeforeResolvingFilesOrCreatingClient) {
+    auto ctx = std::make_shared<StreamLoadContext>(nullptr);
+    TKafkaLoadInfo info;
+    info.__set_properties(
+            {{"https.ca.location", "FILE:123:checksum"}, {"plugin.library.paths", "unused.so"}});
+    ctx->kafka_info = std::make_unique<KafkaLoadInfo>(info);
+    KafkaDataConsumer consumer(ctx);
+    // No ExecEnv or Kafka broker: rejection must precede FILE resolution and
+    // client creation even when an allowed property is present in the same map.
+    auto status = consumer.init(ctx);
+    EXPECT_FALSE(status.ok());
+    EXPECT_NE(status.to_string().find("plugin.library.paths"), std::string::npos);
+}
+
+} // namespace doris

fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java

@@ -2036,7 +2036,7 @@ public void gsonPostProcess() throws IOException {
                         // fall through; let validate() surface the real error
                     }
                 }
-                createRoutineLoadInfo.validate(ctx);
+                createRoutineLoadInfo.validateForRecovery(ctx);
                 setRoutineLoadDesc(createRoutineLoadInfo.getRoutineLoadDesc());
             } finally {
                 ctx.cleanup();

fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaDataSourceProperties.java

@@ -23,6 +23,7 @@
 import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.load.routineload.AbstractDataSourceProperties;
 import org.apache.doris.load.routineload.LoadDataSourceType;
+import org.apache.doris.thrift.BackendServiceConstants;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableSet;
@@ -253,10 +254,33 @@ private void analyzeCustomProperties() throws AnalysisException {
             }
         }
 
+        validateCustomProperties(customKafkaProperties);
+
         // Validate AWS MSK IAM authentication configuration if present
         validateAwsMskIamConfig();
     }
 
+    // Also validate persisted jobs before converting FILE references or making BE metadata requests.
+    public static void validateCustomProperties(Map<String, String> properties) throws AnalysisException {
+        for (Map.Entry<String, String> property : properties.entrySet()) {
+            String key = property.getKey();
+            // Only zero is safe: it prevents the default kinit command from executing.
+            if (key.equals("sasl.kerberos.min.time.before.relogin") && !property.getValue().equals("0")) {
+                throw new AnalysisException("Kafka property '" + key + "' is not allowed in Routine Load"
+                        + " with a nonzero value; Kerberos tickets must be managed externally");
+            }
+            if (BackendServiceConstants.KAFKA_FILE_PROPERTIES.contains(key)) {
+                if (!property.getValue().startsWith("FILE:") || property.getValue().length() == 5) {
+                    throw new AnalysisException("Kafka property '" + key
+                            + "' must reference a file registered with CREATE FILE using FILE:<name>");
+                }
+            } else if (!BackendServiceConstants.KAFKA_ALLOWED_PROPERTIES.contains(key)
+                    && !BackendServiceConstants.KAFKA_DORIS_PROPERTIES.contains(key)) {
+                throw new AnalysisException("Kafka property '" + key + "' is not allowed in Routine Load");
+            }
+        }
+    }
+
     private void validateAwsMskIamConfig() throws AnalysisException {
         String securityProtocol = customKafkaProperties.get(KafkaConfiguration.SECURITY_PROTOCOL);
         String saslMechanism = customKafkaProperties.get(KafkaConfiguration.SASL_MECHANISM);