Merge pull request #90 from teunw/feature/failure-destination-kafka

feat: add a failure destination to the Kafka configuration

Author: mnapoli

docs/events/kafka.md

@@ -153,6 +153,25 @@ functions:
             - eventName: INSERT
 ```
 
+## Failure destinations
+
+A failure destination allows you to send messages that failed processing repeatedly to be sent to an SNS Topic, SQS Queue or another lambda function.
+By default, Lambda discards oversized records (> 6MB) and records that fail all retry attempts.
+Configuring an on-failure destination ensures that you don't lose data when your Lambda function encounters errors.
+For more information, see [capturing discarded batches for a self-managed Apache Kafka event source](https://docs.aws.amazon.com/lambda/latest/dg/with-kafka-on-failure.html)
+
+```yml
+functions:
+  compute:
+    handler: handler.compute
+    events:
+      - kafka:
+          accessConfigurations:
+            saslScram512Auth: arn:aws:secretsmanager:us-east-1:01234567890:secret:MyBrokerSecretName
+          topic: MySelfManagedKafkaTopic
+          onFailureDestination: arn:aws:sqs:us-east-1:01234567890:my-queue
+```
+
 ## IAM Permissions
 
 The Serverless Framework will automatically configure the most minimal set of IAM permissions for you. However you can still add additional permissions if you need to. Read the official [AWS documentation](https://docs.aws.amazon.com/lambda/latest/dg/kafka-smaa.html) for more information about IAM Permissions for Kafka events.

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 = {

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

@@ -550,6 +550,38 @@ 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;
           };
         }
         | {