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

# Teradata Sink Connector for Confluent Platform

The Kafka Connect Teradata Sink connector allows you to export data from
Apache Kafka® topics to Teradata. The connector polls data from Kafka to write to the
database based on the topics subscription. Table auto-creation and limited
auto-evolution are also supported.

#### IMPORTANT


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 Teradata Sink connector includes the following features:

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

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

### Multiple tasks

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

### Data mapping

The Teradata 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). If present, Kafka record keys can be
primitive types or a Connect struct. 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="teradata-sink-key-handling"></a>

### Key handling

The default is for primary keys to not be extracted with `pk.mode` set to
`none`. This is not suitable when the connector is responsible for
auto-creating the destination table. There are different modes that can be
enabled 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](configuration.md#teradata-sink-pk-config-options) for additional information.

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

### Auto-creation and auto-evolution

#### NOTE
The Teradata user requires `CREATE TABLE` permissions for this feature.

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](configuration.md#teradata-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. In contrast, if `auto.evolve` is disabled, no evolution is performed and the connector task fails with a missing columns error.

For both auto-creation and auto-evolution, the nullability of a column is based
on the optionality of the corresponding field in the schema. Default values are
also specified based on the default value of the corresponding field (if
applicable). Confluent uses the following mapping from Connect schema types to
Teradata types:

| Schema Type   | Teradata                           |
|---------------|------------------------------------|
| INT8          | BYTEINT                            |
| INT16         | SMALLINT                           |
| INT32         | INTEGER                            |
| INT64         | BIGINT                             |
| FLOAT32       | FLOAT                              |
| FLOAT64       | DOUBLE PRECISION                   |
| BOOLEAN       | BYTEINT                            |
| STRING        | LONG VARCHAR CHARACTER SET UNICODE |
| BYTES         | VARBYTES (64000)                   |
| ‘Decimal’     | DECIMAL(38,s)                      |
| ‘Date’        | DATE                               |
| ‘Time’        | TIME WITH TIME ZONE                |
| ‘Timestamp’   | TIMESTAMP WITH TIME ZONE           |

#### 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="teradata-sink-limitations"></a>

## Limitations

* The Teradata Sink connector does not support nested data types. Confluent uses the mapping shown in table in the [Auto-creation and auto-evolution](#teradata-sink-auto-creation) section.
* The Teradata Sink connector does not support upserts.

## 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](../source-connector/configuration.md#teradata-connector-license-config), and for
information about the license topic, see
[License topic configuration](../source-connector/configuration.md#teradata-license-topic-configuration).

## Configuration Properties

For a complete list of configuration properties for this connector, see [Configuration Reference for Teradata Sink Connector for Confluent Platform](configuration.md#teradata-sink-config-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).

## Installing Teradata 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.
- 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
- Teradata 15.00 or later
- You must run the Teradata connector with a default timezone that does not
  observe Daylight Saving Time. This is a functional limitation of the Teradata
  JDBC driver and has no workaround. We recommend running your connect workers
  with the system property `-Duser.timezone=UTC` set.

- Set up your [Teradata Development Environment](../tde_setup.md#teradata-development-setup). Leave the database running while you complete the remainder of this guide.
- If you are running a multi-node Connect cluster, the Teradata connector and
  JDBC driver JAR must be installed on every Connect worker in the
  cluster.Install the [Teradata JDBC driver](#teradata-sink-jdbc-driver-setup).

### Installing the connector using Confluent Marketplace

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-hub install confluentinc/kafka-connect-teradata:latest
```

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

```bash
confluent-hub install confluentinc/kafka-connect-teradata:2.0.0
```

### Installing the connector manually

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

<a id="teradata-sink-jdbc-driver-setup"></a>

## Installing JDBC Drivers

The Teradata Source and Sink connectors use the [Java Database Connectivity
(JDBC) API](https://www.oracle.com/technetwork/java/javase/jdbc/index.html)
that enables applications to connect to and use a wide range of database
systems. In order for this to work, the connectors must have a *JDBC Driver* for
Teradata.

1. Navigate to the [Teradata Downloads page](https://downloads.teradata.com/download/connectivity/jdbc-driver)
2. Log in with your Teradata account if you are not signed in already.
3. Download the JDBC driver corresponding to your Teradata version.
4. Extract the archive to get a `terajdbc4.jar` file.
5. Place this JAR file into the `share/confluent-hub-components/confluentinc-kafka-connect-teradata/lib` directory in
   your Confluent Platform installation on each of the Connect worker nodes.
6. Restart all of the Connect worker nodes.

The rest of this section outlines the specific steps for more common database
management systems.

### General guidelines

The following are additional guidelines to consider:

* The `share/confluent-hub-components/confluentinc-kafka-connect-teradata/lib` directory mentioned above is for Confluent Platform.
  If you are using a different installation, find the location where the Confluent Teradata source and sink connector
  JAR files are located, and place the JDBC driver JAR file(s) for the target databases into the same directory.
* If the JDBC driver is not installed correctly, the Teradata source or sink connector will fail on startup.
  Typically, the system throws the error `No suitable driver found`.
  If this happens, install the JDBC driver again by following the instructions.
* The connector must be run on a connect worker with a default timezone which does not observe daylight savings.
  If daylight savings is observed in your timezone, then the connector will fail to configure or start.
  You can manually override the worker timezone by setting the TZ environment variable before starting the worker, or on the command line with the JVM argument -Duser.timezone=UTC

## Quick Start

To see the basic functionality of the connector, you copy Avro data from a single topic to a local Teradata development environment.

### Load the Teradata Sink Connector

1. Create a properties file for your Teradata Sink connector
   ```none
   name=teradata-sink
   confluent.topic.bootstrap.servers=localhost:9092
   confluent.topic.replication.factor=1
   connector.class=io.confluent.connect.teradata.TeradataSinkConnector
   tasks.max=1
   topics=orders
   teradata.url=jdbc:teradata://localhost
   teradata.database=dev
   teradata.username=dev
   teradata.password=dev
   pk.mode=kafka
   auto.create=true

   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
   ```
2. Load the `teradata-sink` connector:
   ```bash
   confluent local load teradata-sink --config teradata-sink.properties
   ```

   Your output should resemble:
   ```none
   {
      "name": "teradata-sink",
      "config": {
        "confluent.topic.bootstrap.servers": "localhost:9092",
        "confluent.topic.replication.factor": "1",
        "connector.class": "io.confluent.connect.teradata.TeradataSinkConnector",
        "tasks.max": "1",
        "topics": "orders",
        "teradata.url": "jdbc:teradata://localhost",
        "teradata.database": "dev",
        "teradata.username": "dev",
        "teradata.password": "dev",
        "pk.mode": "kafka",
        "auto.create": "true",
        "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",
        "name": "teradata-sink"
      },
      "tasks": [],
      "type": "sink"
    }
   ```

   ```none
   ${CONFLUENT_HOME}/bin/connect-standalone \
   ${CONFLUENT_HOME}/etc/schema-registry/connect-avro-standalone.properties \
   teradata-sink.properties
   ```

### Produce a record in Teradata

1. Produce a record into the `orders` topic.
   ```none
   ./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:
   ```none
   {"id": 999, "product": "foo", "quantity": 100, "price": 50}
   ```
3. Log into database user that you created earlier
   ```none
   bteq

   .logon dev

   dev
   ```
4. In the BTEQ command prompt you should see that the `orders` table was automatically created and contains the record:
   ```none
   SELECT * FROM orders;
   ```

   Your output should resemble:
   ```none
   SELECT * FROM orders;

   *** Query completed. 1 rows found. 7 columns returned.
   *** Total elapsed time was 1 second.

   __connect_topic  product   quantity  __connect_partition   __connect_offset  price                   id
   --------------------------------------------------------------------------------------------------------
   orders           foo       100       0                     0                 5.00000000000000E 001   999
   ```

## Troubleshooting

### Daylight saving time

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

```bash
{
  "error_code": 500,
  "message": "This Connector must be used on a connect worker with a default timezone which does not observe daylight savings time. This is a functional limitation of the Teradata JDBC driver and has no workaround. On the JVM arguments, specify -Duser.timezone=UTC to override the system default.
  ...
 }
```

In order to avoid this error, you must change the default timezone of the connect worker by adding -Duser.timezone=UTC to KAFKA_OPTS environment variable. If you start Kafka Connect worker from the command line, you can export the KAFKA_OPTS environment variable before starting Kafka Connect worker.

::
: export KAFKA_OPTS=”-Duser.timezone=UTC”
  connect-distributed -daemon /etc/kafka/connect-distributed.properties

If Kafka Connect is started by systemd, add this to your Kafka Connect service file:

::
: [Service]
  …
  Environment=KAFKA_OPTS=”-Duser.timezone=UTC”
  …