aws-msk-alpha: support multi-VPC private connectivity

[manuwaik]

Describe the feature

As per the announcement: https://aws.amazon.com/about-aws/whats-new/2023/04/amazon-msk-multi-vpc-private-connectivity-cross-account-access/

Amazon Managed Streaming for Apache Kafka (MSK) now offers multi-VPC private connectivity that simplifies connectivity and access to your Amazon MSK clusters from your Apache Kafka clients hosted in any VPC or AWS account.

Use Case

Customers currently need to manually turn on the Multi-VPC option within the console. Some customers have requirements around doing all actions via Infrastructure as Code.

Proposed Solution

No response

Other Information

Currently there are some limitations to keep in mind: https://docs.aws.amazon.com/msk/latest/developerguide/aws-access-mult-vpc.html#mvpc-requirements

• Multi-VPC only on Apache Kafka 2.7.1 or higher.
• Unauthenticated clusters can't use multi-VPC private connectivity.
• Doesn’t support the t3.small instance type.
• Isn’t supported across AWS Regions, only on AWS accounts within the same Region.
• Amazon MSK doesn't support multi-VPC private connectivity to Zookeeper nodes.

Acknowledgements

• I may be able to implement this feature request
• This feature might incur a breaking change

CDK version used

2.118.0

Environment details (OS name and version, etc.)

N/A

[pahud]

Just checked with Amazon Q, this would be possible using cloudformation so it should be possible to be included in aws-msk-alpha modules.

Please help us prioritize with 👍 and we welcome any PRs for that.

[pahud]

This is my current solution for MSK multi-vpc:

• initial deploy

    const cluster = new mskAlpha.Cluster(this, 'msk-cluster', {
      clusterName: `cluster${Stack.of(this).stackName}`,   
      kafkaVersion: mskAlpha.KafkaVersion.V3_5_1,
      clientAuthentication: mskAlpha.ClientAuthentication.sasl({ iam: true }),
      vpc,
      numberOfBrokerNodes: 1,
      vpcSubnets: { subnetType: ec2.SubnetType.PRIVATE_ISOLATED },
    });

    (cluster.node.defaultChild as msk.CfnCluster).brokerNodeGroupInfo = {
      connectivityInfo: {
        vpcConnectivity: {
          clientAuthentication: {
            sasl: {
              iam: { enabled: false },
              scram: { enabled: false},
            },
            tls: { enabled: false }
          }
        }
      },
      clientSubnets: vpc.selectSubnets({ subnetType: SubnetType.PRIVATE_ISOLATED }).subnetIds,
      instanceType: "kafka.m5.large"
    }

update - enable iam

    const cluster = new mskAlpha.Cluster(this, 'msk-cluster', {
      clusterName: `cluster${Stack.of(this).stackName}`,   
      kafkaVersion: mskAlpha.KafkaVersion.V3_5_1,
      clientAuthentication: mskAlpha.ClientAuthentication.sasl({ iam: true }),
      vpc,
      numberOfBrokerNodes: 1,
      vpcSubnets: { subnetType: ec2.SubnetType.PRIVATE_ISOLATED },
    });

    (cluster.node.defaultChild as msk.CfnCluster).brokerNodeGroupInfo = {
      connectivityInfo: {
        vpcConnectivity: {
          clientAuthentication: {
            sasl: {
              iam: { enabled: true },
              scram: { enabled: false},
            },
            tls: { enabled: false }
          }
        }
      },
      clientSubnets: vpc.selectSubnets({ subnetType: SubnetType.PRIVATE_ISOLATED }).subnetIds,
      instanceType: "kafka.m5.large"
    }

For some reason, MSK cluster does not allow you to enable vpcConnectivity auth scheme in the initial deployment otherwise you get the InvalidRequest error:

11:35:10 PM | CREATE_FAILED | AWS::MSK::Cluster | msk-cluster
Resource handler returned message: "[ClientRequestToken: 823e4af3-b557-c3ae-2f52-e93c70425b92] When creating a cluster, all vpcConnectivity auth scheme
s must be disabled ('enabled' : false). You can enable auth schemes after the cluster is created. (Service: Kafka, Status Code: 400, Request ID: 803416
7b-7f43-491d-8a93-2b8837200699) 'vpcConnectivity.clientAuthentication'" (RequestToken: 823e4af3-b557-c3ae-2f52-e93c70425b92, HandlerErrorCode: InvalidRequest)

[lemon0333]

Hi @pahud — I'd like to pick this up and implement it in aws-msk-alpha following your prototype in this thread.

Proposed API shape (keeps default behavior unchanged when unset):

// Cluster side: turn on multi-VPC private connectivity + the client auth it requires
const cluster = new msk.Cluster(this, 'Cluster', {
  clusterName: 'myCluster',
  kafkaVersion: msk.KafkaVersion.V3_5_1,
  vpc,
  clientAuthentication: msk.ClientAuthentication.sasl({ iam: true }),
  vpcConnectivity: msk.VpcConnectivity.sasl({ iam: true }), // .tls() / .saslTls({...}) also available
});

// Cross-account: attach a resource-based cluster policy (AWS::MSK::ClusterPolicy)
cluster.addClusterPolicy(new iam.PolicyDocument({ /* ... */ }));

This renders BrokerNodeGroupInfo.ConnectivityInfo.VpcConnectivity.ClientAuthentication (SASL/IAM, SASL/SCRAM, TLS) exactly as in your prototype, and adds a helper to surface AWS::MSK::ClusterPolicy for cross-account access. The client-side AWS::MSK::VpcConnection (L1) usage will be documented in the README. I'll also note the deploy-time caveat you hit (auth schemes must be disabled on initial create, enabled on a follow-up update).

Will open a PR shortly — would appreciate your review. Thanks!