diff --git a/be/src/load/routine_load/data_consumer.cpp b/be/src/load/routine_load/data_consumer.cpp index b305ba6c9aae38..7bee3077b07618 100644 --- a/be/src/load/routine_load/data_consumer.cpp +++ b/be/src/load/routine_load/data_consumer.cpp @@ -18,6 +18,7 @@ #include "load/routine_load/data_consumer.h" #include +#include #include #include #include @@ -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 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 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; } diff --git a/be/src/load/routine_load/kafka_properties.cpp b/be/src/load/routine_load/kafka_properties.cpp new file mode 100644 index 00000000000000..85899e05be923d --- /dev/null +++ b/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 +#include + +#include "common/status.h" +#include "util/string_util.h" + +namespace doris { + +Status configure_kafka_security(RdKafka::Conf& conf, + const std::map& 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 diff --git a/be/src/load/routine_load/kafka_properties.h b/be/src/load/routine_load/kafka_properties.h new file mode 100644 index 00000000000000..94f2adc9e71697 --- /dev/null +++ b/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 +#include + +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& properties); +} // namespace doris diff --git a/be/test/load/routine_load/kafka_properties_test.cpp b/be/test/load/routine_load/kafka_properties_test.cpp new file mode 100644 index 00000000000000..15f314cd976e41 --- /dev/null +++ b/be/test/load/routine_load/kafka_properties_test.cpp @@ -0,0 +1,121 @@ +// 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 +#include +#include + +#include + +#include "common/status.h" +#include "load/routine_load/data_consumer.h" + +namespace doris { + +TEST(KafkaPropertiesTest, RejectExecutableAndUnknownProperties) { + std::unique_ptr 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 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 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 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 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, AcceptOidcCredentialAliases) { + std::unique_ptr conf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL)); + ASSERT_TRUE(configure_kafka_security( + *conf, + {{"security.protocol", "SASL_SSL"}, + {"sasl.mechanism", "OAUTHBEARER"}, + {"sasl.oauthbearer.method", "oidc"}, + {"sasl.oauthbearer.client.credentials.client.id", "test-client"}, + {"sasl.oauthbearer.client.credentials.client.secret", "test-secret"}, + {"sasl.oauthbearer.token.endpoint.url", "https://issuer.example/token"}}) + .ok()); + std::string value; + ASSERT_EQ(RdKafka::Conf::CONF_OK, conf->get("sasl.kerberos.min.time.before.relogin", value)); + EXPECT_EQ("0", value); +} + +TEST(KafkaPropertiesTest, ConsumerRejectsBeforeResolvingFilesOrCreatingClient) { + auto ctx = std::make_shared(nullptr); + TKafkaLoadInfo info; + info.__set_properties( + {{"https.ca.location", "FILE:123:checksum"}, {"plugin.library.paths", "unused.so"}}); + ctx->kafka_info = std::make_unique(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 diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java index 4cd4434a73e6c3..d617ad5d1546f2 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java +++ b/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(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaDataSourceProperties.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaDataSourceProperties.java index 9cab113563da61..3b7d14364e115c 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaDataSourceProperties.java +++ b/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 properties) throws AnalysisException { + for (Map.Entry 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:"); + } + } 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); diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java index 0570b356edf133..a5d2dea9782e24 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java +++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java @@ -27,6 +27,7 @@ import org.apache.doris.catalog.RandomDistributionInfo; import org.apache.doris.cloud.proto.Cloud; import org.apache.doris.cloud.rpc.MetaServiceProxy; +import org.apache.doris.common.AnalysisException; import org.apache.doris.common.Config; import org.apache.doris.common.DdlException; import org.apache.doris.common.InternalErrorCode; @@ -208,30 +209,48 @@ public void prepare() throws UserException { } private void convertCustomProperties(boolean rebuild) throws DdlException { - if (customProperties.isEmpty()) { - return; - } - + validateCustomProperties(customProperties); if (!rebuild && !convertedCustomProperties.isEmpty()) { return; } + setConvertedCustomProperties(resolveCustomProperties(customProperties)); + } - if (rebuild) { - convertedCustomProperties.clear(); + private void validateCustomProperties(Map properties) throws DdlException { + // Replay can restore properties accepted by an older FE. Reject them before + // metadata requests, including during a rolling upgrade with older BEs. + try { + KafkaDataSourceProperties.validateCustomProperties(properties); + } catch (AnalysisException e) { + throw new DdlException(e.getMessage(), e); } + } - SmallFileMgr smallFileMgr = Env.getCurrentEnv().getSmallFileMgr(); - for (Map.Entry entry : customProperties.entrySet()) { + private Map resolveCustomProperties(Map properties) throws DdlException { + // Publish only a complete conversion. A failed FILE lookup must not leave + // a partial cache that skips validation/conversion on the next metadata request. + Map convertedProperties = Maps.newHashMap(); + for (Map.Entry entry : properties.entrySet()) { if (entry.getValue().startsWith("FILE:")) { // convert FILE:file_name -> FILE:file_id:md5 String file = entry.getValue().substring(entry.getValue().indexOf(":") + 1); + SmallFileMgr smallFileMgr = Env.getCurrentEnv().getSmallFileMgr(); SmallFile smallFile = smallFileMgr.getSmallFile(dbId, KAFKA_FILE_CATALOG, file, true); - convertedCustomProperties.put(entry.getKey(), "FILE:" + smallFile.id + ":" + smallFile.md5); + convertedProperties.put(entry.getKey(), "FILE:" + smallFile.id + ":" + smallFile.md5); } else { - convertedCustomProperties.put(entry.getKey(), entry.getValue()); + convertedProperties.put(entry.getKey(), entry.getValue()); } } + // Older BEs still use librdkafka's default shell command. Send the safe + // setting explicitly so new FE -> old BE also disables ticket execution. + convertedProperties.put("sasl.kerberos.min.time.before.relogin", "0"); + return convertedProperties; + } + + private void setConvertedCustomProperties(Map convertedProperties) { + convertedCustomProperties = convertedProperties; + // This is mainly for compatibility. In the previous version, we directly obtained the value of the // KAFKA_DEFAULT_OFFSETS attribute. In the new version, we support date time as the value of // KAFKA_DEFAULT_OFFSETS, and this attribute will be converted into a timestamp during the analyzing phase, @@ -240,6 +259,9 @@ private void convertCustomProperties(boolean rebuild) throws DdlException { if (convertedCustomProperties.containsKey(KafkaConfiguration.KAFKA_ORIGIN_DEFAULT_OFFSETS.getName())) { kafkaDefaultOffSet = convertedCustomProperties .remove(KafkaConfiguration.KAFKA_ORIGIN_DEFAULT_OFFSETS.getName()); + // Both entries describe the same default. Do not leave the millis value + // behind to overwrite the datetime when an ALTER reuses this cache. + convertedCustomProperties.remove(KafkaConfiguration.KAFKA_DEFAULT_OFFSETS.getName()); return; } if (convertedCustomProperties.containsKey(KafkaConfiguration.KAFKA_DEFAULT_OFFSETS.getName())) { @@ -790,7 +812,9 @@ public void modifyProperties(AlterRoutineLoadCommand command) throws UserExcepti throw new DdlException("Only supports modification of PAUSED jobs"); } - modifyPropertiesInternal(jobProperties, dataSourceProperties); + Map convertedProperties = prepareModifyProperties(dataSourceProperties); + applyProperties(jobProperties, dataSourceProperties); + setConvertedCustomProperties(convertedProperties); AlterRoutineLoadJobOperationLog log = new AlterRoutineLoadJobOperationLog(this.id, jobProperties, dataSourceProperties); @@ -805,21 +829,41 @@ private void convertOffset(KafkaDataSourceProperties dataSourceProperties) throw if (partitionOffsets.isEmpty()) { return; } + Map properties; + String brokerListSnapshot; + String topicSnapshot; + writeLock(); + try { + // An ALTER can repair properties accepted by an older FE. Validate and + // resolve the merged map before lookup without publishing it to the job. + Map mergedProperties = Maps.newHashMap(customProperties); + mergedProperties.putAll(dataSourceProperties.getCustomKafkaProperties()); + validateCustomProperties(mergedProperties); + properties = resolveCustomProperties(mergedProperties); + properties.remove(KafkaConfiguration.KAFKA_ORIGIN_DEFAULT_OFFSETS.getName()); + properties.remove(KafkaConfiguration.KAFKA_DEFAULT_OFFSETS.getName()); + brokerListSnapshot = brokerList; + topicSnapshot = topic; + } finally { + writeUnlock(); + } String computeGroupName = getComputeGroupName(); List> newOffsets; if (dataSourceProperties.isOffsetsForTimes()) { newOffsets = KafkaUtil.getOffsetsForTimes( - brokerList, topic, convertedCustomProperties, partitionOffsets, computeGroupName); + brokerListSnapshot, topicSnapshot, properties, partitionOffsets, computeGroupName); } else { newOffsets = KafkaUtil.getRealOffsets( - brokerList, topic, convertedCustomProperties, partitionOffsets, computeGroupName); + brokerListSnapshot, topicSnapshot, properties, partitionOffsets, computeGroupName); } dataSourceProperties.setKafkaPartitionOffsets(newOffsets); } - private void modifyPropertiesInternal(Map jobProperties, - KafkaDataSourceProperties dataSourceProperties) + // Only a live ALTER validates Kafka properties, resolves files and resets cloud progress. + // Journal replay must apply the already committed transition without these effects. + private Map prepareModifyProperties(KafkaDataSourceProperties dataSourceProperties) throws UserException { + Map convertedProperties = convertedCustomProperties; if (null != dataSourceProperties) { List> kafkaPartitionOffsets = Lists.newArrayList(); Map customKafkaProperties = Maps.newHashMap(); @@ -829,10 +873,13 @@ private void modifyPropertiesInternal(Map jobProperties, customKafkaProperties = dataSourceProperties.getCustomKafkaProperties(); } - // convertCustomProperties and check partitions before reset progress to make modify operation atomic + // Prepare the merged properties without changing durable job state. For + // example, a missing CREATE FILE must not partially apply the ALTER. if (!customKafkaProperties.isEmpty()) { - this.customProperties.putAll(customKafkaProperties); - convertCustomProperties(true); + Map properties = Maps.newHashMap(customProperties); + properties.putAll(customKafkaProperties); + validateCustomProperties(properties); + convertedProperties = resolveCustomProperties(properties); } if (!kafkaPartitionOffsets.isEmpty()) { @@ -858,7 +905,22 @@ private void modifyPropertiesInternal(Map jobProperties, } resetCloudProgress(builder); } + } + return convertedProperties; + } + private void applyProperties(Map jobProperties, + KafkaDataSourceProperties dataSourceProperties) + throws UserException { + if (null != dataSourceProperties) { + List> kafkaPartitionOffsets = Lists.newArrayList(); + if (MapUtils.isNotEmpty(dataSourceProperties.getOriginalDataSourceProperties())) { + kafkaPartitionOffsets = dataSourceProperties.getKafkaPartitionOffsets(); + this.customProperties.putAll(dataSourceProperties.getCustomKafkaProperties()); + } + // A historical ALTER may contain a property rejected by today's policy. + // Restore all fields (including topic/offsets), then validate before use. + convertedCustomProperties = Maps.newHashMap(); // It is necessary to reset the Kafka progress cache if topic change, // and should reset cache before modifying partition offset. if (!Strings.isNullOrEmpty(dataSourceProperties.getTopic())) { @@ -919,7 +981,7 @@ private void resetCloudProgress(Cloud.ResetRLProgressRequest.Builder builder) th @Override public void replayModifyProperties(AlterRoutineLoadJobOperationLog log) { try { - modifyPropertiesInternal(log.getJobProperties(), (KafkaDataSourceProperties) log.getDataSourceProperties()); + applyProperties(log.getJobProperties(), (KafkaDataSourceProperties) log.getDataSourceProperties()); } catch (UserException e) { // should not happen LOG.error("failed to replay modify kafka routine load job: {}", id, e); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java index e8a75c0299b4d6..053ce9ae989400 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java @@ -378,10 +378,22 @@ public String getWorkloadGroupName() { return workloadGroupName; } + /** + * Rebuild load expressions without validating stale data source properties in the original CREATE SQL. + * The job's current effective data source properties are validated before runtime use. + */ + public void validateForRecovery(ConnectContext ctx) throws UserException { + validate(ctx, false); + } + /** * analyze create table info */ public void validate(ConnectContext ctx) throws UserException { + validate(ctx, true); + } + + private void validate(ConnectContext ctx, boolean validateDataSource) throws UserException { // check dbName and tableName checkDBTable(ctx); // check name @@ -398,7 +410,9 @@ public void validate(ConnectContext ctx) throws UserException { // check routine load job properties include desired concurrent number etc. checkJobProperties(); // check data source properties - checkDataSourceProperties(); + if (validateDataSource) { + checkDataSourceProperties(); + } // analyze merge type if (routineLoadDesc != null) { if (mergeType != LoadTask.MergeType.MERGE && routineLoadDesc.getDeleteCondition() != null) { diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaAwsMskIamAuthTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaAwsMskIamAuthTest.java index 18764b97328574..f10b04a0202eb2 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaAwsMskIamAuthTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaAwsMskIamAuthTest.java @@ -255,9 +255,9 @@ public void testPlaintextConfigWithoutSasl() throws UserException { public void testSslConfigWithoutSasl() throws UserException { // Test SSL configuration without SASL dataSourceProperties.put("property.security.protocol", "SSL"); - dataSourceProperties.put("property.ssl.ca.location", "/path/to/ca-cert"); - dataSourceProperties.put("property.ssl.certificate.location", "/path/to/client-cert"); - dataSourceProperties.put("property.ssl.key.location", "/path/to/client-key"); + dataSourceProperties.put("property.ssl.ca.location", "FILE:ca-cert"); + dataSourceProperties.put("property.ssl.certificate.location", "FILE:client-cert"); + dataSourceProperties.put("property.ssl.key.location", "FILE:client-key"); KafkaDataSourceProperties props = new KafkaDataSourceProperties(dataSourceProperties); props.setTimezone("UTC"); diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaPropertyValidationTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaPropertyValidationTest.java new file mode 100644 index 00000000000000..a25b38a4cb75e6 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaPropertyValidationTest.java @@ -0,0 +1,612 @@ +// 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. + +package org.apache.doris.load.routineload; + +import org.apache.doris.analysis.UserIdentity; +import org.apache.doris.catalog.Database; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.Table; +import org.apache.doris.cloud.rpc.MetaServiceProxy; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.Config; +import org.apache.doris.common.DdlException; +import org.apache.doris.common.LoadException; +import org.apache.doris.common.Pair; +import org.apache.doris.common.jmockit.Deencapsulation; +import org.apache.doris.common.util.SmallFileMgr; +import org.apache.doris.common.util.SmallFileMgr.SmallFile; +import org.apache.doris.common.util.TimeUtils; +import org.apache.doris.datasource.CatalogMgr; +import org.apache.doris.datasource.InternalCatalog; +import org.apache.doris.datasource.kafka.KafkaUtil; +import org.apache.doris.load.routineload.kafka.KafkaDataSourceProperties; +import org.apache.doris.load.routineload.kafka.KafkaProgress; +import org.apache.doris.load.routineload.kafka.KafkaRoutineLoadJob; +import org.apache.doris.nereids.parser.NereidsParser; +import org.apache.doris.nereids.trees.plans.commands.AlterRoutineLoadCommand; +import org.apache.doris.nereids.trees.plans.commands.load.CreateRoutineLoadCommand; +import org.apache.doris.persist.AlterRoutineLoadJobOperationLog; +import org.apache.doris.persist.EditLog; +import org.apache.doris.persist.RoutineLoadOperation; +import org.apache.doris.persist.gson.GsonUtils; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.OriginStatement; +import org.apache.doris.thrift.BackendServiceConstants; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +public class KafkaPropertyValidationTest { + private KafkaDataSourceProperties properties(String key, String value, boolean alter) { + Map source = new HashMap<>(); + if (!alter) { + source.put("kafka_broker_list", "127.0.0.1:9092"); + source.put("kafka_topic", "test_topic"); + } + source.put("property." + key, value); + KafkaDataSourceProperties properties = new KafkaDataSourceProperties(source); + properties.setAlter(alter); + properties.setTimezone("UTC"); + return properties; + } + + @Test + public void testCreateAndAlterRejectExecutableAndUnknownProperties() { + for (boolean alter : new boolean[] {false, true}) { + for (String key : List.of("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", + "security.protocol\0suffix")) { + AnalysisException error = Assertions.assertThrows(AnalysisException.class, + () -> properties(key, "sensitive-value", alter).analyze()); + Assertions.assertTrue(error.getMessage().contains("is not allowed in Routine Load"), key); + Assertions.assertFalse(error.getMessage().contains("sensitive-value")); + } + } + } + + @Test + public void testRejectKerberosRefreshReenable() { + for (boolean alter : new boolean[] {false, true}) { + for (String value : List.of("1", "60000", "00", "0\0suffix", "FILE:interval")) { + Assertions.assertThrows(AnalysisException.class, + () -> properties("sasl.kerberos.min.time.before.relogin", value, alter).analyze()); + } + } + } + + @Test + public void testCertificateFilesRequireManagedReferences() throws Exception { + for (boolean alter : new boolean[] {false, true}) { + for (String key : BackendServiceConstants.KAFKA_FILE_PROPERTIES) { + for (String value : List.of("/etc/ssl/private/key.pem", "relative.pem", "", "FILE:")) { + AnalysisException error = Assertions.assertThrows(AnalysisException.class, + () -> properties(key, value, alter).analyze()); + Assertions.assertTrue(error.getMessage().contains("CREATE FILE")); + } + KafkaDataSourceProperties properties = properties(key, "FILE:client.pem", alter); + properties.analyze(); + Assertions.assertEquals("FILE:client.pem", properties.getCustomKafkaProperties().get(key)); + } + } + } + + @Test + public void testNormalConsumerPropertiesAndDefaultOffsets() throws Exception { + Map accepted = Map.of( + "group.id", "routine-load-group", + "client.id", "routine-load-client", + "security.protocol", "SASL_SSL", + "sasl.mechanisms", "SCRAM-SHA-256", + "sasl.username", "user", + "sasl.password", "secret", + "isolation.level", "read_committed", + "fetch.max.bytes", "1048576", + "kafka_default_offsets", "OFFSET_BEGINNING"); + for (boolean alter : new boolean[] {false, true}) { + for (Map.Entry property : accepted.entrySet()) { + properties(property.getKey(), property.getValue(), alter).analyze(); + } + } + } + + @Test + public void testOidcCredentialAliasesCreateAndAlter() throws Exception { + for (boolean alter : new boolean[] {false, true}) { + Map source = new HashMap<>(); + source.put("kafka_broker_list", "127.0.0.1:9092"); + source.put("kafka_topic", "test_topic"); + oidcAliasProperties().forEach((key, value) -> source.put("property." + key, value)); + KafkaDataSourceProperties properties = new KafkaDataSourceProperties(source); + properties.setAlter(alter); + properties.setTimezone("UTC"); + properties.analyze(); + oidcAliasProperties().forEach((key, value) -> + Assertions.assertEquals(value, properties.getCustomKafkaProperties().get(key))); + } + } + + private Map oidcAliasProperties() { + return Map.of("security.protocol", "SASL_SSL", + "sasl.mechanism", "OAUTHBEARER", + "sasl.oauthbearer.method", "oidc", + "sasl.oauthbearer.client.credentials.client.id", "test-client", + "sasl.oauthbearer.client.credentials.client.secret", "test-secret", + "sasl.oauthbearer.token.endpoint.url", "https://issuer.example/token"); + } + + @Test + public void testRecoverAndAlterJobWithOidcCredentialAliases() throws Exception { + Env env = Mockito.mock(Env.class); + CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class); + InternalCatalog catalog = Mockito.mock(InternalCatalog.class); + Database db = Mockito.mock(Database.class); + Table table = Mockito.mock(Table.class); + EditLog editLog = Mockito.mock(EditLog.class); + String originalSql = "CREATE ROUTINE LOAD test_db.test_job ON test_table COLUMNS(id) " + + "FROM KAFKA (\"kafka_broker_list\" = \"127.0.0.1:9092\", \"kafka_topic\" = \"test_topic\", " + + "\"property.security.protocol\" = \"SASL_SSL\", \"property.sasl.mechanism\" = \"OAUTHBEARER\", " + + "\"property.sasl.oauthbearer.method\" = \"oidc\", " + + "\"property.sasl.oauthbearer.client.credentials.client.id\" = \"test-client\", " + + "\"property.sasl.oauthbearer.client.credentials.client.secret\" = \"test-secret\", " + + "\"property.sasl.oauthbearer.token.endpoint.url\" = \"https://issuer.example/token\")"; + + try (MockedStatic envStatic = Mockito.mockStatic(Env.class)) { + envStatic.when(Env::getCurrentEnv).thenReturn(env); + envStatic.when(Env::getCurrentInternalCatalog).thenReturn(catalog); + Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr); + Mockito.when(catalogMgr.getCatalog(InternalCatalog.INTERNAL_CATALOG_NAME)).thenReturn(catalog); + Mockito.when(env.getInternalCatalog()).thenReturn(catalog); + Mockito.when(env.getEditLog()).thenReturn(editLog); + Mockito.when(catalog.getName()).thenReturn(InternalCatalog.INTERNAL_CATALOG_NAME); + Mockito.doReturn(Optional.of(db)).when(catalog).getDb(1L); + Mockito.doReturn(Optional.of(db)).when(catalog).getDb("test_db"); + Mockito.doReturn(db).when(catalog).getDbOrAnalysisException("test_db"); + Mockito.when(db.getId()).thenReturn(1L); + Mockito.when(db.getName()).thenReturn("test_db"); + Mockito.doReturn(Optional.of(table)).when(db).getTable(2L); + Mockito.doReturn(table).when(db).getTableOrAnalysisException("test_table"); + Mockito.when(table.getName()).thenReturn("test_table"); + + KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 1L, 2L, + "127.0.0.1:9092", "test_topic", UserIdentity.ADMIN); + job.setOrigStmt(new OriginStatement(originalSql, 0)); + Deencapsulation.setField(job, "state", RoutineLoadJob.JobState.PAUSED); + Deencapsulation.setField(job, "customProperties", new HashMap<>(oidcAliasProperties())); + KafkaRoutineLoadJob restored = (KafkaRoutineLoadJob) GsonUtils.GSON.fromJson( + GsonUtils.GSON.toJson(job), RoutineLoadJob.class); + Assertions.assertEquals(RoutineLoadJob.JobState.PAUSED, restored.getState()); + Assertions.assertEquals(-1L, restored.getEndTimestamp()); + restored.prepare(); + oidcAliasProperties().forEach((key, value) -> + Assertions.assertEquals(value, restored.getConvertedCustomProperties().get(key))); + + // Adding short names must not make the retained historical aliases fail validation. + KafkaDataSourceProperties update = properties("sasl.oauthbearer.client.secret", "test-secret", true); + update.analyze(); + AlterRoutineLoadCommand command = Mockito.mock(AlterRoutineLoadCommand.class); + Mockito.when(command.getAnalyzedJobProperties()).thenReturn(Map.of()); + Mockito.when(command.getDataSourceProperties()).thenReturn(update); + restored.modifyProperties(command); + Assertions.assertEquals("test-secret", restored.getConvertedCustomProperties() + .get("sasl.oauthbearer.client.credentials.client.secret")); + Assertions.assertEquals("test-secret", restored.getConvertedCustomProperties() + .get("sasl.oauthbearer.client.secret")); + Assertions.assertEquals("******", restored.getCustomProperties() + .get("property.sasl.oauthbearer.client.credentials.client.secret")); + Assertions.assertFalse(restored.customPropertiesJsonToString().contains("test-secret")); + Mockito.verify(editLog).logAlterRoutineLoadJob(Mockito.any(AlterRoutineLoadJobOperationLog.class)); + } + } + + @Test + public void testAwsPropertiesWithAndWithoutPrefix() throws Exception { + for (String prefix : List.of("", "property.")) { + Map source = new HashMap<>(); + source.put("kafka_broker_list", "127.0.0.1:9098"); + source.put("kafka_topic", "test_topic"); + source.put("property.security.protocol", "SASL_SSL"); + source.put("property.sasl.mechanism", "OAUTHBEARER"); + source.put(prefix + "aws.region", "us-east-1"); + source.put(prefix + "aws.role_arn", "arn:aws:iam::123456789012:role/test"); + KafkaDataSourceProperties properties = new KafkaDataSourceProperties(source); + properties.setTimezone("UTC"); + properties.analyze(); + Assertions.assertEquals("us-east-1", properties.getCustomKafkaProperties().get("aws.region")); + } + } + + @Test + public void testDisableKerberosCommandForOlderBackends() throws Exception { + properties("sasl.kerberos.min.time.before.relogin", "0", false).analyze(); + for (Map custom : List.of(Map.of(), + Map.of("group.id", "test", "sasl.kerberos.principal", "user;unused"), + Map.of("kafka_default_offsets", "OFFSET_BEGINNING", + "kafka_origin_default_offsets", "OFFSET_BEGINNING"))) { + KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 1L, 1L, + "127.0.0.1:9092", "test_topic", UserIdentity.ADMIN); + Deencapsulation.setField(job, "customProperties", custom); + Deencapsulation.invoke(job, "convertCustomProperties", true); + Assertions.assertEquals("0", job.getConvertedCustomProperties() + .get("sasl.kerberos.min.time.before.relogin")); + } + } + + @Test + public void testPersistedPropertiesRejectedBeforeMetadataRequests() { + KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 1L, 1L, + "127.0.0.1:9092", "test_topic", UserIdentity.ADMIN); + Deencapsulation.setField(job, "customProperties", Map.of("plugin.library.paths", "unused.so")); + // Even a previously populated conversion cache must not bypass validation. + Deencapsulation.setField(job, "convertedCustomProperties", Map.of("group.id", "cached")); + DdlException error = Assertions.assertThrows(DdlException.class, + () -> Deencapsulation.invoke(job, "convertCustomProperties", false)); + Assertions.assertTrue(error.getMessage().contains("is not allowed in Routine Load")); + } + + // Build an already analyzed, persisted ALTER from an older FE without applying + // today's CREATE/ALTER admission rules to the historical input. + private AlterRoutineLoadJobOperationLog historicalAlter(Map custom, + String topic, String brokers, List> offsets) { + Map source = new HashMap<>(); + custom.forEach((key, value) -> source.put("property." + key, value)); + source.put("kafka_topic", topic); + source.put("kafka_broker_list", brokers); + KafkaDataSourceProperties properties = new KafkaDataSourceProperties(source); + properties.setAlter(true); + Deencapsulation.setField(properties, "dataSourceType", "KAFKA"); + Deencapsulation.setField(properties, "customKafkaProperties", custom); + Deencapsulation.setField(properties, "topic", topic); + Deencapsulation.setField(properties, "brokerList", brokers); + properties.setKafkaPartitionOffsets(offsets); + AlterRoutineLoadJobOperationLog log = new AlterRoutineLoadJobOperationLog(1L, + Map.of("max_batch_rows", "2000000"), properties); + return GsonUtils.GSON.fromJson(GsonUtils.GSON.toJson(log), AlterRoutineLoadJobOperationLog.class); + } + + @Test + public void testReplayHistoricalAlterRestoresCompleteState() throws Exception { + for (boolean cloud : new boolean[] {false, true}) { + try (MockedStatic config = Mockito.mockStatic(Config.class); + MockedStatic env = Mockito.mockStatic(Env.class); + MockedStatic metaService = Mockito.mockStatic(MetaServiceProxy.class); + MockedStatic kafka = Mockito.mockStatic(KafkaUtil.class)) { + config.when(Config::isCloudMode).thenReturn(cloud); + KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 1L, 2L, + "127.0.0.1:9092", "topic_a", UserIdentity.ADMIN); + Deencapsulation.setField(job, "progress", new KafkaProgress(Map.of(0, 5L, 1, 6L))); + job.prepare(); + + AlterRoutineLoadJobOperationLog log = historicalAlter( + Map.of("sasl.kerberos.min.time.before.relogin", "60000"), + "topic_b", "127.0.0.2:9092", List.of(Pair.of(0, 100L))); + job.replayModifyProperties(log); + + Assertions.assertEquals("topic_b", job.getTopic()); + Assertions.assertEquals("127.0.0.2:9092", job.getBrokerList()); + Assertions.assertEquals(Map.of(0, 100L), + ((KafkaProgress) job.getProgress()).getOffsetByPartition()); + Assertions.assertEquals(2000000L, job.getMaxBatchRows()); + Assertions.assertEquals("60000", job.getCustomProperties() + .get("property.sasl.kerberos.min.time.before.relogin")); + Assertions.assertTrue(job.getConvertedCustomProperties().isEmpty()); + + DdlException error = Assertions.assertThrows(DdlException.class, job::prepare); + Assertions.assertTrue(error.getMessage().contains("Kerberos tickets must be managed externally")); + Assertions.assertThrows(DdlException.class, + () -> Deencapsulation.invoke(job, "getAllKafkaPartitions")); + Assertions.assertThrows(DdlException.class, + () -> Deencapsulation.invoke(job, "convertOffset", log.getDataSourceProperties())); + + // A later committed repair must retain topic B and its offsets. + job.replayModifyProperties(historicalAlter( + Map.of("sasl.kerberos.min.time.before.relogin", "0"), "", "", List.of())); + job.prepare(); + Assertions.assertEquals("topic_b", job.getTopic()); + Assertions.assertEquals("127.0.0.2:9092", job.getBrokerList()); + Assertions.assertEquals(Map.of(0, 100L), + ((KafkaProgress) job.getProgress()).getOffsetByPartition()); + Assertions.assertEquals("0", job.getConvertedCustomProperties() + .get("sasl.kerberos.min.time.before.relogin")); + env.verifyNoInteractions(); + metaService.verifyNoInteractions(); + kafka.verifyNoInteractions(); + } + } + } + + @Test + public void testReplayDefersManagedFileResolution() throws Exception { + Env env = Mockito.mock(Env.class); + SmallFileMgr smallFileMgr = Mockito.mock(SmallFileMgr.class); + try (MockedStatic envStatic = Mockito.mockStatic(Env.class)) { + envStatic.when(Env::getCurrentEnv).thenReturn(env); + Mockito.when(env.getSmallFileMgr()).thenReturn(smallFileMgr); + Mockito.when(smallFileMgr.getSmallFile(1L, "kafka", "missing.pem", true)) + .thenThrow(new DdlException("missing managed file")); + KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 1L, 2L, + "127.0.0.1:9092", "topic_a", UserIdentity.ADMIN); + + // Partition 3 need not exist in the pre-replay progress. These offsets + // were checked when the ALTER was committed, not during recovery. + job.replayModifyProperties(historicalAlter(Map.of("ssl.ca.location", "FILE:missing.pem"), + "topic_b", "127.0.0.2:9092", List.of(Pair.of(3, 100L)))); + Assertions.assertEquals("topic_b", job.getTopic()); + Assertions.assertEquals(Map.of(3, 100L), + ((KafkaProgress) job.getProgress()).getOffsetByPartition()); + Assertions.assertEquals(2000000L, job.getMaxBatchRows()); + envStatic.verifyNoInteractions(); + Mockito.verifyNoInteractions(smallFileMgr); + + DdlException error = Assertions.assertThrows(DdlException.class, job::prepare); + Assertions.assertTrue(error.getMessage().contains("missing managed file")); + Assertions.assertTrue(job.getConvertedCustomProperties().isEmpty()); + } + } + + @Test + public void testSchedulerPausesReplayedJobWithRejectedProperties() throws Exception { + Env env = Mockito.mock(Env.class); + EditLog editLog = Mockito.mock(EditLog.class); + RoutineLoadManager manager = Mockito.mock(RoutineLoadManager.class); + try (MockedStatic envStatic = Mockito.mockStatic(Env.class); + MockedStatic kafka = Mockito.mockStatic(KafkaUtil.class)) { + envStatic.when(Env::getCurrentEnv).thenReturn(env); + Mockito.when(env.getEditLog()).thenReturn(editLog); + KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 1L, 2L, + "127.0.0.1:9092", "topic_a", UserIdentity.ADMIN); + job.replayModifyProperties(historicalAlter(Map.of("plugin.library.paths", "unused.so"), + "topic_b", "127.0.0.2:9092", List.of())); + Mockito.when(manager.getRoutineLoadJobByState(Mockito.anySet())).thenReturn(List.of(job)); + + Deencapsulation.invoke(new RoutineLoadScheduler(manager), "process"); + + Assertions.assertEquals(RoutineLoadJob.JobState.PAUSED, job.getState()); + Assertions.assertTrue(job.getPauseReason().getMsg().contains("plugin.library.paths")); + Assertions.assertEquals("topic_b", job.getTopic()); + kafka.verifyNoInteractions(); + Mockito.verify(editLog).logOpRoutineLoadJob(Mockito.any(RoutineLoadOperation.class)); + } + } + + @Test + public void testLiveAlterPublishesOnlyPreparedProperties() throws Exception { + Env env = Mockito.mock(Env.class); + SmallFileMgr smallFileMgr = Mockito.mock(SmallFileMgr.class); + try (MockedStatic envStatic = Mockito.mockStatic(Env.class)) { + envStatic.when(Env::getCurrentEnv).thenReturn(env); + Mockito.when(env.getSmallFileMgr()).thenReturn(smallFileMgr); + Mockito.when(smallFileMgr.getSmallFile(1L, "kafka", "missing.pem", true)) + .thenThrow(new DdlException("missing managed file")); + KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 1L, 2L, + "127.0.0.1:9092", "topic_a", UserIdentity.ADMIN); + Deencapsulation.setField(job, "state", RoutineLoadJob.JobState.PAUSED); + job.prepare(); + Map converted = new HashMap<>(job.getConvertedCustomProperties()); + AlterRoutineLoadJobOperationLog log = historicalAlter(Map.of("ssl.ca.location", "FILE:missing.pem"), + "topic_b", "127.0.0.2:9092", List.of()); + AlterRoutineLoadCommand command = Mockito.mock(AlterRoutineLoadCommand.class); + Mockito.when(command.getAnalyzedJobProperties()).thenReturn(log.getJobProperties()); + Mockito.when(command.getDataSourceProperties()).thenReturn(log.getDataSourceProperties()); + + Assertions.assertThrows(DdlException.class, () -> job.modifyProperties(command)); + Assertions.assertEquals("topic_a", job.getTopic()); + Assertions.assertEquals("127.0.0.1:9092", job.getBrokerList()); + Assertions.assertTrue(job.getCustomProperties().isEmpty()); + Assertions.assertEquals(converted, job.getConvertedCustomProperties()); + Mockito.verify(env, Mockito.never()).getEditLog(); + + KafkaDataSourceProperties safe = properties("sasl.kerberos.min.time.before.relogin", "0", true); + safe.analyze(); + Mockito.when(command.getDataSourceProperties()).thenReturn(safe); + EditLog editLog = Mockito.mock(EditLog.class); + Mockito.when(env.getEditLog()).thenReturn(editLog); + job.modifyProperties(command); + Assertions.assertEquals(Map.of("property.sasl.kerberos.min.time.before.relogin", "0", + "property.kafka_default_offsets", KafkaProgress.OFFSET_END), + job.getCustomProperties()); + Assertions.assertEquals("0", job.getConvertedCustomProperties() + .get("sasl.kerberos.min.time.before.relogin")); + Assertions.assertEquals(2000000L, job.getMaxBatchRows()); + Mockito.verify(editLog).logAlterRoutineLoadJob(Mockito.any(AlterRoutineLoadJobOperationLog.class)); + } + } + + @Test + public void testJobOnlyAlterPreservesDatetimeDefaultForNewPartitions() throws Exception { + Env env = Mockito.mock(Env.class); + EditLog editLog = Mockito.mock(EditLog.class); + try (MockedStatic envStatic = Mockito.mockStatic(Env.class); + MockedStatic kafka = Mockito.mockStatic(KafkaUtil.class)) { + envStatic.when(Env::getCurrentEnv).thenReturn(env); + Mockito.when(env.getEditLog()).thenReturn(editLog); + KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 1L, 2L, + "127.0.0.1:9092", "topic_a", UserIdentity.ADMIN); + String datetime = "2026-10-01 00:00:00"; + KafkaDataSourceProperties source = properties("kafka_default_offsets", datetime, false); + source.analyze(); + Deencapsulation.setField(job, "customProperties", source.getCustomKafkaProperties()); + Deencapsulation.setField(job, "state", RoutineLoadJob.JobState.PAUSED); + job.prepare(); + + AlterRoutineLoadCommand command = Mockito.mock(AlterRoutineLoadCommand.class); + Mockito.when(command.getAnalyzedJobProperties()).thenReturn(Map.of("max_batch_rows", "2000000")); + job.modifyProperties(command); + job.modifyProperties(command); + Assertions.assertEquals(datetime, Deencapsulation.getField(job, "kafkaDefaultOffSet")); + Assertions.assertFalse(job.getConvertedCustomProperties().containsKey("kafka_default_offsets")); + Assertions.assertFalse(job.getConvertedCustomProperties().containsKey("kafka_origin_default_offsets")); + + // New partitions are initialized before prepare() rebuilds the cache on resume. + long timestamp = TimeUtils.timeStringToLong(datetime, TimeUtils.getOrSystemTimeZone(job.getTimezone())); + List> requested = List.of(Pair.of(1, timestamp)); + List> expected = List.of(Pair.of(1, 42L)); + kafka.when(() -> KafkaUtil.getOffsetsForTimes("127.0.0.1:9092", "topic_a", + job.getConvertedCustomProperties(), requested, null)).thenReturn(expected); + List> offsets = Deencapsulation.invoke(job, + "getNewPartitionOffsetsFromDefaultOffset", List.of(1)); + Assertions.assertEquals(expected, offsets); + kafka.verify(() -> KafkaUtil.getOffsetsForTimes("127.0.0.1:9092", "topic_a", + job.getConvertedCustomProperties(), requested, null)); + kafka.verifyNoMoreInteractions(); + } + } + + @Test + public void testRepairCertificateAndOffsetsInSameAlter() throws Exception { + for (boolean offsetsForTimes : new boolean[] {false, true}) { + Env env = Mockito.mock(Env.class); + SmallFileMgr smallFileMgr = Mockito.mock(SmallFileMgr.class); + EditLog editLog = Mockito.mock(EditLog.class); + try (MockedStatic envStatic = Mockito.mockStatic(Env.class); + MockedStatic kafka = Mockito.mockStatic(KafkaUtil.class)) { + envStatic.when(Env::getCurrentEnv).thenReturn(env); + Mockito.when(env.getSmallFileMgr()).thenReturn(smallFileMgr); + Mockito.when(env.getEditLog()).thenReturn(editLog); + SmallFile ca = new SmallFile(1L, "kafka", "ca.pem", 3L, "", 0L, "checksum", true); + Mockito.when(smallFileMgr.getSmallFile(1L, "kafka", "ca.pem", true)).thenReturn(ca); + KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 1L, 2L, + "127.0.0.1:9092", "topic_a", UserIdentity.ADMIN); + Deencapsulation.setField(job, "customProperties", + new HashMap<>(Map.of("ssl.ca.location", "/etc/ca.pem"))); + Deencapsulation.setField(job, "progress", new KafkaProgress(Map.of(0, 5L))); + Deencapsulation.setField(job, "state", RoutineLoadJob.JobState.PAUSED); + + Map input = Map.of("property.ssl.ca.location", "FILE:ca.pem", + "kafka_partitions", "0", "kafka_offsets", offsetsForTimes ? "2026-10-01 00:00:00" : "100"); + KafkaDataSourceProperties source = new KafkaDataSourceProperties(input); + source.setAlter(true); + source.setTimezone("UTC"); + source.analyze(); + Map converted = Map.of("ssl.ca.location", "FILE:3:checksum", + "sasl.kerberos.min.time.before.relogin", "0"); + List> requested = source.getKafkaPartitionOffsets(); + List> expected = List.of(Pair.of(0, 100L)); + if (offsetsForTimes) { + kafka.when(() -> KafkaUtil.getOffsetsForTimes("127.0.0.1:9092", "topic_a", + converted, requested, null)).thenThrow(new LoadException("metadata unavailable")) + .thenReturn(expected); + } else { + kafka.when(() -> KafkaUtil.getRealOffsets("127.0.0.1:9092", "topic_a", + converted, requested, null)).thenThrow(new LoadException("metadata unavailable")) + .thenCallRealMethod(); + } + AlterRoutineLoadCommand command = Mockito.mock(AlterRoutineLoadCommand.class); + Mockito.when(command.getAnalyzedJobProperties()).thenReturn(Map.of()); + Mockito.when(command.getDataSourceProperties()).thenReturn(source); + + Assertions.assertThrows(LoadException.class, () -> job.modifyProperties(command)); + Assertions.assertEquals("/etc/ca.pem", job.getCustomProperties().get("property.ssl.ca.location")); + Assertions.assertTrue(job.getConvertedCustomProperties().isEmpty()); + Assertions.assertEquals(Map.of(0, 5L), ((KafkaProgress) job.getProgress()).getOffsetByPartition()); + Mockito.verifyNoInteractions(editLog); + + job.modifyProperties(command); + + Assertions.assertEquals("FILE:ca.pem", job.getCustomProperties().get("property.ssl.ca.location")); + Assertions.assertEquals(converted, job.getConvertedCustomProperties()); + Assertions.assertEquals(Map.of(0, 100L), ((KafkaProgress) job.getProgress()).getOffsetByPartition()); + if (offsetsForTimes) { + kafka.verify(() -> KafkaUtil.getOffsetsForTimes("127.0.0.1:9092", "topic_a", + converted, requested, null), Mockito.times(2)); + } else { + kafka.verify(() -> KafkaUtil.getRealOffsets("127.0.0.1:9092", "topic_a", + converted, requested, null), Mockito.times(2)); + } + kafka.verifyNoMoreInteractions(); + Mockito.verify(editLog).logAlterRoutineLoadJob(Mockito.any(AlterRoutineLoadJobOperationLog.class)); + } + } + } + + @Test + public void testRecoverMigratedCertificateProperties() throws Exception { + Env env = Mockito.mock(Env.class); + CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class); + InternalCatalog catalog = Mockito.mock(InternalCatalog.class); + Database db = Mockito.mock(Database.class); + Table table = Mockito.mock(Table.class); + SmallFileMgr smallFileMgr = Mockito.mock(SmallFileMgr.class); + String originalSql = "CREATE ROUTINE LOAD test_db.test_job ON test_table COLUMNS(id) " + + "FROM KAFKA (\"kafka_broker_list\" = \"127.0.0.1:9092\", " + + "\"kafka_topic\" = \"test_topic\", \"property.ssl.ca.location\" = \"/etc/ssl/ca.pem\")"; + + try (MockedStatic envStatic = Mockito.mockStatic(Env.class)) { + envStatic.when(Env::getCurrentEnv).thenReturn(env); + envStatic.when(Env::getCurrentInternalCatalog).thenReturn(catalog); + Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr); + Mockito.when(catalogMgr.getCatalog(InternalCatalog.INTERNAL_CATALOG_NAME)).thenReturn(catalog); + Mockito.when(env.getInternalCatalog()).thenReturn(catalog); + Mockito.when(env.getSmallFileMgr()).thenReturn(smallFileMgr); + Mockito.when(catalog.getName()).thenReturn(InternalCatalog.INTERNAL_CATALOG_NAME); + Mockito.doReturn(Optional.of(db)).when(catalog).getDb(1L); + Mockito.doReturn(Optional.of(db)).when(catalog).getDb("test_db"); + Mockito.doReturn(db).when(catalog).getDbOrAnalysisException("test_db"); + Mockito.when(db.getId()).thenReturn(1L); + Mockito.when(db.getName()).thenReturn("test_db"); + Mockito.doReturn(Optional.of(table)).when(db).getTable(2L); + Mockito.doReturn(table).when(db).getTableOrAnalysisException("test_table"); + Mockito.when(table.getName()).thenReturn("test_table"); + SmallFile ca = new SmallFile(1L, "kafka", "ca.pem", 3L, "", 0L, "checksum", true); + Mockito.when(smallFileMgr.getSmallFile(1L, "kafka", "ca.pem", true)).thenReturn(ca); + + // ALTER changes the effective map, but the persisted CREATE SQL still contains the old path. + KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 1L, 2L, + "127.0.0.1:9092", "test_topic", UserIdentity.ADMIN); + job.setOrigStmt(new OriginStatement(originalSql, 0)); + Deencapsulation.setField(job, "state", RoutineLoadJob.JobState.PAUSED); + Deencapsulation.setField(job, "customProperties", Map.of("ssl.ca.location", "FILE:ca.pem")); + KafkaRoutineLoadJob restored = (KafkaRoutineLoadJob) GsonUtils.GSON.fromJson( + GsonUtils.GSON.toJson(job), RoutineLoadJob.class); + + Assertions.assertEquals(RoutineLoadJob.JobState.PAUSED, restored.getState()); + Assertions.assertEquals(-1L, restored.getEndTimestamp()); + Assertions.assertEquals("id", restored.getColumnExprDescs().descs.get(0).getColumnName()); + restored.prepare(); + Assertions.assertEquals("FILE:3:checksum", restored.getConvertedCustomProperties().get("ssl.ca.location")); + + // Recovery also preserves an unmigrated job, while runtime validation still rejects its current path. + Deencapsulation.setField(job, "customProperties", Map.of("ssl.ca.location", "/etc/ssl/ca.pem")); + KafkaRoutineLoadJob unmigrated = (KafkaRoutineLoadJob) GsonUtils.GSON.fromJson( + GsonUtils.GSON.toJson(job), RoutineLoadJob.class); + Assertions.assertEquals(RoutineLoadJob.JobState.PAUSED, unmigrated.getState()); + DdlException error = Assertions.assertThrows(DdlException.class, unmigrated::prepare); + Assertions.assertTrue(error.getMessage().contains("CREATE FILE")); + + // The recovery path must not weaken CREATE admission for the same SQL. + ConnectContext ctx = new ConnectContext(); + try { + ctx.setThreadLocalInfo(); + CreateRoutineLoadCommand command = (CreateRoutineLoadCommand) new NereidsParser() + .parseSingle(originalSql); + AnalysisException createError = Assertions.assertThrows(AnalysisException.class, + () -> command.getCreateRoutineLoadInfo().validate(ctx)); + Assertions.assertTrue(createError.getMessage().contains("CREATE FILE")); + } finally { + ctx.cleanup(); + } + } + } +} diff --git a/gensrc/thrift/BackendService.thrift b/gensrc/thrift/BackendService.thrift index 19d80620702f5c..e63450e6085e75 100644 --- a/gensrc/thrift/BackendService.thrift +++ b/gensrc/thrift/BackendService.thrift @@ -47,6 +47,110 @@ struct TTabletStatResult { 2: optional list tablet_stat_list } +// Custom Routine Load properties cross the SQL-user to BE-process trust boundary. +// Keep this allowlist shared by FE analysis and BE consumer initialization. New +// librdkafka options must be reviewed before being exposed to SQL users. Never +// allow shell commands, loadable modules/providers, or arbitrary local paths. +const set KAFKA_ALLOWED_PROPERTIES = [ + "allow.auto.create.topics", + "api.version.fallback.ms", + "api.version.request", + "api.version.request.timeout.ms", + "auto.offset.reset", + "broker.address.family", + "broker.address.ttl", + "broker.version.fallback", + "check.crcs", + "client.dns.lookup", + "client.id", + "connections.max.idle.ms", + "enable.partition.eof", + "enable.sasl.oauthbearer.unsecure.jwt", + "enable.ssl.certificate.verification", + "fetch.error.backoff.ms", + "fetch.max.bytes", + "fetch.message.max.bytes", + "fetch.min.bytes", + "fetch.wait.max.ms", + "group.id", + "group.instance.id", + "heartbeat.interval.ms", + "isolation.level", + "max.partition.fetch.bytes", + "max.poll.interval.ms", + "metadata.max.age.ms", + "partition.assignment.strategy", + "queued.max.messages.kbytes", + "queued.min.messages", + "receive.message.max.bytes", + "reconnect.backoff.max.ms", + "reconnect.backoff.ms", + "retry.backoff.max.ms", + "retry.backoff.ms", + // Only the value "0" is accepted; ticket management is external. + "sasl.kerberos.min.time.before.relogin", + "sasl.kerberos.principal", + "sasl.kerberos.service.name", + "sasl.mechanism", + "sasl.mechanisms", + "sasl.oauthbearer.client.credentials.client.id", + "sasl.oauthbearer.client.credentials.client.secret", + "sasl.oauthbearer.client.id", + "sasl.oauthbearer.client.secret", + "sasl.oauthbearer.config", + "sasl.oauthbearer.extensions", + "sasl.oauthbearer.method", + "sasl.oauthbearer.scope", + "sasl.oauthbearer.token.endpoint.url", + "sasl.password", + "sasl.username", + "security.protocol", + "session.timeout.ms", + "socket.connection.setup.timeout.ms", + "socket.keepalive.enable", + "socket.max.fails", + "socket.nagle.disable", + "socket.receive.buffer.bytes", + "socket.send.buffer.bytes", + "socket.timeout.ms", + "ssl.ca.pem", + "ssl.certificate.pem", + "ssl.cipher.suites", + "ssl.curves.list", + "ssl.endpoint.identification.algorithm", + "ssl.key.password", + "ssl.key.pem", + "ssl.keystore.password", + "ssl.sigalgs.list", + "topic.metadata.propagation.max.ms", + "topic.metadata.refresh.fast.interval.ms", + "topic.metadata.refresh.interval.ms", + "topic.metadata.refresh.sparse", +] + +// These data files must be registered through CREATE FILE, never raw BE paths. +const set KAFKA_FILE_PROPERTIES = [ + "https.ca.location", + "ssl.ca.location", + "ssl.certificate.location", + "ssl.crl.location", + "ssl.key.location", + "ssl.keystore.location", +] + +// Doris consumes these itself; they must not be forwarded to librdkafka. +const set KAFKA_DORIS_PROPERTIES = [ + "aws.access_key", + "aws.credentials_provider", + "aws.external_id", + "aws.profile_name", + "aws.region", + "aws.role_arn", + "aws.secret_key", + "kafka_default_offsets", + "kafka_origin_default_offsets", +] + struct TKafkaLoadInfo { 1: required string brokers; 2: required string topic; diff --git a/regression-test/suites/load_p0/routine_load/test_kafka_property_validation.groovy b/regression-test/suites/load_p0/routine_load/test_kafka_property_validation.groovy new file mode 100644 index 00000000000000..f06fbf385db68e --- /dev/null +++ b/regression-test/suites/load_p0/routine_load/test_kafka_property_validation.groovy @@ -0,0 +1,57 @@ +// 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. + +suite("test_kafka_property_validation", "p0") { + sql "DROP TABLE IF EXISTS test_kafka_property_validation" + sql """ + CREATE TABLE test_kafka_property_validation (id INT) + DUPLICATE KEY(id) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ("replication_num" = "1") + """ + + // All failures must happen during analysis, without contacting a Kafka broker. + def rejected = ["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"] + rejected.each { key -> + test { + sql """ + CREATE ROUTINE LOAD test_kafka_property_validation_job ON test_kafka_property_validation + FROM KAFKA ( + "kafka_broker_list" = "127.0.0.1:1", + "kafka_topic" = "unused", + "property.${key}" = "unused" + ) + """ + exception "Kafka property '${key}' is not allowed in Routine Load" + } + } + ["ssl.ca.location", "ssl.certificate.location", "ssl.key.location"].each { key -> + test { + sql """ + CREATE ROUTINE LOAD test_kafka_property_validation_job ON test_kafka_property_validation + FROM KAFKA ( + "kafka_broker_list" = "127.0.0.1:1", + "kafka_topic" = "unused", + "property.${key}" = "/etc/ssl/private/unused.pem" + ) + """ + exception "must reference a file registered with CREATE FILE" + } + } +} diff --git a/regression-test/suites/load_p0/routine_load/test_show_routine_load.groovy b/regression-test/suites/load_p0/routine_load/test_show_routine_load.groovy index fc330bb7b1dc18..574dc006cbea7d 100644 --- a/regression-test/suites/load_p0/routine_load/test_show_routine_load.groovy +++ b/regression-test/suites/load_p0/routine_load/test_show_routine_load.groovy @@ -205,31 +205,26 @@ suite("test_show_routine_load","p0") { } // test show routine load and information_schema mask sensitive custom properties + def kafkaPassword = "doris_routine_load_password_should_be_hidden" + sql """ + CREATE ROUTINE LOAD testShowSensitiveProperties ON ${tableName} + COLUMNS TERMINATED BY "," + FROM KAFKA + ( + "kafka_broker_list" = "${externalEnvIp}:${kafka_port}", + "kafka_topic" = "${kafkaCsvTpoics[0]}", + "property.kafka_default_offsets" = "OFFSET_BEGINNING", + "property.security.protocol" = "SASL_PLAINTEXT", + "property.sasl.mechanism" = "PLAIN", + "property.sasl.username" = "doris", + "property.sasl.password" = "${kafkaPassword}" + ); + """ try { - def kafkaPassword = "doris_routine_load_password_should_be_hidden" - def kafkaJaasSecret = "doris_routine_load_jaas_secret_should_be_hidden" - sql """ - CREATE ROUTINE LOAD testShowSensitiveProperties ON ${tableName} - COLUMNS TERMINATED BY "," - FROM KAFKA - ( - "kafka_broker_list" = "${externalEnvIp}:${kafka_port}", - "kafka_topic" = "${kafkaCsvTpoics[0]}", - "property.kafka_default_offsets" = "OFFSET_BEGINNING", - "property.security.protocol" = "SASL_PLAINTEXT", - "property.sasl.mechanism" = "PLAIN", - "property.sasl.username" = "doris", - "property.sasl.password" = "${kafkaPassword}", - "property.sasl.jaas.config" = "password=${kafkaJaasSecret}" - ); - """ - def showResult = sql "show routine load for testShowSensitiveProperties" def showCustomProperties = parseJson(showResult[0][13]) assertEquals("******", showCustomProperties["sasl.password"].toString()) - assertEquals("******", showCustomProperties["sasl.jaas.config"].toString()) assertFalse(showResult[0][13].toString().contains(kafkaPassword)) - assertFalse(showResult[0][13].toString().contains(kafkaJaasSecret)) def systemTableResult = sql """ SELECT CUSTOM_PROPERTIES @@ -238,9 +233,7 @@ suite("test_show_routine_load","p0") { """ def systemTableCustomProperties = parseJson(systemTableResult[0][0]) assertEquals("******", systemTableCustomProperties["sasl.password"].toString()) - assertEquals("******", systemTableCustomProperties["sasl.jaas.config"].toString()) assertFalse(systemTableResult[0][0].toString().contains(kafkaPassword)) - assertFalse(systemTableResult[0][0].toString().contains(kafkaJaasSecret)) } finally { sql "stop routine load for testShowSensitiveProperties" }