<a id="connect-redshift"></a>

# Amazon Redshift Sink Connector for Confluent Platform

The Kafka Connect Amazon Redshift Sink connector allows you to export
data from Apache Kafka® topics to Amazon Redshift. The connector polls data from
Kafka and writes this data to an Amazon Redshift database. Polling data is
based on subscribed topics. Auto-creation of tables and limited auto-evolution
are supported.

#### 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 Amazon Redshift Sink connector for Confluent Platform includes the following features:

- [At least once delivery](#redshift-sink-at-least-once-delivery)
- [Dead Letter Queue](#redshift-sink-dead-letter-queue)
- [Multiple tasks](#redshift-sink-multiple-tasks)
- [Client-side encryption](#redshift-sink-csfle-sm)

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

### At least once delivery

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

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

### Dead Letter Queue

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

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

### Multiple tasks

The Amazon Redshift 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="redshift-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).

### Data mapping

The sink connector requires knowledge of schemas, so you should use a suitable
converter (for example, the Avro converter that comes with Schema Registry, or the JSON
converter with schemas enabled. Kafka record keys, if present, can be primitive
types or a Connect struct, and the record value must be a Connect struct.
Fields being selected from Connect structs must be of primitive types. If the
data in the topic is not of a compatible format, implementing a custom
`Converter` or using [Single Message Transforms
(SMTs)](/platform/current/connect/concepts.html#transforms) may be necessary.

### Key handling

The default is for primary keys to not be extracted with `pk.mode` set to
`none`, which is not suitable for advanced usage such as upsert semantics and
when the connector is responsible for auto-creating the destination table. There
are different modes that enable to use fields from the Kafka record key, the Kafka
record value, or the Kafka coordinates for the record.

Refer to [primary key configuration options](https://docs.confluent.io/kafka-connect-jdbc/current/sink-connector/sink_config_options.html#data-mapping)
for more details.

### Delete mode

The connector can delete rows in a database table when it consumes a tombstone
record, which is a Kafka record that has a non-null key and null value. This
behavior is disabled by default, meaning that any tombstone records will result
in a failure of the connector, making it easy to upgrade the JDBC connector and
keep prior behavior.

You can enable deletes by setting `delete.enabled=true` when `pk.mode` is
set to `record_key`. This is because deleting a row from the table requires
the primary key be used as criteria.

Enabling delete mode does not affect the `insert.mode`.

<!-- The following is currently unsupported and commented out.
-----------------
Idempotent writes
-----------------
The default ``insert.mode`` is ``insert``. If it is configured as ``upsert``,
the connector will use upsert semantics rather than plain ``INSERT``
statements. Upsert semantics refer to atomically adding a new row or updating
the existing row if there is a primary key constraint violation, which
provides idempotence.
If there are failures, the |ak| offset used for recovery may not be
up-to-date with what was committed as of the time of the failure, which can
lead to re-processing during recovery. The upsert mode is highly recommended
as it helps avoid constraint violations or duplicate data if records need to
be re-processed.
Aside from failure recovery, the source topic may also naturally contain
multiple records over time with the same primary key, making upserts
desirable.
Redshift uses the following PostgreSQL-specific DML for upserts: ``INSERT ..
ON CONFLICT .. DO UPDATE SET ..`` -->

### Auto-creation and auto-evolution

If `auto.create` is enabled, the connector can create the destination table
if it is found to be missing. The creation takes place online with records being
consumed from the topic, since the connector uses the record schema as a basis
for the table definition. Primary keys are specified based on the key
[configuration settings](sink_config_options.md#redshift-sink-config-options).

If `auto.evolve` is enabled, the connector can perform limited auto-evolution
by issuing ALTER on the destination table when it encounters a record for
which a column is found to be missing. Since data-type changes and removal of
columns can be dangerous, the connector does not attempt to perform such
evolutions on the table. Addition of primary key constraints is also not
attempted.

For both auto-creation and auto-evolution, the nullability of a column is based
on the optionality of the corresponding field in the schema, and default values
are also specified based on the default value of the corresponding field if
applicable. You can use the following mapping from Connect schema types to
database types:

| Schema Type   | Redshift         |
|---------------|------------------|
| INT8          | SMALLINT         |
| INT16         | SMALLINT         |
| INT32         | INT              |
| INT64         | BIGINT           |
| FLOAT32       | REAL             |
| FLOAT64       | DOUBLE PRECISION |
| BOOLEAN       | BOOLEAN          |
| ‘Decimal’     | DECIMAL          |
| ‘Date’        | DATE             |
| ‘Time’        | TIME             |
| ‘Timestamp’   | TIMESTAMP        |
| BYTES         | Not supported    |
| ‘Struct’      | Not supported    |
| ‘Map’         | Not supported    |
| ‘Array’       | Not supported    |

#### IMPORTANT
For backward-compatible table schema evolution, new fields in record
schemas must be optional or have a default value. If you need to delete a
field, the table schema should be manually altered to either drop the
corresponding column, assign it a default value, or make it nullable.

## Limitations

The Amazon Redshift Sink connector does not support Avro schemas that contain
`decimal` logical types. For a better understanding of numeric data types, see this
blog post: [Bytes, Decimals, Numerics and oh my](https://www.confluent.io/blog/kafka-connect-deep-dive-jdbc-source-connector/#bytes-decimals-numerics).

## Install the Amazon Redshift 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.
- If you want to install the connector using Confluent Marketplace, you must install
  the [Confluent Hub Client](https://docs.confluent.io/home/connect/confluent-hub/client.html). This
  is installed by default with Confluent Enterprise.
- Confluent Platform 3.3.0 or later, or Kafka 0.11.0 or later
- Java 1.8
- At minimum, `INSERT` access privilege is required for this connector. See
  [Amazon Redshift Grant](https://docs.aws.amazon.com/redshift/latest/dg/r_GRANT.html). If
  `delete.enabled=true`, `DELETE` access privilege is required.
- The connector configuration requires a Redshift user (and password) with
  Redshift database privileges. For example:
  ```text
  CREATE DATABASE <DB_NAME>;

  CREATE USER <DB_USER> PASSWORD '<DB_PASSWORD>';

  GRANT USAGE ON SCHEMA public TO <DB_USER>;
  GRANT CREATE ON SCHEMA public TO <DB_USER>;
  GRANT SELECT ON ALL TABLES IN SCHEMA public TO <DB_USER>;
  GRANT ALL ON SCHEMA public TO <DB_USER>;

  GRANT CREATE ON DATABASE <DB_NAME> TO <DB_USER>;
  ```

  For more information, see the [Redshift docs](https://docs.aws.amazon.com/redshift/latest/gsg/database-tasks.html).

### Install the connector using the Confluent CLI

To install the `latest` connector version using [Confluent Hub Client](https://www.confluent.io/hub/), navigate to your Confluent Platform installation directory
and run the following command:

```bash
confluent connect plugin install confluentinc/kafka-connect-aws-redshift: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-redshift:1.2.2
```

If you are running a multi-node Connect cluster, the Redshift connector and
JDBC driver JARs must be installed on every Connect worker in the cluster.

### Install the connector manually

[Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-aws-redshift) 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](sink_config_options.md#aws-redshift-sink-connector-license-config)
and for information about the license topic, see
[License topic configuration](sink_config_options.md#redshift-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 Amazon Redshift Sink Connector for Confluent Platform](sink_config_options.md#redshift-sink-config-options).

<a id="connect-redshift-sink"></a>

## Quick start

To see the basic functionality of the connector, this quick start demonstrates
how to copy Avro data from a single topic to a Redshift instance.

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

### Prerequisites

- [Confluent Platform](/platform/current/installation/installing_cp/index.html) is
  installed and services are running by using the Confluent CLI
  [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) commands.
- Kafka and Schema Registry are running locally on the default ports.

Note that this quick start assumes that you are using the
[Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) commands; however,
[standalone installations](/platform/current/installation/installing_cp/index.html) are
also supported. By default ZooKeeper, Kafka, Schema Registry, Kafka Connect REST API, and
Kafka Connect are started with the `confluent local start` command.
Note that as of Confluent Platform 7.5, ZooKeeper is deprecated for new deployments. Confluent
recommends KRaft mode for new deployments.

<a id="redshift-instructions"></a>

### Create an Amazon Redshift instance

1. Log into your AWS Management Console.
2. Navigate to [Redshift](https://console.aws.amazon.com/redshift/home).
   Note that your account needs permission to create and administer Redshift
   instances. If you see **User <you> is not authorized to describe clusters**,
   then you must contact your account administrator to set up your Redshift cluster.
3. Navigate to Clusters.
4. Click **Quick Launch Cluster**.
5. Set the **Master User Password**. You must remember this password for a later step.
6. Click **Launch Cluster** to complete the setup.
7. Wait for your cluster to be in the “available” state (this takes
   approximately 5 minutes). Note that you need the information in the
   **Cluster Configuration** screen to complete the connector configuration.

### Load the Amazon Redshift Sink connector

1. Create a properties file for your Redshift Sink connector.
   ```text
   name=redshift-sink
   confluent.topic.bootstrap.servers=localhost:9092
   confluent.topic.replication.factor=1
   connector.class=io.confluent.connect.aws.redshift.RedshiftSinkConnector
   tasks.max=1
   topics=orders
   aws.redshift.domain=< Required Configuration >
   aws.redshift.port=< Required Configuration >
   aws.redshift.database=< Required Configuration >
   aws.redshift.user=< Required Configuration >
   aws.redshift.password=< Required Configuration >
   pk.mode=kafka
   auto.create=true
   ```

   Fill in the configuration parameters of your cluster as they appear in your
   [Cluster Details](https://console.aws.amazon.com/redshift/home#cluster-list:).
2. Load the `redshift-sink` connector:
   ```bash
   confluent local load redshift-sink --config redshift-sink.properties
   ```

   Your output should resemble the following:
   ```text
   {
     "name": "redshift-sink",
     "config": {
       "confluent.topic.bootstrap.servers": "localhost:9092",
       "connector.class": "io.confluent.connect.aws.redshift.RedshiftSinkConnector",
       "tasks.max": "1",
       "topics": "orders",
       "aws.redshift.domain": "cluster-name.cluster-id.region.redshift.amazonaws.com",
       "aws.redshift.port": "5439",
       "aws.redshift.database": "dev",
       "aws.redshift.user": "awsuser",
       "aws.redshift.password": "your-password",
       "auto.create": "true",
       "pk.mode": "kafka",
       "name": "redshift-sink"
     },
     "tasks": [],
     "type": "sink"
   }
   ```

   Note that non-CLI users can load the Redshift Sink connector by using the
   following command:
   ```text
   ${CONFLUENT_HOME}/bin/connect-standalone \
   ${CONFLUENT_HOME}/etc/schema-registry/connect-avro-standalone.properties \
   redshift-sink.properties
   ```

### Produce a record in Kafka

1. Produce a record into the `orders` topic.
   ```text
   ./bin/kafka-avro-console-producer \
   --broker-list localhost:9092 --topic orders \
   --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"id","type":"int"},{"name":"product", "type": "string"}, {"name":"quantity", "type": "int"}, {"name":"price", "type": "float"}]}'
   ```

   The console producer waits for input.
2. Copy and paste the following record into the terminal and press **Enter**:
   ```text
   {"id": 999, "product": "foo", "quantity": 100, "price": 50}
   ```
3. Open the [Query Editor](https://console.aws.amazon.com/redshift/home) and
   run the following query:
   ```text
   SELECT * from orders;
   ```