JDBC Sink Connector

Summary

The JDBC Sink Connector uses standard JDBC to write Kafka messages to any database with a JDBC driver, including MySQL, PostgreSQL, Oracle, SQL Server, and many others. It handles INSERT, UPDATE, and UPSERT operations automatically.

WHAT IS THE JDBC SINK CONNECTOR?

The JDBC Sink Connector consumes messages from Kafka topics and writes them as rows in database tables. It automatically creates tables, handles schema evolution, and provides configurable write modes.

Let's visualize how the JDBC Sink Connector works. It consumes from Kafka and writes to database tables.

The connector consumes messages in batches, converts them to SQL INSERT or UPDATE statements, and executes them against the database. Batching improves throughput by reducing the number of database round-trips.

INSERT MODE

In INSERT mode, the connector simply inserts every consumed message as a new row in the database table.

INSERT mode is straightforward but can lead to duplicate rows if the same record is consumed multiple times due to retries or reprocessing. It's best for append-only scenarios.

UPSERT MODE

UPSERT mode is more sophisticated. It uses a primary key to determine whether to INSERT a new row or UPDATE an existing one.

UPSERT mode provides idempotent writes. If you re-consume the same message, it updates the existing row instead of creating duplicates. This is crucial for reliable, exactly-once semantics.

CONNECTOR CONFIGURATION

The following shows a typical JDBC Sink Connector configuration.

1{
2  "name": "jdbc-sink-users",
3  "config": {
4    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
5    "tasks.max": "1",
6    "connection.url": "jdbc:mysql://localhost:3306/mydb",
7    "connection.user": "kafka_connect",
8    "connection.password": "password",
9    "topics": "users-topic",
10    "auto.create": "true",
11    "auto.evolve": "true",
12    "insert.mode": "upsert",
13    "pk.mode": "record_key",
14    "pk.fields": "id",
15    "table.name.format": "kafka_${topic}"
16  }
17}

connector.class: The JDBC Sink Connector class.

connection.url, user, password: Database connection details.

topics: Kafka topics to consume from.

auto.create: Automatically create tables if they don't exist.

auto.evolve: Automatically add columns when schema changes.

insert.mode: Write mode - insert, update, or upsert.

pk.mode: How to determine primary key - record_key, record_value, or kafka (topic+partition+offset).

pk.fields: Column names for the primary key.

table.name.format: Template for table names. ${topic} is replaced with the topic name.

AUTOMATIC TABLE CREATION

With auto.create enabled, the connector automatically creates tables based on the Kafka message schema.

When the connector consumes the first message from a topic, it checks if the table exists. If not, it extracts the schema from the message and generates a CREATE TABLE statement with appropriate column types.

SCHEMA EVOLUTION

With auto.evolve enabled, the connector handles schema changes by adding new columns automatically.

When a message contains a new field not present in the table, the connector automatically adds the column. This allows your database schema to evolve with your Kafka messages without manual intervention.

PRIMARY KEY MODES

The pk.mode configuration determines how the connector identifies rows for updates.

record_key: The primary key fields come from the Kafka message key. This is common when the key is a struct containing the ID.

record_value: The primary key fields come from the message value. Use this when the ID is part of the message body.

kafka: Uses Kafka metadata (topic, partition, offset) as the primary key. This ensures uniqueness but isn't meaningful for application queries.

BATCHING AND PERFORMANCE

The JDBC Sink Connector batches writes for better performance.

Configuration

1{
2  "batch.size": "100",
3  "max.retries": "10",
4  "retry.backoff.ms": "3000"
5}

batch.size: Number of records to batch before writing to the database. Larger batches improve throughput but increase latency.

max.retries: Number of retries for failed writes. Important for handling transient database issues.

retry.backoff.ms: Time to wait between retries, in milliseconds.

Batching significantly improves performance by reducing database round-trips, but finding the right batch size requires testing based on your workload.

HANDLING NULL VALUES

The connector provides options for handling null values in messages.

delete.enabled: When true, records with all null values (except the key) are treated as deletes, and the corresponding row is removed from the database.

This enables CDC-style patterns where tombstone records indicate deletions.

Let's recap what we learned about the JDBC Sink Connector:

Key ideas

The JDBC Sink Connector writes Kafka messages to relational databases using standard JDBC.

It supports INSERT mode for append-only writes and UPSERT mode for idempotent updates.

Automatic table creation and schema evolution eliminate manual database maintenance.

Primary key modes determine how rows are identified for updates: record_key, record_value, or kafka metadata.

Batching improves performance by reducing database round-trips. Configure batch size based on throughput requirements.

The connector handles retries, schema evolution, and null values automatically for production-ready reliability.