sink: support AWS IAM authentication for Amazon MSK

[wk989898]

What problem does this PR solve?

Issue Number: close #5872

What is changed and how it works?

TiCDC supports Kafka authentication through SASL/SCRAM and standard SASL/OAUTHBEARER token endpoints, but it cannot connect to Amazon MSK Serverless or MSK Provisioned clusters that use IAM access control.

Amazon MSK IAM authentication requires non-Java Kafka clients to generate short-lived AWS SigV4 tokens for the SASL/OAUTHBEARER mechanism. The existing OAuth client ID, client secret, and token URL configuration cannot generate these tokens.

This PR adds an Amazon MSK IAM OAuth token provider backed by the official AWS MSK IAM signer. It supports the AWS SDK default credential provider chain and optional STS AssumeRole credentials, including role session name and external ID.

configuration

Create changefeed.toml:

  [sink.kafka-config]
  sasl-mechanism = "OAUTHBEARER"
  sasl-oauth-provider = "AWS_MSK_IAM"
  enable-tls = true
  auto-create-topic = true
  partition-num = 6

  [sink.kafka-config.aws-msk-iam]
  region = "us-west-2"
  #role-arn = "arn:aws:iam::495580073302:role/TiCDCMSKProducer"
  #role-session-name = "ticdc"
  #external-id = "optional-external-id"

Check List

Tests

• Unit test
• Integration test
• Manual test (add detailed scripts or steps below)

1. Create MSK serverless and create AWS EC2 Instance
2. Create a tidb cluster in EC2
3. Create a changefeed with config

  [sink.kafka-config]
  sasl-mechanism = "OAUTHBEARER"
  sasl-oauth-provider = "AWS_MSK_IAM"
  enable-tls = true
  auto-create-topic = true
  partition-num = 6

  [sink.kafka-config.aws-msk-iam]
  region = "us-west-2"

4. Write some data
5. Changefeed is normal

 ./bin/cdc cli changefeed list   --server=http://127.0.0.1:8300
[
  {
    "id": "msk-iam-test",
    "keyspace": "default",
    "summary": {
      "state": "normal",
      "tso": 468172488878850057,
      "checkpoint": "2026-08-05 13:25:13.167",
      "error": null
    }
  }
]

Questions

Will it cause performance regression or break compatibility?

Do you need to update user documentation, design documentation or monitoring documentation?

Release note

Support AWS IAM authentication

Summary by CodeRabbit

• New Features

  • Added AWS MSK IAM authentication support for Kafka connections.
  • Added configuration for AWS region, IAM role, session name, and external ID.
  • Added role-based and default AWS credential support for generating access tokens.

• Bug Fixes

  • Added validation requiring TLS and complete, compatible IAM/OAuth settings.
  • Improved error handling for invalid AWS MSK IAM configurations.

[ti-chi-bot]

Skipping CI for Draft Pull Request.
If you want CI signal for your change, please convert it to an actual PR.
You can still manually trigger a test run with /test all

[ti-chi-bot]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
Once this PR has been reviewed and has the lgtm label, please assign lidezhu for approval. For more information see the Code Review Process.
Please ensure that each of them provides their approval before proceeding.

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• OWNERS
• pkg/config/OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 585eacf3-de63-4beb-9dde-b496145d7036

📥 Commits

Reviewing files that changed from the base of the PR and between c3c98e7 and 8b219db.

⛔ Files ignored due to path filters (1)

• go.sum is excluded by !**/*.sum

📒 Files selected for processing (11)

• api/v2/model.go
• api/v2/model_test.go
• go.mod
• pkg/config/sink.go
• pkg/config/sink_test.go
• pkg/security/sasl.go
• pkg/sink/kafka/aws_msk_iam_token_provider.go
• pkg/sink/kafka/aws_msk_iam_token_provider_test.go
• pkg/sink/kafka/options.go
• pkg/sink/kafka/options_test.go
• pkg/sink/kafka/sarama_config.go

---

📝 Walkthrough

Walkthrough

This PR adds AWS MSK IAM authentication to Kafka sink configuration. It exposes IAM settings through API and TOML models, validates provider-specific options, generates AWS SigV4 tokens, and integrates the provider with Sarama OAuth SASL configuration.

Changes

AWS MSK IAM authentication

Layer / File(s)	Summary
Configuration contracts and conversion api/v2/model.go, api/v2/model_test.go, pkg/config/sink.go, pkg/config/sink_test.go, pkg/security/sasl.go	Adds AWS MSK IAM fields to API, sink, and SASL models. Conversion and TOML decoding tests verify the configuration values.
Provider validation and option application pkg/sink/kafka/options.go, pkg/sink/kafka/options_test.go	Adds the AWS_MSK_IAM provider, requires TLS, normalizes settings, and validates provider, mechanism, region, role, and OAuth combinations.
Token generation and Sarama integration pkg/sink/kafka/aws_msk_iam_token_provider.go, pkg/sink/kafka/aws_msk_iam_token_provider_test.go, pkg/sink/kafka/sarama_config.go, go.mod	Adds AWS MSK IAM token generation using default credentials or role assumption. Sarama selects the new provider for AWS MSK IAM OAuth configuration.

Estimated code review effort: 4 (Complex) | ~45 minutes

Sequence Diagram(s)

sequenceDiagram
  participant KafkaOptions
  participant SaramaConfig
  participant AWSMSKIAMTokenProvider
  KafkaOptions->>SaramaConfig: apply OAUTHBEARER and AWS_MSK_IAM settings
  SaramaConfig->>AWSMSKIAMTokenProvider: create provider with IAM configuration
  AWSMSKIAMTokenProvider->>AWSMSKIAMTokenProvider: generate SigV4 access token
  AWSMSKIAMTokenProvider-->>SaramaConfig: return Sarama access token

Loading

Possibly related PRs

• pingcap/ticdc#5786: Modifies the OAuth/SASL validation and token-provider initialization paths.

Suggested labels: lgtm, approved

Suggested reviewers: 3aceshowhand, hongyunyan, asddongmen

Poem

I’m a rabbit with IAM keys,
Signing tokens through the trees.
Regions, roles, TLS in flight,
Sarama hops to MSK right.
OAuth burrows now align.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly and concisely describes the added AWS IAM authentication support for Amazon MSK.
Description check	✅ Passed	The description includes the issue reference, implementation details, testing steps, and release note; the required question sections remain unanswered.
Linked Issues check	✅ Passed	The changes implement AWS MSK IAM token generation, Sarama OAUTHBEARER support, default credentials, AssumeRole options, and related configuration.
Out of Scope Changes check	✅ Passed	The configuration, dependency, implementation, conversion logic, validation, and tests all support the linked AWS MSK IAM authentication objective.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}