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

# Kudu Sink Connector [Deprecated] for Confluent Platform

#### IMPORTANT
- This connector is deprecated and reaches end of life (EOL) on the Confluent Platform 7.9 end of support (EOS) date.
  For more information, see [Connector support lifecycle policy](https://docs.confluent.io/platform/current/connect/supported.html#support-policy-for-self-managed-connectors).
- <!-- 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 Kafka Connect Kudu Sink connector allows you to export data from an
Apache Kafka® topic to a Kudo columnar relational database using an Impala JDBC
driver. The connector polls data from Kafka to write to Kudu based on the topics
subscription. Auto-creation of tables, and limited auto-evolution is also
supported.

## Features

- [At least once delivery](#kudu-sink-at-least-once-delivery)
- [Dead Letter Queue](#kudu-sink-dead-letter-queue)
- [Multiple tasks](#kudu-sink-multiple-tasks)
- [Data mapping](#kudu-sink-data-mapping)
- [Key handling](#kudu-sink-key-handling)
- [Auto-creation and auto-evolution](#kudu-sink-auto-creation)
- [Client-side encryption](#kudu-sink-sink-csfle-sm)

<a id="kudu-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="kudu-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="kudu-sink-multiple-tasks"></a>

### Multiple tasks

The Kudu 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="kudu-sink-data-mapping"></a>

### Data mapping

The Kudu 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` may
be necessary.

<a id="kudu-sink-key-handling"></a>

### 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 further detail.

<a id="kudu-sink-auto-creation"></a>

### 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#kudu-sink-configs).

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. In contrast, if `auto.evolve` is disabled no evolution is performed
and the connector task fails with an error stating the missing columns.

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. We use the following mapping from Connect schema types to Impala and
Kudu types:

| Schema Type   | Impala        | Kudu            |
|---------------|---------------|-----------------|
| Int8          | TINYINT       | int8            |
| Int16         | SMALLINT      | int16           |
| Int32         | INT           | int32           |
| int64         | BIGINT        | int64           |
| Float32       | FLOAT         | float           |
| Float64       | DOUBLE        | double          |
| Boolean       | BOOLEAN       | bool            |
| String        | STRING        | string          |
| ‘Decimal’     | DECIMAL(38,s) | decimal         |
| ‘Date’        | TIMESTAMP     | unixtime_micros |
| ‘Time’        | TIMESTAMP     | unixtime_micros |
| ‘Timestamp’   | TIMESTAMP     | unixtime_micros |

#### IMPORTANT
For backwards-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.

<a id="kudu-sink-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).

## Limitations

1. Despite Impala supports `ARRAY` and `MAP` types, we currently do not
   support any structured types for column types.
2. Impala will convert all column names to lowercase, so it would be better just
   use lowercase column names in the first place to avoid name mismatches.
3. Upsert and idempotent writes are not supported for now.
4. Delete is not supported for now.

## Install the Kudu 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.

If you are running a multi-node Connect cluster, the Kudu Sink connector and
Impala JDBC driver JARs must be installed on every Connect worker in the
cluster. See below for details.

### Prerequisites

- You must install the connector on every machine where Connect will run.
- An installation of the latest (`latest`) connector version.

  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-kudu: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-kudu:1.0.0-preview
  ```

### Install the connector manually

[Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-kudu) for your
connector and then follow the manual connector installation [instructions](https://docs.confluent.io/home/connect/install.html).

## Install the Impala JDBC Driver

The Kudu Sink connector uses the [Java Database Connectivity (JDBC) API](https://www.oracle.com/technetwork/java/javase/jdbc/index.html). In order
for this to work, the connectors must use Impala to query Kudu database, and
have *Impala JDBC Driver* installed.

The basic steps of installation are:

1. Download [Impala JDBC Connector](https://www.cloudera.com/downloads/connectors/impala/jdbc/2-6-15.html),
   and unzip to get the JAR files.
2. Place these JAR files into the
   `share/confluent-hub-components/confluentinc-kafka-connect-kudu/lib`
   directory in your Confluent Platform installation on each of the Connect worker nodes.
3. Restart all of the Connect worker nodes.

### General guidelines

The following are additional guidelines to consider:

* Use the most recent version of the Impala JDBC driver available.
* Use the correct JAR file for the Java version used to run Connect workers.
  Ensure you use the correct JAR file for the Java version in use. If you
  install and try to use the Impala JDBC driver JAR file for the wrong version
  of Java, starting any Kudu source connector or Kudu sink connector will likely
  fail with `UnsupportedClassVersionError`. If this happens, remove the Impala
  JDBC driver JAR file you installed and repeat the driver installation process
  with the correct JAR file.
* The `share/confluent-hub-components/confluentinc-kafka-connect-kudu/lib`
  directory mentioned above is for Confluent Platform. If you are using a different
  installation, find the location where the Confluent Kudu Source and Sink
  connector JAR files are located, and place the Impala JDBC driver JAR file(s)
  for the target databases into the same directory.
* If the Impala JDBC driver is not installed correctly, the Kudu Source or Sink
  Connector will fail on startup. Typically, the system throws the error `No
  suitable driver found`. If this happens, install the Impala JDBC driver
  again.

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

See [Confluent Platform license](sink_config_options.md#kudo-connector-license-config) for license properties and
[License topic configuration](sink_config_options.md#kudo-license-topic-configuration) for information about the license topic.
License requirements are the same for both the sink and source connector.

## Configuration Properties

For a complete list of configuration properties for this connector, see
[Configuration Reference for Kudu Sink Connector for Confluent Platform](sink_config_options.md#kudu-sink-configs).

## Prerequisites

- [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) commands. This quick start assumes that you are using the
  Confluent CLI. By default ZooKeeper, Kafka, Schema Registry, Kafka Connect REST API, and
  Kafka Connect are started with the `confluent local start` command.
  For more information, see [Confluent
  Platform](/platform/current/installation/installing_cp/index.html). Note that as of Confluent Platform 7.5,
  ZooKeeper is deprecated for new deployments. Confluent recommends KRaft mode for
  new deployments.
- [Kudu](https://kudu.apache.org/releases/) and [Impala](https://impala.apache.org/downloads.html) are installed and configured properly ([Using Kudu with Impala](https://kudu.apache.org/docs/kudu_impala_integration.html)). For DECIMAL type support, we need at least Kudu 1.7.0, and Impala 3.0.
- Verify that the [Impala JDBC driver](https://www.cloudera.com/downloads/connectors/impala/jdbc/2-6-15.html) is available on the Kafka Connect process’s `CLASSPATH`.
- Kafka and Schema Registry are running locally on the default ports.

## Quick Start

The following procedure steps you through copying Avro data from a single Kafka
topic to a local Kudu database.

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

### Create Kudu database

1. Start Impala shell.
   ```bash
   impala-shell -i localhost:21000 -l -u <ldap-username> --ldap_password_cmd="echo -n <ldap-password>" --auth_creds_ok_in_clear
   ```
2. Create a database with this command:
   ```bash
   CREATE DATABASE test;
   ```

   Your output should resemble:
   ```bash
   Query: create DATABASE test
   Fetched 0 row(s) in 0.80s
   ```

### Load the Kudu Sink Connector

Load the predefined Kudu Sink connector.

1. Optional: View the available predefined connectors with this command:
   ```bash
   confluent local list
   ```

   Your output should resemble:
   ```bash
   Bundled Predefined Connectors (edit configuration under etc/):
     elasticsearch-sink
     file-source
     file-sink
     jdbc-source
     jdbc-sink
     kudu-source
     kudu-sink
     hdfs-sink
     s3-sink
   ```
2. Create a `kudu-sink.json` file for your Kudu Sink connector.
   ```text
   {
         "name": "kudu-sink",
         "config": {
           "connector.class": "io.confluent.connect.kudu.KuduSinkConnector",
           "tasks.max": "1",
           "topics": "orders",
           "impala.server": "127.0.0.1",
           "impala.port": "21050",
           "kudu.database": "test",
           "auto.create": "true",
           "pk.mode":"record_value",
           "pk.fields":"id",

           "key.converter": "io.confluent.connect.avro.AvroConverter",
           "key.converter.schema.registry.url": "http://localhost:8081",
           "value.converter": "io.confluent.connect.avro.AvroConverter",
           "value.converter.schema.registry.url": "http://localhost:8081",
           "confluent.topic.bootstrap.servers": "localhost:9092",
           "confluent.topic.replication.factor": "1",
           "impala.ldap.password": "<ladp-password>",
           "impala.ldap.user": "<ldap-user>",
           "kudu.tablet.replicas": "1",
           "name": "kudu-sink"
         }
     }
   ```
3. Load the `kudu-sink` connector:
   ```bash
   confluent local load kudu-sink --config kudu-sink.json
   ```

   Your output should resemble:
   ```bash
   {
       "name": "kudu-sink",
       "config": {
         "connector.class": "io.confluent.connect.kudu.KuduSinkConnector",
         "tasks.max": "1",
         "topics": "orders",
         "impala.server": "127.0.0.1",
         "impala.port": "21050",
         "kudu.database": "test",
         "auto.create": "true",
         "pk.mode":"record_value",
         "pk.fields":"id",

         "key.converter": "io.confluent.connect.avro.AvroConverter",
         "key.converter.schema.registry.url": "http://localhost:8081",
         "value.converter": "io.confluent.connect.avro.AvroConverter",
         "value.converter.schema.registry.url": "http://localhost:8081",
         "confluent.topic.bootstrap.servers": "localhost:9092",
         "confluent.topic.replication.factor": "1",
         "impala.ldap.password": "secret",
         "impala.ldap.user": "kudu",
         "kudu.tablet.replicas": "1",
         "name": "kudu-sink"
         },
         "tasks": [],
         "type": "sink"
       }
   ```

### Produce a record in Kafka

1. Produce a record into the `orders` topic.
   ```bash
    ./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**:
   ```bash
   {"id": 999, "product": "foo", "quantity": 100, "price": 50}
   ```
3. Use Impala shell to query the Kudu database and you should see that the
   `orders` table was automatically created and contains the record.
   ```bash
   USE test;
   SELECT * from orders;
   foo|50.0|100|999
   ```

## Troubleshooting

### HiveServer2 error

When you run this connector, you might see the following error message.

```bash
java.sql.SQLException: [Cloudera][ImpalaJDBCDriver](500176) Error connecting to HiveServer2, please verify connection settings.
```

It means you haven’t set an LDAP in Impala or a username and a password for LDAP
is not valid.

### Not enough live tablet servers

When you run this connector, you might see the following error message:

```bash
com.cloudera.impala.support.exceptions.GeneralException: [Cloudera][ImpalaJDBCDriver](500051) ERROR processing query/statement. Error Code: 0, SQL state: TStatus(statusCode:ERROR_STATUS, sqlState:HY000, errorMessage:ImpalaRuntimeException: Error creating Kudu table 'impala::test.orders'
CAUSED BY: NonRecoverableException: not enough live tablet servers to create a table with the requested replication factor 3; 1 tablet servers are alive
), Query: CREATE TABLE `orders` (
`id` INT NOT NULL,
`product` STRING NOT NULL,
`quantity` INT NOT NULL,
`price` INT NOT NULL,
PRIMARY KEY(`id`)) PARTITION BY HASH PARTITIONS 2 STORED AS KUDU TBLPROPERTIES ('kudu.num_tablet_replicas' = '3').
```

It means you don’t have enough tablet servers to support 3 replicas. Either you
need to increase your tablet servers to 3 or you can add the following property
to your connector.

```bash
"kudu.tablet.replicas":"1"
```