Skip to main content

Writing to AWS MSK

Amazon MSK (Managed Streaming for Apache Kafka) is a fully managed service for Apache Kafka. Striim Kafka Writer supports connecting to MSK using three authentication types: None (plaintext), SCRAM-SHA-512, and AWS_MSK_IAM.

Setting up Connection Encryption

See Setting up Connection Encryption for the general self-managed Kafka walkthrough. AWS MSK recommends enabling TLS encryption for data in transit. Set Use SSL to True in the Kafka Connection Profile when connecting to MSK.

Note

SCRAM-SHA-512 and AWS_MSK_IAM authentication types require SSL to be enabled.

Using the Striim Vault

See Using the Striim Vault for general Striim Vault instructions. The same steps apply when storing MSK certificates.

Connecting to Schema Registry

Kafka Writer does not currently support the AWS Glue Schema Registry. When writing to Amazon MSK with the Avro Formatter, connect to a Confluent Schema Registry or a Karapace registry — the connection steps are the same regardless of whether the Kafka cluster is self-managed, Confluent, or Amazon MSK. See Connecting to Schema Registry above.

Setting up Connection to AWS MSK

Prerequisites:

  • MSK Cluster: Provisioned and running.

  • Bootstrap Server URL: Available in the AWS MSK Console under cluster details (navigate to Amazon MSK → Clusters → Choose the cluster → View Client Information).

  • Appropriate IAM permissions or SCRAM credentials configured.

Required ACLs

These are the AWS IAM-based equivalents of the Required ACLs for self-managed Kafka. The IAM policies required depend on whether E1P (exactly-once processing) or A1P (at-least-once processing) is used.

IAM Policy for KafkaWriter with E1P:

{
"Version": "<versionNo>",
"Statement": [
{
	"Sid": "VisualEditor0",
	"Effect": "Allow",
	"Action": [
	"kafka:GetBootstrapBrokers",
	"kafka:ListTopics",
	"kafka:DescribeTopic",
	"kafka:DescribeTopicPartitions"
	],
	"Resource": "*"
},
{
	"Sid": "VisualEditor1",
	"Effect": "Allow",
	"Action": [
	"kafka-cluster:DeleteGroup",
	"kafka-cluster:WriteDataIdempotently",
	"kafka-cluster:DescribeCluster",
	"kafka-cluster:ReadData",
	"kafka-cluster:DescribeTransactionalId",
	"kafka-cluster:AlterTransactionalId",
	"kafka-cluster:DescribeTopicDynamicConfiguration",
	"kafka-cluster:AlterTopicDynamicConfiguration",
	"kafka-cluster:AlterGroup",
	"kafka-cluster:AlterClusterDynamicConfiguration",
	"kafka-cluster:AlterTopic",
	"kafka-cluster:CreateTopic",
	"kafka-cluster:DescribeTopic",
	"kafka-cluster:AlterCluster",
	"kafka-cluster:DescribeGroup",
	"kafka-cluster:DescribeClusterDynamicConfiguration",
	"kafka-cluster:Connect",
	"kafka-cluster:DeleteTopic",
	"kafka-cluster:WriteData"
	],
	"Resource": [
	"arn:aws:kafka:*:{AWS_ACCOUNT_ID}:transactional-id/*/*/*",
	"arn:aws:kafka:*:{AWS_ACCOUNT_ID}:cluster/*/*",
	"arn:aws:kafka:*:{AWS_ACCOUNT_ID}:topic/*/*/*",
	"arn:aws:kafka:*:{AWS_ACCOUNT_ID}:group/*/*/*"
	]
}
]
}

IAM Policy for KafkaWriter with A1P (at-least-once processing):

{
"Version": "<versionNo>",
"Statement": [	
{
	"Sid": "VisualEditor0",
	"Effect": "Allow",
	"Action": [
	"kafka:GetBootstrapBrokers",
		"kafka:ListTopics",
		"kafka:DescribeTopic",
		"kafka:DescribeTopicPartitions"
	],
	"Resource": "*"
},
{
	"Sid": "VisualEditor1",
	"Effect": "Allow",
	"Action": [
	"kafka-cluster:DeleteGroup",
	"kafka-cluster:DescribeCluster",
	"kafka-cluster:ReadData",
	"kafka-cluster:DescribeTopicDynamicConfiguration",
	"kafka-cluster:AlterTopicDynamicConfiguration",
	"kafka-cluster:AlterGroup",
	"kafka-cluster:AlterClusterDynamicConfiguration",
	"kafka-cluster:AlterTopic",
	"kafka-cluster:CreateTopic",
	"kafka-cluster:DescribeTopic",
	"kafka-cluster:AlterCluster",
	"kafka-cluster:DescribeGroup",
	"kafka-cluster:DescribeClusterDynamicConfiguration",
	"kafka-cluster:Connect",
	"kafka-cluster:DeleteTopic",
	"kafka-cluster:WriteData"
	],
	"Resource": [
	"arn:aws:kafka:*:{AWS_ACCOUNT_ID}:cluster/*/*",
	"arn:aws:kafka:*:{AWS_ACCOUNT_ID}:topic/*/*/*",
	"arn:aws:kafka:*:{AWS_ACCOUNT_ID}:group/*/*/*"
	]
}
]
}

Connecting with None Authentication Type (plaintext):

Property Name

Value

Broker Address

Use the non-TLS plaintext port. Example: b-1.msk-cluster-xxxx.c10.kafka.us-east-1.amazonaws.com:9092

Authentication Type

NONE

Use SSL

True (recommended --- AWS MSK recommends enabling TLS encryption for data in transit)

Connecting with SCRAM-SHA-512 Authentication:

Note

SCRAM-SHA-256 is not supported on AWS MSK; use SCRAM-SHA-512 instead.

Property Name

Value

Broker Address

Use the TLS-enabled port. Example: b-1.msk-cluster-xxxx.c10.kafka.us-east-1.amazonaws.com:9196

Authentication Type

SCRAM-SHA-512

Use SSL

True (required --- SASL/SCRAM authentication requires an encrypted connection)

JAAS Config

org.apache.kafka.common.security.scram.ScramLoginModule required username="<<MSK_USERNAME>>" password="<<MSK_PASSWORD>>"; Replace with credentials provisioned via AWS Secrets Manager.

Connecting with AWS_MSK_IAM Authentication:

Property Name

Value

Broker Address

Use the IAM-enabled port. Example: b-1.msk-cluster-xxxx.c10.kafka.us-east-1.amazonaws.com:9198

Authentication Type

AWS_MSK_IAM

Use SSL

True

JAAS Config

Default: software.amazon.msk.auth.iam.IAMLoginModule required; To use a specific AWS profile: software.amazon.msk.auth.iam.IAMLoginModule required awsProfileName="my-profile";