Definition
Source and sink connectors are the two directions of Kafka Connect.
- Source Connector – reads from an external system (database, file, API) and writes records into Kafka topics; behaves like a managed producer.
- Sink Connector – reads from Kafka topics and writes records to an external system (S3, database, Elasticsearch); behaves like a managed consumer.
- Source offset – the connector's own position in the external system (e.g., last row ID), stored by Connect in Kafka.
- Sink offset – the standard Kafka consumer offset; committed only after the write to the external system succeeds.
- Converter – applied symmetrically: source serializes into the wire format, sink deserializes from it.
Think of Kafka as the central sorting office at National Bank of India. A source connector is the intake clerk who walks out to the account database, collects new records, and brings them into the sorting office. A sink connector is the dispatch clerk who takes sorted records out of the office and files them into cloud storage or a search index. The intake clerk fills the office; the dispatch clerk empties it to somewhere useful.
Understanding the Concepts
What is a source connector?
A source connector is on the import side. It talks to an external system and turns its data into Kafka records. Internally it implements a poll method that returns a batch of source records, and the Connect runtime produces those into topics.
{
"name": "accounts-jdbc-source",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"connection.url": "jdbc:postgresql://db-1:5432/banking",
"table.whitelist": "accounts",
"mode": "timestamp+incrementing",
"timestamp.column.name": "updated_at",
"incrementing.column.name": "id",
"topic.prefix": "db.",
"tasks.max": "3"
}
}
# Source connector reads accounts table
# New/updated rows -> produced to topic db.accounts
# Source offset = last id + timestamp read, stored in Kafka
# On restart: resumes from last offset, no re-import
What is a sink connector?
A sink connector is on the export side. It reads records from Kafka topics and writes them to an external system. It writes first, then commits the Kafka consumer offset, so a crash never loses records (at-least-once).
{
"name": "accounts-s3-sink",
"config": {
"connector.class": "io.confluent.connect.s3.S3SinkConnector",
"topics": "db.accounts",
"s3.bucket.name": "nbi-data-lake",
"s3.region": "ap-south-1",
"flush.size": "1000",
"format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
"tasks.max": "3"
}
}
# Sink connector reads from db.accounts topic
# Writes 1000 records as a Parquet file to S3
# Commits Kafka offset ONLY after file is safely written
# Crash before commit -> re-read and re-write (at-least-once)
How do source and sink track different offsets?
A source tracks its position in the external system (e.g., last row ID). A sink tracks its position in Kafka (consumer offset). Both are managed by the Connect runtime.
| Aspect | Source Connector | Sink Connector | |---|---|---| | Direction | External -> Kafka | Kafka -> External | | Kafka role | Destination | Origin | | Behaves like | Managed producer | Managed consumer | | Offset type | Source offset (row ID, timestamp) | Kafka consumer offset | | Offset storage | Internal Kafka topic | Standard consumer group | | Resume on restart | From last external position | From last Kafka offset |
# Source offset stored in:
# connect-offsets topic (internal)
# Sink offset stored in:
# __consumer_offsets topic (standard Kafka consumer group)
How do you chain a source and sink through Kafka?
Chaining a source and a sink bridges two systems through Kafka with no custom code.
# Pipeline: PostgreSQL -> Kafka -> S3
# Source: JDBC Source reads accounts table -> db.accounts topic
# Sink: S3 Sink reads db.accounts topic -> S3 Parquet files
# Pipeline: PostgreSQL -> Kafka -> Elasticsearch
# Source: JDBC Source reads accounts table -> db.accounts topic
# Sink: Elasticsearch Sink reads db.accounts topic -> accounts index
# Both sinks read the SAME topic independently with their own offsets
Why must sink targets be idempotent?
Connect sinks are at-least-once. A crash after writing but before committing the Kafka offset causes a re-read and re-write. The target must tolerate duplicates.
package com.bank.kafka;
public class SinkIdempotencyDemo {
public static void main(String[] args) {
// JDBC Sink with upsert mode: idempotent
System.out.println("JDBC Sink: INSERT ON CONFLICT (id) DO UPDATE");
System.out.println("Duplicate write for SAV2001 -> updates existing row (safe)");
// Elasticsearch Sink: document ID = Kafka key
System.out.println("ES Sink: document _id = SAV2001");
System.out.println("Duplicate write for SAV2001 -> overwrites document (safe)");
}
}
JDBC Sink: INSERT ON CONFLICT (id) DO UPDATE
Duplicate write for SAV2001 -> updates existing row (safe)
ES Sink: document _id = SAV2001
Duplicate write for SAV2001 -> overwrites document (safe)
How do converters apply symmetrically?
The source serializes records into the chosen wire format (JSON, Avro, Protobuf), and the sink deserializes from that same format. Both ends must agree.
# Source connector produces in Avro format
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://schema-registry:8081
# Sink connector consumes Avro format (same converter)
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://schema-registry:8081
What about the right source polling mode?
Choosing the correct source mode prevents missed or duplicated rows. For databases, timestamp-plus-incrementing is the safest because it handles both new inserts and updates to existing rows.
# Mode options for JDBC Source:
# bulk: re-read entire table every poll (heavy, simple)
# incrementing: new rows only (misses updates to existing rows)
# timestamp: changed rows (risks collision if timestamps duplicate)
# timestamp+incrementing: best of both (handles inserts and updates safely)
What should a senior engineer know?
Source connectors do not capture deletes unless they use CDC (Debezium). JDBC Source polls by query, which misses deleted rows. Converters must match on both sides. Single-message transforms can reshape records in either direction. The operational model is that a source and multiple sinks reading the same topic give you one-to-many data distribution with no code.
# Architecture:
# JDBC Source -> db.accounts topic -> S3 Sink (archive)
# -> ES Sink (search)
# -> JDBC Sink (reporting DB)
# One source, many sinks, no custom code
Visual Overview
Source vs Sink Connector
Source Connector (external -> Kafka):
┌──────────────┐ ┌────────────┐
│ PostgreSQL │ poll() │ Kafka │
│ accounts │────────→│ db.accounts│
│ table │ source │ topic │
└──────────────┘ offset └────────────┘
tracked
Sink Connector (Kafka -> external):
┌────────────┐ ┌──────────────┐
│ Kafka │ put() │ S3 Bucket │
│ db.accounts│────────→│ Parquet files│
│ topic │ consumer│ │
└────────────┘ offset └──────────────┘
committed
AFTER write
Chained Pipeline:
PostgreSQL ──→ Source ──→ Kafka ──→ Sink-1 ──→ S3
──→ Sink-2 ──→ Elasticsearch
──→ Sink-3 ──→ Reporting DB
Key Takeaway
A source connector and a sink connector are the two directions of Kafka Connect. A source reads from an external system and writes into Kafka, behaving like a managed producer. A sink reads from Kafka and writes out to an external system, behaving like a managed consumer. The two track different kinds of offset: a source tracks its position in the external system, and a sink tracks its Kafka consumer offset.
The critical detail for sink connectors is that they write to the target first and commit the Kafka offset only after the write succeeds, giving at-least-once delivery. This means the target must tolerate duplicates, typically through upsert semantics keyed by a primary key.
At National Bank of India, Pankaj chained a JDBC source and two sinks off the same topic: an Elasticsearch sink for instant account search and a JDBC sink for the reporting database, both reading independently with their own offsets. Because the reporting sink upserted by account ID, a redelivery of Rajesh's record was a harmless overwrite.
Remember one sentence: source produces into Kafka, sink consumes from Kafka, both are config-only, and the sink must be idempotent because Connect is at-least-once.
---