How to load Kafka topics into a database with Kafka Connect

What and why

Kafka Connect is a framework for moving data between Kafka and external systems using reusable connectors instead of bespoke code. A sink connector reads a topic and writes to a destination; a source connector does the reverse. This tutorial uses the JDBC sink to land topic records in PostgreSQL.

Prerequisites

• A Kafka cluster and a topic with records.
• A Kafka Connect worker (distributed mode recommended).
• A target database and the JDBC sink connector plus driver installed in the worker.

Steps

1. Run a Connect worker

Start Connect in distributed mode pointing at your cluster, exposing the REST API on port 8083. Confirm it is up with curl localhost:8083/.

2. Install the sink plugin

Place the JDBC sink connector JARs and the database driver in the worker's plugin path, then restart. Verify it loaded:

curl localhost:8083/connector-plugins

3. Prepare the target table

Create a table matching the record fields, or let the connector auto-create it:

CREATE TABLE orders (id BIGINT PRIMARY KEY, total NUMERIC, status TEXT);

4. Create the sink connector

curl -X POST localhost:8083/connectors -H 'Content-Type: application/json' -d '{
  "name": "orders-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "topics": "orders",
    "connection.url": "jdbc:postgresql://postgres:5432/app",
    "connection.user": "app", "connection.password": "secret",
    "insert.mode": "upsert", "pk.mode": "record_key",
    "auto.create": "true"
  }
}'

5. Configure converters

Converters define how Connect reads keys and values. For schemaless JSON, set value.converter=org.apache.kafka.connect.json.JsonConverter. For typed data with a registry, use the Avro converter so the sink knows column types.

6. Monitor the connector

Check status and the tasks' state:

curl localhost:8083/connectors/orders-sink/status

All tasks should be RUNNING. Connect tracks consumed offsets per topic partition automatically.

Verification

Produce records to the orders topic and confirm rows appear in the database table within seconds. The connector status should stay RUNNING, and re-producing a record with the same key should upsert rather than duplicate.

Next Steps

Add single message transforms to reshape records, route failures to a dead letter queue, and scale throughput by increasing tasks.max up to the topic's partition count.