github-actions[bot] commented on code in PR #69391:
URL: https://github.com/apache/doris/pull/69391#discussion_r4346548947
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaDataSourceProperties.java:
##########
@@ -253,10 +254,33 @@ private void analyzeCustomProperties() throws
AnalysisException {
}
}
+ validateCustomProperties(customKafkaProperties);
Review Comment:
[P1] Preserve migrated jobs when recovering their original CREATE statement
`RoutineLoadJob.gsonPostProcess()` reparses `origStmt` and calls
`CreateRoutineLoadInfo.validate()`, so this check also runs on the original
properties during recovery. For an older job created with a raw
`ssl.ca.location` and successfully ALTERed to `FILE:ca.pem` before upgrading,
`origStmt` still contains the raw path although the persisted
`customProperties` is now safe. Validation throws and `gsonPostProcess()` sets
the job to terminal `CANCELLED`, losing the resumable job on restart.
Reconstruct the expressions without reapplying new admission rules to the stale
CREATE properties, then validate the current effective map before using it.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java:
##########
@@ -208,30 +209,37 @@ public void prepare() throws UserException {
}
private void convertCustomProperties(boolean rebuild) throws DdlException {
- if (customProperties.isEmpty()) {
- return;
+ // 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(customProperties);
Review Comment:
[P1] Keep historical ALTER replay independent of runtime validation
This method also runs inside `replayModifyProperties()`, after
`customProperties.putAll()` but before applying the logged topic, broker,
offsets and job settings. If an old ALTER changes topic A to B and sets
`sasl.kerberos.min.time.before.relogin=60000`, this new check throws and replay
only logs the failure, skipping topic B. A later logged ALTER restoring the
interval to `0` then succeeds, leaving a runnable job pointed at A although the
committed job uses B. Replay the complete durable transition and enforce the
new property policy before runtime use.
##########
be/src/load/routine_load/kafka_properties.cpp:
##########
@@ -0,0 +1,112 @@
+// 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 <utility>
+
+#include "common/kerberos/kerberos_ticket_mgr.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; "
+ "Doris manages keytab login and renewal without shell
commands",
+ key);
+ }
+ // Certificate and keytab 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);
+ }
+ }
+
+ // A keytab without a principal or GSSAPI would otherwise silently use the
+ // process identity or be ignored by a different SASL mechanism.
+ if (properties.contains("sasl.kerberos.keytab")) {
+ auto principal = properties.find("sasl.kerberos.principal");
+ auto protocol = properties.find("security.protocol");
+ if (principal == properties.end() || principal->second.empty() ||
+ principal->second.find('\0') != std::string::npos || protocol ==
properties.end() ||
+ (protocol->second != "SASL_SSL" && protocol->second !=
"SASL_PLAINTEXT")) {
Review Comment:
[P2] Preserve case-insensitive Kafka protocol values
With a managed keytab, `security.protocol=sasl_ssl` (or `sasl_plaintext`),
an explicit principal and `sasl.mechanisms=GSSAPI` pass FE validation but fail
this uppercase-only comparison before login or client creation. Librdkafka
defines these lowercase protocol values and parses the enum without case
sensitivity, so previously working Kerberos jobs using those values now fail
metadata probes or pause after the BE upgrade. Normalize the protocol or
compare it without case sensitivity while retaining the same allowed protocol
set.
##########
be/src/load/routine_load/kafka_properties.cpp:
##########
@@ -0,0 +1,112 @@
+// 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 <utility>
+
+#include "common/kerberos/kerberos_ticket_mgr.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; "
+ "Doris manages keytab login and renewal without shell
commands",
+ key);
+ }
+ // Certificate and keytab 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);
+ }
+ }
+
+ // A keytab without a principal or GSSAPI would otherwise silently use the
+ // process identity or be ignored by a different SASL mechanism.
+ if (properties.contains("sasl.kerberos.keytab")) {
+ auto principal = properties.find("sasl.kerberos.principal");
+ auto protocol = properties.find("security.protocol");
+ if (principal == properties.end() || principal->second.empty() ||
+ principal->second.find('\0') != std::string::npos || protocol ==
properties.end() ||
+ (protocol->second != "SASL_SSL" && protocol->second !=
"SASL_PLAINTEXT")) {
+ return Status::InvalidArgument(
+ "PAUSE: Kafka keytab requires an explicit Kerberos
principal and SASL "
+ "protocol");
+ }
+ for (const auto* mechanism : {"sasl.mechanism", "sasl.mechanisms"}) {
+ auto it = properties.find(mechanism);
+ if (it != properties.end() && it->second != "GSSAPI") {
+ return Status::InvalidArgument("PAUSE: Kafka keytab requires
GSSAPI");
+ }
+ }
+ }
+
+ // 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. Doris handles managed keytabs through
libkrb5.
+ // 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();
+}
+
+Status configure_kafka_kerberos(RdKafka::Conf& conf, const std::string&
principal,
+ const std::string& keytab_path,
+ kerberos::KerberosTicketMgr& manager,
+
std::shared_ptr<kerberos::KerberosTicketCache>* ticket_cache) {
+ kerberos::KerberosConfig config;
+ config.set_principal_and_keytab(principal, keytab_path);
+ config.set_use_default_cache(true);
+ std::shared_ptr<kerberos::KerberosTicketCache> cache;
+ RETURN_IF_ERROR(manager.get_or_set_ticket_cache(config, &cache));
Review Comment:
[P2] Initialize native Kafka caches with the authenticated principal
With `canonicalize=true`, a KDC can accept an alias principal and its valid
keytab entry but return a TGT whose `creds.client` is the canonical principal.
This new path calls `KerberosTicketCache::login()`, which initializes the
default cache with the requested `_principal` and then stores that canonical
TGT. GSSAPI reads the cache's alias as its client and cannot retrieve the TGT
stored for the canonical client, so native login reports success but Kafka
authentication fails. The previous kinit path initializes the cache with the
returned client. Initialize it with `creds.client` here as well, including
refreshes.
##########
be/src/load/routine_load/data_consumer.cpp:
##########
@@ -140,12 +147,24 @@ Status
KafkaDataConsumer::init(std::shared_ptr<StreamLoadContext> ctx) {
return Status::InternalError("PAUSE: failed to get file for
config: {}, error: {}",
item.first, st.to_string());
}
- RETURN_IF_ERROR(set_conf(item.first, file_path));
+ if (item.first == "sasl.kerberos.keytab") {
+ // The keytab is consumed by libkrb5, never by a shell command.
+ kerberos_keytab_path = std::move(file_path);
+ } else {
+ RETURN_IF_ERROR(set_conf(item.first, file_path));
+ }
} else {
RETURN_IF_ERROR(set_conf(item.first, item.second));
}
}
+ if (!kerberos_keytab_path.empty()) {
+ RETURN_IF_ERROR(configure_kafka_kerberos(
+ *conf,
ctx->kafka_info->properties.at("sasl.kerberos.principal"),
Review Comment:
[P2] Release the consumer-pool mutex before native KDC login
This call synchronously reaches `krb5_get_init_creds_keytab()`, but
`DataConsumerPool::get_consumer()` holds its pool-wide `_lock` while running
`init()`. If a managed-keytab job contacts an unavailable KDC, even with its
keytab already cached locally, all other Kafka/Kinesis consumer acquisitions,
metadata probes and consumer returns on that BE wait through the KDC retries.
Previously kinit ran from librdkafka's asynchronous refresh timer. Initialize a
newly allocated consumer outside the pool mutex so one authentication outage
cannot stall unrelated jobs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
