<a id="sqs-overview"></a>

# Amazon SQS Source Connector for Confluent Platform

The Kafka Connect Simple Queue Service (SQS) Source connector is used to move
messages from [AWS SQS Queues](https://aws.amazon.com/sqs/) into Apache Kafka®. It
supports both Standard and FIFO queues.

This connector polls an SQS queue, converts SQS messages into Kafka records, and
pushes the records into a Kafka topic.

Each SQS message is converted into exactly one Kafka record, with the following structure:

1. The key encodes the SQS queue name and message ID in a struct. For FIFO queues, it also includes the message group ID.
2. The value encodes the body of the SQS message and various message attributes
   in a struct. See the [Record schema](#sqs-record-schema) section.
3. Each header encodes the message attributes that may be present in SQS message.

The schema for key, value and headers is described in the
[Record schema](#sqs-record-schema) section.

For standard queues, this connector supports *best-effort* ordering guarantees.
This means that there is a chance records will end up in a different order in
Kafka. Also, the connector supports *at least once* delivery guarantees. This
means there is a chance that the connector can introduce duplicate records in
Kafka.

For FIFO queues, this connector guarantees records are inserted into Kafka in the
order they were inserted in SQS, as long as the destination Kafka topic has
exactly 1 partition. If the destination topic has more than 1 partitions, you
can write a single message transform to set the partition based on the
`MessageGroupId` field in the key.

The SQS Source connector sends delete requests once the SQS messages are
processed and stored in the Kafka topics.

#### IMPORTANT
<!-- 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.

## Features

The SQS Source connector includes the following features:

- [At least once delivery](#sqs-source-at-least-once-delivery)
- [Multiple tasks](#sqs-source-multiple-tasks)
- [Automatic retries](#sqs-source-automatic-retries)
- [Supports HTTPS proxy](#sqs-source-supports-https-proxy)
- [Long polling](#sqs-source-long-polling)
- [Fetch multiple messages](#sqs-source-fetch-multiple-messages)
- [Client-side encryption](#sqs-source-csfle-sm)

<a id="sqs-source-at-least-once-delivery"></a>

### At least once delivery

The connector guarantees that messages from SQS are delivered at least once to
the Kafka topic.

<a id="sqs-source-multiple-tasks"></a>

### Multiple tasks

The SQS Source 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="sqs-source-automatic-retries"></a>

### Automatic retries

The SQS Source connector may experience problems connecting to the SQS endpoint
due to network issues or in the rare occasion due to an AWS outage. The
connector will automatically retry polling SQS. The property `sqs.max.retries`
controls how many times retries will be attempted.

The connector internally uses AWS Java SDK. The AWS Java SDK uses an
exponential backoff strategy, such that there is an increasing delay between
subsequent retries. For more information, refer to [Error Retries and
Exponential Backoff in AWS](https://docs.aws.amazon.com/general/latest/gr/api-retries.html).

<a id="sqs-source-supports-https-proxy"></a>

### Supports HTTPS proxy

The connector can connect to SQS using an https proxy server. To configure the
proxy, you can set `sqs.proxy.url`, `sqs.proxy.user` and
`sqs.proxy.password` in the configuration file.

<a id="sqs-source-long-polling"></a>

### Long polling

This connector uses [long polling to retrieve messages](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-long-polling.html)
from SQS. By default, it will wait up to 20 seconds for a message to be
available in SQS. You can change `sqs.waittime.seconds` to a number between 1s
and 20s.

<a id="sqs-source-fetch-multiple-messages"></a>

### Fetch multiple messages

In every poll cycle, the connector fetches `sqs.messages.max` number of
messages. By default, this value is 10, which is the maximum. This setting works
well for most use cases. However, if your message size is exceptionally large,
you may want to reduce this to a lower number.

<a id="sqs-source-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).

## Install the SQS Source 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.
- Java 8+. Note that Java 8 is deprecated in versions 7.2 and later of Confluent Platform. For
  more details, view [Java compatibility with Confluent Platform](/platform/current/installation/versions-interoperability.html#java)
   by version.
- An installation of the [Confluent Hub Client](https://docs.confluent.io/home/connect/confluent-hub/client.html). Note
  that this is installed by default with Confluent Enterprise.
- An installation of the latest (`latest`) connector version.

### Install the connector using the 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-sqs: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-sqs:2.1.0
```

### Install the connector manually

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

## 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](configuration.md#sqs-source-connector-license-config). For
information about the license topic, refer to
[License topic configuration](configuration.md#sqs-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 Amazon SQS Source Connector for Confluent Platform](configuration.md#sqs-configuration-options).

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).

## Quick start

In this quick start guide, you use the SQS Source connector to export messages
from an SQS FIFO queue to a Kafka topic. Before running the quick start, ensure
the following:

- [Confluent Platform](/platform/current/installation/installing_cp/index.html) is
  installed and services are running by using the [Confluent
  CLI](https://docs.confluent.io/confluent-cli/current/index.html). This quick start assumes that you are using the
  Confluent CLI, but standalone installations are also supported. By default
  ZooKeeper, Kafka, Schema Registry, Connect, and the Connect REST API are started with the
  `confluent local start` command. For more information, see
  [Quick Start for Apache Kafka using Confluent Platform
  (Local)](/platform/current/quickstart/ce-quickstart.html). Note that as of Confluent Platform 7.5, ZooKeeper is
  deprecated for new deployments. Confluent recommends KRaft mode for new
  deployments.
- You must install the [AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-install.html) and
  configure it by running the `aws configure` command.
- Ensure the IAM user or role you configure has full access to SQS.
- Create a Kafka topic called `sqs-quickstart`.

1. Create a FIFO queue by running the following command:
   ```bash
   aws sqs create-queue --queue-name sqs-source-connector-demo
   ```

   You should see output similar to the following:
   ```bash
   {
     "QueueUrl": "https://queue.amazonaws.com/940887362971/sqs-source-connector-demo"
   }
   ```
2. Add some records to the newly created queue by first creating a file called
   `send-message-batch.json` with the following content:
   ```bash
   [
     {
         "Id":"FuelReport-0001-2015-09-16T140731Z",
         "MessageBody":"Fuel report for account 0001 on 2015-09-16 at 02:07:31 PM.",
         "DelaySeconds":10,
         "MessageAttributes":{
           "SellerName":{
               "DataType":"String",
               "StringValue":"Example Store"
           },
           "City":{
               "DataType":"String",
               "StringValue":"Any City"
           },
           "Region":{
               "DataType":"String",
               "StringValue":"WA"
           },
           "PostalCode":{
               "DataType":"String",
               "StringValue":"99065"
           },
           "PricePerGallon":{
               "DataType":"Number",
               "StringValue":"1.99"
           }
         }
     },
     {
         "Id":"FuelReport-0002-2015-09-16T140930Z",
         "MessageBody":"Fuel report for account 0002 on 2015-09-16 at 02:09:30 PM.",
         "DelaySeconds":10,
         "MessageAttributes":{
           "SellerName":{
               "DataType":"String",
               "StringValue":"Example Fuels"
           },
           "City":{
               "DataType":"String",
               "StringValue":"North Town"
           },
           "Region":{
               "DataType":"String",
               "StringValue":"WA"
           },
           "PostalCode":{
               "DataType":"String",
               "StringValue":"99123"
           },
           "PricePerGallon":{
               "DataType":"Number",
               "StringValue":"1.87"
           }
         }
     }
   ]
   ```
3. Add the records to the queue by running the following command:
   ```bash
   aws sqs send-message-batch --queue-url https://queue.amazonaws.com/940887362971/sqs-source-connector-demo --entries file://send-message-batch.json``
   ```
4. Load the SQS Source connector. Note that you must ensure the `sqs.url`
   configuration parameter points to the correct SQS URL. The `sqs.url` parameter
   format is:

   `sqs.url=https://sqs.<region-code>.amazonaws.com/<account_no>/<topic-name.fifo>`.

   For example, if the AWS CLI returns the queue URL:
   `https://eu-central-1.queue.amazonaws.com/829250931565/sqs-source-connector-demo`,
   the `sqs.url` for the SQS Source connector is
   `https://sqs.eu-central-1.amazonaws.com/829250931565/sqs-source-connector-demo`.
   ```bash
   confluent local load sqs-source
   ```

   Your output should resemble:
   ```bash
   {
     "name": "sqs-source",
     "config": {
       "connector.class": "io.confluent.connect.sqs.source.SqsSourceConnector",
       "tasks.max": "1",
       "kafka.topic": "test-sqs-source",
       "sqs.url": "https://sqs.us-east-1.amazonaws.com/942288736285822/sqs-fifo-queue.fifo",
       "name": "sqs-source"
     },
     "tasks": [],
     "type": null
   }
   ```
5. After the connector finishes ingesting data to Kafka, check that the data is
   available in the Kafka topic.
   ```bash
   bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-sqs-source --from-beginning
   ```

   You should see two records, similar to the following:
   ```bash
   {
     "schema":{
         "type":"struct",
         "fields":[
           {
               "type":"int64",
               "optional":false,
               "field":"ApproximateFirstReceiveTimestamp"
           },
           {
               "type":"int32",
               "optional":false,
               "field":"ApproximateReceiveCount"
           },
           {
               "type":"string",
               "optional":false,
               "field":"SenderId"
           },
           {
               "type":"int64",
               "optional":false,
               "field":"SentTimestamp"
           },
           {
               "type":"string",
               "optional":true,
               "field":"MessageDeduplicationId"
           },
           {
               "type":"string",
               "optional":true,
               "field":"MessageGroupId"
           },
           {
               "type":"string",
               "optional":true,
               "field":"SequenceNumber"
           },
           {
               "type":"string",
               "optional":false,
               "field":"Body"
           }
         ],
         "optional":false,
         "version":1
     },
     "payload":{
         "ApproximateFirstReceiveTimestamp":1563430750668,
         "ApproximateReceiveCount":2,
         "SenderId":"AIDA5WEKBZWN3QYIY7KAJ",
         "SentTimestamp":1563430591780,
         "MessageDeduplicationId":null,
         "MessageGroupId":null,
         "SequenceNumber":null,
         "Body":"Fuel report for account 0001 on 2015-09-16 at 02:07:31 PM."
     }
   }
   ```

## SQS credentials

The following sections provide information about configuration SQS credentials.

### Export the AWS region

Before configuring credentials, you must export the AWS region. The
environment variable must be exported where the Connect worker is running and
where the connector is deployed.

```text
export AWS_REGION=<your-aws-region>
```

### Credentials providers

The credentials provided must have permission for the actions `sqs:ReceiveMessage` and `sqs:DeleteMessage`.
By default, the SQS source connector looks for SQS credentials in the following
locations and in the following order:

1. The `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` environment variables accessible to the Connect worker processes
   where the connector will be deployed. These variables are recognized by the AWS CLI and all AWS SDKs (except for the AWS SDK for .NET).
   You use export to set these variables.
   ```bash
   export AWS_ACCESS_KEY_ID=<your_access_key_id>
   export AWS_SECRET_ACCESS_KEY=<your_secret_access_key>
   ```

   The `AWS_ACCESS_KEY` and `AWS_SECRET_KEY` can be used instead, but are
   not recognized by the AWS CLI.
2. The `aws.accessKeyId` and `aws.secretKey` Java system properties on the Connect worker processes where the connector will be deployed.
   However, these variables are only recognized by the AWS SDK for Java and are not recommended.
3. The `~/.aws/credentials` file located in the home directory of the operating system user that runs the Connect worker processes.
   These credentials are recognized by most AWS SDKs and the AWS CLI.
   Use the following AWS CLI command to create the credentials file:
   ```bash
   aws configure
   ```

   You can also manually create the credentials file using a text editor. The
   file should contain lines in the following format:
   ```bash
   [default]
   aws_access_key_id = <your_access_key_id>
   aws_secret_access_key = <your_secret_access_key>
   ```

   #### NOTE
   When creating the credentials file, make sure that the user creating the credentials file is the same user
   that runs the Connect worker processes and that the credentials file is in this user’s home directory.
   Otherwise, the SQS connector will not be able to find the credentials.

   For more details, see [AWS Credentials File Format](https://docs.aws.amazon.com/sdk-for-java/v1/developer-guide/credentials.html#credentials-file-format).
4. A query sent to `http://169.254.170.2${AWS_CONTAINER_CREDENTIALS_RELATIVE_URI}` to return AWS credentials.
   This is applicable only if the Connect worker processes are running in AWS containers.
5. A metadata query that returns credentials from an EC2 instance.
   This is applicable only if the Connect worker processes are running in EC2 instances.

Choose one of the above to define the AWS credentials that the SQS connectors
use, verify the credentials implementation is set correctly, and then restart
all of the Connect worker processes.

Note that Confluent recommends using either environment variables or a
credentials file because they are the most straightforward and can be checked
using the AWS CLI tool before running the connector.

### Custom Credentials Provider Support

You can configure the SQS connector to use a custom credentials provider, instead of the default
one provided in the connector. To do this, you need to implement a custom credentials provider, build it
as a JAR file, and deploy the JAR file to use the custom provider. By default, the SQS connector
configuration property `sqs.credentials.provider.class` uses the
`software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider` class.

Complete the following steps to use a custom credentials provider:

1. **Set a custom credentials provider class**: Set the `sqs.credentials.provider.class` property to a class that implements the `software.amazon.awssdk.auth.credentials.AwsCredentialsProvider` [interface](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/AwsCredentialsProvider.html). Configure the class to the fully qualified name of your custom credentials provider class.
2. **Configure additional settings (Optional)**: For additional configuration, prefix the configuration keys with `sqs.credentials.provider.`. If your custom credentials provider needs to accept additional configuration, implement the `org.apache.kafka.common.Configurable` interface that lets the connector receive configurations that are prefixed with `sqs.credentials.provider.`.
3. **Implement static create() method**: Your custom credentials provider class must implement a public static `create()` method that returns an instance of `AwsCredentialsProvider`. This is necessary because the connector uses this method to create the provider instance.
4. **Package your provider**: Once your custom credentials provider class is implemented, package it into a JAR file.
5. **Copy the JAR file to Connect Worker**: Copy the built JAR file to the `share/java/kafka-connect-sqs` directory on all Connect workers. This step ensures that the SQS connector can access and use your custom credentials provider.

#### NOTE
The static `builder()` method is not supported. Your custom credentials provider must implement a static `create()` method that returns an instance of `AwsCredentialsProvider`.

<a id="sqs-record-schema"></a>

## Record schema

Then source connector creates records in the following format:

### Key schema

The Key is a `struct` with the following fields:

| Field Name     | Schema Type   | Optional?   | Description                                                             |
|----------------|---------------|-------------|-------------------------------------------------------------------------|
| QueueUrl       | String        | No          | The fully qualified SQS queue URL from which this record was generated. |
| MessageId      | String        | No          | The unique message ID of the message within SQS.                        |
| MessageGroupId | String        | Yes         | In case of FIFO queues, the message group ID.                           |

### Value schema

The Value is a `struct` with the following fields:

| Field Name                       | Schema Type   | Optional?   | Description                                                                                              |
|----------------------------------|---------------|-------------|----------------------------------------------------------------------------------------------------------|
| Body                             | String        |             | The body of the message.                                                                                 |
| ApproximateFirstReceiveTimestamp | int64         |             | The timestamp when this message was first received by SQS.                                               |
| ApproximateReceiveCount          | int32         |             | The number of times this message was received by SQS.                                                    |
| SenderId                         | String        |             | The IAM user or role that sent this message to SQS.                                                      |
| SentTimestamp                    | int64         |             | The timestamp at which the sender sent it to SQS. `SentTimestamp` <= `ApproximateFirstReceiveTimestamp`. |
| MessageDeduplicationId           | String        | Yes         | For FIFO queues, a unique ID that can be used to eliminate duplicate messages.                           |
| MessageGroupId                   | String        | Yes         | For FIFO queues, the message group that this message belongs to.                                         |
| SequenceNumber                   | String        |             | A globally unique sequence number generated by SQS.                                                      |

### Header schema

Each message attribute in SQS is converted to a Header in Kafka.

1. The key of the header is the key of the message attribute.
2. The value of the header is the value of the message attribute.
3. The schema of the header depends on the data type of the message attribute.
   1. String message attributes have string schema.
   2. Number message attributes also have string schema.
   3. Binary message attributes have bytes schema.
   4. Custom message attributes will either be string or bytes, depending on the type of custom attribute.