feature: Add AWS MSK IAM Authentication Support for Kafka Engine by kalavt · Pull Request #91118 · ClickHouse/ClickHouse

kalavt · 2025-11-28T11:57:17Z

Changelog category (leave one):

New Feature

Changelog entry (a user-readable short description of the changes that goes into CHANGELOG.md):

Add support for AWS MSK IAM authentication for Kafka.

Documentation entry for user-facing changes

Documentation is written (mandatory for new features)

Description

This PR implements AWS MSK IAM authentication support for the Kafka Table Engine.
It allows ClickHouse to authenticate with Amazon Managed Streaming for Apache Kafka (MSK) using IAM roles.

Motivation:
Many users running ClickHouse on AWS want to connect to MSK using the standard IAM authentication method for better security and management, rather than managing SASL/SCRAM credentials or mTLS certificates.

Changes:

Added AWSMSKIAMAuth class (src/Storages/Kafka/AWSMSKIAMAuth.cpp/h) to handle IAM authentication logic.
Updated StorageKafka and KafkaConsumer to use the new authentication mechanism.
Added configuration settings for AWS MSK IAM auth.
Updated librdkafka-cmake to support the necessary build options.
Updated documentation with usage examples.

CLAassistant · 2025-11-28T11:57:24Z

All committers have signed the CLA.

kalavt · 2025-11-28T14:38:55Z

For implement #80446

kalavt · 2025-12-01T08:28:23Z

@antaljanosbenjamin can you pls have a check on this PR, thanks.

antaljanosbenjamin

In general looks good. There are a few general comments:

Could you please also update the regarding docs?
Have you thought about using "assumed roles" similarly to the S3 table function? I think it is implemented for S3, it shouldn't be hard to enable it here too.
I have to talk with some security people, but we might need a setting to disable this globally if there is some issue regarding security. Please don't act on this yet, I will try to get a precise answer on that and understand it better.

clickhouse-gh · 2025-12-01T12:33:23Z

Workflow [PR], commit [3a8f50a]

Summary: ❌

job_name	test_name	status	info
Style check		failure
	cpp	failure	cidb
Build (arm_tidy)		failure
	Build ClickHouse	failure	cidb
Docs check		dropped
Build (amd_debug)		dropped
Build (amd_asan)		dropped
Build (amd_tsan)		dropped
Build (amd_msan)		dropped
Build (amd_ubsan)		dropped
Build (amd_binary)		dropped
Build (arm_asan)		dropped

antaljanosbenjamin · 2025-12-05T15:43:59Z

The changes in c-ares are intentional?

kalavt · 2025-12-08T04:08:12Z

@antaljanosbenjamin for your comments:

The changes in c-ares are intentional?

Regarding the c-ares submodule changes visible in the PR:

I can confirm that I have not made any direct modifications to the contrib/c-ares submodule. The changes appearing in the PR diff are due to the submodule version difference between my feature branch and the current master branch.

Here's what happened:

My feature branch was created from an earlier master commit
After branch creation, master updated the c-ares submodule to a newer version
The PR comparison now shows this submodule pointer difference as part of the overall diff

Could you please also update the regarding docs?

updated, or any other docs need update, pls let me know.

Have you thought about using "assumed roles" similarly to the S3 table function? I think it is implemented for S3, it shouldn't be hard to enable it here too.

Yes, but should be by another PR.

I have to talk with some security people, but we might need a setting to disable this globally if there is some issue regarding security. Please don't act on this yet, I will try to get a precise answer on that and understand it better.

sure. Thanks for checking,

antaljanosbenjamin · 2025-12-10T14:03:34Z

So for the submodules, please revert those changes locally. The issue is in my opinion is the submodules were updated on master, you didn't updated them locally but committed them. This means if we would merge your PR, it would revert the changes in submodules.

For the credentials, there are two things. First of all, please use our S3CredentialsProviderChain that can be found here:

ClickHouse/src/IO/S3/Credentials.h

Lines 211 to 218 in 7d0b74a

    
           class S3CredentialsProviderChain : public Aws::Auth::AWSCredentialsProviderChain 
        
           { 
        
           public: 
        
               S3CredentialsProviderChain( 
        
                   const DB::S3::PocoHTTPClientConfiguration & configuration, 
        
                   const Aws::Auth::AWSCredentials & credentials, 
        
                   CredentialsConfiguration credentials_configuration); 
        
           };

Second, when creating such provider chain, please make sure use_environment_credentials can be enabled only from server configuration and not by user queries. I think for now it is okay if it is just a boolean flag without any further details, so it is either enabled or disabled to all URL.

If you have any questions, please don't hesitate to ask.

kalavt · 2025-12-10T17:58:24Z

@antaljanosbenjamin for your comments, I've solved the submodules changes, using S3CredentialsProviderChain with use_environment_credentials for loading credentials.

thanks for your guidance and point out the security concern.

antaljanosbenjamin

A few small things.

kalavt · 2025-12-20T14:26:44Z

Hi @antaljanosbenjamin

I've completely re-design the feature and tested it works, would you please review my code and let me know if you have any concern. Thanks.

kalavt · 2025-12-25T07:49:27Z

@antaljanosbenjamin can you pls guide me for next step merge this PR?

antaljanosbenjamin · 2025-12-25T09:20:22Z

I will have a look at it next week on the first day I am working. Sorry for the delay, but I am on vacation.

antaljanosbenjamin · 2026-01-02T14:45:53Z

+# it does NOT require Cyrus SASL library. By enabling it unconditionally
+# when WITH_SSL is set, we can support AWS MSK IAM authentication and
+# other OAUTHBEARER mechanisms without needing the full Cyrus SASL stack.
+set(WITH_SASL_OAUTHBEARER 1)


I have to verify this locally, so this is a note to myself.

antaljanosbenjamin · 2026-01-02T14:58:23Z

+            {
+                std::lock_guard lock(g_provider_cache.mutex);
+                if (!g_provider_cache.provider)
+                {
+                    auto aws_client_configuration = DB::S3::ClientFactory::instance().createClientConfiguration(
+                        region, {}, 0, {}, false, false, false, false, {}, {});
+
+                    S3::CredentialsConfiguration credentials_configuration;
+                    credentials_configuration.use_environment_credentials = use_environment_credentials;
+
+                    g_provider_cache.provider = std::make_shared<S3::S3CredentialsProviderChain>(
+                        aws_client_configuration, Aws::Auth::AWSCredentials{}, credentials_configuration);
+                }
+                provider = g_provider_cache.provider;
+            }


What happens if the region is different in the current config then it was when the cache was created?

I might miss something, but I don't understand why the storage cannot own the credentials provider. it is tight to a single storage, clear ownership and scope. Now we have a global object which is shared among multiple storages.

Hi. @antaljanosbenjamin

sorry for the late replay, thought you still on vacation :)

You are absolutely right. The global cache was problematic because it didn't account for different regions across multiple storages and introduced unnecessary global state.

I have refactored the implementation to ensure proper scoping and ownership:

Removed Global Cache: g_provider_cache has been removed entirely.

Scoped Ownership: The S3CredentialsProviderChain is now stored inside OAuthBearerTokenRefreshContext.

Storage Ownership: StorageKafka (and StorageKafka2) now owns the OAuthBearerTokenRefreshContext (via a std::shared_ptr member). This ensures the credentials provider's lifecycle is tied directly to the storage instance and uses the correct configuration (region) for that specific table.

Safe Context Passing: I'm passing the context pointer via the configuration string to avoid conflicts with cppkafka's use of the opaque pointer, ensuring we can access this scoped context safely in the callback.

This design eliminates the risk of region mismatch and removes the global state as suggested.

- Implements AWS MSK IAM authentication for Kafka storage. - Updates librdkafka-cmake configuration.

antaljanosbenjamin

A few comments, please don't use force push.

antaljanosbenjamin · 2026-01-29T20:27:29Z

    using Base = WriteBufferFromVectorImpl<std::string>;
 public:
-    WriteBufferFromOwnString()
+    WriteBufferFromOwnString() // NOLINT(modernize-use-equals-default, hicpp-use-equals-default)


Is this necessary?

The constructor cannot use = default because it must initialize the base class WriteBufferFromVectorImpl<std::string> with the value member inherited from StringHolder. A defaulted constructor cannot include member initializers, so the NOLINT suppression is required here.

antaljanosbenjamin · 2026-01-29T20:49:39Z

+            String aws_region = kafka_settings[KafkaSetting::kafka_aws_region].value;
+
+            String broker_list = kafka_config.has_property("metadata.broker.list")
+                ? kafka_config.get("metadata.broker.list")
+                : "";
+
+            AWSMSKIAMAuth::setupAuthentication(
+                kafka_config, params.config, aws_region, broker_list, params.log, storage.getOAuthContext());


Correct me if I am wrong, but this should only happen on the first occasion, no?

So I would say a slightly better logic considering my other comment about not returning a non-const reference to internal members:

check if the storage has a context or not (returning a non const raw pointer is perfect for this use-case)

if not, then create one and use the setOAuthContext function to assign it to the storage

if there is already a context, then use that.

You're correct. The OAuth context initialization is expensive and should be cached. I'll refactor this part.

antaljanosbenjamin · 2026-01-29T20:52:11Z


+    // Enable background SASL callbacks for OAUTHBEARER authentication.
+    if (auto * error = rd_kafka_sasl_background_callbacks_enable(consumer->get_handle()))
+    {
+        // If SASL background callbacks cannot be enabled (e.g., not configured or not supported),
+        // we still continue since this is not a critical error for basic Kafka functionality
+        rd_kafka_error_destroy(error);
+    }


I think we don't need this callback. My reasoning is if the consumer is not polled, then we don't need to refresh the auth secrets.

The background SASL callbacks are necessary for handling low-traffic topics where the consumer may be idle for extended periods. Without them, OAuth tokens can expire during idle periods, causing unrecoverable authentication failures when the consumer attempts to poll again. This was specifically implemented to ensure consumers remain operational even on low-volume topics where polls are infrequent.

antaljanosbenjamin · 2026-01-29T21:04:59Z

+        // NOTE: We cannot use the global 'opaque' pointer because it is used by cppkafka
+        // to store the Consumer/Producer instance. Overwriting it would break cppkafka callbacks.
+        // Instead, we pass the context pointer address in the configuration string.


I am not sure about this. This feels a very hacky solution. I think the proper solution is to add this functionality to cppkafka first and then we can utilize it here. Passing opaque pointers is fine, passing pointers as strings is not in my opnion. Technically they are the same, but it seems like a code smell. Feel free open a PR to https://github.com/ClickHouse/cppkafka and tag me.

yah that's a workaround temporary not the ideal solution, for now, should we keep the current implementation as a temporary workaround until the cppkafka enhancement is merged,
or would you prefer to block this PR until the proper solution is in place?

well, I've proposed a PR to https://github.com/ClickHouse/cppkafka to support OAUTHBEARER token refresh callback.
ClickHouse/cppkafka#5
and I'll refactor the Clickhouse AWS MSK IAM auth implementation to new PR with newly added refresh callback function #96100

antaljanosbenjamin · 2026-01-29T21:09:30Z

+        const char * oauthbearer_config,
+        void * /* opaque */)
+    {
+        LoggerPtr log = getLogger("AWSMSKIAMAuth");


And please, remove this log.

antaljanosbenjamin · 2026-01-29T21:17:59Z

+    params.emplace_back("region", effective_region);
+    params.emplace_back("use_environment_credentials", use_environment_credentials ? "1" : "0");


These are also not needed.

antaljanosbenjamin self-assigned this Nov 28, 2025

kalavt marked this pull request as draft November 28, 2025 14:33

kalavt marked this pull request as ready for review December 1, 2025 08:25

antaljanosbenjamin reviewed Dec 1, 2025

View reviewed changes

Comment thread src/Storages/Kafka/KafkaSettings.cpp Outdated

Comment thread src/Storages/Kafka/KafkaConfigLoader.cpp Outdated

antaljanosbenjamin added the can be tested Allows running workflows for external contributors label Dec 1, 2025

clickhouse-gh Bot added pr-feature Pull request with new product feature submodule changed At least one submodule changed in this PR. labels Dec 1, 2025

ilejn reviewed Dec 5, 2025

View reviewed changes

Comment thread src/Storages/Kafka/AWSMSKIAMAuth.cpp

kalavt force-pushed the feature/aws_msk_iam_auth branch from ab166da to abacc8d Compare December 10, 2025 17:06

clickhouse-gh Bot added the manual approve Manual approve required to run CI label Dec 10, 2025

kalavt force-pushed the feature/aws_msk_iam_auth branch 2 times, most recently from a732cbb to b6d319e Compare December 10, 2025 17:53

kalavt force-pushed the feature/aws_msk_iam_auth branch from 0a265e0 to b1bd1c1 Compare December 12, 2025 14:50

antaljanosbenjamin reviewed Dec 16, 2025

View reviewed changes

kalavt changed the title ~~feature: Add AWS MSK IAM Authentication Support for Kafka Engine~~ WIP: feature: Add AWS MSK IAM Authentication Support for Kafka Engine Dec 17, 2025

kalavt force-pushed the feature/aws_msk_iam_auth branch 7 times, most recently from 2eb1077 to e5ee4ac Compare December 18, 2025 18:30

kalavt force-pushed the feature/aws_msk_iam_auth branch 2 times, most recently from cf93f1c to a193144 Compare December 20, 2025 08:42

kalavt changed the title ~~WIP: feature: Add AWS MSK IAM Authentication Support for Kafka Engine~~ feature: Add AWS MSK IAM Authentication Support for Kafka Engine Dec 20, 2025

kalavt force-pushed the feature/aws_msk_iam_auth branch 4 times, most recently from d9fdaf8 to 4ce1766 Compare December 24, 2025 16:57

antaljanosbenjamin reviewed Jan 2, 2026

View reviewed changes

kalavt force-pushed the feature/aws_msk_iam_auth branch 2 times, most recently from e64d84a to b7f3eee Compare January 29, 2026 06:21

feat: Add AWS MSK IAM authentication support for Kafka

bad8cb1

- Implements AWS MSK IAM authentication for Kafka storage. - Updates librdkafka-cmake configuration.

kalavt force-pushed the feature/aws_msk_iam_auth branch from b7f3eee to bad8cb1 Compare January 29, 2026 19:07

antaljanosbenjamin requested changes Jan 29, 2026

View reviewed changes

antaljanosbenjamin reviewed Jan 29, 2026

View reviewed changes

feat: AWS MSK IAM fix review comments

3a8f50a

kalavt mentioned this pull request Feb 5, 2026

feat: add support for OAUTHBEARER token refresh callback ClickHouse/cppkafka#5

Merged

kalavt closed this Feb 5, 2026

		params.emplace_back("region", effective_region);
		params.emplace_back("use_environment_credentials", use_environment_credentials ? "1" : "0");

Conversation

kalavt commented Nov 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Changelog category (leave one):

Changelog entry (a user-readable short description of the changes that goes into CHANGELOG.md):

Documentation entry for user-facing changes

Description

Uh oh!

CLAassistant commented Nov 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

kalavt commented Nov 28, 2025

Uh oh!

kalavt commented Dec 1, 2025

Uh oh!

antaljanosbenjamin left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

clickhouse-gh Bot commented Dec 1, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

antaljanosbenjamin commented Dec 5, 2025

Uh oh!

kalavt commented Dec 8, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

antaljanosbenjamin commented Dec 10, 2025

Uh oh!

kalavt commented Dec 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

antaljanosbenjamin left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

kalavt commented Dec 20, 2025

Uh oh!

kalavt commented Dec 25, 2025

Uh oh!

antaljanosbenjamin commented Dec 25, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

antaljanosbenjamin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

kalavt commented Nov 28, 2025 •

edited

Loading

CLAassistant commented Nov 28, 2025 •

edited

Loading

clickhouse-gh Bot commented Dec 1, 2025 •

edited

Loading

kalavt commented Dec 8, 2025 •

edited

Loading

kalavt commented Dec 10, 2025 •

edited

Loading