@aws-cdk/aws-msk-alpha

Amazon Managed Streaming for Apache Kafka Construct Library

Version 2.273.0-alpha.0

Separate package

@aws-cdk/aws-msk-alpha is not part of aws-cdk-lib — install it separately:

npm install @aws-cdk/aws-msk-alpha
pip install aws-cdk.aws-msk-alpha

This module is experimental (alpha): its API may change in any release, without a major version bump.

---

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.

declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'Cluster', {
  clusterName: 'myCluster',
  kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
  vpc,
});

# vpc: ec2.Vpc

cluster = msk.Cluster(self, "Cluster",
    cluster_name="myCluster",
    kafka_version=msk.KafkaVersion.V4_1_X_KRAFT,
    vpc=vpc
)

Vpc vpc;

Cluster cluster = Cluster.Builder.create(this, "Cluster")
        .clusterName("myCluster")
        .kafkaVersion(KafkaVersion.V4_1_X_KRAFT)
        .vpc(vpc)
        .build();

Vpc vpc;

var cluster = new Cluster(this, "Cluster", new ClusterProps {
    ClusterName = "myCluster",
    KafkaVersion = KafkaVersion.V4_1_X_KRAFT,
    Vpc = vpc
});

var vpc Vpc

cluster := msk.NewCluster(this, jsii.String("Cluster"), &ClusterProps{
	ClusterName: jsii.String("myCluster"),
	KafkaVersion: 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.

declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'Cluster', {
  clusterName: 'myCluster',
  kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
  vpc,
});

cluster.connections.allowFrom(
  ec2.Peer.ipv4('1.2.3.4/8'),
  ec2.Port.tcp(2181),
);
cluster.connections.allowFrom(
  ec2.Peer.ipv4('1.2.3.4/8'),
  ec2.Port.tcp(9094),
);

# 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))

Vpc vpc;

Cluster cluster = Cluster.Builder.create(this, "Cluster")
        .clusterName("myCluster")
        .kafkaVersion(KafkaVersion.V4_1_X_KRAFT)
        .vpc(vpc)
        .build();

cluster.connections.allowFrom(Peer.ipv4("1.2.3.4/8"), Port.tcp(2181));
cluster.connections.allowFrom(Peer.ipv4("1.2.3.4/8"), Port.tcp(9094));

Vpc vpc;

var cluster = new Cluster(this, "Cluster", new ClusterProps {
    ClusterName = "myCluster",
    KafkaVersion = KafkaVersion.V4_1_X_KRAFT,
    Vpc = vpc
});

cluster.Connections.AllowFrom(Peer.Ipv4("1.2.3.4/8"), Port.Tcp(2181));
cluster.Connections.AllowFrom(Peer.Ipv4("1.2.3.4/8"), Port.Tcp(9094));

var vpc Vpc

cluster := msk.NewCluster(this, jsii.String("Cluster"), &ClusterProps{
	ClusterName: jsii.String("myCluster"),
	KafkaVersion: msk.KafkaVersion_V4_1_X_KRAFT(),
	Vpc: Vpc,
})

cluster.connections.AllowFrom(ec2.Peer_Ipv4(jsii.String("1.2.3.4/8")), ec2.Port_Tcp(jsii.Number(2181)))
cluster.connections.AllowFrom(ec2.Peer_Ipv4(jsii.String("1.2.3.4/8")), ec2.Port_Tcp(jsii.Number(9094)))

Cluster Endpoints

You can use the following attributes to get a list of the Kafka broker or ZooKeeper node endpoints

declare const cluster: msk.Cluster;
new CfnOutput(this, 'BootstrapBrokers', { value: cluster.bootstrapBrokers });
new CfnOutput(this, 'BootstrapBrokersTls', { value: cluster.bootstrapBrokersTls });
new CfnOutput(this, 'BootstrapBrokersSaslScram', { value: cluster.bootstrapBrokersSaslScram });
new CfnOutput(this, 'BootstrapBrokerStringSaslIam', { value: cluster.bootstrapBrokersSaslIam });
new CfnOutput(this, 'ZookeeperConnection', { value: cluster.zookeeperConnectionString });
new CfnOutput(this, 'ZookeeperConnectionTls', { value: cluster.zookeeperConnectionStringTls });

# 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)

Cluster cluster;

CfnOutput.Builder.create(this, "BootstrapBrokers").value(cluster.getBootstrapBrokers()).build();
CfnOutput.Builder.create(this, "BootstrapBrokersTls").value(cluster.getBootstrapBrokersTls()).build();
CfnOutput.Builder.create(this, "BootstrapBrokersSaslScram").value(cluster.getBootstrapBrokersSaslScram()).build();
CfnOutput.Builder.create(this, "BootstrapBrokerStringSaslIam").value(cluster.getBootstrapBrokersSaslIam()).build();
CfnOutput.Builder.create(this, "ZookeeperConnection").value(cluster.getZookeeperConnectionString()).build();
CfnOutput.Builder.create(this, "ZookeeperConnectionTls").value(cluster.getZookeeperConnectionStringTls()).build();

Cluster cluster;

new CfnOutput(this, "BootstrapBrokers", new CfnOutputProps { Value = cluster.BootstrapBrokers });
new CfnOutput(this, "BootstrapBrokersTls", new CfnOutputProps { Value = cluster.BootstrapBrokersTls });
new CfnOutput(this, "BootstrapBrokersSaslScram", new CfnOutputProps { Value = cluster.BootstrapBrokersSaslScram });
new CfnOutput(this, "BootstrapBrokerStringSaslIam", new CfnOutputProps { Value = cluster.BootstrapBrokersSaslIam });
new CfnOutput(this, "ZookeeperConnection", new CfnOutputProps { Value = cluster.ZookeeperConnectionString });
new CfnOutput(this, "ZookeeperConnectionTls", new CfnOutputProps { Value = cluster.ZookeeperConnectionStringTls });

var cluster Cluster

awscdk.NewCfnOutput(this, jsii.String("BootstrapBrokers"), &CfnOutputProps{
	Value: cluster.bootstrapBrokers,
})
awscdk.NewCfnOutput(this, jsii.String("BootstrapBrokersTls"), &CfnOutputProps{
	Value: cluster.bootstrapBrokersTls,
})
awscdk.NewCfnOutput(this, jsii.String("BootstrapBrokersSaslScram"), &CfnOutputProps{
	Value: cluster.bootstrapBrokersSaslScram,
})
awscdk.NewCfnOutput(this, jsii.String("BootstrapBrokerStringSaslIam"), &CfnOutputProps{
	Value: cluster.bootstrapBrokersSaslIam,
})
awscdk.NewCfnOutput(this, jsii.String("ZookeeperConnection"), &CfnOutputProps{
	Value: cluster.zookeeperConnectionString,
})
awscdk.NewCfnOutput(this, jsii.String("ZookeeperConnectionTls"), &CfnOutputProps{
	Value: cluster.zookeeperConnectionStringTls,
})

Importing an existing Cluster

To import an existing MSK cluster into your CDK app use the .fromClusterArn() method.

const cluster = msk.Cluster.fromClusterArn(this, 'Cluster',
  'arn:aws:kafka:us-west-2:1234567890:cluster/a-cluster/11111111-1111-1111-1111-111111111111-1',
);

cluster = msk.Cluster.from_cluster_arn(self, "Cluster", "arn:aws:kafka:us-west-2:1234567890:cluster/a-cluster/11111111-1111-1111-1111-111111111111-1")

ICluster cluster = Cluster.fromClusterArn(this, "Cluster", "arn:aws:kafka:us-west-2:1234567890:cluster/a-cluster/11111111-1111-1111-1111-111111111111-1");

var cluster = Cluster.FromClusterArn(this, "Cluster", "arn:aws:kafka:us-west-2:1234567890:cluster/a-cluster/11111111-1111-1111-1111-111111111111-1");

cluster := msk.Cluster_FromClusterArn(this, jsii.String("Cluster"), jsii.String("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 * as acmpca from 'aws-cdk-lib/aws-acmpca';

declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'Cluster', {
  clusterName: 'myCluster',
  kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
  vpc,
  encryptionInTransit: {
    clientBroker: msk.ClientBrokerEncryption.TLS,
  },
  clientAuthentication: msk.ClientAuthentication.tls({
    certificateAuthorities: [
      acmpca.CertificateAuthority.fromCertificateAuthorityArn(
        this,
        'CertificateAuthority',
        'arn:aws:acm-pca:us-west-2:1234567890:certificate-authority/11111111-1111-1111-1111-111111111111',
      ),
    ],
  }),
});

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")
        ]
    )
)

import software.amazon.awscdk.services.acmpca.*;

Vpc vpc;

Cluster cluster = Cluster.Builder.create(this, "Cluster")
        .clusterName("myCluster")
        .kafkaVersion(KafkaVersion.V4_1_X_KRAFT)
        .vpc(vpc)
        .encryptionInTransit(EncryptionInTransitConfig.builder()
                .clientBroker(ClientBrokerEncryption.TLS)
                .build())
        .clientAuthentication(ClientAuthentication.tls(TlsAuthProps.builder()
                .certificateAuthorities(List.of(CertificateAuthority.fromCertificateAuthorityArn(this, "CertificateAuthority", "arn:aws:acm-pca:us-west-2:1234567890:certificate-authority/11111111-1111-1111-1111-111111111111")))
                .build()))
        .build();

using Amazon.CDK.AWS.ACMPCA;

Vpc vpc;

var cluster = new Cluster(this, "Cluster", new ClusterProps {
    ClusterName = "myCluster",
    KafkaVersion = KafkaVersion.V4_1_X_KRAFT,
    Vpc = vpc,
    EncryptionInTransit = new EncryptionInTransitConfig {
        ClientBroker = ClientBrokerEncryption.TLS
    },
    ClientAuthentication = ClientAuthentication.Tls(new TlsAuthProps {
        CertificateAuthorities = new [] { CertificateAuthority.FromCertificateAuthorityArn(this, "CertificateAuthority", "arn:aws:acm-pca:us-west-2:1234567890:certificate-authority/11111111-1111-1111-1111-111111111111") }
    })
});

import acmpca "github.com/aws/aws-cdk-go/awscdk"

var vpc Vpc

cluster := msk.NewCluster(this, jsii.String("Cluster"), &ClusterProps{
	ClusterName: jsii.String("myCluster"),
	KafkaVersion: msk.KafkaVersion_V4_1_X_KRAFT(),
	Vpc: Vpc,
	EncryptionInTransit: &EncryptionInTransitConfig{
		ClientBroker: msk.ClientBrokerEncryption_TLS,
	},
	ClientAuthentication: msk.ClientAuthentication_Tls(&TlsAuthProps{
		CertificateAuthorities: []ICertificateAuthority{
			acmpca.CertificateAuthority_FromCertificateAuthorityArn(this, jsii.String("CertificateAuthority"), jsii.String("arn:aws:acm-pca:us-west-2:1234567890:certificate-authority/11111111-1111-1111-1111-111111111111")),
		},
	}),
})

SASL/SCRAM

Enable client authentication with SASL/SCRAM:

declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'cluster', {
  clusterName: 'myCluster',
  kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
  vpc,
  encryptionInTransit: {
    clientBroker: msk.ClientBrokerEncryption.TLS,
  },
  clientAuthentication: msk.ClientAuthentication.sasl({
    scram: true,
  }),
});

# 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
    )
)

Vpc vpc;

Cluster cluster = Cluster.Builder.create(this, "cluster")
        .clusterName("myCluster")
        .kafkaVersion(KafkaVersion.V4_1_X_KRAFT)
        .vpc(vpc)
        .encryptionInTransit(EncryptionInTransitConfig.builder()
                .clientBroker(ClientBrokerEncryption.TLS)
                .build())
        .clientAuthentication(ClientAuthentication.sasl(SaslAuthProps.builder()
                .scram(true)
                .build()))
        .build();

Vpc vpc;

var cluster = new Cluster(this, "cluster", new ClusterProps {
    ClusterName = "myCluster",
    KafkaVersion = KafkaVersion.V4_1_X_KRAFT,
    Vpc = vpc,
    EncryptionInTransit = new EncryptionInTransitConfig {
        ClientBroker = ClientBrokerEncryption.TLS
    },
    ClientAuthentication = ClientAuthentication.Sasl(new SaslAuthProps {
        Scram = true
    })
});

var vpc Vpc

cluster := msk.NewCluster(this, jsii.String("cluster"), &ClusterProps{
	ClusterName: jsii.String("myCluster"),
	KafkaVersion: msk.KafkaVersion_V4_1_X_KRAFT(),
	Vpc: Vpc,
	EncryptionInTransit: &EncryptionInTransitConfig{
		ClientBroker: msk.ClientBrokerEncryption_TLS,
	},
	ClientAuthentication: msk.ClientAuthentication_Sasl(&SaslAuthProps{
		Scram: jsii.Boolean(true),
	}),
})

IAM

Enable client authentication with IAM:

declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'cluster', {
  clusterName: 'myCluster',
  kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
  vpc,
  encryptionInTransit: {
    clientBroker: msk.ClientBrokerEncryption.TLS,
  },
  clientAuthentication: msk.ClientAuthentication.sasl({
    iam: true,
  }),
});

# 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
    )
)

Vpc vpc;

Cluster cluster = Cluster.Builder.create(this, "cluster")
        .clusterName("myCluster")
        .kafkaVersion(KafkaVersion.V4_1_X_KRAFT)
        .vpc(vpc)
        .encryptionInTransit(EncryptionInTransitConfig.builder()
                .clientBroker(ClientBrokerEncryption.TLS)
                .build())
        .clientAuthentication(ClientAuthentication.sasl(SaslAuthProps.builder()
                .iam(true)
                .build()))
        .build();

Vpc vpc;

var cluster = new Cluster(this, "cluster", new ClusterProps {
    ClusterName = "myCluster",
    KafkaVersion = KafkaVersion.V4_1_X_KRAFT,
    Vpc = vpc,
    EncryptionInTransit = new EncryptionInTransitConfig {
        ClientBroker = ClientBrokerEncryption.TLS
    },
    ClientAuthentication = ClientAuthentication.Sasl(new SaslAuthProps {
        Iam = true
    })
});

var vpc Vpc

cluster := msk.NewCluster(this, jsii.String("cluster"), &ClusterProps{
	ClusterName: jsii.String("myCluster"),
	KafkaVersion: msk.KafkaVersion_V4_1_X_KRAFT(),
	Vpc: Vpc,
	EncryptionInTransit: &EncryptionInTransitConfig{
		ClientBroker: msk.ClientBrokerEncryption_TLS,
	},
	ClientAuthentication: msk.ClientAuthentication_Sasl(&SaslAuthProps{
		Iam: jsii.Boolean(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 * as acmpca from 'aws-cdk-lib/aws-acmpca';

declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'Cluster', {
  clusterName: 'myCluster',
  kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
  vpc,
  encryptionInTransit: {
    clientBroker: msk.ClientBrokerEncryption.TLS,
  },
  clientAuthentication: msk.ClientAuthentication.saslTls({
    iam: true,
    certificateAuthorities: [
      acmpca.CertificateAuthority.fromCertificateAuthorityArn(
        this,
        'CertificateAuthority',
        'arn:aws:acm-pca:us-west-2:1234567890:certificate-authority/11111111-1111-1111-1111-111111111111',
      ),
    ],
  }),
});

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")
        ]
    )
)

import software.amazon.awscdk.services.acmpca.*;

Vpc vpc;

Cluster cluster = Cluster.Builder.create(this, "Cluster")
        .clusterName("myCluster")
        .kafkaVersion(KafkaVersion.V4_1_X_KRAFT)
        .vpc(vpc)
        .encryptionInTransit(EncryptionInTransitConfig.builder()
                .clientBroker(ClientBrokerEncryption.TLS)
                .build())
        .clientAuthentication(ClientAuthentication.saslTls(SaslTlsAuthProps.builder()
                .iam(true)
                .certificateAuthorities(List.of(CertificateAuthority.fromCertificateAuthorityArn(this, "CertificateAuthority", "arn:aws:acm-pca:us-west-2:1234567890:certificate-authority/11111111-1111-1111-1111-111111111111")))
                .build()))
        .build();

using Amazon.CDK.AWS.ACMPCA;

Vpc vpc;

var cluster = new Cluster(this, "Cluster", new ClusterProps {
    ClusterName = "myCluster",
    KafkaVersion = KafkaVersion.V4_1_X_KRAFT,
    Vpc = vpc,
    EncryptionInTransit = new EncryptionInTransitConfig {
        ClientBroker = ClientBrokerEncryption.TLS
    },
    ClientAuthentication = ClientAuthentication.SaslTls(new SaslTlsAuthProps {
        Iam = true,
        CertificateAuthorities = new [] { CertificateAuthority.FromCertificateAuthorityArn(this, "CertificateAuthority", "arn:aws:acm-pca:us-west-2:1234567890:certificate-authority/11111111-1111-1111-1111-111111111111") }
    })
});

import acmpca "github.com/aws/aws-cdk-go/awscdk"

var vpc Vpc

cluster := msk.NewCluster(this, jsii.String("Cluster"), &ClusterProps{
	ClusterName: jsii.String("myCluster"),
	KafkaVersion: msk.KafkaVersion_V4_1_X_KRAFT(),
	Vpc: Vpc,
	EncryptionInTransit: &EncryptionInTransitConfig{
		ClientBroker: msk.ClientBrokerEncryption_TLS,
	},
	ClientAuthentication: msk.ClientAuthentication_SaslTls(&SaslTlsAuthProps{
		Iam: jsii.Boolean(true),
		CertificateAuthorities: []ICertificateAuthority{
			acmpca.CertificateAuthority_FromCertificateAuthorityArn(this, jsii.String("CertificateAuthority"), jsii.String("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.

declare const vpc: ec2.Vpc;
declare const bucket: s3.IBucket;
const cluster = new msk.Cluster(this, 'cluster', {
  clusterName: 'myCluster',
  kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
  vpc,
  logging: {
    s3: {
      bucket,
    },
  },
});

# 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
        )
    )
)

Vpc vpc;
IBucket bucket;

Cluster cluster = Cluster.Builder.create(this, "cluster")
        .clusterName("myCluster")
        .kafkaVersion(KafkaVersion.V4_1_X_KRAFT)
        .vpc(vpc)
        .logging(BrokerLogging.builder()
                .s3(S3LoggingConfiguration.builder()
                        .bucket(bucket)
                        .build())
                .build())
        .build();

Vpc vpc;
IBucket bucket;

var cluster = new Cluster(this, "cluster", new ClusterProps {
    ClusterName = "myCluster",
    KafkaVersion = KafkaVersion.V4_1_X_KRAFT,
    Vpc = vpc,
    Logging = new BrokerLogging {
        S3 = new S3LoggingConfiguration {
            Bucket = bucket
        }
    }
});

var vpc Vpc
var bucket IBucket

cluster := msk.NewCluster(this, jsii.String("cluster"), &ClusterProps{
	ClusterName: jsii.String("myCluster"),
	KafkaVersion: msk.KafkaVersion_V4_1_X_KRAFT(),
	Vpc: Vpc,
	Logging: &BrokerLogging{
		S3: &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.

declare const vpc: ec2.Vpc;
declare const bucket: s3.IBucket;

const cluster = new msk.Cluster(this, 'cluster', {
  clusterName: 'myCluster',
  kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
  vpc,
  storageMode: msk.StorageMode.TIERED,
});

# 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
)

Vpc vpc;
IBucket bucket;


Cluster cluster = Cluster.Builder.create(this, "cluster")
        .clusterName("myCluster")
        .kafkaVersion(KafkaVersion.V4_1_X_KRAFT)
        .vpc(vpc)
        .storageMode(StorageMode.TIERED)
        .build();

Vpc vpc;
IBucket bucket;


var cluster = new Cluster(this, "cluster", new ClusterProps {
    ClusterName = "myCluster",
    KafkaVersion = KafkaVersion.V4_1_X_KRAFT,
    Vpc = vpc,
    StorageMode = StorageMode.TIERED
});

var vpc Vpc
var bucket IBucket


cluster := msk.NewCluster(this, jsii.String("cluster"), &ClusterProps{
	ClusterName: jsii.String("myCluster"),
	KafkaVersion: msk.KafkaVersion_V4_1_X_KRAFT(),
	Vpc: Vpc,
	StorageMode: 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, 3.9.x, or 4.2.x
• You must specify the instanceType
• The VPC must have at least 3 subnets (across 3 AZs)
• ebsStorageInfo is not supported
• storageMode is not supported
• logging is not supported
• Supported broker sizes: m7g.xlarge, m7g.2xlarge, m7g.4xlarge, m7g.8xlarge, m7g.12xlarge, m7g.16xlarge

declare const vpc: ec2.Vpc;

const expressCluster = new msk.Cluster(this, 'ExpressCluster', {
  clusterName: 'MyExpressCluster',
  kafkaVersion: msk.KafkaVersion.V3_8_X,
  vpc,
  brokerType: msk.BrokerType.EXPRESS,
  instanceType: ec2.InstanceType.of(
    ec2.InstanceClass.M7G,
    ec2.InstanceSize.XLARGE,
  ),
});

# 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)
)

Vpc vpc;


Cluster expressCluster = Cluster.Builder.create(this, "ExpressCluster")
        .clusterName("MyExpressCluster")
        .kafkaVersion(KafkaVersion.V3_8_X)
        .vpc(vpc)
        .brokerType(BrokerType.EXPRESS)
        .instanceType(InstanceType.of(InstanceClass.M7G, InstanceSize.XLARGE))
        .build();

Vpc vpc;


var expressCluster = new Cluster(this, "ExpressCluster", new ClusterProps {
    ClusterName = "MyExpressCluster",
    KafkaVersion = KafkaVersion.V3_8_X,
    Vpc = vpc,
    BrokerType = BrokerType.EXPRESS,
    InstanceType = InstanceType.Of(InstanceClass.M7G, InstanceSize.XLARGE)
});

var vpc Vpc


expressCluster := msk.NewCluster(this, jsii.String("ExpressCluster"), &ClusterProps{
	ClusterName: jsii.String("MyExpressCluster"),
	KafkaVersion: msk.KafkaVersion_V3_8_X(),
	Vpc: Vpc,
	BrokerType: msk.BrokerType_EXPRESS,
	InstanceType: 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.

declare const vpc: ec2.Vpc;

const serverlessCluster = new msk.ServerlessCluster(this, 'ServerlessCluster', {
  clusterName: 'MyServerlessCluster',
  vpcConfigs: [
    { vpc },
  ],
});

# vpc: ec2.Vpc


serverless_cluster = msk.ServerlessCluster(self, "ServerlessCluster",
    cluster_name="MyServerlessCluster",
    vpc_configs=[msk.VpcConfig(vpc=vpc)
    ]
)

Vpc vpc;


ServerlessCluster serverlessCluster = ServerlessCluster.Builder.create(this, "ServerlessCluster")
        .clusterName("MyServerlessCluster")
        .vpcConfigs(List.of(VpcConfig.builder().vpc(vpc).build()))
        .build();

Vpc vpc;


var serverlessCluster = new ServerlessCluster(this, "ServerlessCluster", new ServerlessClusterProps {
    ClusterName = "MyServerlessCluster",
    VpcConfigs = new [] { new VpcConfig { Vpc = vpc } }
});

var vpc Vpc


serverlessCluster := msk.NewServerlessCluster(this, jsii.String("ServerlessCluster"), &ServerlessClusterProps{
	ClusterName: jsii.String("MyServerlessCluster"),
	VpcConfigs: []VpcConfig{
		&VpcConfig{
			Vpc: *Vpc,
		},
	},
})

Types

Name	Kind	Summary
Cluster	Construct (L2)	Create a MSK Cluster.
ClusterBase	Construct (L2)	A new or imported MSK Cluster.
ServerlessCluster	Construct (L2)	Create a MSK Serverless Cluster.
BrokerLogging	Struct	Configuration details related to broker logs.
BrokerType	Enum	The broker type for the cluster.
ClientAuthentication	Class	Configuration properties for client authentication.
ClientBrokerEncryption	Enum	Indicates the encryption setting for data in transit between clients and brokers.
ClusterConfigurationInfo	Struct	The Amazon MSK configuration to use for the cluster.
ClusterMonitoringLevel	Enum	The level of monitoring for the MSK cluster.
ClusterProps	Struct	Properties for a MSK Cluster.
EbsStorageInfo	Struct	EBS volume information.
EncryptionInTransitConfig	Struct	The settings for encrypting data in transit.
ICluster	Interface	Represents a MSK Cluster.
KafkaVersion	Class	Kafka cluster version.
KafkaVersionFeatures	Struct	Available features for a given Kafka version.
MonitoringConfiguration	Struct	Monitoring Configuration.
S3LoggingConfiguration	Struct	Details of the Amazon S3 destination for broker logs.
SaslAuthProps	Struct	SASL authentication properties.
SaslTlsAuthProps	Struct	SASL + TLS authentication properties.
ServerlessClusterProps	Struct	Properties for a MSK Serverless Cluster.
StorageMode	Enum	The storage mode for the cluster brokers.
TlsAuthProps	Struct	TLS authentication properties.
VpcConfig	Struct	The configuration of the Amazon VPCs for the cluster.