sink: support AWS IAM authentication for Amazon MSK

api/v2/model.go

@@ -458,6 +458,15 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 					Token:           c.Sink.KafkaConfig.GlueSchemaRegistryConfig.Token,
 				}
 			}
+			var awsMSKIAMConfig *config.AWSMSKIAMConfig
+			if c.Sink.KafkaConfig.AWSMSKIAM != nil {
+				awsMSKIAMConfig = &config.AWSMSKIAMConfig{
+					Region:          c.Sink.KafkaConfig.AWSMSKIAM.Region,
+					RoleARN:         c.Sink.KafkaConfig.AWSMSKIAM.RoleARN,
+					RoleSessionName: c.Sink.KafkaConfig.AWSMSKIAM.RoleSessionName,
+					ExternalID:      c.Sink.KafkaConfig.AWSMSKIAM.ExternalID,
+				}
+			}
 
 			kafkaConfig = &config.KafkaConfig{
 				PartitionNum:                 c.Sink.KafkaConfig.PartitionNum,
@@ -488,6 +497,8 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 				SASLOAuthScopes:              c.Sink.KafkaConfig.SASLOAuthScopes,
 				SASLOAuthGrantType:           c.Sink.KafkaConfig.SASLOAuthGrantType,
 				SASLOAuthAudience:            c.Sink.KafkaConfig.SASLOAuthAudience,
+				SASLOAuthProvider:            c.Sink.KafkaConfig.SASLOAuthProvider,
+				AWSMSKIAM:                    awsMSKIAMConfig,
 				EnableTLS:                    c.Sink.KafkaConfig.EnableTLS,
 				CA:                           c.Sink.KafkaConfig.CA,
 				Cert:                         c.Sink.KafkaConfig.Cert,
@@ -789,6 +800,15 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 					Token:           cloned.Sink.KafkaConfig.GlueSchemaRegistryConfig.Token,
 				}
 			}
+			var awsMSKIAMConfig *AWSMSKIAMConfig
+			if cloned.Sink.KafkaConfig.AWSMSKIAM != nil {
+				awsMSKIAMConfig = &AWSMSKIAMConfig{
+					Region:          cloned.Sink.KafkaConfig.AWSMSKIAM.Region,
+					RoleARN:         cloned.Sink.KafkaConfig.AWSMSKIAM.RoleARN,
+					RoleSessionName: cloned.Sink.KafkaConfig.AWSMSKIAM.RoleSessionName,
+					ExternalID:      cloned.Sink.KafkaConfig.AWSMSKIAM.ExternalID,
+				}
+			}
 
 			kafkaConfig = &KafkaConfig{
 				PartitionNum:                 cloned.Sink.KafkaConfig.PartitionNum,
@@ -819,6 +839,8 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 				SASLOAuthScopes:              cloned.Sink.KafkaConfig.SASLOAuthScopes,
 				SASLOAuthGrantType:           cloned.Sink.KafkaConfig.SASLOAuthGrantType,
 				SASLOAuthAudience:            cloned.Sink.KafkaConfig.SASLOAuthAudience,
+				SASLOAuthProvider:            cloned.Sink.KafkaConfig.SASLOAuthProvider,
+				AWSMSKIAM:                    awsMSKIAMConfig,
 				EnableTLS:                    cloned.Sink.KafkaConfig.EnableTLS,
 				CA:                           cloned.Sink.KafkaConfig.CA,
 				Cert:                         cloned.Sink.KafkaConfig.Cert,
@@ -1506,6 +1528,8 @@ type KafkaConfig struct {
 	SASLOAuthScopes              []string                  `json:"sasl_oauth_scopes,omitempty" toml:"sasl-oauth-scopes,omitempty"`
 	SASLOAuthGrantType           *string                   `json:"sasl_oauth_grant_type,omitempty" toml:"sasl-oauth-grant-type,omitempty"`
 	SASLOAuthAudience            *string                   `json:"sasl_oauth_audience,omitempty" toml:"sasl-oauth-audience,omitempty"`
+	SASLOAuthProvider            *string                   `json:"sasl_oauth_provider,omitempty" toml:"sasl-oauth-provider,omitempty"`
+	AWSMSKIAM                    *AWSMSKIAMConfig          `json:"aws_msk_iam,omitempty" toml:"aws-msk-iam,omitempty"`
 	EnableTLS                    *bool                     `json:"enable_tls,omitempty" toml:"enable-tls,omitempty"`
 	CA                           *string                   `json:"ca,omitempty" toml:"ca,omitempty"`
 	Cert                         *string                   `json:"cert,omitempty" toml:"cert,omitempty"`
@@ -1573,6 +1597,14 @@ type GlueSchemaRegistryConfig struct {
 	Token           string `json:"token,omitempty" toml:"token,omitempty"`
 }
 
+// AWSMSKIAMConfig represents an Amazon MSK IAM authentication configuration.
+type AWSMSKIAMConfig struct {
+	Region          string `json:"region" toml:"region"`
+	RoleARN         string `json:"role_arn,omitempty" toml:"role-arn,omitempty"`
+	RoleSessionName string `json:"role_session_name,omitempty" toml:"role-session-name,omitempty"`
+	ExternalID      string `json:"external_id,omitempty" toml:"external-id,omitempty"`
+}
+
 // OpenProtocolConfig represents the configurations for open protocol encoding
 type OpenProtocolConfig struct {
 	OutputOldValue bool `json:"output_old_value" toml:"output-old-value"`

api/v2/model_test.go

@@ -41,6 +41,16 @@ func TestReplicaConfigConversion(t *testing.T) {
 				SpoolDiskQuota:   util.AddressOf(int64(1024)),
 				SpoolBaseDir:     util.AddressOf("/tmp/ticdc-spool"),
 			},
+			KafkaConfig: &KafkaConfig{
+				SASLMechanism:     util.AddressOf("OAUTHBEARER"),
+				SASLOAuthProvider: util.AddressOf("AWS_MSK_IAM"),
+				AWSMSKIAM: &AWSMSKIAMConfig{
+					Region:          "us-west-2",
+					RoleARN:         "arn:aws:iam::123456789012:role/ticdc-msk",
+					RoleSessionName: "ticdc",
+					ExternalID:      "external-id",
+				},
+			},
 		},
 		Mounter: &MounterConfig{
 			WorkerNum: util.AddressOf(16),
@@ -73,6 +83,11 @@ func TestReplicaConfigConversion(t *testing.T) {
 	require.True(t, util.GetOrZero(internalCfg.Sink.CloudStorageConfig.UseTableIDAsPath))
 	require.Equal(t, int64(1024), util.GetOrZero(internalCfg.Sink.CloudStorageConfig.SpoolDiskQuota))
 	require.Equal(t, "/tmp/ticdc-spool", util.GetOrZero(internalCfg.Sink.CloudStorageConfig.SpoolBaseDir))
+	require.Equal(t, "AWS_MSK_IAM", util.GetOrZero(internalCfg.Sink.KafkaConfig.SASLOAuthProvider))
+	require.Equal(t, "us-west-2", internalCfg.Sink.KafkaConfig.AWSMSKIAM.Region)
+	require.Equal(t, "arn:aws:iam::123456789012:role/ticdc-msk", internalCfg.Sink.KafkaConfig.AWSMSKIAM.RoleARN)
+	require.Equal(t, "ticdc", internalCfg.Sink.KafkaConfig.AWSMSKIAM.RoleSessionName)
+	require.Equal(t, "external-id", internalCfg.Sink.KafkaConfig.AWSMSKIAM.ExternalID)
 	require.Equal(t, internalCfg.Mounter.WorkerNum, *apiCfg.Mounter.WorkerNum)
 	require.True(t, util.GetOrZero(internalCfg.Scheduler.EnableTableAcrossNodes))
 	require.Equal(t, 1000, util.GetOrZero(internalCfg.Scheduler.RegionThreshold))
@@ -100,6 +115,11 @@ func TestReplicaConfigConversion(t *testing.T) {
 	require.True(t, *apiCfgBack.Sink.CloudStorageConfig.UseTableIDAsPath)
 	require.Equal(t, int64(1024), *apiCfgBack.Sink.CloudStorageConfig.SpoolDiskQuota)
 	require.Equal(t, "/tmp/ticdc-spool", *apiCfgBack.Sink.CloudStorageConfig.SpoolBaseDir)
+	require.Equal(t, "AWS_MSK_IAM", *apiCfgBack.Sink.KafkaConfig.SASLOAuthProvider)
+	require.Equal(t, "us-west-2", apiCfgBack.Sink.KafkaConfig.AWSMSKIAM.Region)
+	require.Equal(t, "arn:aws:iam::123456789012:role/ticdc-msk", apiCfgBack.Sink.KafkaConfig.AWSMSKIAM.RoleARN)
+	require.Equal(t, "ticdc", apiCfgBack.Sink.KafkaConfig.AWSMSKIAM.RoleSessionName)
+	require.Equal(t, "external-id", apiCfgBack.Sink.KafkaConfig.AWSMSKIAM.ExternalID)
 	require.Equal(t, 16, *apiCfgBack.Mounter.WorkerNum)
 	require.True(t, *apiCfgBack.Scheduler.EnableTableAcrossNodes)
 	require.Equal(t, "correctness", *apiCfgBack.Integrity.IntegrityCheckLevel)

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/KimMachineGun/automemlimit v0.2.4
 	github.com/agiledragon/gomonkey/v2 v2.11.0
 	github.com/apache/pulsar-client-go v0.13.0
+	github.com/aws/aws-msk-iam-sasl-signer-go v1.0.4
 	github.com/aws/aws-sdk-go v1.55.7
 	github.com/aws/aws-sdk-go-v2 v1.41.5
 	github.com/aws/aws-sdk-go-v2/config v1.32.2

go.sum

@@ -138,6 +138,8 @@ github.com/ardielle/ardielle-tools v1.5.4/go.mod h1:oZN+JRMnqGiIhrzkRN9l26Cej9dE
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
 github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 h1:DklsrG3dyBCFEj5IhUbnKptjxatkF07cF2ak3yi77so=
 github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2/go.mod h1:WaHUgvxTVq04UNunO+XhnAqY/wQc+bxr74GqbsZ/Jqw=
+github.com/aws/aws-msk-iam-sasl-signer-go v1.0.4 h1:2jAwFwA0Xgcx94dUId+K24yFabsKYDtAhCgyMit6OqE=
+github.com/aws/aws-msk-iam-sasl-signer-go v1.0.4/go.mod h1:MVYeeOhILFFemC/XlYTClvBjYZrg/EPd3ts885KrNTI=
 github.com/aws/aws-sdk-go v1.32.6/go.mod h1:5zCpMtNQVjRREroY7sYe8lOMRSxkhG6MZveU8YkpAk0=
 github.com/aws/aws-sdk-go v1.44.204/go.mod h1:aVsgQcEevwlmQ7qHE9I3h+dtQgpqhFB+i8Phjh7fkwI=
 github.com/aws/aws-sdk-go v1.55.7 h1:UJrkFq7es5CShfBwlWAC8DA077vp8PyVbQd3lqLiztE=

pkg/config/sink.go

@@ -457,6 +457,14 @@ type CodecConfig struct {
 	OutputRowKey                   *bool   `toml:"output-row-key" json:"output-row-key,omitempty"`
 }
 
+// AWSMSKIAMConfig represents an Amazon MSK IAM authentication configuration.
+type AWSMSKIAMConfig struct {
+	Region          string `toml:"region" json:"region"`
+	RoleARN         string `toml:"role-arn" json:"role-arn,omitempty"`
+	RoleSessionName string `toml:"role-session-name" json:"role-session-name,omitempty"`
+	ExternalID      string `toml:"external-id" json:"external-id,omitempty"`
+}
+
 // KafkaConfig represents a kafka sink configuration
 type KafkaConfig struct {
 	PartitionNum                 *int32                    `toml:"partition-num" json:"partition-num,omitempty"`
@@ -487,6 +495,8 @@ type KafkaConfig struct {
 	SASLOAuthScopes              []string                  `toml:"sasl-oauth-scopes" json:"sasl-oauth-scopes,omitempty"`
 	SASLOAuthGrantType           *string                   `toml:"sasl-oauth-grant-type" json:"sasl-oauth-grant-type,omitempty"`
 	SASLOAuthAudience            *string                   `toml:"sasl-oauth-audience" json:"sasl-oauth-audience,omitempty"`
+	SASLOAuthProvider            *string                   `toml:"sasl-oauth-provider" json:"sasl-oauth-provider,omitempty"`
+	AWSMSKIAM                    *AWSMSKIAMConfig          `toml:"aws-msk-iam" json:"aws-msk-iam,omitempty"`
 	EnableTLS                    *bool                     `toml:"enable-tls" json:"enable-tls,omitempty"`
 	CA                           *string                   `toml:"ca" json:"ca,omitempty"`
 	Cert                         *string                   `toml:"cert" json:"cert,omitempty"`

pkg/config/sink_test.go

@@ -17,11 +17,40 @@ import (
 	"net/url"
 	"testing"
 
+	"github.com/BurntSushi/toml"
 	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/util"
 	"github.com/stretchr/testify/require"
 )
 
+func TestDecodeAWSMSKIAMConfig(t *testing.T) {
+	t.Parallel()
+
+	var replicaConfig ReplicaConfig
+	metadata, err := toml.Decode(`
+[sink.kafka-config]
+sasl-mechanism = "OAUTHBEARER"
+sasl-oauth-provider = "AWS_MSK_IAM"
+enable-tls = true
+
+[sink.kafka-config.aws-msk-iam]
+region = "ap-northeast-1"
+role-arn = "arn:aws:iam::123456789012:role/TiCDCMSKProducer"
+role-session-name = "ticdc"
+external-id = "external-id"
+`, &replicaConfig)
+	require.NoError(t, err)
+	require.Empty(t, metadata.Undecoded())
+	require.NotNil(t, replicaConfig.Sink)
+	require.NotNil(t, replicaConfig.Sink.KafkaConfig)
+	require.Equal(t, &AWSMSKIAMConfig{
+		Region:          "ap-northeast-1",
+		RoleARN:         "arn:aws:iam::123456789012:role/TiCDCMSKProducer",
+		RoleSessionName: "ticdc",
+		ExternalID:      "external-id",
+	}, replicaConfig.Sink.KafkaConfig.AWSMSKIAM)
+}
+
 func TestValidateTxnAtomicity(t *testing.T) {
 	t.Parallel()
 	testCases := []struct {

pkg/security/sasl.go

@@ -64,6 +64,16 @@ type SASL struct {
 	SASLMechanism SASLMechanism
 	GSSAPI        GSSAPI
 	OAuth2        OAuth2
+	OAuthProvider string
+	AWSMSKIAM     AWSMSKIAM
+}
+
+// AWSMSKIAM holds parameters for Amazon MSK IAM authentication.
+type AWSMSKIAM struct {
+	Region          string
+	RoleARN         string
+	RoleSessionName string
+	ExternalID      string
 }
 
 // OAuth2 holds necessary parameters to support sasl-oauth2.

pkg/sink/kafka/aws_msk_iam_token_provider.go

@@ -0,0 +1,54 @@
+// Copyright 2026 PingCAP, Inc.
+//
+// Licensed 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kafka
+
+import (
+	"context"
+
+	"github.com/IBM/sarama"
+	"github.com/aws/aws-msk-iam-sasl-signer-go/signer"
+	"github.com/pingcap/ticdc/pkg/errors"
+	"github.com/pingcap/ticdc/pkg/security"
+)
+
+type awsMSKIAMTokenProvider struct {
+	ctx    context.Context
+	config security.AWSMSKIAM
+}
+
+var _ sarama.AccessTokenProvider = (*awsMSKIAMTokenProvider)(nil)
+
+func newAWSMSKIAMTokenProvider(
+	ctx context.Context, config security.AWSMSKIAM,
+) sarama.AccessTokenProvider {
+	return &awsMSKIAMTokenProvider{ctx: ctx, config: config}
+}
+
+func (p *awsMSKIAMTokenProvider) Token() (*sarama.AccessToken, error) {
+	var (
+		token string
+		err   error
+	)
+	if p.config.RoleARN == "" {
+		token, _, err = signer.GenerateAuthToken(p.ctx, p.config.Region)
+	} else {
+		token, _, err = signer.GenerateAuthTokenFromRoleWithExternalId(
+			p.ctx, p.config.Region, p.config.RoleARN,
+			p.config.RoleSessionName, p.config.ExternalID)
+	}
+	if err != nil {
+		return nil, errors.WrapError(errors.ErrKafkaInvalidConfig, err)
+	}
+	return &sarama.AccessToken{Token: token}, nil
+}

pkg/sink/kafka/aws_msk_iam_token_provider_test.go

@@ -0,0 +1,55 @@
+// Copyright 2026 PingCAP, Inc.
+//
+// Licensed 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kafka
+
+import (
+	"context"
+	"testing"
+
+	"github.com/IBM/sarama"
+	"github.com/pingcap/ticdc/pkg/security"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAWSMSKIAMTokenProvider(t *testing.T) {
+	t.Setenv("AWS_ACCESS_KEY_ID", "test-access-key")
+	t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret-key")
+	t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
+
+	provider := newAWSMSKIAMTokenProvider(t.Context(), security.AWSMSKIAM{
+		Region: "us-east-1",
+	})
+	token, err := provider.Token()
+	require.NoError(t, err)
+	require.NotEmpty(t, token.Token)
+}
+
+func TestCompleteSaramaSASLConfigUsesAWSMSKIAMProvider(t *testing.T) {
+	saramaConfig := sarama.NewConfig()
+	options := NewOptions()
+	options.SASL.SASLMechanism = security.OAuthMechanism
+	options.SASL.OAuthProvider = SASLOAuthProviderAWSMSKIAM
+	options.SASL.AWSMSKIAM = security.AWSMSKIAM{
+		Region:          "ap-northeast-1",
+		RoleARN:         "arn:aws:iam::123456789012:role/TiCDCMSKProducer",
+		RoleSessionName: "ticdc",
+		ExternalID:      "external-id",
+	}
+
+	err := completeSaramaSASLConfig(context.Background(), saramaConfig, options)
+	require.NoError(t, err)
+	provider, ok := saramaConfig.Net.SASL.TokenProvider.(*awsMSKIAMTokenProvider)
+	require.True(t, ok)
+	require.Equal(t, options.SASL.AWSMSKIAM, provider.config)
+}