<a id="lambda-sink-connector"></a>

# AWS Lambda Sink Connector for Confluent Platform

The Kafka Connect AWS Lambda Sink connector pulls records from one or more
Apache Kafka® topics, converts them to JSON, and executes an AWS Lambda function. The
response of the AWS Lambda can optionally be written to another Kafka topic.

#### IMPORTANT
- The AWS Lambda Sink connector is designed to access the general AWS Lambda
  endpoints. The connector doesn’t support gateway endpoints.
- <!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

  Effective July 6, 2025, only self-managed connector versions that meet or exceed the minimum version listed on the
  [Supported Connector Versions](https://docs.confluent.io/platform/7.8/connect/supported-connector-version.html#supported-connector-versions-till-cp-7-8)
  page receive support from Confluent. Older, unsupported connector versions have been removed from Confluent Marketplace and
  are no longer available for download.

The AWS Lambda function can be invoked either synchronously or asynchronously.

In synchronous mode, records within a topic and partition are processed
sequentially. Records within different topic partitions, though, can be
processed in parallel. If configured, the response from AWS Lambda can be written
to a Kafka topic. In case of errors during Lambda execution, the connector can be
configured to either ignore and proceed, log the error, or stop the connector
completely.

In asynchronous mode, the connector operates in a fire-and-forget mode.
Records are processed on a best-effort, sequential basis. The connector does
not attempt any retries. AWS Lambda automatically retries  up to two times, after
which AWS Lambda can move the request to a [dead letter queue](https://docs.aws.amazon.com/lambda/latest/dg/dlq.html).

The connector guarantees at-least-once processing semantics. Under certain
circumstances,  it is possible that a record is processed more than once. You
should [design your |aws| Lambda function to be idempotent](https://aws.amazon.com/premiumsupport/knowledge-center/lambda-function-idempotent/).
If you have configured the connector to log the response from AWS Lambda to a
Kafka topic, the topic can contain duplicate records.  You can enable Kafka log
compaction on the topic to remove duplicate records.  Alternatively, you can
write a ksqlDB query to detect duplicate records in a time window.

## Features

The AWS Lambda Sink connector includes the following features:

- [At least once delivery](#lambda-sink-at-least-once-delivery)
- [Dead Letter Queue](#lambda-sink-dead-letter-queue)
- [Multiple tasks](#lambda-sink-multiple-tasks)
- [Input data formats](#lambda-sink-data-formats)
- [AWS Lambda function versions and aliases](#lambda-sink-alias-support)
- [Client-side encryption](#lambda-sink-csfle-sm)

<a id="lambda-sink-at-least-once-delivery"></a>

### At least once delivery

This connector guarantees that records are delivered at least once from the Kafka
topic.

<a id="lambda-sink-dead-letter-queue"></a>

### Dead Letter Queue

This connector supports the Dead Letter Queue (DLQ) functionality. If an `InvalidRequestContentException` occurs, the entire batch containing the failed record is sent to the DLQ.

For information about accessing and using the DLQ, see [Confluent Platform Dead Letter Queue](/platform/current/connect/concepts.html#dead-letter-queue).

<a id="lambda-sink-multiple-tasks"></a>

### Multiple tasks

The AWS Lambda Sink connector supports running one or more tasks. You can
specify the number of tasks in the `tasks.max` configuration parameter. This
can lead to performance gains when multiple files need to be parsed.

<a id="lambda-sink-data-formats"></a>

### Input data formats

The AWS AWS Lambda Sink connector for Confluent Platform supports the following input data
formats:

- Avro
- JSON Schema (JSON_SR)
- Protobuf
- JSON (schemaless)

Note that you must enable Schema Registry to use a Schema Registry-based format (for example, Avro,
JSON Schema, or Protobuf). If no schema is defined, values are encoded as plain
strings. For example, `"name": "Kimberley Human"` is encoded as
`name=Kimberley Human`.

<a id="lambda-sink-alias-support"></a>

### AWS Lambda function versions and aliases

The connector supports invoking specific AWS Lambda function versions or aliases by appending a colon and the desired version or alias to the function name (for example, `function:1` for a version or `function:alias` for an alias).

<a id="lambda-sink-csfle-sm"></a>

### Client-side encryption

This connector supports Client-Side Field Level Encryption (CSFLE) and Client-Side Payload Encryption (CSPE). For more information, see [Manage Client-Side Encryption](https://docs.confluent.io/platform/current/connect/manage-csfle.html).

## License

You can use this connector for a 30-day trial period without a license key.

After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information.

For license properties, see [Confluent Platform license](lambda_sink_connector_config.md#aws-lambda-sink-connector-license-config). For
information about the license topic, see
[License topic configuration](lambda_sink_connector_config.md#lambda-sink-license-topic-configuration).

## Linux on IBM Z (s390x) support

Starting with Confluent Platform 8.2, this connector supports Linux on IBM Z (s390x).
The connector supports the same capability available on x86_64 unless
otherwise noted. For more information, see [Linux on IBM Z (s390x) support](https://docs.confluent.io/platform/current/installation/versions-interoperability.html#linux-on-ibm-z-s390x-support).

## Configuration properties

For a complete list of configuration properties for this connector, see
[Configuration Reference for AWS Lambda Sink Connector for Confluent Platform](lambda_sink_connector_config.md#lambda-sink-connector-config).

For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see
[Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster).

<a id="lambda-sink-migration"></a>

## Upgrading to Version 3.x

Starting with version 3.x, the AWS Lambda Sink connector uses the AWS SDK v2, upgrading from v1.
This upgrade does not allow backward compatibility with versions 1.x and 2.x. If you use any custom or
built-in AWS credentials providers, you must update your implementation to ensure compatibility
with the new AWS SDK v2.

Follow the steps below based on the type of credentials provider your implementation uses.

### Custom credentials provider

- Update [credentials provider](#lambda-sink-aws-creds-export) interface: If you have implemented a
  custom credentials provider, you must update your code to implement the new AWS SDK v2 interface.

  Old interface: [com.amazon.auth.AWSCredentialsProvider](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/AWSCredentialsProvider.html?session_ref=direct)

  New interface: [software.amazon.awssdk.auth.credentials.AwsCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/AwsCredentialsProvider.html)
- Set a new instantiation method: The connector no longer uses a default no-args constructor to create
  an instance of your custom provider. Instead, your credentials provider class must now implement
  a static, no-argument `create()` method that returns a new instance of your provider.

### AWS built-in credentials provider

- Use v2 equivalent: If you use a credentials provider built into the AWS SDK itself, you must ensure you use the v2
  equivalent of that provider.
- Use `create()` method: This provider must implement a static, no-argument `create()`
  method that returns a new provider instance.

### Default credentials provider

- Review changes to the [default chain](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/credentials-chain.html):
  The default credentials provider chain in the AWS SDK v2 has changed the order in which
  it searches for credentials. The SDK v2 version now checks for system properties before checking
  for environment variables. This change may affect your connector’s authentication if you
  previously relied on environment variables and also have system properties set.

For more information about credentials provider migration, see
[Credentials provider changes for AWS SDK v2](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-client-credentials.html).

## Install the AWS Lambda Sink connector

You can install this connector by using the [confluent connect
plugin
install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html)
command, or by manually downloading the ZIP file.

### Prerequisites

- You must install the connector on every machine where Connect will run.
- Kafka Broker: Confluent Platform 3.3.0 or later, or Kafka 0.11.0 or later.
- Connect: Confluent Platform 4.0.0 or later, or Kafka 1.0.0 or later.
- Java 1.8.
- AWS credentials (For more details, see [Access Key ID and Secret Access Key](https://docs.aws.amazon.com/general/latest/gr/aws-sec-cred-types.html#access-keys-and-secret-access-keys)).
- An installation of the latest (`latest`) connector version.

### Install the connector using Confluent CLI

To install the `latest` connector version, navigate to your Confluent Platform installation
directory and run the following command:

```bash
confluent connect plugin install confluentinc/kafka-connect-aws-lambda:latest
```

You can install a specific version by replacing `latest` with a version number
as shown in the following example:

```bash
confluent connect plugin install confluentinc/kafka-connect-aws-lambda:3.0.1
```

<a id="lambda-sink-aws-creds-export"></a>

### Exporting AWS credentials and region

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 for the following:

- `lambda:InvokeFunction` and `lambda:GetFunction`.
- Add resource to allow invoking all aliases and versions of the function, including `$LATEST`. When you specify function name without a version or alias suffix, all underlying versions, aliases, and `$LATEST` are implicitly included and accessible.

The following shows a JSON example for setting this policy:

```bash
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "lambda:InvokeFunction",
                "lambda:GetFunction"
            ],
            "Resource": [
              "arn:aws:lambda:*:*:function:<function-name>"
            ]
        }
    ]
}
```

#### NOTE
If you want to restrict the connector to a particular alias or version, update the permission with alias or versions appended at the end as show below:

```none
arn:aws:lambda:*:*:function:functionName:alias
OR
arn:aws:lambda:*:*:function:functionName:1
```

Exporting environment variables is sufficient for a development and testing
environment. However, in a production environment, you should provide
credentials as part of the worker process itself using the configuration
property `aws.credentials.provider.class`. This is the credentials provider or
provider chain to use for authentication to AWS. By default, the connector
uses `DefaultCredentialsProvider`. For details on configuring a
credentials provider, see [AWS Credentials](https://docs.confluent.io/kafka-connect-s3-sink/current/#aws-credentials).
The information provided in [AWS Credentials](https://docs.confluent.io/kafka-connect-s3-sink/current/#aws-credentials) is
applicable for most connectors accessing resources in Amazon Web Services, including the
AWS Lambda Sink connector.

Export the following AWS environment variables to allow the connector to
access AWS Lambda. Note that these environment variables must be exported where
the Kafka Connect worker processes and the connector are deployed.

* `AWS_DEFAULT_REGION`

  To export, run the following command:
  ```bash
  export AWS_DEFAULT_REGION=<your-aws-lambda-region>
  ```
* `AWS_ACCESS_KEY_ID`

  To export, run the following command:
  ```bash
  export AWS_ACCESS_KEY_ID=<your-accesskey-id>
  ```
* `AWS_SECRET_ACCESS_KEY`

  To export, run the following command:
  ```bash
  export AWS_SECRET_ACCESS_KEY=<your-secret-access-key>
  ```

You can also set the region and the credentials using the `aws.lambda.region`,
`aws.access.key.id`, and `aws.secret.access.key` configurations. If you add
these, they are used by the connector. Additional credentials coming from the
`aws.credentials.provider.class` configuration are ignored.

### Using trusted account credentials

This connector can assume a role and use credentials from a separate trusted
account. This is a default feature provided with recent versions of this
connector that include an updated version of the AWS SDK.

#### IMPORTANT
You cannot use assumed role credentials to access AWS through a proxy
server without first passing [environment variables](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-proxy.html)
or [system properties](https://confluence.atlassian.com/kb/how-to-configure-outbound-http-and-https-proxy-for-your-atlassian-application-834000120.html).
This is due to an AWS SDK [limitation](https://github.com/aws/aws-sdk-java/issues/2558).

After you create the trust relationship, an IAM user or an application from the
trusted account can use the [AWS Security Token Service (AWS STS)](https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRole.html)
`AssumeRole` API operation. This operation provides temporary security
credentials that enable access to AWS resources for the connector. For
details, see [Creating a Role to Delegate Permissions to an IAM User](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user.html).

**Example**:

```bash
Profile in ~/.aws/credentials:

[default]
role_arn=arn:aws:iam::037803949979:role/kinesis_cross_account_role
source_profile=staging
role_session_name = OPTIONAL_SESSION_NAME

[staging]
 aws_access_key_id = <STAGING KEY>
 aws_secret_access_key = <STAGING SECRET>
```

To allow the connector to assume a role with the right permissions, set the
[Amazon Resource Name (ARN)](https://docs.aws.amazon.com/credref/latest/refdocs/setting-global-role_arn.html)
for the role. Additionally, you must use `source_profile` as the way to get
credentials that have permission to assume the role in the environment where the
connector is running.

Note that when setting up trusted account credentials, loading profiles from
both `~/.aws/credentials` and `~/.aws/config` does not work when configuring
this connector. Assumed role settings and credentials must be placed in the
`~/.aws/credentials` file.

## Install the connector manually

[Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-aws-lambda) for your
connector and then follow the manual connector installation
[instructions](/kafka-connectors/self-managed/userguide.html#connect-installing-plugins).

### Property-based example

This configuration is typically used with [standalone
workers](/platform/current/connect/concepts.html#standalone-workers).

```properties
 name=LambdaSinkConnector
 connector.class=io.confluent.connect.aws.lambda.AwsLambdaSinkConnector
 tasks.max=1

 topics=<Required Configuration>

 aws.lambda.function.name=<Required Configuration>
 aws.lambda.invocation.type=sync
 aws.lambda.batch.size=50

 behavior.on.error=fail

 confluent.topic.bootstrap.servers=localhost:9092
 confluent.topic.replication.factor=1
```

For details about using this connector with Kafka Connect Reporter, see
[Connect Reporter](/kafka-connectors/self-managed/userguide.html#userguide-connect-reporter).

### REST-based example

This configuration is typically used with [distributed
workers](/platform/current/connect/concepts.html#distributed-workers). Write the following JSON to
`connector.json`, configure all of the required values. Use the command below
to post the configuration to one of the distributed Kafka Connect worker(s).
See Kafka Connect [REST API](/platform/current/connect/references/restapi.html) for
more information.

```bash
 {
   "name": "LambdaSinkConnector",
   "config" : {
     "connector.class" : "io.confluent.connect.aws.lambda.AwsLambdaSinkConnector",
     "tasks.max" : "1",

     "topics" : "< Required Configuration >",

     "aws.lambda.function.name" : "< Required Configuration >",
     "aws.lambda.invocation.type" : "sync",
     "aws.lambda.batch.size" : "50",

     "behavior.on.error" : "fail",

     "confluent.topic.bootstrap.servers" : "localhost:9092",
     "confluent.topic.replication.factor" : "1"
   }
 }
```

Use curl to post the configuration to one of the Kafka Connect workers. Change
`http://localhost:8083/` the endpoint of one of your Kafka Connect workers.

```bash
curl -s -X POST -H 'Content-Type: application/json' --data @connector.json http://localhost:8083/connectors
```

For details about using this connector with Kafka Connect Reporter, see
[Connect Reporter](/kafka-connectors/self-managed/userguide.html#userguide-connect-reporter).

## AWS Lambda payload

The default payload converter converts Kafka records to payload in the form of
JSON Array, following is a sample payload:

```bash
[
    {
        "payload": {
            "topic": "mytopic",
            "partition": 1,
            "offset": 101,
            "key": "key_from_sink_record",
            "value": "value_from_sink_record",
            "headers": [{"key":"key1","value":"value1"},{"key":"key2","value":"value2"},{...},...],
            "timestamp": 1562844607000
        }
    },
    {
        "payload": {
            "topic": "mytopic",
            "partition": 1,
            "offset": 102,
            "key": "key_from_sink_record",
            "value": "value_from_sink_record",
            "headers": [{...},{...},...],
            "timestamp": 1562844608000
        }
    }
]
```

The `key` and `value` are converted to either JSON primitives or objects
according to their schema. If no schema is defined, they are encoded as plain
strings.

For any AWS Lambda invocation, all the records belong to the same topic and
partition, and the offset will be in a strictly increasing order.

When the connector is configured, it validates if the Lambda function exists,
and if the AWS credentials used can invoke the Lambda function.

## Batching records

The AWS Lambda Sink connector combines multiple records into the input payload for
the AWS 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](https://docs.aws.amazon.com/lambda/latest/dg/limits.html).
- To disable batching, set `aws.lambda.batch.size` to 1.

## Response topic

In `sync` mode, the connector can optionally log the response from AWS Lambda in
a Kafka topic using Kafka Connect Reporter.

The connector attempts to map each response to a single record before producing
it to the corresponding topic. It can receive the responses from the AWS
AWS Lambda function in the following three formats.

1. The first format is JSON:
   ```bash
   [
     {
       "payload": {
         "result": ...,
         "topic": string,
         "partition": <number>,
         "offset": <number>,
       }
     },
     ...
   ]
   ```

   This list can be out of order relative to the order that the records were
   provided. The connector will correctly match the record to the result based
   off its Kafka coordinates. However the list must be one-to-one to the list of
   records that were sent in the request.
2. The second format is a JSON list:
   ```bash
   [
     ...,
     ...,
     ...
   ]
   ```

   As long as the list is one-to-one to the list of records, the list will be
   assumed to be ordered and matched with the corresponding records.
3. The third format can be any format that does not satisfy either of the above
   formats. The connector will report the entire response for each individual
   record (one-to-many correlation).

The connector logs the output of each record from the AWS Lambda function response
in the configured Kafka topic as JSON. The following is an example of a logged
JSON response from a AWS Lambda function:

```bash
[
    {
        "topic": "lambda-success",
        "partition": 1,
        "offset": 101,
        "timestamp": 1562844608000,
        "headers": [{...},{...},...],
        "key": "sink_record_key",
        "value": {
            "body": "Successfully executed Lambda!",
            "status_code": 200
        }
    },
    ...
]
```

AWS Lambda function:

```bash
def lambda_handler(event, context):
    result_list = []
    for obj in event:
        result = {}
        result["body"] = "Successfully executed Lambda!";
        result["status_code"] = 200;
        result_list.append(result)
    return result_list
```

To enable a response topic, set `reporter.result.topic.name` to the topic
where you want to log the responses. The following shows a sample set of
properties to add to configure a response topic:

```properties
aws.lambda.invocation.type=sync

reporter.result.topic.name=<Required Configuration>
reporter.bootstrap.servers=<Required Configuration>
```

Note that in `async` mode, the connector can’t log the response.

For more information about using this connector with Kafka Connect Reporter,
see [Connect Reporter](/kafka-connectors/self-managed/userguide.html#userguide-connect-reporter).

## Error handling

The AWS Lambda Sink connector may encounter the following types of errors:

### Transient errors

Transient errors such as network timeouts or errors because of rate limiting.
The connector handles transient errors through multiple layers:

- **AWS SDK automatic retries**: The connector relies on the AWS Lambda SDK to automatically
  handle transient errors such as network timeouts or server errors using the SDK’s built-in
  retry mechanisms.
- **Connector-level retries**: The connector additionally handles specific AWS Lambda service
  limits (`Ec2ThrottledException`, `EniLimitReachedException`) by marking them as retriable,
  allowing Kafka Connect to retry the entire batch. In asynchronous mode, such retries are
  not applicable as the connector uses fire-and-forget invocation.
- **Socket timeout configuration**: Configure `aws.lambda.socket.timeout` (default: 50 seconds,
  range: 1-600 seconds) to control how long the client waits for AWS Lambda responses, preventing hangs
  on network issues. This setting is not applicable in asynchronous mode.

### Configuration errors

Configuration errors such as an incorrect AWS Lambda function names, access-related issues, or
invalid credentials are handled as follows:

- **Startup validation**: During connector startup, the connector validates function existence,
  permissions, and credentials. If validation fails, the connector fails to start.
- **Runtime configuration errors**: If configuration-related errors occur during record
  processing (for example, function suddenly becomes unavailable, permission changes), the connector
  does not retry these errors. Instead:
  - **Most configuration errors**: The connector throws a `ConnectException` and stops the task.
  - **Invalid content errors** (specifically `InvalidRequestContentException`): You can optionally
    send records to a dead letter queue (DLQ, if configured) instead of stopping the connector.
- **No automatic recovery**: Unlike transient errors, configuration errors require manual
  intervention to fix the underlying configuration issue before restarting the connector.

### AWS Lambda function execution errors

Errors encountered during execution of the AWS Lambda function are classified as handled or
unhandled, with different handling behaviors depending on the invocation mode.

- **Handled errors**: Exceptions intentionally thrown by your application or business logic
  (for example, via the `throw` in a try-catch block). Best practice for sync mode as follows:
  - Throw exceptions on failure (avoid returning an error string).
  - The connector treats thrown exceptions as failures, using `behavior.on.error` to decide
    whether to `fail`, `log`, or `ignore`.
- **Unhandled errors**: Errors from AWS Lambda itself (for example, timeouts, throttling,
  runtime errors).
  - In **sync** mode, the connector interprets similarly to handled errors, and
    processed according to `behavior.on.error`.
  - In **async** mode, AWS Lambda performs internal retries before dropping the request.

#### AWS Lambda function execution error scenarios

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

| aws.lambda.invocation.type   | behavior.on.error   | Error Handling                                                                                                                                                                                                                                                                                                                                       |
|------------------------------|---------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| async                        |                     | The connector relies on AWS Lambda to perform retries<br/>and error handling. AWS Lambda retries the function twice (a total of three<br/>attempts), after which it discards the event. You should configure a<br/>[dead letter queue](https://docs.aws.amazon.com/lambda/latest/dg/dlq.html) if you want to<br/>track the input events that failed. |
| sync                         | fail                | `fail` is the default mode. The connector stops processing records for<br/>that TopicPartition. Records for other TopicPartitions will continue to<br/>process.                                                                                                                                                                                      |
| sync                         | log                 | The connector logs the error message and continues to process the next<br/>batch of records. For help on how to configure the Kafka Connect<br/>Reporter to report errors in a separate topic, see<br/>[Connect<br/>Reporter](/kafka-connectors/self-managed/userguide.html#userguide-connect-reporter).                                             |
| sync                         | ignore              | The connector continues to process next set of records without logging<br/>the error message 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:

```properties
aws.lambda.invocation.type=sync
behavior.on.error=log

reporter.error.topic.name=<Required Configuration>
reporter.bootstrap.servers=<Required Configuration>
```

For details about using Kafka Connect Reporter, see [Connect
Reporter](/kafka-connectors/self-managed/userguide.html#userguide-connect-reporter).

## RecordConverter Interface

The AWS Lambda Sink connector can use a custom implementation of the
RecordConverter interface to convert a batch of records to string payload. The
connector sends the string payload to AWS AWS Lambda in a single invocation.
The following interface can be implemented and the corresponding class is
specified using the `record.converter.class` connector configuration property.

```properties
import java.util.List;

import org.apache.kafka.connect.sink.SinkRecord;

public interface RecordConverter {
    /**
    * Converts a Batch of SinkRecords to string payload sent to AWS lambda in single invocation.
    * The list of SinkRecords is converted into one single string payload. This payload is
    * sent while invoking AWS Lambda.
    * @param records : List of Sink Records.
    * @return payload: Payload string sent to Lambda
    */
    String convertRecordsToPayloadString(List<SinkRecord> records);
}
```