This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 4fc67338ab8 branch-4.1: [feature](RoutineLoad) Support RoutineLoad IAM
auth #61324 (#62245)
4fc67338ab8 is described below
commit 4fc67338ab880cea23e513a1c7baf62a7a227d85
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Apr 9 11:27:27 2026 +0800
branch-4.1: [feature](RoutineLoad) Support RoutineLoad IAM auth #61324
(#62245)
Cherry-picked from #61324
Co-authored-by: 黄瑞鑫 <[email protected]>
---
be/src/load/routine_load/data_consumer.cpp | 43 +-
be/src/load/routine_load/data_consumer.h | 4 +
be/src/runtime/aws_msk_iam_auth.cpp | 517 +++++++++++++++++++++
be/src/runtime/aws_msk_iam_auth.h | 169 +++++++
be/test/runtime/aws_msk_iam_auth_test.cpp | 244 ++++++++++
.../load/routineload/kafka/KafkaConfiguration.java | 15 +
.../kafka/KafkaDataSourceProperties.java | 80 +++-
.../load/routineload/KafkaAwsMskIamAuthTest.java | 491 +++++++++++++++++++
regression-test/conf/regression-conf.groovy | 1 +
.../load_p0/msk_load/test_routine_load_msk.groovy | 177 +++++++
.../test_routine_load_msk_public_access.groovy | 105 +++++
11 files changed, 1839 insertions(+), 7 deletions(-)
diff --git a/be/src/load/routine_load/data_consumer.cpp
b/be/src/load/routine_load/data_consumer.cpp
index c3732ffee39..1f7265b2d8a 100644
--- a/be/src/load/routine_load/data_consumer.cpp
+++ b/be/src/load/routine_load/data_consumer.cpp
@@ -33,6 +33,7 @@
#include "common/config.h"
#include "common/metrics/doris_metrics.h"
#include "common/status.h"
+#include "runtime/aws_msk_iam_auth.h"
#include "runtime/exec_env.h"
#include "runtime/small_file_mgr.h"
#include "service/backend_options.h"
@@ -99,6 +100,15 @@ 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;
+ continue;
+ }
+
if (starts_with(item.second, "FILE:")) {
// file property should has format: FILE:file_id:md5
std::vector<std::string> parts =
@@ -118,13 +128,12 @@ Status
KafkaDataConsumer::init(std::shared_ptr<StreamLoadContext> ctx) {
} else {
RETURN_IF_ERROR(set_conf(item.first, item.second));
}
- _custom_properties.emplace(item.first, item.second);
}
// if not specified group id, generate a random one.
// ATTN: In the new version, we have set a group.id on the FE side for
jobs that have not set a groupid,
// but in order to ensure compatibility, we still do a check here.
- if (_custom_properties.find(PROP_GROUP_ID) == _custom_properties.end()) {
+ if (!_custom_properties.contains(PROP_GROUP_ID)) {
std::stringstream ss;
ss << BackendOptions::get_localhost() << "_";
std::string group_id = ss.str() + UniqueId::gen_uid().to_string();
@@ -140,6 +149,24 @@ Status
KafkaDataConsumer::init(std::shared_ptr<StreamLoadContext> ctx) {
return Status::InternalError(ss.str());
}
+ // Set up AWS MSK IAM authentication if configured
+ _aws_msk_oauth_callback = AwsMskIamOAuthCallback::create_from_properties(
+ _custom_properties, ctx->kafka_info->brokers);
+ if (_aws_msk_oauth_callback) {
+ // Enable SASL queue to support background callbacks
+ if (conf->enable_sasl_queue(true, errstr) != RdKafka::Conf::CONF_OK) {
+ LOG(WARNING) << "PAUSE: failed to enable SASL queue: " << errstr;
+ return Status::InternalError("PAUSE: failed to enable SASL queue:
" + errstr);
+ }
+
+ if (conf->set("oauthbearer_token_refresh_cb",
_aws_msk_oauth_callback.get(), errstr) !=
+ RdKafka::Conf::CONF_OK) {
+ LOG(WARNING) << "PAUSE: failed to set OAuth callback: " << errstr;
+ return Status::InternalError("PAUSE: failed to set OAuth callback:
" + errstr);
+ }
+ LOG(INFO) << "AWS MSK IAM authentication enabled successfully";
+ }
+
// create consumer
_k_consumer = RdKafka::KafkaConsumer::create(conf, errstr);
if (!_k_consumer) {
@@ -147,6 +174,18 @@ Status
KafkaDataConsumer::init(std::shared_ptr<StreamLoadContext> ctx) {
return Status::InternalError("PAUSE: failed to create kafka consumer:
" + errstr);
}
+ // If AWS MSK IAM auth is enabled, inject initial token and enable
background refresh
+ if (_aws_msk_oauth_callback) {
+ RETURN_IF_ERROR(_aws_msk_oauth_callback->refresh_now(_k_consumer));
+
+ std::unique_ptr<RdKafka::Error>
bg_err(_k_consumer->sasl_background_callbacks_enable());
+ if (bg_err) {
+ return Status::InternalError("Failed to enable SASL background
callbacks: " +
+ bg_err->str());
+ }
+ LOG(INFO) << "AWS MSK IAM: initial token set, background refresh
enabled";
+ }
+
VLOG_NOTICE << "finished to init kafka consumer. " << ctx->brief();
_init = true;
diff --git a/be/src/load/routine_load/data_consumer.h
b/be/src/load/routine_load/data_consumer.h
index 6e223e07d98..2be8e3d23ec 100644
--- a/be/src/load/routine_load/data_consumer.h
+++ b/be/src/load/routine_load/data_consumer.h
@@ -32,6 +32,7 @@
#include "common/status.h"
#include "librdkafka/rdkafkacpp.h"
#include "load/stream_load/stream_load_context.h"
+#include "runtime/aws_msk_iam_auth.h"
#include "util/uid_util.h"
namespace doris {
@@ -167,6 +168,9 @@ private:
KafkaEventCb _k_event_cb;
RdKafka::KafkaConsumer* _k_consumer = nullptr;
+
+ // AWS MSK IAM authentication callback (must outlive _k_consumer)
+ std::unique_ptr<AwsMskIamOAuthCallback> _aws_msk_oauth_callback;
};
} // end namespace doris
diff --git a/be/src/runtime/aws_msk_iam_auth.cpp
b/be/src/runtime/aws_msk_iam_auth.cpp
new file mode 100644
index 00000000000..6d6292b7b01
--- /dev/null
+++ b/be/src/runtime/aws_msk_iam_auth.cpp
@@ -0,0 +1,517 @@
+// 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 "runtime/aws_msk_iam_auth.h"
+
+#include <aws/core/auth/AWSCredentials.h>
+#include <aws/core/auth/AWSCredentialsProvider.h>
+#include <aws/core/auth/AWSCredentialsProviderChain.h>
+#include <aws/core/auth/STSCredentialsProvider.h>
+#include <aws/core/platform/Environment.h>
+#include <aws/identity-management/auth/STSAssumeRoleCredentialsProvider.h>
+#include <aws/sts/STSClient.h>
+#include <aws/sts/model/AssumeRoleRequest.h>
+#include <openssl/hmac.h>
+#include <openssl/sha.h>
+
+#include <algorithm>
+#include <chrono>
+#include <iomanip>
+#include <sstream>
+
+#include "common/logging.h"
+
+namespace doris {
+
+AwsMskIamAuth::AwsMskIamAuth(Config config) : _config(std::move(config)) {
+ _credentials_provider = _create_credentials_provider();
+}
+
+std::shared_ptr<Aws::Auth::AWSCredentialsProvider>
AwsMskIamAuth::_create_provider_from_type(
+ const std::string& provider_type) {
+ std::string provider_upper = provider_type;
+ std::transform(provider_upper.begin(), provider_upper.end(),
provider_upper.begin(), ::toupper);
+
+ if (provider_upper == "ENV" || provider_upper == "ENVIRONMENT") {
+ return
std::make_shared<Aws::Auth::EnvironmentAWSCredentialsProvider>();
+ } else if (provider_upper == "INSTANCE_PROFILE" || provider_upper ==
"INSTANCEPROFILE") {
+ return
std::make_shared<Aws::Auth::InstanceProfileCredentialsProvider>();
+ } else if (provider_upper == "CONTAINER" || provider_upper == "ECS") {
+ return std::make_shared<Aws::Auth::TaskRoleCredentialsProvider>(
+
Aws::Environment::GetEnv("AWS_CONTAINER_CREDENTIALS_RELATIVE_URI").c_str());
+ } else if (provider_upper == "SYSTEM_PROPERTIES" || provider_upper ==
"SYSTEMPROPERTIES") {
+ return
std::make_shared<Aws::Auth::ProfileConfigFileAWSCredentialsProvider>();
+ } else if (provider_upper == "WEB_IDENTITY" || provider_upper ==
"WEBIDENTITY" ||
+ provider_upper == "WEB_IDENTITY_TOKEN_FILE") {
+ return
std::make_shared<Aws::Auth::STSAssumeRoleWebIdentityCredentialsProvider>();
+ } else if (provider_upper == "ANONYMOUS") {
+ return std::make_shared<Aws::Auth::AnonymousAWSCredentialsProvider>();
+ } else if (provider_upper.empty() || provider_upper == "DEFAULT") {
+ return
std::make_shared<Aws::Auth::DefaultAWSCredentialsProviderChain>();
+ }
+
+ LOG(WARNING) << "Unknown credentials provider type: " << provider_type
+ << ", falling back to default credentials provider chain";
+ return std::make_shared<Aws::Auth::DefaultAWSCredentialsProviderChain>();
+}
+
+std::shared_ptr<Aws::Auth::AWSCredentialsProvider>
+AwsMskIamAuth::_create_assume_role_base_provider() {
+ if (!_config.access_key.empty() && !_config.secret_key.empty()) {
+ Aws::Auth::AWSCredentials base_credentials(_config.access_key,
_config.secret_key);
+ return
std::make_shared<Aws::Auth::SimpleAWSCredentialsProvider>(base_credentials);
+ }
+
+ if (!_config.profile_name.empty()) {
+ return
std::make_shared<Aws::Auth::ProfileConfigFileAWSCredentialsProvider>(
+ _config.profile_name.c_str());
+ }
+
+ if (!_config.credentials_provider.empty()) {
+ return _create_provider_from_type(_config.credentials_provider);
+ }
+
+ return std::make_shared<Aws::Auth::DefaultAWSCredentialsProviderChain>();
+}
+
+std::shared_ptr<Aws::Auth::AWSCredentialsProvider>
AwsMskIamAuth::_create_credentials_provider() {
+ if (!_config.role_arn.empty()) {
+ Aws::Client::ClientConfiguration client_config;
+ if (!_config.region.empty()) {
+ client_config.region = _config.region;
+ }
+
+ auto base_provider = _create_assume_role_base_provider();
+ LOG(INFO) << "Using AWS STS Assume Role: " << _config.role_arn;
+
+ auto sts_client = std::make_shared<Aws::STS::STSClient>(base_provider,
client_config);
+
+ Aws::String external_id = _config.external_id.empty()
+ ? Aws::String()
+ :
Aws::String(_config.external_id.c_str());
+ return std::make_shared<Aws::Auth::STSAssumeRoleCredentialsProvider>(
+ _config.role_arn, Aws::String(), external_id,
+ Aws::Auth::DEFAULT_CREDS_LOAD_FREQ_SECONDS, sts_client);
+ }
+ // 2. Explicit AK/SK credentials (direct access)
+ if (!_config.access_key.empty() && !_config.secret_key.empty()) {
+ LOG(INFO) << "Using explicit AWS credentials (Access Key ID: "
+ << _config.access_key.substr(0, 4) << "****)";
+
+ Aws::Auth::AWSCredentials credentials(_config.access_key,
_config.secret_key);
+
+ return
std::make_shared<Aws::Auth::SimpleAWSCredentialsProvider>(credentials);
+ }
+ // 3. AWS Profile (reads from ~/.aws/credentials)
+ if (!_config.profile_name.empty()) {
+ LOG(INFO) << "Using AWS Profile: " << _config.profile_name;
+
+ return
std::make_shared<Aws::Auth::ProfileConfigFileAWSCredentialsProvider>(
+ _config.profile_name.c_str());
+ }
+ // 4. Custom Credentials Provider
+ if (!_config.credentials_provider.empty()) {
+ LOG(INFO) << "Using custom credentials provider: " <<
_config.credentials_provider;
+ return _create_provider_from_type(_config.credentials_provider);
+ }
+ // No valid credentials configuration found
+ LOG(ERROR) << "AWS MSK IAM authentication requires credentials. Please
provide.";
+ return nullptr;
+}
+
+Status AwsMskIamAuth::get_credentials(Aws::Auth::AWSCredentials* credentials) {
+ std::lock_guard<std::mutex> lock(_mutex);
+
+ if (!_credentials_provider) {
+ return Status::InternalError("AWS credentials provider not
initialized");
+ }
+
+ // Refresh if needed
+ if (_should_refresh_credentials()) {
+ _cached_credentials = _credentials_provider->GetAWSCredentials();
+ if (_cached_credentials.GetAWSAccessKeyId().empty()) {
+ return Status::InternalError("Failed to get AWS credentials");
+ }
+
+ // Set expiry time (assume 1 hour for instance profile, or use the
credentials expiration)
+ _credentials_expiry = std::chrono::system_clock::now() +
std::chrono::hours(1);
+
+ LOG(INFO) << "Refreshed AWS credentials for MSK IAM authentication";
+ }
+
+ *credentials = _cached_credentials;
+ return Status::OK();
+}
+
+bool AwsMskIamAuth::_should_refresh_credentials() {
+ auto now = std::chrono::system_clock::now();
+ auto refresh_time =
+ _credentials_expiry -
std::chrono::milliseconds(_config.token_refresh_margin_ms);
+ return now >= refresh_time ||
_cached_credentials.GetAWSAccessKeyId().empty();
+}
+
+Status AwsMskIamAuth::generate_token(const std::string& broker_hostname,
std::string* token,
+ int64_t* token_lifetime_ms) {
+ Aws::Auth::AWSCredentials credentials;
+ RETURN_IF_ERROR(get_credentials(&credentials));
+
+ std::string timestamp = _get_timestamp();
+ std::string date_stamp = _get_date_stamp(timestamp);
+
+ // AWS MSK IAM token is a base64-encoded presigned URL
+ // Reference: https://github.com/aws/aws-msk-iam-sasl-signer-python
+
+ // Token expiry in seconds (900 seconds = 15 minutes, matching AWS MSK IAM
signer reference)
+ static constexpr int TOKEN_EXPIRY_SECONDS = 900;
+
+ // Build the endpoint URL
+ std::string endpoint_url = "https://kafka."; + _config.region +
".amazonaws.com/";
+
+ // Build credential scope
+ std::string credential_scope =
+ date_stamp + "/" + _config.region + "/kafka-cluster/aws4_request";
+
+ // Build the canonical query string (sorted alphabetically)
+ // IMPORTANT: All query parameters must be included in the signature
calculation
+ // Session Token must be in canonical query string if using temporary
credentials
+ std::stringstream canonical_query_ss;
+ canonical_query_ss << "Action=kafka-cluster%3AConnect"; // URL-encoded :
+
+ // Add Algorithm
+ canonical_query_ss << "&X-Amz-Algorithm=AWS4-HMAC-SHA256";
+
+ // Add Credential
+ std::string credential = std::string(credentials.GetAWSAccessKeyId()) +
"/" + credential_scope;
+ canonical_query_ss << "&X-Amz-Credential=" << _url_encode(credential);
+
+ // Add Date
+ canonical_query_ss << "&X-Amz-Date=" << timestamp;
+
+ // Add Expires
+ canonical_query_ss << "&X-Amz-Expires=" << TOKEN_EXPIRY_SECONDS;
+
+ // Add Security Token if present (MUST be before signature calculation)
+ if (!credentials.GetSessionToken().empty()) {
+ canonical_query_ss << "&X-Amz-Security-Token="
+ <<
_url_encode(std::string(credentials.GetSessionToken()));
+ }
+
+ // Add SignedHeaders
+ canonical_query_ss << "&X-Amz-SignedHeaders=host";
+
+ std::string canonical_query_string = canonical_query_ss.str();
+
+ // Build the canonical headers
+ std::string host = "kafka." + _config.region + ".amazonaws.com";
+ std::string canonical_headers = "host:" + host + "\n";
+ std::string signed_headers = "host";
+
+ // Build the canonical request
+ std::string method = "GET";
+ std::string uri = "/";
+ std::string payload_hash = _sha256("");
+
+ std::string canonical_request = method + "\n" + uri + "\n" +
canonical_query_string + "\n" +
+ canonical_headers + "\n" + signed_headers
+ "\n" + payload_hash;
+
+ // Build the string to sign
+ std::string algorithm = "AWS4-HMAC-SHA256";
+ std::string canonical_request_hash = _sha256(canonical_request);
+ std::string string_to_sign =
+ algorithm + "\n" + timestamp + "\n" + credential_scope + "\n" +
canonical_request_hash;
+
+ // Calculate signature
+ std::string signing_key =
_calculate_signing_key(std::string(credentials.GetAWSSecretKey()),
+ date_stamp,
_config.region, "kafka-cluster");
+ std::string signature = _hmac_sha256_hex(signing_key, string_to_sign);
+
+ // Build the final presigned URL
+ // All parameters are already in canonical_query_string, just add signature
+ // Then add User-Agent AFTER signature (not part of signed content,
matching reference impl)
+ std::string signed_url = endpoint_url + "?" + canonical_query_string +
+ "&X-Amz-Signature=" + signature +
+ "&User-Agent=doris-msk-iam-auth%2F1.0";
+
+ // Base64url encode the signed URL (without padding)
+ *token = _base64url_encode(signed_url);
+
+ // Token lifetime in milliseconds
+ *token_lifetime_ms = TOKEN_EXPIRY_SECONDS * 1000;
+
+ VLOG_DEBUG << "Generated AWS MSK IAM token for region: " << _config.region;
+ return Status::OK();
+}
+
+std::string AwsMskIamAuth::_hmac_sha256_hex(const std::string& key, const
std::string& data) {
+ std::string raw = _hmac_sha256(key, data);
+ std::stringstream ss;
+ for (unsigned char c : raw) {
+ ss << std::hex << std::setw(2) << std::setfill('0') <<
static_cast<int>(c);
+ }
+ return ss.str();
+}
+
+std::string AwsMskIamAuth::_url_encode(const std::string& value) {
+ std::ostringstream escaped;
+ escaped.fill('0');
+ escaped << std::hex;
+
+ for (char c : value) {
+ // Keep alphanumeric and other accepted characters intact
+ if (isalnum(static_cast<unsigned char>(c)) || c == '-' || c == '_' ||
c == '.' ||
+ c == '~') {
+ escaped << c;
+ } else {
+ // Any other characters are percent-encoded
+ escaped << std::uppercase;
+ escaped << '%' << std::setw(2) <<
static_cast<int>(static_cast<unsigned char>(c));
+ escaped << std::nouppercase;
+ }
+ }
+
+ return escaped.str();
+}
+
+std::string AwsMskIamAuth::_base64url_encode(const std::string& input) {
+ // Standard base64 alphabet
+ static const char* base64_chars =
+ "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
+
+ std::string result;
+ result.reserve(((input.size() + 2) / 3) * 4);
+
+ const unsigned char* bytes = reinterpret_cast<const unsigned
char*>(input.c_str());
+ size_t len = input.size();
+
+ for (size_t i = 0; i < len; i += 3) {
+ uint32_t n = static_cast<uint32_t>(bytes[i]) << 16;
+ if (i + 1 < len) n |= static_cast<uint32_t>(bytes[i + 1]) << 8;
+ if (i + 2 < len) n |= static_cast<uint32_t>(bytes[i + 2]);
+
+ result += base64_chars[(n >> 18) & 0x3F];
+ result += base64_chars[(n >> 12) & 0x3F];
+ if (i + 1 < len) result += base64_chars[(n >> 6) & 0x3F];
+ if (i + 2 < len) result += base64_chars[n & 0x3F];
+ }
+
+ // Convert to URL-safe base64 (replace + with -, / with _)
+ // and remove padding (=)
+ for (char& c : result) {
+ if (c == '+')
+ c = '-';
+ else if (c == '/')
+ c = '_';
+ }
+
+ return result;
+}
+
+std::string AwsMskIamAuth::_calculate_signing_key(const std::string&
secret_key,
+ const std::string&
date_stamp,
+ const std::string& region,
+ const std::string& service) {
+ std::string k_secret = "AWS4" + secret_key;
+ std::string k_date = _hmac_sha256(k_secret, date_stamp);
+ std::string k_region = _hmac_sha256(k_date, region);
+ std::string k_service = _hmac_sha256(k_region, service);
+ std::string k_signing = _hmac_sha256(k_service, "aws4_request");
+ return k_signing;
+}
+
+std::string AwsMskIamAuth::_hmac_sha256(const std::string& key, const
std::string& data) {
+ unsigned char digest[EVP_MAX_MD_SIZE];
+ unsigned int digest_len = 0;
+ HMAC(EVP_sha256(), key.c_str(), static_cast<int>(key.length()),
+ reinterpret_cast<const unsigned char*>(data.c_str()), data.length(),
digest, &digest_len);
+ return {reinterpret_cast<char*>(digest), digest_len};
+}
+
+std::string AwsMskIamAuth::_sha256(const std::string& data) {
+ unsigned char hash[SHA256_DIGEST_LENGTH];
+ SHA256(reinterpret_cast<const unsigned char*>(data.c_str()),
data.length(), hash);
+
+ std::stringstream ss;
+ for (unsigned char i : hash) {
+ ss << std::hex << std::setw(2) << std::setfill('0') << (int)i;
+ }
+ return ss.str();
+}
+
+std::string AwsMskIamAuth::_get_timestamp() {
+ auto now = std::chrono::system_clock::now();
+ auto time_t_now = std::chrono::system_clock::to_time_t(now);
+ std::tm tm_now;
+ gmtime_r(&time_t_now, &tm_now);
+
+ std::stringstream ss;
+ ss << std::put_time(&tm_now, "%Y%m%dT%H%M%SZ");
+ return ss.str();
+}
+
+std::string AwsMskIamAuth::_get_date_stamp(const std::string& timestamp) {
+ // Extract YYYYMMDD from YYYYMMDDTHHMMSSz
+ return timestamp.substr(0, 8);
+}
+
+// AwsMskIamOAuthCallback implementation
+
+namespace {
+// Property keys for AWS MSK IAM authentication
+constexpr const char* PROP_SECURITY_PROTOCOL = "security.protocol";
+constexpr const char* PROP_SASL_MECHANISM = "sasl.mechanism";
+constexpr const char* PROP_AWS_REGION = "aws.region";
+constexpr const char* PROP_AWS_ACCESS_KEY = "aws.access_key";
+constexpr const char* PROP_AWS_SECRET_KEY = "aws.secret_key";
+constexpr const char* PROP_AWS_ROLE_ARN = "aws.role_arn";
+constexpr const char* PROP_AWS_EXTERNAL_ID = "aws.external_id";
+constexpr const char* PROP_AWS_PROFILE_NAME = "aws.profile_name";
+constexpr const char* PROP_AWS_CREDENTIALS_PROVIDER =
"aws.credentials_provider";
+} // namespace
+
+std::unique_ptr<AwsMskIamOAuthCallback>
AwsMskIamOAuthCallback::create_from_properties(
+ const std::unordered_map<std::string, std::string>& custom_properties,
+ const std::string& brokers) {
+ auto security_protocol_it = custom_properties.find(PROP_SECURITY_PROTOCOL);
+ auto sasl_mechanism_it = custom_properties.find(PROP_SASL_MECHANISM);
+ bool is_sasl_ssl = security_protocol_it != custom_properties.end() &&
+ security_protocol_it->second == "SASL_SSL";
+ bool is_oauthbearer = sasl_mechanism_it != custom_properties.end() &&
+ sasl_mechanism_it->second == "OAUTHBEARER";
+
+ if (!is_sasl_ssl || !is_oauthbearer) {
+ return nullptr;
+ }
+
+ // Extract broker hostname for token generation.
+ std::string broker_hostname = brokers;
+ // If there are multiple brokers, we use the first one (Refrain : is this
ok?)
+ if (broker_hostname.find(',') != std::string::npos) {
+ broker_hostname = broker_hostname.substr(0, broker_hostname.find(','));
+ }
+ // Remove port if present
+ if (broker_hostname.find(':') != std::string::npos) {
+ broker_hostname = broker_hostname.substr(0, broker_hostname.find(':'));
+ }
+
+ AwsMskIamAuth::Config auth_config;
+
+ auto region_it = custom_properties.find(PROP_AWS_REGION);
+ if (region_it != custom_properties.end()) {
+ auth_config.region = region_it->second;
+ }
+
+ auto access_key_it = custom_properties.find(PROP_AWS_ACCESS_KEY);
+ auto secret_key_it = custom_properties.find(PROP_AWS_SECRET_KEY);
+ if (access_key_it != custom_properties.end() && secret_key_it !=
custom_properties.end()) {
+ auth_config.access_key = access_key_it->second;
+ auth_config.secret_key = secret_key_it->second;
+ LOG(INFO) << "AWS MSK IAM: using explicit credentials (region: " <<
auth_config.region
+ << ")";
+ }
+
+ auto role_arn_it = custom_properties.find(PROP_AWS_ROLE_ARN);
+ if (role_arn_it != custom_properties.end()) {
+ auth_config.role_arn = role_arn_it->second;
+ LOG(INFO) << "AWS MSK IAM: using role " << auth_config.role_arn
+ << " (region: " << auth_config.region << ")";
+ }
+
+ auto external_id_it = custom_properties.find(PROP_AWS_EXTERNAL_ID);
+ if (external_id_it != custom_properties.end()) {
+ auth_config.external_id = external_id_it->second;
+ LOG(INFO) << "AWS MSK IAM: using external id with role assumption
(region: "
+ << auth_config.region << ")";
+ }
+
+ auto profile_name_it = custom_properties.find(PROP_AWS_PROFILE_NAME);
+ if (profile_name_it != custom_properties.end()) {
+ auth_config.profile_name = profile_name_it->second;
+ LOG(INFO) << "AWS MSK IAM: using profile " << auth_config.profile_name
+ << " (region: " << auth_config.region << ")";
+ }
+
+ auto credentials_provider_it =
custom_properties.find(PROP_AWS_CREDENTIALS_PROVIDER);
+ if (credentials_provider_it != custom_properties.end()) {
+ auth_config.credentials_provider = credentials_provider_it->second;
+ LOG(INFO) << "AWS MSK IAM: using credentials provider " <<
auth_config.credentials_provider
+ << " (region: " << auth_config.region << ")";
+ }
+
+ if (!auth_config.external_id.empty() && auth_config.role_arn.empty()) {
+ LOG(ERROR) << "AWS MSK IAM authentication: 'aws.external_id' requires
'aws.role_arn'";
+ return nullptr;
+ }
+
+ // Validate that at least one credential source is configured
+ bool has_credentials = !auth_config.access_key.empty() ||
!auth_config.role_arn.empty() ||
+ !auth_config.profile_name.empty() ||
+ !auth_config.credentials_provider.empty();
+
+ if (!has_credentials) {
+ LOG(ERROR) << "AWS MSK IAM authentication enabled but no credentials
configured. "
+ << "Please provide one of: access_key/secret_key, role_arn,
profile_name, or "
+ "credentials_provider";
+ return nullptr;
+ }
+
+ LOG(INFO) << "Enabling AWS MSK IAM authentication for broker: " <<
broker_hostname
+ << ", region: " << auth_config.region;
+
+ auto auth = std::make_shared<AwsMskIamAuth>(auth_config);
+ return std::make_unique<AwsMskIamOAuthCallback>(std::move(auth),
std::move(broker_hostname));
+}
+
+AwsMskIamOAuthCallback::AwsMskIamOAuthCallback(std::shared_ptr<AwsMskIamAuth>
auth,
+ std::string broker_hostname)
+ : _auth(std::move(auth)), _broker_hostname(std::move(broker_hostname))
{}
+
+Status AwsMskIamOAuthCallback::refresh_now(RdKafka::Handle* handle) {
+ std::string token;
+ int64_t token_lifetime_ms = 0;
+
+ RETURN_IF_ERROR(_auth->generate_token(_broker_hostname, &token,
&token_lifetime_ms));
+
+ std::string principal = "doris-consumer";
+ std::list<std::string> extensions;
+ std::string errstr;
+
+ auto now = std::chrono::system_clock::now();
+ auto now_ms =
+
std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count();
+ int64_t token_expiry_ms = now_ms + token_lifetime_ms;
+
+ auto err = handle->oauthbearer_set_token(token, token_expiry_ms,
principal, extensions, errstr);
+ if (err != RdKafka::ERR_NO_ERROR) {
+ return Status::InternalError("Failed to set OAuth token: {}, detail:
{}",
+ RdKafka::err2str(err), errstr);
+ }
+
+ LOG(INFO) << "Successfully set AWS MSK IAM OAuth token, lifetime: " <<
token_lifetime_ms
+ << "ms";
+ return Status::OK();
+}
+
+void AwsMskIamOAuthCallback::oauthbearer_token_refresh_cb(
+ RdKafka::Handle* handle, const std::string& /*oauthbearer_config*/) {
+ Status st = refresh_now(handle);
+ if (!st.ok()) {
+ LOG(WARNING) << st;
+ handle->oauthbearer_set_token_failure(st.to_string());
+ }
+}
+
+} // namespace doris
diff --git a/be/src/runtime/aws_msk_iam_auth.h
b/be/src/runtime/aws_msk_iam_auth.h
new file mode 100644
index 00000000000..79ceb03dd77
--- /dev/null
+++ b/be/src/runtime/aws_msk_iam_auth.h
@@ -0,0 +1,169 @@
+// 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 <aws/core/auth/AWSCredentialsProvider.h>
+#include <librdkafka/rdkafkacpp.h>
+
+#include <memory>
+#include <mutex>
+#include <string>
+#include <unordered_map>
+
+#include "common/status.h"
+
+namespace doris {
+/**
+ * AWS MSK IAM authentication token generator.
+ *
+ * This class generates SASL/OAUTHBEARER tokens for AWS MSK IAM authentication.
+ * It uses AWS SDK for C++ to obtain IAM credentials and generates signed
tokens
+ * that can be used with librdkafka's OAUTHBEARER mechanism.
+ */
+class AwsMskIamAuth {
+public:
+ struct Config {
+ std::string region; // AWS region (e.g., "us-east-1"), required
+ std::string access_key; // AWS Access Key ID (optional, for explicit
credentials)
+ std::string secret_key; // AWS Secret Access Key (optional, for
explicit credentials)
+ std::string role_arn; // IAM role ARN (optional, for assume role)
+ std::string external_id; // Optional external ID for STS AssumeRole
+ std::string profile_name; // AWS profile name (optional, reads from
~/.aws/credentials)
+ std::string
+ credentials_provider; // Credentials provider type (optional,
e.g., "ENV", "INSTANCE_PROFILE")
+ int token_refresh_margin_ms = 60000; // Refresh token 60s before expiry
+ };
+
+ explicit AwsMskIamAuth(Config config);
+ ~AwsMskIamAuth() = default;
+
+ /**
+ * Generate AWS MSK IAM authentication token.
+ *
+ * The token is a base64url-encoded presigned URL following AWS SigV4
format:
+ * https://kafka.<region>.amazonaws.com/?Action=kafka-cluster:Connect
+ * &X-Amz-Algorithm=AWS4-HMAC-SHA256
+ *
&X-Amz-Credential=<access-key>/<date>/<region>/kafka-cluster/aws4_request
+ * &X-Amz-Date=<timestamp>
+ * &X-Amz-Expires=900
+ * &X-Amz-SignedHeaders=host
+ * &X-Amz-Signature=<signature>
+ * &X-Amz-Security-Token=<session-token> // if using temporary
credentials
+ * &User-Agent=doris-msk-iam-auth/1.0
+ *
+ * Reference: https://github.com/aws/aws-msk-iam-sasl-signer-python
+ *
+ * @param broker_hostname The MSK broker hostname (used for logging, not
in token)
+ * @param token Output: base64url-encoded signed URL token
+ * @param token_lifetime_ms Output: token lifetime in milliseconds
(3600000ms = 1 hour)
+ * @return Status indicating success or failure
+ */
+ Status generate_token(const std::string& broker_hostname, std::string*
token,
+ int64_t* token_lifetime_ms);
+
+ /**
+ * Get current AWS credentials.
+ * This will refresh credentials if they are expired or about to expire.
+ */
+ Status get_credentials(Aws::Auth::AWSCredentials* credentials);
+
+private:
+ // Create AWS credentials provider based on configuration
+ std::shared_ptr<Aws::Auth::AWSCredentialsProvider>
_create_credentials_provider();
+ std::shared_ptr<Aws::Auth::AWSCredentialsProvider>
_create_provider_from_type(
+ const std::string& provider_type);
+ std::shared_ptr<Aws::Auth::AWSCredentialsProvider>
_create_assume_role_base_provider();
+
+ // HMAC-SHA256 returning hex string
+ std::string _hmac_sha256_hex(const std::string& key, const std::string&
data);
+
+ std::string _url_encode(const std::string& value);
+
+ std::string _base64url_encode(const std::string& input);
+
+ // Calculate AWS SigV4 signing key
+ std::string _calculate_signing_key(const std::string& secret_key, const
std::string& date_stamp,
+ const std::string& region, const
std::string& service);
+
+ std::string _hmac_sha256(const std::string& key, const std::string& data);
+
+ std::string _sha256(const std::string& data);
+
+ std::string _get_timestamp();
+
+ std::string _get_date_stamp(const std::string& timestamp);
+
+ bool _should_refresh_credentials();
+
+ Config _config;
+ std::shared_ptr<Aws::Auth::AWSCredentialsProvider> _credentials_provider;
+ std::mutex _mutex;
+ Aws::Auth::AWSCredentials _cached_credentials;
+ std::chrono::time_point<std::chrono::system_clock> _credentials_expiry;
+};
+
+/**
+ * librdkafka OAUTHBEARER callback for AWS MSK IAM authentication.
+ *
+ * This callback is invoked by librdkafka when it needs to refresh the
+ * OAUTHBEARER token. It uses AwsMskIamAuth to generate the token.
+ */
+class AwsMskIamOAuthCallback : public RdKafka::OAuthBearerTokenRefreshCb {
+public:
+ /**
+ * Create an OAuth callback from Kafka custom properties.
+ *
+ * This factory method checks if AWS MSK IAM authentication is configured
+ * (security.protocol=SASL_SSL and sasl.mechanism=OAUTHBEARER) and creates
+ * the callback with proper configuration.
+ *
+ * @param custom_properties Kafka custom properties map
+ * @param brokers Kafka broker list (comma-separated)
+ * @return unique_ptr to callback if IAM auth is configured, nullptr
otherwise
+ */
+ static std::unique_ptr<AwsMskIamOAuthCallback> create_from_properties(
+ const std::unordered_map<std::string, std::string>&
custom_properties,
+ const std::string& brokers);
+
+ explicit AwsMskIamOAuthCallback(std::shared_ptr<AwsMskIamAuth> auth,
+ std::string broker_hostname);
+
+ /**
+ * Synchronously refresh and set OAuth token.
+ * Can be called directly during initialization or by the callback.
+ *
+ * @param handle The Kafka handle (consumer or producer)
+ * @return Status indicating success or failure
+ */
+ Status refresh_now(RdKafka::Handle* handle);
+
+ /**
+ * Callback invoked by librdkafka to refresh OAuth token.
+ *
+ * @param handle The Kafka handle (consumer or producer)
+ * @param oauthbearer_config Configuration string from
'sasl.oauthbearer.config'
+ */
+ void oauthbearer_token_refresh_cb(RdKafka::Handle* handle,
+ const std::string& oauthbearer_config)
override;
+
+private:
+ std::shared_ptr<AwsMskIamAuth> _auth;
+ std::string _broker_hostname;
+};
+
+} // namespace doris
diff --git a/be/test/runtime/aws_msk_iam_auth_test.cpp
b/be/test/runtime/aws_msk_iam_auth_test.cpp
new file mode 100644
index 00000000000..f1eb1ec37ed
--- /dev/null
+++ b/be/test/runtime/aws_msk_iam_auth_test.cpp
@@ -0,0 +1,244 @@
+// 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 "runtime/aws_msk_iam_auth.h"
+
+#include <gtest/gtest.h>
+
+#include <memory>
+#include <string>
+#include <unordered_map>
+
+#include "common/status.h"
+
+namespace doris {
+
+class AwsMskIamAuthTest : public testing::Test {
+protected:
+ void SetUp() override {
+ // Setup test configuration
+ config.region = "us-east-1";
+ }
+
+ AwsMskIamAuth::Config config;
+};
+
+TEST_F(AwsMskIamAuthTest, TestConfigCreation) {
+ // Test basic configuration creation
+ AwsMskIamAuth auth(config);
+
+ // This test just ensures the object can be created without crashing
+ ASSERT_TRUE(true);
+}
+
+TEST_F(AwsMskIamAuthTest, TestTokenGeneration) {
+ // This test requires AWS credentials to be available
+ // In a real environment, you would mock the AWS SDK or use test
credentials
+
+ AwsMskIamAuth auth(config);
+ std::string token;
+ int64_t token_lifetime_ms;
+
+ std::string broker_hostname = "b-1.test-msk.us-east-1.amazonaws.com";
+
+ // In a real test environment with credentials, this should succeed
+ // For CI/CD without credentials, we expect it to fail gracefully
+ Status status = auth.generate_token(broker_hostname, &token,
&token_lifetime_ms);
+
+ if (status.ok()) {
+ // If we have credentials, verify token properties
+ ASSERT_FALSE(token.empty());
+ ASSERT_GT(token_lifetime_ms, 0);
+ ASSERT_LT(token_lifetime_ms, 3600000); // Less than 1 hour
+
+ // Token should be valid JSON
+ ASSERT_NE(token.find("version"), std::string::npos);
+ ASSERT_NE(token.find("host"), std::string::npos);
+ ASSERT_NE(token.find(broker_hostname), std::string::npos);
+ } else {
+ // Without credentials, we expect an error but no crash
+ ASSERT_FALSE(status.ok());
+ LOG(INFO) << "Token generation failed (expected without AWS
credentials): "
+ << status.to_string();
+ }
+}
+
+TEST_F(AwsMskIamAuthTest, TestConfigWithRoleArn) {
+ config.role_arn = "arn:aws:iam::123456789012:role/TestRole";
+
+ AwsMskIamAuth auth(config);
+
+ // Should create without error even if role doesn't exist
+ // (actual assumption will fail later when trying to get credentials)
+ ASSERT_TRUE(true);
+}
+
+TEST_F(AwsMskIamAuthTest, TestConfigWithRoleArnAndExternalId) {
+ config.role_arn = "arn:aws:iam::123456789012:role/TestRole";
+ config.external_id = "external-id-123";
+
+ AwsMskIamAuth auth(config);
+
+ // Should create without error even if role doesn't exist
+ ASSERT_TRUE(true);
+}
+
+TEST_F(AwsMskIamAuthTest, TestConfigWithProfile) {
+ config.profile_name = "test-profile";
+
+ AwsMskIamAuth auth(config);
+
+ // Should create without error even if profile doesn't exist
+ ASSERT_TRUE(true);
+}
+
+TEST_F(AwsMskIamAuthTest, TestConfigWithCredentialsProvider) {
+ config.credentials_provider = "INSTANCE_PROFILE";
+
+ AwsMskIamAuth auth(config);
+
+ // Should create without error
+ ASSERT_TRUE(true);
+}
+
+TEST_F(AwsMskIamAuthTest, TestConfigWithExplicitCredentials) {
+ config.access_key = "AKIAIOSFODNN7EXAMPLE";
+ config.secret_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY";
+
+ AwsMskIamAuth auth(config);
+
+ // Should create without error
+ ASSERT_TRUE(true);
+}
+
+TEST_F(AwsMskIamAuthTest, TestCrossAccountAssumeRole) {
+ config.role_arn = "arn:aws:iam::123456789012:role/CrossAccountRole";
+ config.access_key = "AKIAIOSFODNN7EXAMPLE";
+ config.secret_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY";
+
+ AwsMskIamAuth auth(config);
+
+ // Should create without error (actual assume role will fail without real
credentials)
+ ASSERT_TRUE(true);
+}
+
+TEST_F(AwsMskIamAuthTest, TestRoleArnWithCredentialsProvider) {
+ config.role_arn = "arn:aws:iam::123456789012:role/TestRole";
+ config.credentials_provider = "ENV";
+
+ AwsMskIamAuth auth(config);
+
+ // Should create without error. ENV provider will be used as AssumeRole
base credentials provider.
+ ASSERT_TRUE(true);
+}
+
+TEST_F(AwsMskIamAuthTest, TestRoleArnWithProfileName) {
+ config.role_arn = "arn:aws:iam::123456789012:role/TestRole";
+ config.profile_name = "default";
+
+ AwsMskIamAuth auth(config);
+
+ // Should create without error. Profile provider will be used as
AssumeRole base credentials provider.
+ ASSERT_TRUE(true);
+}
+
+TEST_F(AwsMskIamAuthTest, TestMultipleRegions) {
+ std::vector<std::string> regions = {"us-east-1", "us-west-2", "eu-west-1",
"ap-southeast-1"};
+
+ for (const auto& region : regions) {
+ config.region = region;
+ AwsMskIamAuth auth(config);
+
+ std::string broker_hostname = "b-1.test-msk." + region +
".amazonaws.com";
+ std::string token;
+ int64_t token_lifetime_ms;
+
+ Status status = auth.generate_token(broker_hostname, &token,
&token_lifetime_ms);
+
+ // We don't expect this to succeed without credentials,
+ // but it should fail gracefully without crashing
+ if (!status.ok()) {
+ LOG(INFO) << "Token generation for region " << region
+ << " failed (expected): " << status.to_string();
+ }
+ }
+}
+
+TEST_F(AwsMskIamAuthTest, TestOAuthCallbackCreation) {
+ auto auth = std::make_shared<AwsMskIamAuth>(config);
+ std::string broker_hostname = "b-1.test-msk.us-east-1.amazonaws.com";
+
+ AwsMskIamOAuthCallback callback(auth, broker_hostname);
+
+ // Callback should be created successfully
+ ASSERT_TRUE(true);
+}
+
+TEST_F(AwsMskIamAuthTest, TestOAuthCallbackCreationWithExternalIdAndRoleArn) {
+ std::unordered_map<std::string, std::string> properties {
+ {"security.protocol", "SASL_SSL"},
+ {"sasl.mechanism", "OAUTHBEARER"},
+ {"aws.region", "us-east-1"},
+ {"aws.role_arn", "arn:aws:iam::123456789012:role/TestRole"},
+ {"aws.external_id", "external-id-123"},
+ };
+
+ auto callback = AwsMskIamOAuthCallback::create_from_properties(
+ properties, "b-1.test-msk.us-east-1.amazonaws.com:9098");
+ ASSERT_NE(callback, nullptr);
+}
+
+TEST_F(AwsMskIamAuthTest,
TestOAuthCallbackCreationWithExternalIdWithoutRoleArn) {
+ std::unordered_map<std::string, std::string> properties {
+ {"security.protocol", "SASL_SSL"},
+ {"sasl.mechanism", "OAUTHBEARER"},
+ {"aws.region", "us-east-1"},
+ {"aws.external_id", "external-id-123"},
+ };
+
+ auto callback = AwsMskIamOAuthCallback::create_from_properties(
+ properties, "b-1.test-msk.us-east-1.amazonaws.com:9098");
+ ASSERT_EQ(callback, nullptr);
+}
+
+// Integration test - only runs if AWS credentials are available
+TEST_F(AwsMskIamAuthTest, DISABLED_IntegrationTestWithRealCredentials) {
+ // This test is disabled by default
+ // To run it, you need:
+ // 1. AWS credentials configured (environment variables,
~/.aws/credentials, or IAM role)
+ // 2. Run with: --gtest_also_run_disabled_tests
+
+ config.credentials_provider = "INSTANCE_PROFILE";
+
+ AwsMskIamAuth auth(config);
+ std::string token;
+ int64_t token_lifetime_ms;
+
+ std::string broker_hostname =
"b-1.real-msk-cluster.us-east-1.amazonaws.com";
+
+ Status status = auth.generate_token(broker_hostname, &token,
&token_lifetime_ms);
+
+ ASSERT_TRUE(status.ok()) << "Token generation failed: " <<
status.to_string();
+ ASSERT_FALSE(token.empty());
+ ASSERT_GT(token_lifetime_ms, 0);
+
+ LOG(INFO) << "Generated token (first 100 chars): "
+ << token.substr(0, std::min(size_t(100), token.size()));
+ LOG(INFO) << "Token lifetime: " << token_lifetime_ms << "ms";
+}
+
+} // namespace doris
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaConfiguration.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaConfiguration.java
index 959d88bb5da..259bb797b6d 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaConfiguration.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaConfiguration.java
@@ -46,6 +46,21 @@ public enum KafkaConfiguration {
KAFKA_TEXT_TABLE_NAME_FIELD_DELIMITER("kafka_text_table_name_field_delimiter",
",",
value -> value.replace(" ", ""));
+
+ // Common Kafka property names used for security and authentication
+ // These are passed through property.* prefix and validated for correctness
+ public static final String SECURITY_PROTOCOL = "security.protocol";
+ public static final String SASL_MECHANISM = "sasl.mechanism";
+ public static final String SASL_JAAS_CONFIG = "sasl.jaas.config";
+
+ // AWS MSK IAM authentication properties
+ public static final String AWS_REGION = "aws.region";
+ public static final String AWS_ACCESS_KEY = "aws.access_key";
+ public static final String AWS_SECRET_KEY = "aws.secret_key";
+ public static final String AWS_MSK_IAM_ROLE_ARN = "aws.role_arn";
+ public static final String AWS_EXTERNAL_ID = "aws.external_id";
+ public static final String AWS_PROFILE_NAME = "aws.profile_name";
+ public static final String AWS_CREDENTIALS_PROVIDER =
"aws.credentials_provider";
private final String name;
public String getName() {
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 47c4c3b19be..a5b1afb2cd3 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
@@ -34,6 +34,8 @@ import lombok.Setter;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
import java.util.Arrays;
import java.util.HashMap;
@@ -47,10 +49,12 @@ import java.util.regex.Pattern;
* Kafka data source properties
*/
public class KafkaDataSourceProperties extends AbstractDataSourceProperties {
+ private static final Logger LOG =
LogManager.getLogger(KafkaDataSourceProperties.class);
private static final String ENDPOINT_REGEX =
"[-A-Za-z0-9+&@#/%?=~_|!:,.;]+[-A-Za-z0-9+&@#/%=~_|]";
private static final String CUSTOM_KAFKA_PROPERTY_PREFIX = "property.";
+ private static final String AWS_PROPERTY_PREFIX = "aws.";
@Getter
@Setter
@@ -124,7 +128,8 @@ public class KafkaDataSourceProperties extends
AbstractDataSourceProperties {
public void convertAndCheckDataSourceProperties() throws UserException {
Optional<String> optional = originalDataSourceProperties.keySet()
.stream().filter(entity ->
!CONFIGURABLE_DATA_SOURCE_PROPERTIES_SET.contains(entity))
- .filter(entity ->
!entity.startsWith(CUSTOM_KAFKA_PROPERTY_PREFIX)).findFirst();
+ .filter(entity ->
!entity.startsWith(CUSTOM_KAFKA_PROPERTY_PREFIX))
+ .filter(entity ->
!entity.startsWith(AWS_PROPERTY_PREFIX)).findFirst();
if (optional.isPresent()) {
throw new AnalysisException(optional.get() + " is invalid kafka
property or can not be set");
}
@@ -233,16 +238,81 @@ public class KafkaDataSourceProperties extends
AbstractDataSourceProperties {
private void analyzeCustomProperties() throws AnalysisException {
this.customKafkaProperties = new HashMap<>();
for (Map.Entry<String, String> dataSourceProperty :
originalDataSourceProperties.entrySet()) {
- if
(dataSourceProperty.getKey().startsWith(CUSTOM_KAFKA_PROPERTY_PREFIX)) {
- String propertyKey = dataSourceProperty.getKey();
- String propertyValue = dataSourceProperty.getValue();
+ String propertyKey = dataSourceProperty.getKey();
+ String propertyValue = dataSourceProperty.getValue();
+ if (propertyKey.startsWith(CUSTOM_KAFKA_PROPERTY_PREFIX)) {
+ // property.xxx -> xxx (strip "property." prefix)
String[] propertyValueArr = propertyKey.split("\\.");
if (propertyValueArr.length < 2) {
throw new AnalysisException("kafka property value could
not be a empty string");
}
this.customKafkaProperties.put(propertyKey.substring(propertyKey.indexOf(".") +
1), propertyValue);
+ } else if (propertyKey.startsWith(AWS_PROPERTY_PREFIX)) {
+ // aws.xxx -> aws.xxx (keep as-is, no prefix stripping)
+ // AWS properties are Doris-specific and passed directly to BE
+ this.customKafkaProperties.put(propertyKey, propertyValue);
+ }
+ }
+
+ // Validate AWS MSK IAM authentication configuration if present
+ validateAwsMskIamConfig();
+ }
+
+ private void validateAwsMskIamConfig() throws AnalysisException {
+ String securityProtocol =
customKafkaProperties.get(KafkaConfiguration.SECURITY_PROTOCOL);
+ String saslMechanism =
customKafkaProperties.get(KafkaConfiguration.SASL_MECHANISM);
+
+ // Check if AWS MSK IAM authentication is being used
+ boolean hasAwsRegion =
customKafkaProperties.containsKey(KafkaConfiguration.AWS_REGION);
+ boolean hasAccessKey =
customKafkaProperties.containsKey(KafkaConfiguration.AWS_ACCESS_KEY);
+ boolean hasSecretKey =
customKafkaProperties.containsKey(KafkaConfiguration.AWS_SECRET_KEY);
+ boolean hasRoleArn =
customKafkaProperties.containsKey(KafkaConfiguration.AWS_MSK_IAM_ROLE_ARN);
+ boolean hasExternalId =
customKafkaProperties.containsKey(KafkaConfiguration.AWS_EXTERNAL_ID);
+ boolean hasProfileName =
customKafkaProperties.containsKey(KafkaConfiguration.AWS_PROFILE_NAME);
+ boolean hasCredentialsProvider =
customKafkaProperties.containsKey(KafkaConfiguration.AWS_CREDENTIALS_PROVIDER);
+
+ boolean isAwsMskIam = hasAwsRegion || hasAccessKey || hasSecretKey ||
hasRoleArn
+ || hasExternalId
+ || hasProfileName || hasCredentialsProvider;
+
+ // If AWS-related property is set, validate the complete configuration
+ if (isAwsMskIam) {
+ if (hasExternalId && !hasRoleArn) {
+ throw new AnalysisException("When using AWS MSK IAM
authentication, "
+ + "'aws.external_id' must be used with
'aws.role_arn'.");
+ }
+
+ // aws.region is required for AWS MSK IAM authentication
+ if (!hasAwsRegion) {
+ throw new AnalysisException("When using AWS MSK IAM
authentication, "
+ + "'aws.region' is required. Please specify the AWS
region.");
+ }
+
+ // security.protocol should be SASL_SSL for AWS MSK IAM
+ if (securityProtocol == null ||
!"SASL_SSL".equalsIgnoreCase(securityProtocol)) {
+ throw new AnalysisException("For AWS MSK IAM authentication, "
+ + "'property.security.protocol' must be 'SASL_SSL' "
+ + "(SSL encryption + SASL authentication required), "
+ + "but got: " + securityProtocol);
+ }
+
+ // sasl.mechanism should be OAUTHBEARER for AWS MSK IAM
+ if (saslMechanism == null ||
!"OAUTHBEARER".equalsIgnoreCase(saslMechanism)) {
+ throw new AnalysisException("For AWS MSK IAM authentication, "
+ + "'property.sasl.mechanism' must be 'OAUTHBEARER',
but got: " + saslMechanism + ". "
+ + "Other SASL mechanisms are not supported now.");
+ }
+
+ // Detect if this is public (internet) access or internal AWS
access
+ boolean isPublicAccess = brokerList != null &&
brokerList.contains("-public.");
+
+ if (isPublicAccess) {
+ if (!(hasAccessKey && hasSecretKey)) {
+ throw new AnalysisException(
+ "When using explicit AWS credentials for public
(internet) access, "
+ + "both 'aws.access_key' and 'aws.secret_key' must
be provided together.");
+ }
}
- // can be extended in the future which other prefix
}
}
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
new file mode 100644
index 00000000000..ed77128d02f
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaAwsMskIamAuthTest.java
@@ -0,0 +1,491 @@
+// 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.common.UserException;
+import org.apache.doris.load.routineload.kafka.KafkaDataSourceProperties;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Test cases for AWS MSK IAM authentication configuration in Routine Load.
+ */
+public class KafkaAwsMskIamAuthTest {
+
+ private Map<String, String> dataSourceProperties;
+
+ @Before
+ public void setUp() {
+ dataSourceProperties = new HashMap<>();
+ dataSourceProperties.put("kafka_broker_list",
"b-1.msk-cluster.xxx.kafka.us-east-1.amazonaws.com:9098");
+ dataSourceProperties.put("kafka_topic", "test_topic");
+ }
+
+ @Test
+ public void testValidAwsMskIamConfig() throws UserException {
+ // Test with OAUTHBEARER mechanism (correct way for AWS MSK IAM)
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+ dataSourceProperties.put("property.aws.role_arn",
"arn:aws:iam::123456789012:role/MyMskRole");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+ props.analyze();
+
+ Assert.assertNotNull(props.getCustomKafkaProperties());
+ Assert.assertEquals("SASL_SSL",
props.getCustomKafkaProperties().get("security.protocol"));
+ Assert.assertEquals("OAUTHBEARER",
props.getCustomKafkaProperties().get("sasl.mechanism"));
+ Assert.assertEquals("us-east-1",
props.getCustomKafkaProperties().get("aws.region"));
+ }
+
+ @Test
+ public void testValidOAuthBearerConfig() throws UserException {
+ // Test with OAUTHBEARER mechanism (alternative approach)
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+ props.analyze();
+
+ Assert.assertNotNull(props.getCustomKafkaProperties());
+ Assert.assertEquals("SASL_SSL",
props.getCustomKafkaProperties().get("security.protocol"));
+ Assert.assertEquals("OAUTHBEARER",
props.getCustomKafkaProperties().get("sasl.mechanism"));
+ }
+
+ @Test
+ public void testMissingSecurityProtocol() {
+ // Test missing security.protocol when using AWS MSK IAM
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+
+ try {
+ props.analyze();
+ Assert.fail("Should throw AnalysisException for missing
security.protocol");
+ } catch (UserException e) {
+ Assert.assertTrue(e.getMessage().contains("security.protocol"));
+ Assert.assertTrue(e.getMessage().contains("SASL_SSL"));
+ }
+ }
+
+ @Test
+ public void testInvalidSecurityProtocol() {
+ // Test invalid security.protocol value
+ dataSourceProperties.put("property.security.protocol", "PLAINTEXT");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+
+ try {
+ props.analyze();
+ Assert.fail("Should throw AnalysisException for invalid
security.protocol");
+ } catch (UserException e) {
+ Assert.assertTrue(e.getMessage().contains("SASL_SSL"));
+ }
+ }
+
+ @Test
+ public void testMissingSaslMechanism() {
+ // Test missing sasl.mechanism when using SASL_SSL with AWS
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+
+ try {
+ props.analyze();
+ Assert.fail("Should throw AnalysisException for missing
sasl.mechanism");
+ } catch (UserException e) {
+ Assert.assertTrue(e.getMessage().contains("sasl.mechanism"));
+ Assert.assertTrue(e.getMessage().contains("OAUTHBEARER"));
+ }
+ }
+
+ @Test
+ public void testInvalidSaslMechanism() {
+ // Test invalid sasl.mechanism value when using AWS MSK IAM
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "PLAIN");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+
+ try {
+ props.analyze();
+ Assert.fail("Should throw AnalysisException for invalid
sasl.mechanism with AWS config");
+ } catch (UserException e) {
+ Assert.assertTrue(e.getMessage().contains("OAUTHBEARER"));
+ }
+ }
+
+ @Test
+ public void testMissingRegionWithRoleArn() {
+ // Test with role ARN but missing region - should fail
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.role_arn",
"arn:aws:iam::123456789012:role/MyMskRole");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+
+ try {
+ props.analyze();
+ Assert.fail("Should throw AnalysisException for missing region
with AWS properties");
+ } catch (UserException e) {
+ Assert.assertTrue(e.getMessage().contains("aws.region"));
+ Assert.assertTrue(e.getMessage().contains("required"));
+ }
+ }
+
+ @Test
+ public void testCompleteAwsIamConfigWithRoleArn() throws UserException {
+ // Test complete configuration with IAM role ARN
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+ dataSourceProperties.put("property.aws.role_arn",
"arn:aws:iam::123456789012:role/MyMskRole");
+ dataSourceProperties.put("property.aws.profile_name", "default");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+ props.analyze();
+
+ Map<String, String> customProps = props.getCustomKafkaProperties();
+ Assert.assertEquals("SASL_SSL", customProps.get("security.protocol"));
+ Assert.assertEquals("OAUTHBEARER", customProps.get("sasl.mechanism"));
+ Assert.assertEquals("us-east-1", customProps.get("aws.region"));
+ Assert.assertEquals("arn:aws:iam::123456789012:role/MyMskRole",
customProps.get("aws.role_arn"));
+ Assert.assertEquals("default", customProps.get("aws.profile_name"));
+ }
+
+ @Test
+ public void testExternalIdWithRoleArn() throws UserException {
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+ dataSourceProperties.put("property.aws.role_arn",
"arn:aws:iam::123456789012:role/MyMskRole");
+ dataSourceProperties.put("property.aws.external_id",
"external-id-123");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+ props.analyze();
+
+ Map<String, String> customProps = props.getCustomKafkaProperties();
+ Assert.assertEquals("external-id-123",
customProps.get("aws.external_id"));
+ }
+
+ @Test
+ public void testExternalIdWithoutRoleArn() {
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+ dataSourceProperties.put("property.aws.external_id",
"external-id-123");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+
+ try {
+ props.analyze();
+ Assert.fail("Should throw AnalysisException for external_id
without role_arn");
+ } catch (UserException e) {
+ Assert.assertTrue(e.getMessage().contains("aws.external_id"));
+ Assert.assertTrue(e.getMessage().contains("aws.role_arn"));
+ }
+ }
+
+ @Test
+ public void testScramSha256Config() throws UserException {
+ // Test SCRAM-SHA-256 configuration (non-IAM SASL)
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "SCRAM-SHA-256");
+ dataSourceProperties.put("property.sasl.username", "myuser");
+ dataSourceProperties.put("property.sasl.password", "mypassword");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+ props.analyze();
+
+ Assert.assertNotNull(props.getCustomKafkaProperties());
+ Assert.assertEquals("SASL_SSL",
props.getCustomKafkaProperties().get("security.protocol"));
+ Assert.assertEquals("SCRAM-SHA-256",
props.getCustomKafkaProperties().get("sasl.mechanism"));
+ }
+
+ @Test
+ public void testPlaintextConfigWithoutSasl() throws UserException {
+ // Test plain PLAINTEXT configuration (no SASL)
+ dataSourceProperties.put("property.security.protocol", "PLAINTEXT");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+ props.analyze();
+
+ Assert.assertNotNull(props.getCustomKafkaProperties());
+ Assert.assertEquals("PLAINTEXT",
props.getCustomKafkaProperties().get("security.protocol"));
+ }
+
+ @Test
+ 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");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+ props.analyze();
+
+ Assert.assertNotNull(props.getCustomKafkaProperties());
+ Assert.assertEquals("SSL",
props.getCustomKafkaProperties().get("security.protocol"));
+ }
+
+ @Test
+ public void testPublicAccessWithExplicitCredentials() throws UserException
{
+ // Test public MSK access (broker contains "-public.") with explicit
credentials
+ dataSourceProperties.put("kafka_broker_list",
"b-1-public.msk-cluster.xxx.kafka.us-east-1.amazonaws.com:9198");
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+ dataSourceProperties.put("property.aws.access_key",
"AKIAIOSFODNN7EXAMPLE");
+ dataSourceProperties.put("property.aws.secret_key",
"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+ props.analyze();
+
+ Assert.assertNotNull(props.getCustomKafkaProperties());
+ Assert.assertEquals("us-east-1",
props.getCustomKafkaProperties().get("aws.region"));
+ Assert.assertEquals("AKIAIOSFODNN7EXAMPLE",
props.getCustomKafkaProperties().get("aws.access_key"));
+ }
+
+ @Test
+ public void testPublicAccessWithoutCredentials() {
+ // Test public MSK access without credentials - should fail
+ dataSourceProperties.put("kafka_broker_list",
"b-1-public.msk-cluster.xxx.kafka.us-east-1.amazonaws.com:9198");
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+
+ try {
+ props.analyze();
+ Assert.fail("Should throw AnalysisException for public access
without credentials");
+ } catch (UserException e) {
+ Assert.assertTrue(e.getMessage().contains("public"));
+ Assert.assertTrue(e.getMessage().contains("credentials"));
+ }
+ }
+
+ @Test
+ public void testInternalAccessWithInstanceProfile() throws UserException {
+ // Test internal MSK access with instance profile
(credentials_provider)
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+ dataSourceProperties.put("property.aws.credentials_provider",
"INSTANCE_PROFILE");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+ props.analyze();
+
+ Assert.assertNotNull(props.getCustomKafkaProperties());
+ Assert.assertEquals("INSTANCE_PROFILE",
props.getCustomKafkaProperties().get("aws.credentials_provider"));
+ }
+
+ @Test
+ public void testRoleArnWithCredentialsProvider() throws UserException {
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+ dataSourceProperties.put("property.aws.role_arn",
"arn:aws:iam::123456789012:role/MyMskRole");
+ dataSourceProperties.put("property.aws.credentials_provider", "ENV");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+ props.analyze();
+
+ Assert.assertNotNull(props.getCustomKafkaProperties());
+ Assert.assertEquals("arn:aws:iam::123456789012:role/MyMskRole",
+ props.getCustomKafkaProperties().get("aws.role_arn"));
+ Assert.assertEquals("ENV",
props.getCustomKafkaProperties().get("aws.credentials_provider"));
+ }
+
+ @Test
+ public void testInternalAccessWithProfile() throws UserException {
+ // Test internal MSK access with AWS profile
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+ dataSourceProperties.put("property.aws.profile_name", "default");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+ props.analyze();
+
+ Assert.assertNotNull(props.getCustomKafkaProperties());
+ Assert.assertEquals("default",
props.getCustomKafkaProperties().get("aws.profile_name"));
+ }
+
+ @Test
+ public void testCrossAccountAccessWithRoleArnAndCredentials() throws
UserException {
+ // Test cross-account access: Account B's credentials + Account A's
role ARN
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+ dataSourceProperties.put("property.aws.access_key",
"AKIAIOSFODNN7EXAMPLE");
+ dataSourceProperties.put("property.aws.secret_key",
"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY");
+ dataSourceProperties.put("property.aws.role_arn",
"arn:aws:iam::111111111111:role/AccountAMskRole");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+ props.analyze();
+
+ Map<String, String> customProps = props.getCustomKafkaProperties();
+ Assert.assertEquals("AKIAIOSFODNN7EXAMPLE",
customProps.get("aws.access_key"));
+ Assert.assertEquals("arn:aws:iam::111111111111:role/AccountAMskRole",
customProps.get("aws.role_arn"));
+ }
+
+ @Test
+ public void testMissingAccessKeyWithSecretKeyPublicAccess() {
+ // Test with only secret key but missing access key for PUBLIC access
- should fail
+ dataSourceProperties.put("kafka_broker_list",
"b-1-public.msk-cluster.xxx.kafka.us-east-1.amazonaws.com:9198");
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+ dataSourceProperties.put("property.aws.secret_key",
"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+
+ try {
+ props.analyze();
+ Assert.fail("Should throw AnalysisException for missing access
key");
+ } catch (UserException e) {
+ Assert.assertTrue(e.getMessage().contains("aws.access_key"));
+ Assert.assertTrue(e.getMessage().contains("aws.secret_key"));
+ Assert.assertTrue(e.getMessage().contains("together"));
+ }
+ }
+
+ @Test
+ public void testMissingSecretKeyWithAccessKeyPublicAccess() {
+ // Test with only access key but missing secret key for PUBLIC access
- should fail
+ dataSourceProperties.put("kafka_broker_list",
"b-1-public.msk-cluster.xxx.kafka.us-east-1.amazonaws.com:9198");
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+ dataSourceProperties.put("property.aws.access_key",
"AKIAIOSFODNN7EXAMPLE");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+
+ try {
+ props.analyze();
+ Assert.fail("Should throw AnalysisException for missing secret
key");
+ } catch (UserException e) {
+ Assert.assertTrue(e.getMessage().contains("aws.access_key"));
+ Assert.assertTrue(e.getMessage().contains("aws.secret_key"));
+ Assert.assertTrue(e.getMessage().contains("together"));
+ }
+ }
+
+ @Test
+ public void testPublicAccessWithProfileOnly() {
+ // Test public access with AWS profile only (no explicit credentials)
- should fail
+ // because public access requires explicit credentials
+ dataSourceProperties.put("kafka_broker_list",
"b-1-public.msk-cluster.xxx.kafka.us-east-1.amazonaws.com:9198");
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+ dataSourceProperties.put("property.aws.profile_name", "default");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+
+ try {
+ props.analyze();
+ Assert.fail("Should throw AnalysisException for public access
without explicit credentials");
+ } catch (UserException e) {
+ Assert.assertTrue(e.getMessage().contains("aws.access_key"));
+ Assert.assertTrue(e.getMessage().contains("aws.secret_key"));
+ Assert.assertTrue(e.getMessage().contains("together"));
+ }
+ }
+
+ @Test
+ public void testPublicAccessWithCredentialsProviderOnly() {
+ // Test public access with credentials provider only (no explicit
credentials) - should fail
+ // because public access requires explicit credentials
+ dataSourceProperties.put("kafka_broker_list",
"b-1-public.msk-cluster.xxx.kafka.us-east-1.amazonaws.com:9198");
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+ dataSourceProperties.put("property.aws.credentials_provider", "ENV");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+
+ try {
+ props.analyze();
+ Assert.fail("Should throw AnalysisException for public access
without explicit credentials");
+ } catch (UserException e) {
+ Assert.assertTrue(e.getMessage().contains("aws.access_key"));
+ Assert.assertTrue(e.getMessage().contains("aws.secret_key"));
+ Assert.assertTrue(e.getMessage().contains("together"));
+ }
+ }
+
+ @Test
+ public void testMultipleCredentialsSources() throws UserException {
+ // Test with multiple credentials sources - all should be allowed
+ dataSourceProperties.put("property.security.protocol", "SASL_SSL");
+ dataSourceProperties.put("property.sasl.mechanism", "OAUTHBEARER");
+ dataSourceProperties.put("property.aws.region", "us-east-1");
+ dataSourceProperties.put("property.aws.access_key",
"AKIAIOSFODNN7EXAMPLE");
+ dataSourceProperties.put("property.aws.secret_key",
"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY");
+ dataSourceProperties.put("property.aws.role_arn",
"arn:aws:iam::123456789012:role/MyRole");
+ dataSourceProperties.put("property.aws.external_id",
"external-id-123");
+ dataSourceProperties.put("property.aws.profile_name", "default");
+
+ KafkaDataSourceProperties props = new
KafkaDataSourceProperties(dataSourceProperties);
+ props.setTimezone("UTC");
+ props.analyze();
+
+ // All properties should be preserved (BE will use them in priority
order)
+ Map<String, String> customProps = props.getCustomKafkaProperties();
+ Assert.assertEquals("AKIAIOSFODNN7EXAMPLE",
customProps.get("aws.access_key"));
+ Assert.assertEquals("arn:aws:iam::123456789012:role/MyRole",
customProps.get("aws.role_arn"));
+ Assert.assertEquals("external-id-123",
customProps.get("aws.external_id"));
+ Assert.assertEquals("default", customProps.get("aws.profile_name"));
+ }
+}
diff --git a/regression-test/conf/regression-conf.groovy
b/regression-test/conf/regression-conf.groovy
index 17257831333..21f4bd047d8 100644
--- a/regression-test/conf/regression-conf.groovy
+++ b/regression-test/conf/regression-conf.groovy
@@ -90,6 +90,7 @@ excludeDirectories = "000_the_start_sentinel_do_not_touch," +
// keep this line
"segcompaction_p2," +
"workload_manager_p1," +
"cloud_p0/cache," +
+ "load_p0/msk_load," +
"zzz_the_end_sentinel_do_not_touch," +
"dialect_compatible"// keep this line as the last line
diff --git
a/regression-test/suites/load_p0/msk_load/test_routine_load_msk.groovy
b/regression-test/suites/load_p0/msk_load/test_routine_load_msk.groovy
new file mode 100644
index 00000000000..63fc38e6e13
--- /dev/null
+++ b/regression-test/suites/load_p0/msk_load/test_routine_load_msk.groovy
@@ -0,0 +1,177 @@
+// 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.
+
+import com.google.common.base.Strings;
+
+suite("test_routine_load_msk_access") {
+ def topicName = "routineload-test"
+ def tableName = "test_routine_load_iam_aksk_table"
+ def jobName = "test_routine_load_iam_aksk_job"
+
+ if
(Strings.isNullOrEmpty(context.config.otherConfigs.get("mskBrokerList"))) {
+ return
+ }
+
+ String ak = context.config.otherConfigs.get("awsMskAccessKey")
+ String sk = context.config.otherConfigs.get("awsMskSecretKey")
+ String region = context.config.otherConfigs.get("awsMskRegion")
+ String bootstrapBrokers = context.config.otherConfigs.get("mskBrokerList")
+ String role = context.config.otherConfigs.get("awsMskRole")
+ String sslCaLocation = context.config.otherConfigs.get("MskSslCaLocation")
+ String credProvider =
context.config.otherConfigs.get("CredentialsProvider")
+
+ logger.info("Using topic: ${topicName}")
+ logger.info("Bootstrap brokers: ${bootstrapBrokers}")
+ logger.info("AWS region: ${region}")
+
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ sql """
+ CREATE TABLE ${tableName} (
+ id INT,
+ name VARCHAR(50)
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ sql """
+ CREATE ROUTINE LOAD ${jobName} ON ${tableName}
+ COLUMNS TERMINATED BY ",",
+ COLUMNS(id, name)
+ FROM KAFKA (
+ "kafka_broker_list" = "${bootstrapBrokers}",
+ "kafka_topic" = "${topicName}",
+ "aws.region" = "${region}",
+ "aws.access_key" = "${ak}",
+ "aws.secret_key" = "${sk}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING",
+ "property.security.protocol" = "SASL_SSL",
+ "property.sasl.mechanism" = "OAUTHBEARER",
+ "property.ssl.ca.location" = "${sslCaLocation}"
+ )
+ """
+ logger.info("Created routine load job: ${jobName}")
+
+ def maxRetries = 60
+ def retryCount = 0
+ def jobState = ""
+
+ while (retryCount < maxRetries) {
+ Thread.sleep(5000)
+
+ def jobStatus = sql """ SHOW ROUTINE LOAD FOR ${jobName} """
+ if (jobStatus.size() > 0) {
+ jobState = jobStatus[0][8]
+ logger.info("Job state: ${jobState}, retry: ${retryCount}")
+
+ if (jobState == "RUNNING") {
+ def result = sql """ SELECT COUNT(*) FROM ${tableName} """
+ if (result[0][0] > 0) {
+ logger.info("Data loaded successfully: ${result[0][0]}
rows")
+ break
+ }
+ } else if (jobState == "PAUSED" || jobState == "CANCELLED") {
+ logger.error("Job failed with state: ${jobState}")
+ logger.error("Job status: ${jobStatus[0]}")
+ break
+ }
+ }
+ retryCount++
+ }
+
+ def result = sql """ SELECT COUNT(*) FROM ${tableName} """
+ logger.info("Final row count: ${result[0][0]}")
+
+ assertTrue(result[0][0] > 0, "No data loaded from MSK. Job state:
${jobState}")
+
+ def sampleData = sql """ SELECT * FROM ${tableName} LIMIT 5 """
+ logger.info("Sample data: ${sampleData}")
+
+ sql """ STOP ROUTINE LOAD FOR ${jobName} """
+
+ // Test with role arn if configured
+ if (role) {
+ def roleTableName = "test_routine_load_iam_role_table"
+ def roleJobName = "test_routine_load_iam_role_job"
+
+ logger.info("Testing with AWS Role ARN: ${role}")
+
+ sql """ DROP TABLE IF EXISTS ${roleTableName} """
+ sql """
+ CREATE TABLE ${roleTableName} (
+ id INT,
+ name VARCHAR(50)
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ sql """
+ CREATE ROUTINE LOAD ${roleJobName} ON ${roleTableName}
+ COLUMNS TERMINATED BY ",",
+ COLUMNS(id, name)
+ FROM KAFKA (
+ "kafka_broker_list" = "${bootstrapBrokers}",
+ "kafka_topic" = "${topicName}",
+ "aws.region" = "${region}",
+ "aws.role_arn" = "${role}",
+ "aws.credentials_provider" = "${credProvider}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING",
+ "property.security.protocol" = "SASL_SSL",
+ "property.sasl.mechanism" = "OAUTHBEARER",
+ "property.ssl.ca.location" = "${sslCaLocation}"
+ )
+ """
+ logger.info("Created routine load job with role: ${roleJobName}")
+
+ retryCount = 0
+ jobState = ""
+
+ while (retryCount < maxRetries) {
+ Thread.sleep(5000)
+
+ def jobStatus = sql """ SHOW ROUTINE LOAD FOR ${roleJobName} """
+ if (jobStatus.size() > 0) {
+ jobState = jobStatus[0][8]
+ logger.info("Role job state: ${jobState}, retry:
${retryCount}")
+
+ if (jobState == "RUNNING") {
+ def roleResult = sql """ SELECT COUNT(*) FROM
${roleTableName} """
+ if (roleResult[0][0] > 0) {
+ logger.info("Data loaded with role successfully:
${roleResult[0][0]} rows")
+ break
+ }
+ } else if (jobState == "PAUSED" || jobState == "CANCELLED") {
+ logger.error("Role job failed with state: ${jobState}")
+ logger.error("Job status: ${jobStatus[0]}")
+ break
+ }
+ }
+ retryCount++
+ }
+
+ def roleResult = sql """ SELECT COUNT(*) FROM ${roleTableName} """
+ logger.info("Role job final row count: ${roleResult[0][0]}")
+
+ assertTrue(roleResult[0][0] > 0, "No data loaded with role ARN. Job
state: ${jobState}")
+
+ sql """ STOP ROUTINE LOAD FOR ${roleJobName} """
+ logger.info("Role ARN test completed successfully")
+ }
+}
diff --git
a/regression-test/suites/load_p0/msk_load/test_routine_load_msk_public_access.groovy
b/regression-test/suites/load_p0/msk_load/test_routine_load_msk_public_access.groovy
new file mode 100644
index 00000000000..cb25324b119
--- /dev/null
+++
b/regression-test/suites/load_p0/msk_load/test_routine_load_msk_public_access.groovy
@@ -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.
+
+import com.google.common.base.Strings;
+
+suite("test_routine_load_msk_public_access") {
+ def topicName = "routineload-test"
+ def tableName = "test_routine_load_iam_aksk_table"
+ def jobName = "test_routine_load_iam_aksk_job"
+
+ if
(Strings.isNullOrEmpty(context.config.otherConfigs.get("publicMskBrokerList")))
{
+ return
+ }
+
+ def ak = context.config.awsMskAccessKey
+ def sk = context.config.awsMskSecretKey
+ def region = context.config.awsMskRegion
+ String bootstrapBrokers = context.config.otherConfigs.get("mskBrokerList")
+ String role = context.config.otherConfigs.get("awsMskRole")
+ String sslCaLocation = context.config.otherConfigs.get("MskSslCaLocation")
+
+ logger.info("Using topic: ${topicName}")
+ logger.info("Bootstrap brokers: ${bootstrapBrokers}")
+ logger.info("AWS region: ${region}")
+
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ sql """
+ CREATE TABLE ${tableName} (
+ id INT,
+ name VARCHAR(50)
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ sql """
+ CREATE ROUTINE LOAD ${jobName} ON ${tableName}
+ COLUMNS TERMINATED BY ",",
+ COLUMNS(id, name)
+ FROM KAFKA (
+ "kafka_broker_list" = "${bootstrapBrokers}",
+ "kafka_topic" = "${topicName}",
+ "aws.region" = "${region}",
+ "aws.access_key" = "${ak}",
+ "aws.secret_key" = "${sk}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING",
+ "property.security.protocol" = "SASL_SSL",
+ "property.sasl.mechanism" = "OAUTHBEARER",
+ "property.ssl.ca.location" = "${sslCaLocation}"
+ )
+ """
+ logger.info("Created routine load job: ${jobName}")
+
+ def maxRetries = 60
+ def retryCount = 0
+ def jobState = ""
+
+ while (retryCount < maxRetries) {
+ Thread.sleep(5000)
+
+ def jobStatus = sql """ SHOW ROUTINE LOAD FOR ${jobName} """
+ if (jobStatus.size() > 0) {
+ jobState = jobStatus[0][8]
+ logger.info("Job state: ${jobState}, retry: ${retryCount}")
+
+ if (jobState == "RUNNING") {
+ def result = sql """ SELECT COUNT(*) FROM ${tableName} """
+ if (result[0][0] > 0) {
+ logger.info("Data loaded successfully: ${result[0][0]}
rows")
+ break
+ }
+ } else if (jobState == "PAUSED" || jobState == "CANCELLED") {
+ logger.error("Job failed with state: ${jobState}")
+ logger.error("Job status: ${jobStatus[0]}")
+ break
+ }
+ }
+ retryCount++
+ }
+
+ def result = sql """ SELECT COUNT(*) FROM ${tableName} """
+ logger.info("Final row count: ${result[0][0]}")
+
+ assertTrue(result[0][0] > 0, "No data loaded from MSK. Job state:
${jobState}")
+
+ def sampleData = sql """ SELECT * FROM ${tableName} LIMIT 5 """
+ logger.info("Sample data: ${sampleData}")
+
+ sql """ STOP ROUTINE LOAD FOR ${jobName} """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
