From 4eee62d8f560b78ed484b324265dcbf5bae094c6 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Tue, 4 Aug 2026 10:04:52 +0000 Subject: [PATCH 1/4] init Signed-off-by: wk989898 --- go.mod | 1 + go.sum | 2 + pkg/config/sink.go | 5 + pkg/security/sasl.go | 10 ++ pkg/sink/kafka/aws_msk_iam_token_provider.go | 54 ++++++++ .../kafka/aws_msk_iam_token_provider_test.go | 55 ++++++++ pkg/sink/kafka/options.go | 59 +++++++++ pkg/sink/kafka/options_test.go | 125 ++++++++++++++++++ pkg/sink/kafka/sarama_config.go | 10 +- 9 files changed, 320 insertions(+), 1 deletion(-) create mode 100644 pkg/sink/kafka/aws_msk_iam_token_provider.go create mode 100644 pkg/sink/kafka/aws_msk_iam_token_provider_test.go diff --git a/go.mod b/go.mod index 564652b685..59ab6cf676 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 2a9ecdef0e..d5a403efb6 100644 --- a/go.sum +++ b/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= diff --git a/pkg/config/sink.go b/pkg/config/sink.go index dca4d50572..3395d43a81 100644 --- a/pkg/config/sink.go +++ b/pkg/config/sink.go @@ -487,6 +487,11 @@ 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"` + AWSRegion *string `toml:"aws-region" json:"aws-region,omitempty"` + AWSRoleARN *string `toml:"aws-role-arn" json:"aws-role-arn,omitempty"` + AWSRoleSessionName *string `toml:"aws-role-session-name" json:"aws-role-session-name,omitempty"` + AWSExternalID *string `toml:"aws-external-id" json:"aws-external-id,omitempty"` EnableTLS *bool `toml:"enable-tls" json:"enable-tls,omitempty"` CA *string `toml:"ca" json:"ca,omitempty"` Cert *string `toml:"cert" json:"cert,omitempty"` diff --git a/pkg/security/sasl.go b/pkg/security/sasl.go index 6b503b5bea..aa80b9cabe 100644 --- a/pkg/security/sasl.go +++ b/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. diff --git a/pkg/sink/kafka/aws_msk_iam_token_provider.go b/pkg/sink/kafka/aws_msk_iam_token_provider.go new file mode 100644 index 0000000000..594a52987f --- /dev/null +++ b/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 +} diff --git a/pkg/sink/kafka/aws_msk_iam_token_provider_test.go b/pkg/sink/kafka/aws_msk_iam_token_provider_test.go new file mode 100644 index 0000000000..d67a329bae --- /dev/null +++ b/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) +} diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index da1dfb3640..387c51b269 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -70,6 +70,8 @@ const ( SASLTypeGSSAPI = "GSSAPI" // SASLTypeOAuth represents the SASL/OAUTHBEARER mechanism (Kafka 2.0.0+) SASLTypeOAuth = "OAUTHBEARER" + // saslOAuthProviderAWSMSKIAM uses AWS IAM credentials to authenticate with Amazon MSK. + saslOAuthProviderAWSMSKIAM = "aws-msk-iam" ) // RequiredAcks is used in Produce Requests to tell the broker how many replica acknowledgements @@ -338,6 +340,10 @@ func (o *options) Apply(changefeedID common.ChangeFeedID, if err != nil { return err } + if o.SASL.OAuthProvider == saslOAuthProviderAWSMSKIAM && !o.EnableTLS { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "AWS MSK IAM authentication requires TLS to be enabled") + } return nil } @@ -483,6 +489,35 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf } if sinkConfig != nil && sinkConfig.KafkaConfig != nil { + kafkaConfig := sinkConfig.KafkaConfig + if kafkaConfig.SASLOAuthProvider != nil { + o.SASL.OAuthProvider = strings.ToLower(strings.TrimSpace(*kafkaConfig.SASLOAuthProvider)) + if o.SASL.OAuthProvider == "" { + return errors.ErrKafkaInvalidConfig.GenWithStack("SASL OAuth provider cannot be empty") + } + } + if kafkaConfig.AWSRegion != nil { + o.SASL.AWSMSKIAM.Region = strings.TrimSpace(*kafkaConfig.AWSRegion) + } + if kafkaConfig.AWSRoleARN != nil { + o.SASL.AWSMSKIAM.RoleARN = strings.TrimSpace(*kafkaConfig.AWSRoleARN) + } + if kafkaConfig.AWSRoleSessionName != nil { + o.SASL.AWSMSKIAM.RoleSessionName = strings.TrimSpace(*kafkaConfig.AWSRoleSessionName) + } + if kafkaConfig.AWSExternalID != nil { + o.SASL.AWSMSKIAM.ExternalID = strings.TrimSpace(*kafkaConfig.AWSExternalID) + } + standardOAuthConfigPresent := kafkaConfig.SASLOAuthClientID != nil || + kafkaConfig.SASLOAuthClientSecret != nil || kafkaConfig.SASLOAuthTokenURL != nil || + kafkaConfig.SASLOAuthScopes != nil || kafkaConfig.SASLOAuthGrantType != nil || + kafkaConfig.SASLOAuthAudience != nil + if o.SASL.OAuthProvider != "" && standardOAuthConfigPresent { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "SASL OAuth provider %s cannot be used with standard OAuth2 configuration", + o.SASL.OAuthProvider) + } + if sinkConfig.KafkaConfig.SASLOAuthClientID != nil { clientID := *sinkConfig.KafkaConfig.SASLOAuthClientID if clientID == "" { @@ -539,6 +574,30 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf if sinkConfig.KafkaConfig.SASLOAuthAudience != nil { o.SASL.OAuth2.Audience = *sinkConfig.KafkaConfig.SASLOAuthAudience } + + awsMSKIAM := o.SASL.AWSMSKIAM + awsConfigPresent := o.SASL.OAuthProvider != "" || awsMSKIAM.Region != "" || + awsMSKIAM.RoleARN != "" || awsMSKIAM.RoleSessionName != "" || awsMSKIAM.ExternalID != "" + if awsConfigPresent { + if o.SASL.OAuthProvider != saslOAuthProviderAWSMSKIAM { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "unsupported SASL OAuth provider %q", o.SASL.OAuthProvider) + } + if o.SASL.SASLMechanism != security.OAuthMechanism { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "AWS MSK IAM is only supported with SASL mechanism type OAUTHBEARER, but got %s", + o.SASL.SASLMechanism) + } + if awsMSKIAM.Region == "" { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "AWS region cannot be empty for AWS MSK IAM authentication") + } + if awsMSKIAM.RoleARN == "" && + (awsMSKIAM.RoleSessionName != "" || awsMSKIAM.ExternalID != "") { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "AWS role ARN is required when role session name or external ID is configured") + } + } } return nil diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index 025df9086b..1d1e0c0c70 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -345,6 +345,131 @@ func TestApplyRejectsNonPositiveMaxMessageBytes(t *testing.T) { } } +func TestApplyAWSMSKIAM(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + kafkaConfig *config.KafkaConfig + expectedErr string + }{ + { + name: "default credential chain", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + AWSRegion: aws.String("ap-northeast-1"), + EnableTLS: aws.Bool(true), + }, + }, + { + name: "assume role", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + AWSRegion: aws.String("ap-northeast-1"), + AWSRoleARN: aws.String("arn:aws:iam::123456789012:role/TiCDCMSKProducer"), + AWSRoleSessionName: aws.String("ticdc"), + AWSExternalID: aws.String("external-id"), + EnableTLS: aws.Bool(true), + }, + }, + { + name: "TLS disabled", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + AWSRegion: aws.String("ap-northeast-1"), + }, + expectedErr: "AWS MSK IAM authentication requires TLS", + }, + { + name: "missing region", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + EnableTLS: aws.Bool(true), + }, + expectedErr: "AWS region cannot be empty", + }, + { + name: "unsupported provider", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String("unknown"), + AWSRegion: aws.String("ap-northeast-1"), + EnableTLS: aws.Bool(true), + }, + expectedErr: "unsupported SASL OAuth provider", + }, + { + name: "wrong SASL mechanism", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypePlaintext), + SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + AWSRegion: aws.String("ap-northeast-1"), + EnableTLS: aws.Bool(true), + }, + expectedErr: "AWS MSK IAM is only supported with SASL mechanism type OAUTHBEARER", + }, + { + name: "external ID without role ARN", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + AWSRegion: aws.String("ap-northeast-1"), + AWSExternalID: aws.String("external-id"), + EnableTLS: aws.Bool(true), + }, + expectedErr: "AWS role ARN is required", + }, + { + name: "mixed OAuth providers", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + SASLOAuthClientID: aws.String("client-id"), + AWSRegion: aws.String("ap-northeast-1"), + EnableTLS: aws.Bool(true), + }, + expectedErr: "cannot be used with standard OAuth2 configuration", + }, + { + name: "AWS provider with standard OAuth scopes", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + SASLOAuthScopes: []string{"scope"}, + AWSRegion: aws.String("ap-northeast-1"), + EnableTLS: aws.Bool(true), + }, + expectedErr: "cannot be used with standard OAuth2 configuration", + }, + } + + changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") + sinkURI, err := url.Parse("kafka://127.0.0.1:9092/topic") + require.NoError(t, err) + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + options := NewOptions() + err := options.Apply(changefeedID, sinkURI, &config.SinkConfig{ + KafkaConfig: test.kafkaConfig, + }) + if test.expectedErr != "" { + require.ErrorContains(t, err, test.expectedErr) + require.ErrorIs(t, err, errors.ErrKafkaInvalidConfig) + return + } + require.NoError(t, err) + require.Equal(t, saslOAuthProviderAWSMSKIAM, options.SASL.OAuthProvider) + require.Equal(t, *test.kafkaConfig.AWSRegion, options.SASL.AWSMSKIAM.Region) + require.True(t, options.EnableTLS) + }) + } +} + func TestSetPartitionNum(t *testing.T) { options := NewOptions() changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") diff --git a/pkg/sink/kafka/sarama_config.go b/pkg/sink/kafka/sarama_config.go index 51dbd2384e..f64bfae4bc 100644 --- a/pkg/sink/kafka/sarama_config.go +++ b/pkg/sink/kafka/sarama_config.go @@ -172,7 +172,15 @@ func completeSaramaSASLConfig(ctx context.Context, config *sarama.Config, o *opt } case SASLTypeOAuth: - p, err := newTokenProvider(ctx, o) + var ( + p sarama.AccessTokenProvider + err error + ) + if o.SASL.OAuthProvider == saslOAuthProviderAWSMSKIAM { + p = newAWSMSKIAMTokenProvider(ctx, o.SASL.AWSMSKIAM) + } else { + p, err = newTokenProvider(ctx, o) + } if err != nil { return err } From bbae54164382b31c815dc300f9456e51bceb5c37 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Tue, 4 Aug 2026 11:06:20 +0000 Subject: [PATCH 2/4] update Signed-off-by: wk989898 --- .../kafka/aws_msk_iam_token_provider_test.go | 2 +- pkg/sink/kafka/options.go | 10 +++++----- pkg/sink/kafka/options_test.go | 18 +++++++++--------- pkg/sink/kafka/sarama_config.go | 2 +- 4 files changed, 16 insertions(+), 16 deletions(-) diff --git a/pkg/sink/kafka/aws_msk_iam_token_provider_test.go b/pkg/sink/kafka/aws_msk_iam_token_provider_test.go index d67a329bae..aa0c7f43d8 100644 --- a/pkg/sink/kafka/aws_msk_iam_token_provider_test.go +++ b/pkg/sink/kafka/aws_msk_iam_token_provider_test.go @@ -39,7 +39,7 @@ func TestCompleteSaramaSASLConfigUsesAWSMSKIAMProvider(t *testing.T) { saramaConfig := sarama.NewConfig() options := NewOptions() options.SASL.SASLMechanism = security.OAuthMechanism - options.SASL.OAuthProvider = saslOAuthProviderAWSMSKIAM + options.SASL.OAuthProvider = SASLOAuthProviderAWSMSKIAM options.SASL.AWSMSKIAM = security.AWSMSKIAM{ Region: "ap-northeast-1", RoleARN: "arn:aws:iam::123456789012:role/TiCDCMSKProducer", diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index 387c51b269..48d8e50cfd 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -70,8 +70,8 @@ const ( SASLTypeGSSAPI = "GSSAPI" // SASLTypeOAuth represents the SASL/OAUTHBEARER mechanism (Kafka 2.0.0+) SASLTypeOAuth = "OAUTHBEARER" - // saslOAuthProviderAWSMSKIAM uses AWS IAM credentials to authenticate with Amazon MSK. - saslOAuthProviderAWSMSKIAM = "aws-msk-iam" + // SASLOAuthProviderAWSMSKIAM uses AWS IAM credentials to authenticate with Amazon MSK. + SASLOAuthProviderAWSMSKIAM = "AWS_MSK_IAM" ) // RequiredAcks is used in Produce Requests to tell the broker how many replica acknowledgements @@ -340,7 +340,7 @@ func (o *options) Apply(changefeedID common.ChangeFeedID, if err != nil { return err } - if o.SASL.OAuthProvider == saslOAuthProviderAWSMSKIAM && !o.EnableTLS { + if o.SASL.OAuthProvider == SASLOAuthProviderAWSMSKIAM && !o.EnableTLS { return errors.ErrKafkaInvalidConfig.GenWithStack( "AWS MSK IAM authentication requires TLS to be enabled") } @@ -491,7 +491,7 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf if sinkConfig != nil && sinkConfig.KafkaConfig != nil { kafkaConfig := sinkConfig.KafkaConfig if kafkaConfig.SASLOAuthProvider != nil { - o.SASL.OAuthProvider = strings.ToLower(strings.TrimSpace(*kafkaConfig.SASLOAuthProvider)) + o.SASL.OAuthProvider = strings.TrimSpace(*kafkaConfig.SASLOAuthProvider) if o.SASL.OAuthProvider == "" { return errors.ErrKafkaInvalidConfig.GenWithStack("SASL OAuth provider cannot be empty") } @@ -579,7 +579,7 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf awsConfigPresent := o.SASL.OAuthProvider != "" || awsMSKIAM.Region != "" || awsMSKIAM.RoleARN != "" || awsMSKIAM.RoleSessionName != "" || awsMSKIAM.ExternalID != "" if awsConfigPresent { - if o.SASL.OAuthProvider != saslOAuthProviderAWSMSKIAM { + if o.SASL.OAuthProvider != SASLOAuthProviderAWSMSKIAM { return errors.ErrKafkaInvalidConfig.GenWithStack( "unsupported SASL OAuth provider %q", o.SASL.OAuthProvider) } diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index 1d1e0c0c70..c440891a5f 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -357,7 +357,7 @@ func TestApplyAWSMSKIAM(t *testing.T) { name: "default credential chain", kafkaConfig: &config.KafkaConfig{ SASLMechanism: aws.String(SASLTypeOAuth), - SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + SASLOAuthProvider: aws.String("AWS_MSK_IAM"), AWSRegion: aws.String("ap-northeast-1"), EnableTLS: aws.Bool(true), }, @@ -366,7 +366,7 @@ func TestApplyAWSMSKIAM(t *testing.T) { name: "assume role", kafkaConfig: &config.KafkaConfig{ SASLMechanism: aws.String(SASLTypeOAuth), - SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), AWSRegion: aws.String("ap-northeast-1"), AWSRoleARN: aws.String("arn:aws:iam::123456789012:role/TiCDCMSKProducer"), AWSRoleSessionName: aws.String("ticdc"), @@ -378,7 +378,7 @@ func TestApplyAWSMSKIAM(t *testing.T) { name: "TLS disabled", kafkaConfig: &config.KafkaConfig{ SASLMechanism: aws.String(SASLTypeOAuth), - SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), AWSRegion: aws.String("ap-northeast-1"), }, expectedErr: "AWS MSK IAM authentication requires TLS", @@ -387,7 +387,7 @@ func TestApplyAWSMSKIAM(t *testing.T) { name: "missing region", kafkaConfig: &config.KafkaConfig{ SASLMechanism: aws.String(SASLTypeOAuth), - SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), EnableTLS: aws.Bool(true), }, expectedErr: "AWS region cannot be empty", @@ -406,7 +406,7 @@ func TestApplyAWSMSKIAM(t *testing.T) { name: "wrong SASL mechanism", kafkaConfig: &config.KafkaConfig{ SASLMechanism: aws.String(SASLTypePlaintext), - SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), AWSRegion: aws.String("ap-northeast-1"), EnableTLS: aws.Bool(true), }, @@ -416,7 +416,7 @@ func TestApplyAWSMSKIAM(t *testing.T) { name: "external ID without role ARN", kafkaConfig: &config.KafkaConfig{ SASLMechanism: aws.String(SASLTypeOAuth), - SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), AWSRegion: aws.String("ap-northeast-1"), AWSExternalID: aws.String("external-id"), EnableTLS: aws.Bool(true), @@ -427,7 +427,7 @@ func TestApplyAWSMSKIAM(t *testing.T) { name: "mixed OAuth providers", kafkaConfig: &config.KafkaConfig{ SASLMechanism: aws.String(SASLTypeOAuth), - SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), SASLOAuthClientID: aws.String("client-id"), AWSRegion: aws.String("ap-northeast-1"), EnableTLS: aws.Bool(true), @@ -438,7 +438,7 @@ func TestApplyAWSMSKIAM(t *testing.T) { name: "AWS provider with standard OAuth scopes", kafkaConfig: &config.KafkaConfig{ SASLMechanism: aws.String(SASLTypeOAuth), - SASLOAuthProvider: aws.String(saslOAuthProviderAWSMSKIAM), + SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), SASLOAuthScopes: []string{"scope"}, AWSRegion: aws.String("ap-northeast-1"), EnableTLS: aws.Bool(true), @@ -463,7 +463,7 @@ func TestApplyAWSMSKIAM(t *testing.T) { return } require.NoError(t, err) - require.Equal(t, saslOAuthProviderAWSMSKIAM, options.SASL.OAuthProvider) + require.Equal(t, SASLOAuthProviderAWSMSKIAM, options.SASL.OAuthProvider) require.Equal(t, *test.kafkaConfig.AWSRegion, options.SASL.AWSMSKIAM.Region) require.True(t, options.EnableTLS) }) diff --git a/pkg/sink/kafka/sarama_config.go b/pkg/sink/kafka/sarama_config.go index f64bfae4bc..7a2693c4c7 100644 --- a/pkg/sink/kafka/sarama_config.go +++ b/pkg/sink/kafka/sarama_config.go @@ -176,7 +176,7 @@ func completeSaramaSASLConfig(ctx context.Context, config *sarama.Config, o *opt p sarama.AccessTokenProvider err error ) - if o.SASL.OAuthProvider == saslOAuthProviderAWSMSKIAM { + if o.SASL.OAuthProvider == SASLOAuthProviderAWSMSKIAM { p = newAWSMSKIAMTokenProvider(ctx, o.SASL.AWSMSKIAM) } else { p, err = newTokenProvider(ctx, o) From 9b9b7608d3e19eef5dde496449231a631bb585b4 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Wed, 5 Aug 2026 03:44:06 +0000 Subject: [PATCH 3/4] update Signed-off-by: wk989898 --- pkg/config/sink.go | 13 +++++--- pkg/config/sink_test.go | 29 ++++++++++++++++ pkg/sink/kafka/options.go | 19 ++++------- pkg/sink/kafka/options_test.go | 61 ++++++++++++++++++++++------------ 4 files changed, 83 insertions(+), 39 deletions(-) diff --git a/pkg/config/sink.go b/pkg/config/sink.go index 3395d43a81..1bc4fd54ba 100644 --- a/pkg/config/sink.go +++ b/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"` @@ -488,10 +496,7 @@ type KafkaConfig struct { 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"` - AWSRegion *string `toml:"aws-region" json:"aws-region,omitempty"` - AWSRoleARN *string `toml:"aws-role-arn" json:"aws-role-arn,omitempty"` - AWSRoleSessionName *string `toml:"aws-role-session-name" json:"aws-role-session-name,omitempty"` - AWSExternalID *string `toml:"aws-external-id" json:"aws-external-id,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"` diff --git a/pkg/config/sink_test.go b/pkg/config/sink_test.go index 507bb250ef..d02a5aced2 100644 --- a/pkg/config/sink_test.go +++ b/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 { diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index 48d8e50cfd..65016318f5 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -496,17 +496,11 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf return errors.ErrKafkaInvalidConfig.GenWithStack("SASL OAuth provider cannot be empty") } } - if kafkaConfig.AWSRegion != nil { - o.SASL.AWSMSKIAM.Region = strings.TrimSpace(*kafkaConfig.AWSRegion) - } - if kafkaConfig.AWSRoleARN != nil { - o.SASL.AWSMSKIAM.RoleARN = strings.TrimSpace(*kafkaConfig.AWSRoleARN) - } - if kafkaConfig.AWSRoleSessionName != nil { - o.SASL.AWSMSKIAM.RoleSessionName = strings.TrimSpace(*kafkaConfig.AWSRoleSessionName) - } - if kafkaConfig.AWSExternalID != nil { - o.SASL.AWSMSKIAM.ExternalID = strings.TrimSpace(*kafkaConfig.AWSExternalID) + if kafkaConfig.AWSMSKIAM != nil { + o.SASL.AWSMSKIAM.Region = strings.TrimSpace(kafkaConfig.AWSMSKIAM.Region) + o.SASL.AWSMSKIAM.RoleARN = strings.TrimSpace(kafkaConfig.AWSMSKIAM.RoleARN) + o.SASL.AWSMSKIAM.RoleSessionName = strings.TrimSpace(kafkaConfig.AWSMSKIAM.RoleSessionName) + o.SASL.AWSMSKIAM.ExternalID = strings.TrimSpace(kafkaConfig.AWSMSKIAM.ExternalID) } standardOAuthConfigPresent := kafkaConfig.SASLOAuthClientID != nil || kafkaConfig.SASLOAuthClientSecret != nil || kafkaConfig.SASLOAuthTokenURL != nil || @@ -576,8 +570,7 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf } awsMSKIAM := o.SASL.AWSMSKIAM - awsConfigPresent := o.SASL.OAuthProvider != "" || awsMSKIAM.Region != "" || - awsMSKIAM.RoleARN != "" || awsMSKIAM.RoleSessionName != "" || awsMSKIAM.ExternalID != "" + awsConfigPresent := o.SASL.OAuthProvider != "" || kafkaConfig.AWSMSKIAM != nil if awsConfigPresent { if o.SASL.OAuthProvider != SASLOAuthProviderAWSMSKIAM { return errors.ErrKafkaInvalidConfig.GenWithStack( diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index c440891a5f..ce79fba05b 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -358,20 +358,24 @@ func TestApplyAWSMSKIAM(t *testing.T) { kafkaConfig: &config.KafkaConfig{ SASLMechanism: aws.String(SASLTypeOAuth), SASLOAuthProvider: aws.String("AWS_MSK_IAM"), - AWSRegion: aws.String("ap-northeast-1"), - EnableTLS: aws.Bool(true), + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + }, + EnableTLS: aws.Bool(true), }, }, { name: "assume role", kafkaConfig: &config.KafkaConfig{ - SASLMechanism: aws.String(SASLTypeOAuth), - SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), - AWSRegion: aws.String("ap-northeast-1"), - AWSRoleARN: aws.String("arn:aws:iam::123456789012:role/TiCDCMSKProducer"), - AWSRoleSessionName: aws.String("ticdc"), - AWSExternalID: aws.String("external-id"), - EnableTLS: aws.Bool(true), + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + RoleARN: "arn:aws:iam::123456789012:role/TiCDCMSKProducer", + RoleSessionName: "ticdc", + ExternalID: "external-id", + }, + EnableTLS: aws.Bool(true), }, }, { @@ -379,7 +383,9 @@ func TestApplyAWSMSKIAM(t *testing.T) { kafkaConfig: &config.KafkaConfig{ SASLMechanism: aws.String(SASLTypeOAuth), SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), - AWSRegion: aws.String("ap-northeast-1"), + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + }, }, expectedErr: "AWS MSK IAM authentication requires TLS", }, @@ -388,6 +394,7 @@ func TestApplyAWSMSKIAM(t *testing.T) { kafkaConfig: &config.KafkaConfig{ SASLMechanism: aws.String(SASLTypeOAuth), SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), + AWSMSKIAM: &config.AWSMSKIAMConfig{}, EnableTLS: aws.Bool(true), }, expectedErr: "AWS region cannot be empty", @@ -397,8 +404,10 @@ func TestApplyAWSMSKIAM(t *testing.T) { kafkaConfig: &config.KafkaConfig{ SASLMechanism: aws.String(SASLTypeOAuth), SASLOAuthProvider: aws.String("unknown"), - AWSRegion: aws.String("ap-northeast-1"), - EnableTLS: aws.Bool(true), + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + }, + EnableTLS: aws.Bool(true), }, expectedErr: "unsupported SASL OAuth provider", }, @@ -407,8 +416,10 @@ func TestApplyAWSMSKIAM(t *testing.T) { kafkaConfig: &config.KafkaConfig{ SASLMechanism: aws.String(SASLTypePlaintext), SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), - AWSRegion: aws.String("ap-northeast-1"), - EnableTLS: aws.Bool(true), + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + }, + EnableTLS: aws.Bool(true), }, expectedErr: "AWS MSK IAM is only supported with SASL mechanism type OAUTHBEARER", }, @@ -417,9 +428,11 @@ func TestApplyAWSMSKIAM(t *testing.T) { kafkaConfig: &config.KafkaConfig{ SASLMechanism: aws.String(SASLTypeOAuth), SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), - AWSRegion: aws.String("ap-northeast-1"), - AWSExternalID: aws.String("external-id"), - EnableTLS: aws.Bool(true), + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + ExternalID: "external-id", + }, + EnableTLS: aws.Bool(true), }, expectedErr: "AWS role ARN is required", }, @@ -429,8 +442,10 @@ func TestApplyAWSMSKIAM(t *testing.T) { SASLMechanism: aws.String(SASLTypeOAuth), SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), SASLOAuthClientID: aws.String("client-id"), - AWSRegion: aws.String("ap-northeast-1"), - EnableTLS: aws.Bool(true), + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + }, + EnableTLS: aws.Bool(true), }, expectedErr: "cannot be used with standard OAuth2 configuration", }, @@ -440,8 +455,10 @@ func TestApplyAWSMSKIAM(t *testing.T) { SASLMechanism: aws.String(SASLTypeOAuth), SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), SASLOAuthScopes: []string{"scope"}, - AWSRegion: aws.String("ap-northeast-1"), - EnableTLS: aws.Bool(true), + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + }, + EnableTLS: aws.Bool(true), }, expectedErr: "cannot be used with standard OAuth2 configuration", }, @@ -464,7 +481,7 @@ func TestApplyAWSMSKIAM(t *testing.T) { } require.NoError(t, err) require.Equal(t, SASLOAuthProviderAWSMSKIAM, options.SASL.OAuthProvider) - require.Equal(t, *test.kafkaConfig.AWSRegion, options.SASL.AWSMSKIAM.Region) + require.Equal(t, test.kafkaConfig.AWSMSKIAM.Region, options.SASL.AWSMSKIAM.Region) require.True(t, options.EnableTLS) }) } From 8b219dbbe8b4590fc8d1ea225a2668c762a82728 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Wed, 5 Aug 2026 12:53:25 +0000 Subject: [PATCH 4/4] update Signed-off-by: wk989898 --- api/v2/model.go | 32 ++++++++++++++++++++++++++++++++ api/v2/model_test.go | 20 ++++++++++++++++++++ 2 files changed, 52 insertions(+) diff --git a/api/v2/model.go b/api/v2/model.go index 0575a73e23..50f916ab79 100644 --- a/api/v2/model.go +++ b/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"` diff --git a/api/v2/model_test.go b/api/v2/model_test.go index 88fb821cdb..ffd5faeef0 100644 --- a/api/v2/model_test.go +++ b/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)