feat: add a failure destination to the Kafka configuration

Author: Teun Willems

lib/plugins/aws/package/compile/events/kafka.js

@@ -94,6 +94,9 @@ class AwsCompileKafkaEvents {
         topic: {
           type: 'string',
         },
+        onFailureDestination: {
+          type: 'string',
+        },
         consumerGroupId: {
           type: 'string',
           maxLength: 200,
@@ -286,6 +289,15 @@ class AwsCompileKafkaEvents {
           };
         }
 
+        const onFailureDestination = event.kafka.onFailureDestination;
+        if (onFailureDestination) {
+          kafkaResource.Properties.DestinationConfig = {
+            OnFailure: {
+              Destination: onFailureDestination,
+            },
+          };
+        }
+
         const provisionedPollerConfig = event.kafka.provisionedPollerConfig;
         if (provisionedPollerConfig) {
           kafkaResource.Properties.ProvisionedPollerConfig = {

lib/plugins/aws/package/compile/events/msk/index.js

@@ -44,6 +44,9 @@ class AwsCompileMSKEvents {
         startingPositionTimestamp: {
           type: 'number',
         },
+        onFailureDestination: {
+          type: 'string',
+        },
         topic: {
           type: 'string',
         },

package.json

@@ -113,6 +113,7 @@
     "mocha": "^9.2.2",
     "mock-require": "^3.0.3",
     "ncjsm": "^4.3.2",
+    "nyc": "^17.1.0",
     "pkg": "^5.8.1",
     "prettier": "^2.8.8",
     "proxyquire": "^2.1.3",

test/unit/lib/plugins/aws/package/compile/events/kafka.test.js

@@ -550,6 +550,36 @@ describe('test/unit/lib/plugins/aws/package/compile/events/kafka.test.js', () =>
       });
     });
 
+    describe('onFailureDestination', () => {
+      it('should correctly compile EventSourceMapping resource properties for onFailureDestination', async () => {
+        const { awsNaming, cfTemplate } = await runServerless({
+          fixture: 'function',
+          configExt: {
+            functions: {
+              basic: {
+                role: { 'Fn::ImportValue': 'MyImportedRole' },
+                events: [
+                  {
+                    kafka: {
+                      topic,
+                      bootstrapServers: ['abc.xyz:9092'],
+                      accessConfigurations: { saslScram256Auth: saslScram256AuthArn },
+                      onFailureDestination: 'arn:aws:sqs:eu-central-1:000000000000:some-queue'
+                    },
+                  },
+                ],
+              },
+            },
+          },
+          command: 'package',
+        });
+
+        const eventSourceMappingResource =
+          cfTemplate.Resources[awsNaming.getKafkaEventLogicalId('basic', 'TestingTopic')];
+        expect(eventSourceMappingResource.Properties.DestinationConfig.OnFailure.Destination).to.equal('arn:aws:sqs:eu-central-1:000000000000:some-queue');
+      });
+    });
+
     describe('startingPositionTimestamp', () => {
       it('should fail to compile EventSourceMapping resource properties for startingPosition AT_TIMESTAMP with no startingPositionTimestamp', async () => {
         await expect(

types/index.d.ts

@@ -384,6 +384,7 @@ export interface AWS {
             topic: string;
             consumerGroupId?: string;
             filterPatterns?: FilterPatterns;
+            onFailureDestination?: string;
           };
         }
         | {