Sink connectors save weeks: patterns, pitfalls and fixes

The case for built-in connectors

We've all been tempted to write a custom Kafka consumer that reads events and pushes them somewhere—a database, S3, Elasticsearch. The problem is that sink connectors save weeks of development and maintenance by handling the hard parts: retries, exactly-once semantics, offset management, dead-letter queues, and monitoring. Sink connectors come with production-grade patterns, pitfalls and fixes built in.

When you write custom code to move data from Kafka to external systems, you inherit problems that sink connectors already solved. Our team learned this the hard way across e-commerce and trading platforms. By the time we implemented distributed tracing, backpressure handling, and connection pooling, we'd spent what a mature connector framework had already provided. This article covers when and how to use three essential sinks—S3, JDBC, and Elasticsearch—and how to keep them healthy in production.

S3 sink for data lakes

Data lakes live on object storage. S3 is the obvious choice, and the Confluent S3 sink connector (or open-source alternatives like Aiven's S3 connector) handles the mechanics of batching, partitioning, and fault tolerance so you don't have to write a consumer that uploads to S3.

Real-world scenario: An e-commerce platform ingests checkout events—order ID, timestamp, customer segment, total amount—into a Kafka topic orders-events. You want to land those events in S3 for analytics and machine learning teams to query with Athena or BigQuery federation.

A naive custom consumer would:

• Buffer events in memory until a batch threshold is hit.
• Serialize and compress them.
• Upload to S3 with hardcoded retry logic.
• Lose data if the process crashes mid-batch.
• Provide no visibility into how many events are pending.

The S3 sink connector does all this:

{
  "name": "s3-orders-sink",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "topics": "orders-events",
    "s3.bucket.name": "data-lake-prod",
    "s3.region": "us-east-1",
    "s3.part.size": 5242880,
    "flush.size": 10000,
    "rotate.interval.ms": 3600000,
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "path.format": "yyyy/MM/dd/HH",
    "timestamp.extractor": "RecordField",
    "timestamp.field": "event_ts",
    "s3.compression.type": "snappy",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "orders-events-dlq",
    "tasks.max": 4
  }
}

This configuration:

• Partitions by hour on disk, making downstream queries efficient.
• Compresses with Snappy for storage cost reduction.
• Routes unparseable records to a dead-letter queue instead of crashing.
• Scales to four parallel tasks, each handling a subset of partitions.
• Flushes every 10,000 records or 1 hour, whichever comes first.

The connector handles offset commits safely: it only advances Kafka offsets after S3 writes succeed. If the task fails, it resumes from the last committed offset. Your analytics team gets reliable, queryable Parquet files partitioned by date—with almost no custom code.

JDBC sink for databases

Many teams need to replicate Kafka events into operational databases: order status into a PostgreSQL write-replica, user profile updates into MySQL, trading positions into a data warehouse. The JDBC sink connector (part of Confluent Platform or available open-source) manages connection pooling, batching, and idempotent writes.

Real-world scenario: A ride-hailing service publishes driver-location updates to topic driver-locations. A search and dispatch service needs these locations in PostgreSQL to answer "drivers within 5 miles" queries. A naive consumer would open a connection per event, execute an INSERT or UPDATE, and close—burning through connection limits and database CPU.

The JDBC sink connector batches and reuses connections:

{
  "name": "jdbc-driver-locations-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "topics": "driver-locations",
    "connection.url": "jdbc:postgresql://db-primary:5432/dispatch",
    "connection.user": "kafka_sink",
    "connection.password": "${JDBC_PASSWORD}",
    "table.name.format": "driver_locations",
    "insert.mode": "upsert",
    "pk.fields": "driver_id",
    "pk.mode": "record_key",
    "batch.size": 500,
    "connection.attempts": 3,
    "connection.backoff.ms": 1000,
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "driver-locations-dlq",
    "max.retries": 3,
    "retry.backoff.ms": 100,
    "tasks.max": 2
  }
}

Key details:

• pk.mode: record_key tells the connector to extract the primary key (driver ID) from the Kafka message key, not from the value.
• insert.mode: upsert means "insert if not present, update if it is"—idempotent by design.
• batch.size: 500 groups writes into batches, dramatically reducing round-trips and database load.
• Connection pooling is handled internally; the connector maintains a pool and reuses connections across batches.
• Failed writes go to the dead-letter queue; the connector keeps running rather than crashing on a single bad record.

The driver-locations table stays fresh without you writing SQL or managing connection pools. If a write fails due to a constraint violation (e.g., a driver ID that doesn't exist in the drivers table), it's logged and sent to the DLQ, not lost.

Elasticsearch sink for search

Product catalogs, logs, documents—anything searchable often lands in Elasticsearch. The Elasticsearch sink connector (open-source or Confluent) handles index creation, document routing, and bulk indexing.

Real-world scenario: A travel platform ingests hotel listings—name, location, amenities, star rating, price—into topic hotel-inventory. The search API queries Elasticsearch to show results sorted by rating and price. Each update should hit Elasticsearch within seconds for users to see new availability.

Elasticsearch sink configuration:

{
  "name": "es-hotel-inventory-sink",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "topics": "hotel-inventory",
    "connection.url": "https://elasticsearch-prod:9200",
    "connection.username": "kafka_sink",
    "connection.password": "${ES_PASSWORD}",
    "type.name": "_doc",
    "index.name.format": "hotels-${topic}-v1",
    "topic.index.map": "hotel-inventory:hotels-prod-v1",
    "key.ignore": false,
    "key.field": "hotel_id",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": false,
    "batch.size": 1000,
    "linger.ms": 100,
    "max.in.flight.requests": 5,
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "hotel-inventory-dlq",
    "tasks.max": 3
  }
}

Why this works:

• key.field: hotel_id ensures each hotel document uses its ID as the Elasticsearch document ID. Updates to the same hotel overwrite the old document.
• batch.size: 1000 groups indexing requests, reducing API calls.
• linger.ms: 100 waits up to 100ms for more records before flushing a partial batch, balancing latency and throughput.
• max.in.flight.requests: 5 allows up to 5 concurrent bulk requests, parallelizing writes without overwhelming Elasticsearch.
• Index name is templated per topic, making it easy to spin up new indices for new event types.

Your search API queries hotels-prod-v1 and sees fresh data. When Elasticsearch is briefly unavailable, the connector queues writes and retries. There's no custom code managing connection pools, retries, or bulk request batching.

Monitor connector task failures

Connectors aren't "set and forget." Tasks fail, and if you don't notice, data silently stops flowing. Monitoring is essential.

Key metrics to watch:

• Task status: Is each task RUNNING or FAILED?
• Offset lag: How far behind the connector is relative to the topic's latest offset.
• Dead-letter queue size: Are records accumulating in the DLQ, indicating a systemic parsing or validation issue?
• Connector restart rate: Does the task restart frequently, suggesting flaky external connectivity?
• Batch latency: How long does each batch take? Spikes indicate resource contention or network issues.

Kafka Connect exposes metrics via JMX. Hook them into Prometheus and Grafana:

// Prometheus scrape config for Kafka Connect
{
  "job_name": "kafka-connect",
  "scrape_interval": "15s",
  "static_configs": [
    {
      "targets": ["kafka-connect:8080"]
    }
  ]
}

// Key metrics to alert on
kafka_connect_connector_status_connector_status == 2  // FAILED
kafka_connect_sink_task_put_batch_total_records_processed < threshold  // No progress
kafka_connect_sink_task_offset_lag_max > 100000  // Falling behind

Set up alerts for task failures, large offset lags, and DLQ growth. Many teams run a simple health-check dashboard: Is each critical sink task RUNNING, and is lag under 100K records? A single pager alert catches the common case.

Common mistakes

• Ignoring the dead-letter queue: Configure errors.deadletterqueue.topic.name so unparseable records don't stop the connector. Monitor that queue; if it grows, investigate schema mismatches or corrupt data upstream.
• Over-tuning batch size without testing: A large batch size reduces API calls but increases memory and latency. Start conservative (500–1000), measure, and adjust.
• Not committing offsets safely: Connectors commit offsets after writes succeed. If you restart the connector, it resumes where it left off. But if the external system (S3, database, Elasticsearch) is slow, lag builds. Monitor it.
• Mixing schema evolution with rigid connectors: If your Kafka messages add new fields over time, ensure your database tables or Elasticsearch indices can handle them. Use schema registry and evolve schemas carefully.
• Running single-threaded sinks on high-volume topics: By default, connectors run one task. For topics with millions of events per minute, increase tasks.max and let the connector partition work across tasks.
• Forgetting to rotate credentials: JDBC passwords and Elasticsearch credentials are often hardcoded. Use secrets management (AWS Secrets Manager, Vault) and rotate regularly.

Key takeaways

• Sink connectors handle retries, batching, offset management, and dead-letter queues—the hard parts of moving data reliably. Writing custom consumers for standard destinations costs weeks of development and ongoing maintenance.
• S3 sinks partition data by time and compress for analytics. Elasticsearch sinks keep search indices fresh. JDBC sinks replicate into databases with idempotent upserts and connection pooling.
• Configure error tolerance and dead-letter queues. A single unparseable record shouldn't crash your pipeline; route it and keep running.
• Monitor task status, offset lag, and dead-letter queue growth. Connectors fail silently; metrics and alerts catch problems early.
• Batch size, parallel tasks, and linger intervals are levers you tune based on your throughput and latency SLOs. Start with sensible defaults, measure, and adjust.