This release is a pre-release and may not be stable for production use.
Amazon Managed Streaming for Apache Kafka Construct Library
---
The APIs of higher level constructs in this module are experimental and under active development. They are subject to non-backward compatible changes or removal in any future version. These are not subject to the Semantic Versioning model and breaking changes will be announced in the release notes. This means that while you may use them, you may need to update your source code when upgrading to a newer version of this package.
Amazon MSK is a fully managed service that makes it easy for you to build and run applications that use Apache Kafka to process streaming data.
The following example creates an MSK Cluster.
# vpc: ec2.Vpc
cluster = msk.Cluster(self, "Cluster",
cluster_name="myCluster",
kafka_version=msk.KafkaVersion.V4_1_X_KRAFT,
vpc=vpc
)
Allowing Connections
To control who can access the Cluster, use the .connections attribute. For a list of ports used by MSK, refer to the MSK documentation.
# vpc: ec2.Vpc
cluster = msk.Cluster(self, "Cluster",
cluster_name="myCluster",
kafka_version=msk.KafkaVersion.V4_1_X_KRAFT,
vpc=vpc
)
cluster.connections.allow_from(
ec2.Peer.ipv4("1.2.3.4/8"),
ec2.Port.tcp(2181))
cluster.connections.allow_from(
ec2.Peer.ipv4("1.2.3.4/8"),
ec2.Port.tcp(9094))
Cluster Endpoints
You can use the following attributes to get a list of the Kafka broker or ZooKeeper node endpoints
# cluster: msk.Cluster
CfnOutput(self, "BootstrapBrokers", value=cluster.bootstrap_brokers)
CfnOutput(self, "BootstrapBrokersTls", value=cluster.bootstrap_brokers_tls)
CfnOutput(self, "BootstrapBrokersSaslScram", value=cluster.bootstrap_brokers_sasl_scram)
CfnOutput(self, "BootstrapBrokerStringSaslIam", value=cluster.bootstrap_brokers_sasl_iam)
CfnOutput(self, "ZookeeperConnection", value=cluster.zookeeper_connection_string)
CfnOutput(self, "ZookeeperConnectionTls", value=cluster.zookeeper_connection_string_tls)
Importing an existing Cluster
To import an existing MSK cluster into your CDK app use the .fromClusterArn() method.
cluster = msk.Cluster.from_cluster_arn(self, "Cluster", "arn:aws:kafka:us-west-2:1234567890:cluster/a-cluster/11111111-1111-1111-1111-111111111111-1")
Client Authentication
MSK supports the following authentication mechanisms.
TLS
To enable client authentication with TLS set the certificateAuthorityArns property to reference your ACM Private CA. More info on Private CAs.
import aws_cdk.aws_acmpca as acmpca
# vpc: ec2.Vpc
cluster = msk.Cluster(self, "Cluster",
cluster_name="myCluster",
kafka_version=msk.KafkaVersion.V4_1_X_KRAFT,
vpc=vpc,
encryption_in_transit=msk.EncryptionInTransitConfig(
client_broker=msk.ClientBrokerEncryption.TLS
),
client_authentication=msk.ClientAuthentication.tls(
certificate_authorities=[
acmpca.CertificateAuthority.from_certificate_authority_arn(self, "CertificateAuthority", "arn:aws:acm-pca:us-west-2:1234567890:certificate-authority/11111111-1111-1111-1111-111111111111")
]
)
)
SASL/SCRAM
Enable client authentication with SASL/SCRAM:
# vpc: ec2.Vpc
cluster = msk.Cluster(self, "cluster",
cluster_name="myCluster",
kafka_version=msk.KafkaVersion.V4_1_X_KRAFT,
vpc=vpc,
encryption_in_transit=msk.EncryptionInTransitConfig(
client_broker=msk.ClientBrokerEncryption.TLS
),
client_authentication=msk.ClientAuthentication.sasl(
scram=True
)
)
IAM
Enable client authentication with IAM:
# vpc: ec2.Vpc
cluster = msk.Cluster(self, "cluster",
cluster_name="myCluster",
kafka_version=msk.KafkaVersion.V4_1_X_KRAFT,
vpc=vpc,
encryption_in_transit=msk.EncryptionInTransitConfig(
client_broker=msk.ClientBrokerEncryption.TLS
),
client_authentication=msk.ClientAuthentication.sasl(
iam=True
)
)
SASL/IAM + TLS
Enable client authentication with IAM
as well as enable client authentication with TLS by setting the certificateAuthorityArns property to reference your ACM Private CA. More info on Private CAs.
import aws_cdk.aws_acmpca as acmpca
# vpc: ec2.Vpc
cluster = msk.Cluster(self, "Cluster",
cluster_name="myCluster",
kafka_version=msk.KafkaVersion.V4_1_X_KRAFT,
vpc=vpc,
encryption_in_transit=msk.EncryptionInTransitConfig(
client_broker=msk.ClientBrokerEncryption.TLS
),
client_authentication=msk.ClientAuthentication.sasl_tls(
iam=True,
certificate_authorities=[
acmpca.CertificateAuthority.from_certificate_authority_arn(self, "CertificateAuthority", "arn:aws:acm-pca:us-west-2:1234567890:certificate-authority/11111111-1111-1111-1111-111111111111")
]
)
)
Logging
You can deliver Apache Kafka broker logs to one or more of the following destination types: Amazon CloudWatch Logs, Amazon S3, Amazon Data Firehose.
To configure logs to be sent to an S3 bucket, provide a bucket in the logging config.
# vpc: ec2.Vpc
# bucket: s3.IBucket
cluster = msk.Cluster(self, "cluster",
cluster_name="myCluster",
kafka_version=msk.KafkaVersion.V4_1_X_KRAFT,
vpc=vpc,
logging=msk.BrokerLogging(
s3=msk.S3LoggingConfiguration(
bucket=bucket
)
)
)
When the S3 destination is configured, AWS will automatically create an S3 bucket policy
that allows the service to write logs to the bucket. This makes it impossible to later update
that bucket policy. To have CDK create the bucket policy so that future updates can be made,
the @aws-cdk/aws-s3:createDefaultLoggingPolicy feature flag can be used. This can be set
in the cdk.json file.
{
"context": {
"@aws-cdk/aws-s3:createDefaultLoggingPolicy": true
}
}
Storage Mode
You can configure an MSK cluster storage mode using the storageMode property.
Tiered storage is a low-cost storage tier for Amazon MSK that scales to virtually unlimited storage, making it cost-effective to build streaming data applications.
Visit Tiered storage to see the list of compatible Kafka versions and for more details.
# vpc: ec2.Vpc
# bucket: s3.IBucket
cluster = msk.Cluster(self, "cluster",
cluster_name="myCluster",
kafka_version=msk.KafkaVersion.V4_1_X_KRAFT,
vpc=vpc,
storage_mode=msk.StorageMode.TIERED
)
MSK Express Brokers
You can create an MSK cluster with Express Brokers by setting the brokerType property to BrokerType.EXPRESS. Express Brokers are a low-cost option for development, testing, and workloads that don't require the high availability guarantees of standard MSK cluster.
For more information, see Amazon MSK Express Brokers.
Note: When using Express Brokers, the following constraints apply:
- Apache Kafka version must be 3.6.x, 3.8.x, or 3.9.x
- You must specify the
instanceType - The VPC must have at least 3 subnets (across 3 AZs)
ebsStorageInfois not supportedstorageModeis not supportedloggingis not supported- Supported broker sizes:
m7g.xlarge,m7g.2xlarge,m7g.4xlarge,m7g.8xlarge,m7g.12xlarge,m7g.16xlarge
# vpc: ec2.Vpc
express_cluster = msk.Cluster(self, "ExpressCluster",
cluster_name="MyExpressCluster",
kafka_version=msk.KafkaVersion.V3_8_X,
vpc=vpc,
broker_type=msk.BrokerType.EXPRESS,
instance_type=ec2.InstanceType.of(ec2.InstanceClass.M7G, ec2.InstanceSize.XLARGE)
)
MSK Serverless
You can also use MSK Serverless by using ServerlessCluster class.
MSK Serverless is a cluster type for Amazon MSK that makes it possible for you to run Apache Kafka without having to manage and scale cluster capacity.
MSK Serverless requires IAM access control for all clusters.
For more infomation, see Use MSK Serverless clusters.
# vpc: ec2.Vpc
serverless_cluster = msk.ServerlessCluster(self, "ServerlessCluster",
cluster_name="MyServerlessCluster",
vpc_configs=[msk.VpcConfig(vpc=vpc)
]
)
2 maintainers
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file aws_cdk_aws_msk_alpha-2.265.0a0.tar.gz.
File metadata
- Download URL: aws_cdk_aws_msk_alpha-2.265.0a0.tar.gz
- Upload date:
- Size: 119.2 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/7.0.0 CPython/3.11.15
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
6851eff17cfa2f55977d8c08ac08b59f124e46757b378644bb316762c601e113
|
|
| MD5 |
a0dc1cf6f25c38b035c6b2a715addd75
|
|
| BLAKE2b-256 |
e336660b3820b3869416945adc6f30400076c1f3fb5688523d3296c3226fd55e
|
File details
Details for the file aws_cdk_aws_msk_alpha-2.265.0a0-py3-none-any.whl.
File metadata
- Download URL: aws_cdk_aws_msk_alpha-2.265.0a0-py3-none-any.whl
- Upload date:
- Size: 117.9 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/7.0.0 CPython/3.11.15
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
e7fe0b8cf2d4f041b694e11401c54143777e3fa0937032fb00a332c3151810e2
|
|
| MD5 |
19d989422ce32f6773cfbcde719d2c3f
|
|
| BLAKE2b-256 |
62e01698dfc2d78526845a7d35c939a292d94afadb8a3e95122f71f685f1ad64
|
