feat(streaming-kafka-api): Reuse existing service token on MSK Kafka API

framework/src/streaming/README.md

@@ -202,6 +202,8 @@ The construct leverages the [CDK Provider Framework](https://docs.aws.amazon.com
 
 [example kafka api](./examples/kafka-api-default.lit.ts)
 
+When deploying multiple stacks with the Kafka Api, if there is an already existing service token deployed for the custom resource, you can reuse it to reduce the number of resource created, therefore the number of IP assigned to the custom resources.
+
 :::warning
 
 The construct needs to be deployed in the same region as the MSK cluster.

framework/src/streaming/examples/kafka-api-default.lit.ts

@@ -43,6 +43,7 @@ const kafkaApi = new KafkaApi(stack, 'kafkaApi', {
     certificateAuthorities: [certificateAuthority],
   },),
   kafkaClientLogLevel: KafkaClientLogLevel.DEBUG,
+  serviceToken: 'arn:aws:lambda::XXXXXX:function:XXXXXX-kafkaApiMskIamProviderCustomResour-XXXXXX',
 });
 /// !hide
 

framework/src/streaming/lib/msk/kafka-api-props.ts

@@ -88,4 +88,10 @@ export interface KafkaApiProps {
    * @default WARN
    */
   readonly kafkaClientLogLevel?: KafkaClientLogLevel;
+
+  /**
+   * If there is an already existing service token deployed for the custom resource
+   * you can reuse it to reduce the number of resource created
+   */
+  readonly serviceToken?: string;
 }

framework/src/streaming/lib/msk/kafka-api.ts

@@ -59,6 +59,11 @@ export class KafkaApi extends TrackedConstruct {
    * The Security Group used by the Custom Resource provider when MSK is using mTLS authentication
    */
   public readonly mskAclSecurityGroup?: ISecurityGroup [];
+  /**
+   * If there is an already existing service token deployed for the custom resource
+   * you can reuse it to reduce the number of resource created
+   */
+  public readonly serviceToken?: string;
 
   private readonly mskAclServiceToken?: string;
   private readonly mskIamServiceToken?: string;
@@ -96,41 +101,51 @@ export class KafkaApi extends TrackedConstruct {
 
     if (props.clientAuthentication.tlsProps?.certificateAuthorities) {
 
-      const mskAclProvider = mskAclAdminProviderSetup(
-        this,
-        this.removalPolicy,
-        props.vpc,
-        props.subnets || props.vpc.selectSubnets({ subnetType: SubnetType.PRIVATE_WITH_EGRESS }),
-        props.brokerSecurityGroup,
-        props.clusterArn,
-        props.certficateSecret!,
-        props.mtlsHandlerRole,
-      );
-
-      this.mskAclServiceToken = mskAclProvider.serviceToken;
-      this.mskAclRole = mskAclProvider.onEventHandlerRole;
-      this.mskAclLogGroup = mskAclProvider.onEventHandlerLogGroup;
-      this.mskAclFunction = mskAclProvider.onEventHandlerFunction;
-      this.mskAclSecurityGroup = mskAclProvider.securityGroups;
+      if (props.serviceToken) {
+        this.mskAclServiceToken = this.serviceToken = props.serviceToken;
+      } else {
+        const mskAclProvider = mskAclAdminProviderSetup(
+          this,
+          this.removalPolicy,
+          props.vpc,
+          props.subnets || props.vpc.selectSubnets({ subnetType: SubnetType.PRIVATE_WITH_EGRESS }),
+          props.brokerSecurityGroup,
+          props.clusterArn,
+          props.certficateSecret!,
+          props.mtlsHandlerRole,
+        );
+
+        this.mskAclServiceToken = mskAclProvider.serviceToken;
+        this.mskAclRole = mskAclProvider.onEventHandlerRole;
+        this.mskAclLogGroup = mskAclProvider.onEventHandlerLogGroup;
+        this.mskAclFunction = mskAclProvider.onEventHandlerFunction;
+        this.mskAclSecurityGroup = mskAclProvider.securityGroups;
+        this.serviceToken = mskAclProvider.serviceToken;
+      }
     }
 
     if ( props.clientAuthentication.saslProps?.iam) {
 
-      const mskIamProvider = mskIamCrudProviderSetup(
-        this,
-        this.removalPolicy,
-        props.vpc,
-        props.subnets || props.vpc.selectSubnets({ subnetType: SubnetType.PRIVATE_WITH_EGRESS }),
-        props.brokerSecurityGroup,
-        props.clusterArn,
-        props.iamHandlerRole,
-      );
-
-      this.mskIamServiceToken = mskIamProvider.serviceToken;
-      this.mskIamRole = mskIamProvider.onEventHandlerRole;
-      this.mskIamLogGroup = mskIamProvider.onEventHandlerLogGroup;
-      this.mskIamFunction = mskIamProvider.onEventHandlerFunction;
-      this.mskIamSecurityGroup = mskIamProvider.securityGroups;
+      if (props.serviceToken) {
+        this.mskIamServiceToken = this.serviceToken = props.serviceToken;
+      } else {
+        const mskIamProvider = mskIamCrudProviderSetup(
+          this,
+          this.removalPolicy,
+          props.vpc,
+          props.subnets || props.vpc.selectSubnets({ subnetType: SubnetType.PRIVATE_WITH_EGRESS }),
+          props.brokerSecurityGroup,
+          props.clusterArn,
+          props.iamHandlerRole,
+        );
+
+        this.mskIamServiceToken = mskIamProvider.serviceToken;
+        this.mskIamRole = mskIamProvider.onEventHandlerRole;
+        this.mskIamLogGroup = mskIamProvider.onEventHandlerLogGroup;
+        this.mskIamFunction = mskIamProvider.onEventHandlerFunction;
+        this.mskIamSecurityGroup = mskIamProvider.securityGroups;
+        this.serviceToken = mskIamProvider.serviceToken;
+      }
     }
 
   }

framework/src/streaming/lib/msk/msk-provisioned.ts

@@ -5,7 +5,7 @@ import { readFileSync } from 'fs';
 import { join } from 'path';
 
 
-import { CustomResource, Duration, RemovalPolicy, Stack } from 'aws-cdk-lib';
+import { CfnOutput, CustomResource, Duration, RemovalPolicy, Stack, Aws } from 'aws-cdk-lib';
 import { Connections, ISecurityGroup, IVpc, SecurityGroup, SubnetType } from 'aws-cdk-lib/aws-ec2';
 import { IPrincipal, IRole, ManagedPolicy, PolicyDocument, PolicyStatement, Role, ServicePrincipal } from 'aws-cdk-lib/aws-iam';
 import { IKey, Key } from 'aws-cdk-lib/aws-kms';
@@ -174,6 +174,11 @@ export class MskProvisioned extends TrackedConstruct {
    * The Security Group used by the Lambda responsible for CRUD operations via mTLS authentication
    */
   public readonly inClusterAclSecurityGroup?: ISecurityGroup[];
+  /**
+   * If there is an already existing service token deployed for the custom resource
+   * you can reuse it to reduce the number of resource created
+   */
+  public readonly serviceToken?: string;
 
   private readonly removalPolicy: RemovalPolicy;
   private readonly region: string;
@@ -474,6 +479,8 @@ export class MskProvisioned extends TrackedConstruct {
       removalPolicy: this.removalPolicy,
     });
 
+    this.serviceToken = this.kafkaApi.serviceToken;
+
     // Create the configuration
     let clusterConfigurationInfo: ClusterConfigurationInfo;
 
@@ -532,7 +539,7 @@ export class MskProvisioned extends TrackedConstruct {
     this.updateConnectivitySecurityGroup = updateConnectivityProvider.securityGroups;
 
     // Set the CR resource that are used by IAM credentials auth CR
-    // Applly the cluster configuration if provided and the cluster is created without mTLS auth
+    // Apply the cluster configuration if provided and the cluster is created without mTLS auth
     if (this.iamAcl) {
 
       this.iamCrudAdminRole = this.kafkaApi.mskAclRole;
@@ -635,6 +642,11 @@ export class MskProvisioned extends TrackedConstruct {
       }
     }
 
+    new CfnOutput(this, 'ServiceToken', {
+      value: this.serviceToken!,
+      exportName: `${Aws.STACK_NAME}-ServiceToken`,
+    });
+
   }
 
 

framework/src/streaming/lib/msk/msk-serverless.ts

@@ -1,7 +1,7 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
-import { CustomResource, RemovalPolicy } from 'aws-cdk-lib';
+import { CfnOutput, CustomResource, RemovalPolicy, Aws } from 'aws-cdk-lib';
 import { ISecurityGroup, IVpc, SecurityGroup, SubnetSelection } from 'aws-cdk-lib/aws-ec2';
 import { IPrincipal, PolicyDocument } from 'aws-cdk-lib/aws-iam';
 import { CfnClusterPolicy, CfnServerlessCluster } from 'aws-cdk-lib/aws-msk';
@@ -28,6 +28,11 @@ export class MskServerless extends TrackedConstruct {
   public readonly brokerSecurityGroup?: ISecurityGroup;
   public readonly clusterName: string;
   public readonly lambdaSecurityGroup: ISecurityGroup;
+  /**
+   * If there is an already existing service token deployed for the custom resource
+   * you can reuse it to reduce the number of resource created
+   */
+  public readonly serviceToken?: string;
 
   private readonly removalPolicy: RemovalPolicy;
   private readonly kafkaApi: KafkaApi;
@@ -112,6 +117,11 @@ export class MskServerless extends TrackedConstruct {
       kafkaClientLogLevel: props?.kafkaClientLogLevel ?? KafkaClientLogLevel.WARN,
     });
 
+    this.serviceToken = this.kafkaApi.serviceToken;
+    new CfnOutput(this, 'ServiceToken', {
+      value: this.serviceToken!,
+      exportName: `${Aws.STACK_NAME}-ServiceToken`,
+    });
   }
 
   /**