Merge pull request #29 from Demetrios-loutsios/feat-kafka-event-source-support-provisioned-poller-config

Add ProvisionedPollerConfig to Kafka event source

Author: mnapoli

docs/events/kafka.md

@@ -182,3 +182,24 @@ functions:
           maximumBatchingWindow: 30
           startingPosition: LATEST
 ```
+
+## Setting ProvisionedPollerConfig
+
+You can set the `provisionedPollerConfig` to configure your Kafka event source to be in provisioned mode. A `minimumPollers` must be configured from 1 to 200 and a `maximumPollers` from 1 to 2000.
+
+```yml
+functions:
+  compute:
+    handler: handler.compute
+    events:
+      - kafka:
+          accessConfigurations:
+            saslScram512Auth: arn:aws:secretsmanager:us-east-1:01234567890:secret:MyBrokerSecretName
+          topic: MySelfManagedKafkaTopic
+          bootstrapServers:
+            - abc3.xyz.com:9092
+            - abc2.xyz.com:9092
+          provisionedPollerConfig:
+            minimumPollers: 1
+            maximumPollers: 10
+```

docs/guides/serverless.yml.md

@@ -1044,6 +1044,10 @@ functions:
           # Optional, specifies event pattern content filtering
           filterPatterns:
             - eventName: INSERT
+          # Optional, configures provisioned mode, must specify minimumPollers (1-200 range), minimumPollers (1-2000 range). Both or one can be specified.
+          provisionedPollerConfig:
+            minimumPollers: 1
+            maximumPollers: 10
 ```
 
 ### RabbitMQ

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

@@ -100,6 +100,22 @@ class AwsCompileKafkaEvents {
           pattern: '[a-zA-Z0-9-/*:_+=.@-]*',
         },
         filterPatterns: { $ref: '#/definitions/filterPatterns' },
+        provisionedPollerConfig: {
+          type: 'object',
+          minProperties: 1,
+          properties: {
+            minimumPollers: {
+              type: 'number',
+              minimum: 1,
+              maximum: 200,
+            },
+            maximumPollers: {
+              type: 'number',
+              minimum: 1,
+              maximum: 2000,
+            },
+          },
+        },
       },
       additionalProperties: false,
       required: ['accessConfigurations', 'bootstrapServers', 'topic'],
@@ -270,6 +286,14 @@ class AwsCompileKafkaEvents {
           };
         }
 
+        const provisionedPollerConfig = event.kafka.provisionedPollerConfig;
+        if (provisionedPollerConfig) {
+          kafkaResource.Properties.ProvisionedPollerConfig = {
+            MinimumPollers: provisionedPollerConfig.minimumPollers,
+            MaximumPollers: provisionedPollerConfig.maximumPollers,
+          };
+        }
+
         cfTemplate.Resources[kafkaEventLogicalId] = kafkaResource;
       });
 

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

@@ -21,6 +21,7 @@ describe('test/unit/lib/plugins/aws/package/compile/events/kafka.test.js', () =>
   const batchSize = 5000;
   const maximumBatchingWindow = 20;
   const filterPatterns = [{ eventName: 'INSERT' }];
+  const provisionedPollerConfig = { minimumPollers: 1, maximumPollers: 10 };
 
   describe('when there are kafka events defined', () => {
     let minimalEventSourceMappingResource;
@@ -56,6 +57,7 @@ describe('test/unit/lib/plugins/aws/package/compile/events/kafka.test.js', () =>
                     enabled,
                     startingPosition,
                     filterPatterns,
+                    provisionedPollerConfig,
                   },
                 },
               ],
@@ -136,6 +138,10 @@ describe('test/unit/lib/plugins/aws/package/compile/events/kafka.test.js', () =>
             },
           ],
         },
+        ProvisionedPollerConfig: {
+          MinimumPollers: 1,
+          MaximumPollers: 10,
+        },
       });
     });
   });
@@ -673,6 +679,38 @@ describe('test/unit/lib/plugins/aws/package/compile/events/kafka.test.js', () =>
     });
   });
 
+  it('should correctly compile EventSourceMapping resource properties for ProvisionedPollerConfig', 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 },
+                  provisionedPollerConfig: { minimumPollers: 2, maximumPollers: 10 },
+                },
+              },
+            ],
+          },
+        },
+      },
+      command: 'package',
+    });
+
+    const eventSourceMappingResource =
+      cfTemplate.Resources[awsNaming.getKafkaEventLogicalId('basic', 'TestingTopic')];
+
+    expect(eventSourceMappingResource.Properties.ProvisionedPollerConfig).to.deep.equal({
+      MinimumPollers: 2,
+      MaximumPollers: 10,
+    });
+  });
+
   describe('when no kafka events are defined', () => {
     it('should not modify the default IAM role', async () => {
       const { cfTemplate } = await runServerless({