fix(lambda): translate self-managed Kafka Endpoints key to the SDK enum value

docs/_generated/cli-flag-coverage.json

@@ -187,6 +187,7 @@
         "lambda-config-field-removal",
         "lambda-destinations",
         "lambda-env-removal",
+        "lambda-esm-self-managed-kafka",
         "lambda-event-invoke-config-update",
         "lambda-layer-version-update",
         "lambda-log-retention",
@@ -424,6 +425,7 @@
         "lambda-config-field-removal",
         "lambda-destinations",
         "lambda-env-removal",
+        "lambda-esm-self-managed-kafka",
         "lambda-event-invoke-config-update",
         "lambda-layer-version-update",
         "lambda-log-retention",
@@ -637,6 +639,7 @@
         "lambda-config-field-removal",
         "lambda-destinations",
         "lambda-env-removal",
+        "lambda-esm-self-managed-kafka",
         "lambda-event-invoke-config-update",
         "lambda-layer-version-update",
         "lambda-log-retention",
@@ -890,6 +893,7 @@
         "lambda-config-field-removal",
         "lambda-destinations",
         "lambda-env-removal",
+        "lambda-esm-self-managed-kafka",
         "lambda-event-invoke-config-update",
         "lambda-layer-version-update",
         "lambda-log-retention",
@@ -1127,6 +1131,7 @@
         "lambda-config-field-removal",
         "lambda-destinations",
         "lambda-env-removal",
+        "lambda-esm-self-managed-kafka",
         "lambda-event-invoke-config-update",
         "lambda-layer-version-update",
         "lambda-log-retention",

docs/_generated/integ-coverage.json

@@ -3267,6 +3267,7 @@
         "iam-role-prefixed-name-update",
         "infra-security",
         "intrinsic-functions",
+        "lambda-esm-self-managed-kafka",
         "lambda-microvm-image",
         "lambda-snapstart",
         "local-invoke-dotnet",
@@ -3396,6 +3397,9 @@
         "intrinsic-functions": [
           "l2"
         ],
+        "lambda-esm-self-managed-kafka": [
+          "literal"
+        ],
         "lambda-microvm-image": [
           "l2"
         ],
@@ -3655,7 +3659,8 @@
         "dynamodb-stream-filter",
         "dynamodb-streams",
         "eventsourcemapping-race",
-        "fifo-sqs-event-source"
+        "fifo-sqs-event-source",
+        "lambda-esm-self-managed-kafka"
       ],
       "signals": {
         "dynamodb-stream-filter": [
@@ -3669,6 +3674,9 @@
         ],
         "fifo-sqs-event-source": [
           "literal"
+        ],
+        "lambda-esm-self-managed-kafka": [
+          "literal"
         ]
       }
     },
@@ -3719,6 +3727,7 @@
         "lambda-config-field-removal",
         "lambda-destinations",
         "lambda-env-removal",
+        "lambda-esm-self-managed-kafka",
         "lambda-event-invoke-config-update",
         "lambda-layer-version-update",
         "lambda-log-retention",
@@ -3917,6 +3926,10 @@
           "l2",
           "literal"
         ],
+        "lambda-esm-self-managed-kafka": [
+          "l2",
+          "literal"
+        ],
         "lambda-event-invoke-config-update": [
           "l2"
         ],
@@ -5288,6 +5301,7 @@
         "event-driven",
         "full-stack-demo",
         "iam-access-key",
+        "lambda-esm-self-managed-kafka",
         "local-run-task-from-state",
         "replacement-immutable-name",
         "secrets-dynamic-ref",
@@ -5307,6 +5321,10 @@
         "iam-access-key": [
           "l2"
         ],
+        "lambda-esm-self-managed-kafka": [
+          "l2",
+          "literal"
+        ],
         "local-run-task-from-state": [
           "l2",
           "literal"

docs/_generated/integ-last-run.tsv

@@ -145,6 +145,7 @@ lambda-arch-switch	2026-07-26T19:25:15Z	PASS	63	verify.sh	0727b sweep-b10 stalen
 lambda-config-field-removal	2026-07-22T08:19:38Z	PASS	140	verify.sh	issue #1158 post-rebase re-verify; 6-field removal reset + destroy clean
 lambda-destinations	2026-07-21T18:15:21Z	PASS	81	verify.sh	fixture fix: EventInvokeConfig now sdk-routed (#919); rc ok, orph clean
 lambda-env-removal	2026-07-26T19:16:39Z	PASS	70	verify.sh	0727b sweep-b9 staleness re-run (rc=0); account clean
+lambda-esm-self-managed-kafka	2026-08-09T04:24:56Z	PASS	215	verify.sh	#1384 enum key + state-spelling inverse asserted; destroy 5 deleted, 0 errors, 0 orphans
 lambda-event-invoke-config-update	2026-07-26T18:20:27Z	PASS	80	verify.sh	0727b sweep-b3 staleness re-run (rc=0); account clean
 lambda-layer-version-update	2026-07-26T21:13:17Z	PASS	82	verify.sh	re-run post review fixes (tab-safe version list); account clean
 lambda-log-retention	2026-07-21T14:46:30Z	PASS	95	verify.sh	rc ok, orph clean

docs/_generated/scenario-coverage.json

@@ -1139,6 +1139,11 @@
       "annotated": false,
       "scenarios": []
     },
+    {
+      "name": "lambda-esm-self-managed-kafka",
+      "annotated": false,
+      "scenarios": []
+    },
     {
       "name": "lambda-event-invoke-config-update",
       "annotated": false,
@@ -2584,6 +2589,7 @@
     "lambda-config-field-removal",
     "lambda-destinations",
     "lambda-env-removal",
+    "lambda-esm-self-managed-kafka",
     "lambda-event-invoke-config-update",
     "lambda-layer-version-update",
     "lambda-log-retention",

docs/scenario-coverage.md

@@ -4,7 +4,7 @@
 
 Run `vp run scenario-coverage` to regenerate.
 
-**84 / 84 canonical scenarios** have at least one integ fixture exercising them. **180 / 264 integ fixtures** carry a `.scenarios.json` sidecar (with 0+ tags); the rest are un-annotated and contributor-reviewed below.
+**84 / 84 canonical scenarios** have at least one integ fixture exercising them. **180 / 265 integ fixtures** carry a `.scenarios.json` sidecar (with 0+ tags); the rest are un-annotated and contributor-reviewed below.
 
 ## How this is computed
 
@@ -115,7 +115,7 @@ _None._ Every canonical scenario has at least one integ fixture tagged with it.
 | `vpc-lambda-eni-release` | Lambda hyperplane ENI cleanup after DeleteFunction (5-30 min eventually consistent). | [`bench-cdk-sample`](../tests/integration/bench-cdk-sample/)<br>[`destroy-interrupt`](../tests/integration/destroy-interrupt/)<br>[`lambda`](../tests/integration/lambda/)<br>[`vpc-lambda`](../tests/integration/vpc-lambda/) |
 | `wide-dag-throttle-retry` | Wide (~100-resource: 80 SSM Parameters + 10 IAM Roles + 10 SNS Topics, 10-deep SSM Fn::Sub chain) single-stack burst deployed under a HIGH `--concurrency` to stress the concurrency limiter + event-driven DAG executor + throttle/retry classifier: a `TooManyRequests` / `Rate exceeded` / HTTP 429 during the burst must be RETRIED (deploy still succeeds) not fatal, the chained subset proves strict DAG ordering, and the destroy burst absorbs ~100 deletes with 0 orphans. | [`throttle-wide-dag`](../tests/integration/throttle-wide-dag/) |
 
-## Un-annotated fixtures (84)
+## Un-annotated fixtures (85)
 
 These integ fixtures have no `.scenarios.json` sidecar. They may or may not exercise a canonical scenario — contributor review needed. To opt out (per-service smoke tests with no canonical pattern), add a sidecar with `{ "scenarios": [] }`.
 
@@ -165,6 +165,7 @@ These integ fixtures have no `.scenarios.json` sidecar. They may or may not exer
 - [`lambda-config-field-removal`](../tests/integration/lambda-config-field-removal/)
 - [`lambda-destinations`](../tests/integration/lambda-destinations/)
 - [`lambda-env-removal`](../tests/integration/lambda-env-removal/)
+- [`lambda-esm-self-managed-kafka`](../tests/integration/lambda-esm-self-managed-kafka/)
 - [`lambda-event-invoke-config-update`](../tests/integration/lambda-event-invoke-config-update/)
 - [`lambda-layer-version-update`](../tests/integration/lambda-layer-version-update/)
 - [`lambda-log-retention`](../tests/integration/lambda-log-retention/)

src/provisioning/providers/lambda-eventsource-provider.ts

@@ -99,6 +99,68 @@ function classifyEventSourceFromProperties(properties: Record<string, unknown>):
   });
 }
 
+/**
+ * CFn spells the self-managed-Kafka bootstrap-server list
+ * `SelfManagedEventSource.Endpoints.KafkaBootstrapServers`, while the SDK
+ * models `Endpoints` as `Partial<Record<EndPointType, string[]>>` keyed by the
+ * enum VALUE `KAFKA_BOOTSTRAP_SERVERS` (`@aws-sdk/client-lambda`
+ * `EndPointType`). Because `Endpoints` is a MAP rather than a modeled
+ * structure, the serializer forwards the unknown CFn key verbatim and the
+ * service rejects the whole request — so every CDK `SelfManagedKafkaEventSource`
+ * user hit a hard `CreateEventSourceMapping` failure (issue #1384).
+ *
+ * Both directions are pure key renames of the ONE diverging key; every other
+ * member of the blob (both inside `Endpoints` and beside it) is copied through
+ * untouched. Note that a future `EndPointType` whose CFn spelling also diverges
+ * would reproduce this bug exactly — it needs its own entry here, the pass-
+ * through is not a general solution.
+ *
+ * Anything that is not a re-shapeable object — a non-object blob, a missing or
+ * non-object `Endpoints`, an array `Endpoints` — is returned VERBATIM rather
+ * than dropped. Dropping would be a silent-drop regression against the raw
+ * pass-through this replaced: AWS must stay the one that rejects a malformed
+ * template, not this layer.
+ */
+const CFN_KAFKA_ENDPOINTS_KEY = 'KafkaBootstrapServers';
+const SDK_KAFKA_ENDPOINTS_KEY = 'KAFKA_BOOTSTRAP_SERVERS';
+
+function renameEndpointsKey(selfManagedEventSource: unknown, from: string, to: string): unknown {
+  if (typeof selfManagedEventSource !== 'object' || selfManagedEventSource === null) {
+    return selfManagedEventSource;
+  }
+  const source = { ...(selfManagedEventSource as Record<string, unknown>) };
+  const endpoints = source['Endpoints'];
+  if (typeof endpoints !== 'object' || endpoints === null || Array.isArray(endpoints)) {
+    return source;
+  }
+  const renamed: Record<string, unknown> = {};
+  for (const [key, value] of Object.entries(endpoints as Record<string, unknown>)) {
+    renamed[key === from ? to : key] = value;
+  }
+  source['Endpoints'] = renamed;
+  return source;
+}
+
+/** CFn property bag -> `CreateEventSourceMapping` input shape. */
+function toSdkSelfManagedEventSource(
+  selfManagedEventSource: unknown
+): import('@aws-sdk/client-lambda').SelfManagedEventSource {
+  return renameEndpointsKey(
+    selfManagedEventSource,
+    CFN_KAFKA_ENDPOINTS_KEY,
+    SDK_KAFKA_ENDPOINTS_KEY
+  ) as import('@aws-sdk/client-lambda').SelfManagedEventSource;
+}
+
+/** `GetEventSourceMapping` response -> CFn property shape (drift readback). */
+function toCfnSelfManagedEventSource(selfManagedEventSource: unknown): unknown {
+  return renameEndpointsKey(
+    selfManagedEventSource,
+    SDK_KAFKA_ENDPOINTS_KEY,
+    CFN_KAFKA_ENDPOINTS_KEY
+  );
+}
+
 const KINDS_WITH_FUNCTION_RESPONSE_TYPES: ReadonlySet<EventSourceKind> = new Set([
   'sqs',
   'kinesis',
@@ -250,9 +312,9 @@ export class LambdaEventSourceMappingProvider implements ResourceProvider {
           'SourceAccessConfigurations'
         ] as import('@aws-sdk/client-lambda').SourceAccessConfiguration[];
       if (properties['SelfManagedEventSource'])
-        params.SelfManagedEventSource = properties[
-          'SelfManagedEventSource'
-        ] as import('@aws-sdk/client-lambda').SelfManagedEventSource;
+        params.SelfManagedEventSource = toSdkSelfManagedEventSource(
+          properties['SelfManagedEventSource']
+        );
       if (properties['SelfManagedKafkaEventSourceConfig'])
         params.SelfManagedKafkaEventSourceConfig = properties[
           'SelfManagedKafkaEventSourceConfig'
@@ -787,7 +849,11 @@ export class LambdaEventSourceMappingProvider implements ResourceProvider {
       result['SourceAccessConfigurations'] = resp.SourceAccessConfigurations;
     }
     if (resp.SelfManagedEventSource !== undefined) {
-      result['SelfManagedEventSource'] = resp.SelfManagedEventSource;
+      // Inverse of the create-side rename: state holds the CFn spelling
+      // (`Endpoints.KafkaBootstrapServers`), so emitting the SDK's
+      // `KAFKA_BOOTSTRAP_SERVERS` here would fire guaranteed drift on every
+      // clean run of a self-managed-Kafka ESM (issue #1384).
+      result['SelfManagedEventSource'] = toCfnSelfManagedEventSource(resp.SelfManagedEventSource);
     }
     if (resp.SelfManagedKafkaEventSourceConfig !== undefined) {
       result['SelfManagedKafkaEventSourceConfig'] = resp.SelfManagedKafkaEventSourceConfig;

tests/integration/lambda-esm-self-managed-kafka/bin/app.ts

@@ -0,0 +1,9 @@
+#!/usr/bin/env node
+import * as cdk from 'aws-cdk-lib';
+import { LambdaEsmSelfManagedKafkaStack } from '../lib/lambda-esm-self-managed-kafka-stack.ts';
+
+const app = new cdk.App();
+new LambdaEsmSelfManagedKafkaStack(app, 'CdkdLambdaEsmSelfManagedKafkaExample', {
+  description: 'cdkd Lambda self-managed Kafka ESM Endpoints key integ probe',
+  env: { account: process.env.CDK_DEFAULT_ACCOUNT, region: process.env.CDK_DEFAULT_REGION },
+});

tests/integration/lambda-esm-self-managed-kafka/cdk.json

@@ -0,0 +1,3 @@
+{
+  "app": "node bin/app.ts"
+}

tests/integration/lambda-esm-self-managed-kafka/lib/lambda-esm-self-managed-kafka-stack.ts

@@ -0,0 +1,80 @@
+import * as cdk from 'aws-cdk-lib';
+import { Construct } from 'constructs';
+import * as lambda from 'aws-cdk-lib/aws-lambda';
+import * as secretsmanager from 'aws-cdk-lib/aws-secretsmanager';
+import {
+  AuthenticationMethod,
+  SelfManagedKafkaEventSource,
+} from 'aws-cdk-lib/aws-lambda-event-sources';
+
+/**
+ * Lambda event source mapping for a SELF-MANAGED Apache Kafka cluster
+ * (issue #1384).
+ *
+ * CFn spells the bootstrap-server map key
+ * `SelfManagedEventSource.Endpoints.KafkaBootstrapServers`, while the SDK
+ * models `Endpoints` as `Partial<Record<EndPointType, string[]>>` keyed by the
+ * enum VALUE `KAFKA_BOOTSTRAP_SERVERS`. Because `Endpoints` is a MAP rather
+ * than a modeled structure, the AWS SDK v3 serializer forwards the unknown CFn
+ * key verbatim and the service REJECTS the request — so before the fix
+ * `CreateEventSourceMapping` failed outright for every CDK
+ * `SelfManagedKafkaEventSource` user. A successful deploy of this fixture IS
+ * the proof; verify.sh additionally reads the key back from AWS.
+ *
+ * covers: AWS::Lambda::EventSourceMapping
+ * covers: AWS::Lambda::Function
+ * covers: AWS::SecretsManager::Secret
+ * covers: AWS::IAM::Role
+ *
+ * The broker endpoint is deliberately a NON-EXISTENT host: the
+ * CreateEventSourceMapping API validates the request SHAPE, not connectivity,
+ * and the mapping is created `enabled: false` so Lambda never attempts to poll
+ * it. That keeps the fixture free of an actual Kafka cluster (and its cost)
+ * while still exercising the exact wire path the bug broke.
+ *
+ * Self-managed Kafka requires a `SourceAccessConfigurations` auth entry, hence
+ * the SASL/SCRAM secret. It holds placeholder credentials — nothing ever
+ * authenticates with them.
+ *
+ * UPDATE phase (CDKD_TEST_UPDATE=true) bumps `batchSize`, which routes through
+ * `UpdateEventSourceMapping` against the SAME mapping (`SelfManagedEventSource`
+ * is create-only and is deliberately held constant, so the UUID must not
+ * change).
+ */
+export class LambdaEsmSelfManagedKafkaStack extends cdk.Stack {
+  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
+    super(scope, id, props);
+
+    const isUpdate = process.env.CDKD_TEST_UPDATE === 'true';
+
+    const secret = new secretsmanager.Secret(this, 'KafkaAuth', {
+      secretName: `${this.stackName}-kafka-auth`,
+      description: 'cdkd integ placeholder SASL/SCRAM credentials (never used)',
+      secretObjectValue: {
+        username: cdk.SecretValue.unsafePlainText('cdkd-integ'),
+        password: cdk.SecretValue.unsafePlainText('cdkd-integ-placeholder'),
+      },
+      removalPolicy: cdk.RemovalPolicy.DESTROY,
+    });
+
+    const fn = new lambda.Function(this, 'Fn', {
+      functionName: `${this.stackName}-fn`,
+      runtime: lambda.Runtime.NODEJS_20_X,
+      handler: 'index.handler',
+      code: lambda.Code.fromInline('exports.handler=async()=>({});'),
+    });
+
+    fn.addEventSource(
+      new SelfManagedKafkaEventSource({
+        bootstrapServers: ['b-1.cdkd-integ.example.com:9092', 'b-2.cdkd-integ.example.com:9092'],
+        topic: 'cdkd-integ-topic',
+        secret,
+        authenticationMethod: AuthenticationMethod.SASL_SCRAM_512_AUTH,
+        startingPosition: lambda.StartingPosition.TRIM_HORIZON,
+        batchSize: isUpdate ? 20 : 10,
+        // Never poll the (non-existent) brokers.
+        enabled: false,
+      })
+    );
+  }
+}

tests/integration/lambda-esm-self-managed-kafka/package.json

@@ -0,0 +1,20 @@
+{
+  "name": "cdkd-integ-lambda-esm-self-managed-kafka",
+  "version": "1.0.0",
+  "private": true,
+  "description": "cdkd Lambda self-managed Kafka ESM Endpoints key integ probe",
+  "scripts": {
+    "build": "tsc",
+    "watch": "tsc -w"
+  },
+  "devDependencies": {
+    "@types/node": "^20.0.0",
+    "aws-cdk": "^2.1112.0",
+    "typescript": "^5.0.0"
+  },
+  "dependencies": {
+    "aws-cdk-lib": "^2.169.0",
+    "constructs": "^10.0.0"
+  },
+  "type": "module"
+}