Sink Connectors

Intermediate

Kafka Connect

Sink connectors push Kafka records to external systems (Elasticsearch, S3, JDBC); they support upsert semantics and dead-letter queuing for failed records.

Overview

Sink connectors read from Kafka topics and write to external target systems — the opposite of source connectors. Common targets include Elasticsearch (for search indexing), Amazon S3 (data lake archiving), JDBC databases (for OLAP or reporting), ClickHouse (analytics), Snowflake, and Redis. Sink connectors are fully managed by the Kafka Connect framework: offsets are committed back to Kafka, tasks run in parallel for throughput, and the framework handles retries and task restarts. Dead-letter queuing (DLQ) routes records that fail to process (schema mismatch, serialisation error, target unavailability) to a designated Kafka topic instead of halting the connector.

JDBC Sink Connector — writing to a relational database

The Confluent JDBC Sink Connector writes Kafka records to a relational database table. It supports INSERT, UPSERT (based on primary key), and DELETE operations. Column mapping is driven by field names in the Kafka value schema — Avro or JSON Schema is recommended for schema-driven field mapping. Table auto-creation and auto-evolution add and rename columns as the schema evolves.

JSON — JDBC Sink Connector with upsert and DLQ configuration

# JDBC Sink Connector configuration
curl -X POST http://connect:8083/connectors -H 'Content-Type: application/json' -d '{
  "name": "orders-jdbc-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "3",
    "topics": "orders",
    "connection.url": "jdbc:mysql://mysql:3306/analytics",
    "connection.user": "sink_user",
    "connection.password": "secret",
    "insert.mode": "upsert",          // INSERT or upsert (requires pk.mode)
    "pk.mode": "record_key",          // use Kafka record key as PK
    "pk.fields": "order_id",
    "auto.create": "true",            // create table if not exists
    "auto.evolve": "true",            // add new columns on schema change
    "table.name.format": "${topic}",  // table name = topic name
    // Dead-letter queue for failed records
    "errors.tolerance": "all",        // continue on error (vs fail fast)
    "errors.deadletterqueue.topic.name": "orders-sink-dlq",
    "errors.deadletterqueue.context.headers.enable": "true"
  }
}'

S3 Sink Connector — data lake archiving

The S3 Sink Connector batches Kafka records and uploads them to S3 in configurable formats (Parquet, Avro, JSON, CSV). Files are written to partitioned paths (/topic/year/month/day/hour/file) for Athena/Hive-compatible partitioning. The flush.size and rotate.interval.ms settings control when a file is closed and uploaded. This is the standard pattern for landing Kafka events into a data lake.

JSON — S3 Sink Connector with Parquet format and time-based partitioning

# S3 Sink Connector
curl -X POST http://connect:8083/connectors -H 'Content-Type: application/json' -d '{
  "name": "orders-s3-sink",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "4",
    "topics": "orders",
    "s3.region": "eu-west-1",
    "s3.bucket.name": "my-data-lake",
    "s3.part.size": "5242880",       // 5 MB multipart upload part size
    "topics.dir": "raw/kafka",
    "flush.size": "10000",           // write file every 10 000 records
    "rotate.interval.ms": "300000",  // or every 5 minutes
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "parquet.codec": "snappy",
    "locale": "en_US",
    "timezone": "UTC",
    "timestamp.extractor": "RecordField",
    "timestamp.field": "created_at",
    // Partitioning: s3://bucket/raw/kafka/orders/year=2024/month=01/day=15/
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "path.format": "YYYY/MM/dd/HH",
    "partition.duration.ms": "3600000"  // hourly partitions
  }
}'

Dead-letter queuing and error handling

By default, a single bad record halts the connector (errors.tolerance=none). errors.tolerance=all skips bad records and routes them to a DLQ topic. The DLQ record includes error context in headers: connector name, task ID, error message, and the stage (key/value conversion, transformation, or write). Monitor the DLQ topic for schema mismatches and data quality issues.

JSON + CLI — DLQ configuration, header inspection, and status check

# Common DLQ configuration (applicable to any sink connector)
"errors.tolerance": "all",
"errors.deadletterqueue.topic.name": "my-sink-dlq",
"errors.deadletterqueue.topic.replication.factor": "3",
"errors.deadletterqueue.context.headers.enable": "true",
# ↑ Adds headers to DLQ records:
#   connect.errors.topic          — source topic
#   connect.errors.partition       — source partition
#   connect.errors.offset          — source offset
#   connect.errors.connector.name — connector that failed
#   connect.errors.exception.message — error message

# Monitor DLQ for data quality issues
kafka-console-consumer.sh \
  --bootstrap-server broker:9092 \
  --topic my-sink-dlq \
  --from-beginning \
  --property print.headers=true

# Connector status check
curl http://connect:8083/connectors/orders-jdbc-sink/status
# {"name":"orders-jdbc-sink","connector":{"state":"RUNNING"},
#  "tasks":[{"id":0,"state":"RUNNING"},{"id":1,"state":"RUNNING"}]}

Key Points to Remember

1Sink connectors read from Kafka and write to external targets — offsets are committed by the Connect framework, not the connector
2JDBC Sink supports INSERT/UPSERT modes; pk.mode=record_key uses the Kafka record key as the primary key
3S3 Sink batches records into Parquet/Avro files; flush.size and rotate.interval.ms control file roll-over
4Time-based partitioning creates Hive-compatible S3 paths (year/month/day/hour) for efficient Athena/Spark queries
5errors.tolerance=all + DLQ prevents a single bad record from halting the connector — DLQ holds failed records with error headers
6Monitor DLQ topic depth as a data quality metric — growing DLQ indicates schema mismatches or target availability issues

Interview Questions

What is the difference between a source connector and a sink connector in Kafka Connect?

EasyConfluent

How would you configure a JDBC Sink Connector to upsert records based on a primary key?

MediumThoughtworks

What is the dead-letter queue pattern in Kafka Connect and when should you use errors.tolerance=all?

MediumAmazon

How does the S3 Sink Connector partition files for Athena/Hive-compatible queries?

HardNetflix

A sink connector task is in FAILED state. How would you diagnose the issue?

MediumUber

Ask Aria about Sink Connectors

Your personal AI tutor — ask anything about this concept

Revision Status

Personal Notes

Discussion

Loading discussion…

Source Connectors & Debezium CDC

Log Compaction