AWS Lambda Sink Connector

🧠 Free Recall ❓ Quiz 🔍 Cues 🔮 Simulation 🥷 ShadowBox 📐 Encode Cards

Goals, metrics, and success criteria
Goals
Enable Confluent customers to execute a Lambda function with Kafka records as a payload.
Most of the internal AWS cloud infrastructure uses JSON payloads. 
Find any gaps while Confluent team goes through the connector development process, and update the process.
Success Criteria

Customers are able to execute a Lambda function with Kafka records as a payload and use it in production.
DevX can successfully use this connector to build a demo
Assumptions
Users are familiar with AWS Lambda
Some part of S3 Sink Connector can be reused. (i.e. authentication)
We can use the following open source connector as a starting point for AWS Lambda Sink Connector.
https://github.com/llofberg/kafka-connect-aws-lambda
Requirements
We should offer the same (close to same) experience across AWS related connectors.
Upon startup and prior to consuming any messages, the connector will check if the configured lambda function is available; if it is not available an error will be logged. 
The connector will invoke the configured function with a configurable batch size of ‘n‘ ‘maximum events’ (1-->n). The event payload will be batch up until the ‘maximum events’ are met, or, until approaching the relevant sync or async size limitations. The batch size is used to tune the per invocation payload size so that users can control function duration and also control return volumes.
This connector will provide an option for the user to determine how to handle errors. We can use the same option we offer for Salesforce Sink Connectors.
"behavior.on.errors = fail, ignore, and log."
There are two modes of operation: asynchronous (i.e. fire and forget or InvocationType=Event) and synchronous (where the result is sent to the output topic). Sync/async should be implied based on whether the user specifies an output topic configuration. (Note that there is no non-blocking API for Lambda so Tasks will have to used to provide parallelism, so each task will block during function execution.)
Store the output of synchronously invoked functions to the configured output topic.
The Kafka Producer should only be initialized when an output topic is configured. (This is important because a connector configured in async mode should not have to provide all the configs required to create a producer.) 
If a synchronous request fails with error=”EC2ThrottledException, ENILimitReachedException” then a RetryException should be thrown up to the Connect framework to control behavior and obey the behavior.on.errors configuration.
Retry periodicity should be configurable via SinkTaskContext.timeout(long).
Results from synchronous calls will be logged when ‘trace’ level logging is configured.

Configure an AWS Lambda Sink Connector
Developer: I can configure an AWS Lambda Sink Connector with my AWS account.

Connection to Amazon
aws.lambda.credentials.provider.class=
Credentials provider or provider chain to use for authentication to AWS. By default the connector uses DefaultAWSCredentialsProviderChain.
Type: class
Default: com.amazonaws.auth.DefaultAWSCredentialsProviderChain
Valid Values: Any class implementing: interface com.amazonaws.auth.AWSCredentialsProvider
Importance: low
aws.lambda.region=(similar to S3 region)
Specify a Lambda function
aws.lambda.function.name=(e.g. kafka-aws-lambda)
 Proxy
aws.lambda.proxy.url
aws.lambda.proxy.user
aws.lambda.proxy.password
2
Specify a Kafka topic or Kafka topics
Developer: I'd like to specify a Kafka topic or Kafka topics I need to use as a payload for a Lambda function.
The following connect framework properties should be supported.
topics=(comma separated topics)
topics.regex=(Java regex)
aws.lambda.response.topic=
Store the output of synchronously invoked functions to a configured output topic (batched or not).
3
Documentation
Developer: I'm able to follow Confluent documents to read data from a Kafka topic or Kafka topics as a payload and execute a Lambda function with the payload. If something goes wrong, I'm able to debug the issue based on Confluent documents.
Testing
Follow the testing guideline described in the Connector Development Process doc.
Given no function is configured, when the connector is run and an event passed to the input topic, then the appropriate error should be raised. 
Given a function. When an event is sent to the connector with a payload larger than the batch size. Then an error should be logged. 
Given a function is configured but does not exist in AWS, when the connector is run and an event passed to the input topic, then the appropriate error should be raised. 
Given a function is configured but not deployed and the connector is already running. When a function is deployed and an event is sent to the system. Then the function should be run.
Given a function is deployed but not configured and the connector is already running. When a function is configured and an event is sent to the system. Then the function should be run.
Given the input topic does not exist. When the connector is run. Then the appropriate error message should be logged. 
Function failure (code error) and logged error message
Given a function that throws an exception. When the connector is run in sync mode and an event passed. Then the appropriate error message should be logged in Connect. 
Given a function that runs beyond the lambda configured timeout. When the connector is run in sync mode and an event passed. Then the appropriate error message should be logged in Connect. 
Given a function that runs for 15mins and a connect cluster with at least 2 tasks. When the connector is run in sync mode and an event passed. Then connect framework should not attempt to rebalance or log other exceptions. 
Given a connector with standard configs. When messages are sent at a rate above the configured AWS throughput limits (i.e. is throttled). Then the functions should all complete eventually with no message loss.

Expected Behavior
When no function is configured an error should be raised
When a event with payload larger than batch size, error should be logged
When function is set up but not deployed. connector will continue to run

References
S3 Sink Connector Source Code
Kafka Connect Storage Cloud Source Code
AWS Lambda Sink Source Code (Community Connector)
AWS Lambda Connector by Workato

At least once delivery
This connector guarantees that records are delivered at least once to the Kafka topic. If the connector restarts, there may be some duplicate records in the Kafka topic.

Before you can run this connector, you must provide credentials and the region where the AWS Lambda function is located.
The credentials provided need to have permissions to the actions lambda:InvokeFunction and lambda:GetFunction. An example of how this policy may be set up is shown below:

Batching Records
The AWS Lambda sink connector combines multiple records into the input payload for the Lambda Function invocation. The following rules apply:
A batch of records will belong to the same topic and partition.
A batch always has records in increasing order of the offset.
Total number of records in a batch is less than or equal to the configuration aws.lambda.batch.size and the size of the batch is less than the AWS Lambda Payload Limits.
To disable batching, set aws.lambda.batch.size to 1.

Error Handling
The AWS Lambda sink connector may encounter the following types of errors:
Transient errors such as network timeouts or errors because of rate limiting.
Configuration errors such as an incorrect Lambda Function name or access-related issues.
Errors encountered during execution of the Lambda Function. These errors are further classified as handled or unhandled.

The AWS Lambda sink connector automatically handles transient errors such as network timeouts. The connector relies on the AWS Lambda SDK to perform retries.

In case of configuration errors, the connector does not retry. Instead, it throws a ConnectException and stops the task.

In case of errors encountered during execution of the Lambda Function, the behavior of the connector depends on the configuration parameters aws.lambda.invocation.type and behavior.on.error.

In asynchronous mode, the connector relies on AWS Lambda to perform retries and error handling. AWS Lambda retries the function twice (a total of three attempts), after which it discards the event. You should configure a dead letter queue if you want to track the input events that failed.

fail is the default mode. The connector stops processing records for that TopicPartition. Records for other TopicPartitions will continue to process.

The connector will log the error message and continue processing the next batch of records. See Connect Reporter for how to configure the Kafka Connect Reporter to report errors in a separate topic.

Logging Errors to Kafka topic
The connector can optionally log errors to a Kafka topic using Kafka Connect Reporter. Use the following configuration settings to enable error logging:

Native Lambda with AWS

Apache Kafka as an event source operates similarly to using Amazon Simple Queue Service (Amazon SQS) or Amazon Kinesis. Lambda internally polls for new messages from the event source and then synchronously invokes the target Lambda function. Lambda reads the messages in batches and provides these to your function as an event payload. The maximum batch size is configurable. (The default is 100 messages.)

Lambda sends the batch of messages in the event parameter when it invokes your Lambda function. The event payload contains an array of messages. Each array item contains details of the Kafka topic and Kafka partition identifier, together with a timestamp and a base64-encoded message.

For Lambda to poll your Apache Kafka topic and update other cluster resources, your Lambda function—as well as your AWS Identity and Access Management (IAM) users and roles—must have the following permissions.

Required Lambda function permissions
To create and store logs to a log group in Amazon CloudWatch Logs, your Lambda function must have the following permissions in its execution role:

logs:CreateLogGroup

logs:CreateLogStream

logs:PutLogEvents

Secrets Manager and AWS KMS permissions
If your Apache Kafka users access your Kafka brokers over the internet, you must specify a Secrets Manager secret. Your Lambda function might need permission to describe your Secrets Manager secret or to decrypt your AWS KMS customer managed CMK. To access these resources, your function's execution role must have the following permissions:
secretsmanager:GetSecretValue
kms:Decrypt
VPC permissions
If only users within a VPC can access your self-managed Apache Kafka cluster, your Lambda function must have permission to access your Amazon Virtual Private Cloud (Amazon VPC) resources. These resources include your VPC, subnets, security groups, and network interfaces. To access these resources, your function's execution role must have the following permissions:

Unstarred

Dunk (exclude from study)

AWS Lambda Sink Connector

Review Questions

Todos

Sensemaker