Source vs Sink Connector #

In the Apache Kafka ecosystem, moving data efficiently between platforms requires deep understanding of data flow direction. Kafka Connect simplifies this process by providing two main categories of integration plugins: Source Connectors and Sink Connectors. Although both run under the same worker runtime and use similar internal objects, their operational principles, offset management methods, fault tolerance, and data rate handling (backpressure) are very different. This article will dissect in detail the comparison between Source and Sink Connectors, explain each one’s internal mechanisms, compare how offset coordinates are stored, and unpack how data surge handling is done asynchronously in production.

Basic Source Connector Concepts #

Source Connectors act as ingestion agents pulling information from external systems and publishing it into Apache Kafka topics. These external systems can be transactional databases (like PostgreSQL, Oracle, or SQL Server), legacy message queue systems (like IBM MQ or ActiveMQ), local file systems, or even real-time data streams from IoT sensors or third-party APIs.

Source Connector Internal Workflow #

Architecturally, the data writing workflow from external systems to Kafka through a Source Connector can be described as an asynchronous processing chain:

1. Data Polling: The Connect Source Task runs in an infinite loop calling the poll() method periodically to fetch data from the source system.
2. SourceRecord Creation: Each database row, queue message, or log line obtained is then converted by the connector class into an internal Java object called SourceRecord.
3. SMT Application (Single Message Transforms): If configured, the SourceRecord object is sent to the in-memory transformation chain for dynamic modification (like column renaming or data filtering).
4. Serialization via Converter: The cleaned SourceRecord is translated by the Converter (for example AvroConverter or JsonConverter) into a binary representation (byte array).
5. Publish to Broker: The Connect Worker acts as a Kafka Producer sending that binary payload to the Kafka broker using the standard Producer API.

The main goal of a Source Connector is ensuring every newly appearing piece of data in the source system is sent to Kafka with the smallest possible latency without burdening the source system itself.

Basic Sink Connector Concepts #

Conversely, Sink Connectors act as egress agents reading data from one or several Apache Kafka topics and writing it into external storage systems. These target systems are usually search databases (Elasticsearch/OpenSearch), big data analytics systems (ClickHouse, Snowflake, Google BigQuery), cloud object storage (AWS S3, Azure Blob Storage), or operational relational/NoSQL databases.

Sink Connector Internal Workflow #

Sink Connectors operate in the opposite direction of Source Connectors, leveraging standard Kafka consumption functionality:

1. Consumer Fetch: The Connect Worker acts as a Kafka Consumer subscribing to specific topics. The Worker calls the Kafka Consumer API’s poll() method to pull binary bytes from the broker.
2. Deserialization via Converter: The binary bytes pulled from the Kafka broker are translated by the Converter back into an internal Java object called SinkRecord containing the original data schema and values.
3. SMT Application: SMTs process the SinkRecord object in memory to manipulate or add data before handing it to the connector.
4. Buffer & Batching: The Sink Connector Task collects SinkRecord objects into a local memory buffer to do write grouping (batching) so the target system isn’t burdened with too-frequent small transactions.
5. Write to Target: The Task calls the put() method to write the collection of data records to the external storage system (for example running bulk SQL INSERT queries or sending documents via the Elasticsearch Bulk HTTP API).

In Sink mode, the architecture’s main focus is maintaining data delivery continuity and handling target system failures without damaging Kafka offset read order.

Offset Management Comparison #

One of the most fundamental differences between Source and Sink Connectors lies in how they manage their work progress status (offset management). Offsets are pointer coordinates for the last data position successfully processed.

flowchart TD
    subgraph SourceOffsetTracking["Source Offset Tracking"]
        direction TB
        SourceDB[("Source System")] -->|Read data at position X| SrcTask["Source Task"]
        SrcTask -->|Send record + Position X Metadata| WorkerProducer["Worker Producer"]
        WorkerProducer -->|Write data| DataTopic(("Kafka Data Topic"))
        WorkerProducer -->|Record Position X asynchronously| OffsetTopic(("connect-offsets Topic"))
    end
    
    subgraph SinkOffsetTracking["Sink Offset Tracking"]
        direction TB
        ConsumerPoll["Worker Consumer"] -->|Fetch data from Offset Y| DataTopic
        ConsumerPoll -->|Send SinkRecord| SnkTask["Sink Task"]
        SnkTask -->|Write successfully to target| TargetStorage["Target Storage"]
        TargetStorage -.->|Verify success| SnkTask
        SnkTask -->|Commit Offset Y to Kafka| CommitTopic(("__consumer_offsets Topic"))
    end

1. Offset Management on Source Connectors #

On Source Connectors, Kafka doesn’t know the source system’s internal data structure. For example, if we read database table rows, the offset coordinates might be the incrementing_id column value or the last_modified timestamp. If we read text files, the offset is the file cursor position byte index.

• Custom Offset Format: Kafka Connect allows Source Connectors to define their own offset schemas in key-value format. The Key defines the source partition (for example {"table": "orders"} or {"file": "logs.txt"}), and the Value defines the source offset position (for example {"id": 1024} or {"byte_offset": 4560}).
• Storage Location: These custom offset coordinates are periodically written asynchronously by the Connect Worker into a special internal Kafka topic named connect-offsets. The log cleanup policy on this topic is set to compact so every source partition’s last position is persistently maintained.

2. Offset Management on Sink Connectors #

On Sink Connectors, the offset coordinates are standard Kafka partition offsets (linear 64-bit integers showing message positions inside a Kafka topic partition).

• Standard Offset Format: The offsets used are standard Kafka Topic-Partition-Offset pairs.
• Storage Location: Because the Connect Worker acts as a regular Kafka consumer, consumer offsets are managed directly by the Kafka broker through the Group Coordinator mechanism and stored in the internal __consumer_offsets system topic.
• Commit Synchronization: Offset commits to the Kafka broker only happen after the Sink Task confirms all records in the latest put() batch call were successfully written to the external target system. If the target system fails to receive data (for example a network timeout), the offset isn’t committed, and the same records are resent by Kafka for reprocessing (At-Least-Once delivery semantics).

Serialization and Deserialization Mechanism Details in Source & Sink Flows #

To understand how data is physically manipulated, we need to see how binary objects are processed at the converter level. The decoupling model in Kafka Connect guarantees binary writing and schema reading logic run modularly.

1. Data Journey on the Source Side #

When a Source Task pulls a database table row, the task assembles the data structure into a SourceRecord object.

• Stage 1: Java Heap representation: This object carries a defined Schema (like VARCHAR data types mapped to Java strings, INT mapped to Integer) and the actual Value (like "Rudi" and 25).

• Stage 2: Serialization by Converter: If we set value.converter=org.apache.kafka.connect.json.JsonConverter with schemas.enable=true, the binary payload output sent to Kafka carries the entire schema metadata structure in every message:

  {
    "schema": {
      "type": "struct",
      "fields": [
        {"type": "string", "optional": false, "field": "name"},
        {"type": "int32", "optional": true, "field": "age"}
      ],
      "optional": false
    },
    "payload": {
      "name": "Rudi",
      "age": 25
    }
  }
  

  This is an anti-pattern for large-scale architectures because schema data is repeated endlessly, wasting up to 90% of disk space.

  Conversely, if we use AvroConverter, the schema is registered to the Confluent Schema Registry. The schema is stored once in the registry, and the binary sent to Kafka is only: [1-byte Magic Byte] + [4-byte Schema ID] + [Avro Binary Payload (without column names)] This dramatically saves network bandwidth and heap memory.

2. Data Journey on the Sink Side #

When a Sink Task receives a message from the Kafka broker:

• Stage 1: Binary Reading: The Connect Worker takes raw bytes from the Kafka topic.
• Stage 2: Deserialization by Converter: The Converter detects the data format. If using AvroConverter, it reads the Schema ID from the first 5 message bytes, downloads that schema from the Schema Registry (if it’s not already in the worker’s local memory cache), then decodes the binary into a Java SinkRecord object.
• Stage 3: Handover to Connector: The reconstructed SinkRecord object is handed to the Sink Connector to be written to the destination database using syntax compatible with that database.

Backpressure Handling on Sink Connectors #

Backpressure is a distributed system defense mechanism controlling data flow rates when the receiver system can’t keep up with the data rate sent by the sender system. This scenario happens very often on Sink Connectors where the Kafka broker’s message emission speed far exceeds the target database’s disk write speed.

How Does Kafka Connect Handle Backpressure? #

By default, the Kafka consumer inside the Connect Worker uses a pull model that inherently supports very dynamic backpressure coordination. When the external database experiences slowness (for example a PostgreSQL database running an intensive vacuuming process or Elasticsearch at 100% CPU usage):

1. Delayed put() Execution: The Sink Connector Task detects write slowdowns because connection threads or database pools are held waiting for responses. As a result, the put() function call inside the Connect task takes longer to complete.
2. Partition Polling Suspension: The Connect Worker monitors task execution health. If the internal queue buffer for that task is full because previous put() executions haven’t finished, the worker asynchronously calls the Kafka consumer suspension API, KafkaConsumer.pause(Collection<TopicPartition> partitions).
3. Temporary Fetching Stop: The Kafka broker stops sending new data for those paused partitions. No extra JVM heap memory is wasted storing queued messages.
4. Polling Resume: Once the target database returns to normal and the task successfully completes the pending data batch writes, the worker calls KafkaConsumer.resume(Collection<TopicPartition> partitions) to start pulling data from Kafka normally again.

Important Parameters for Backpressure Tuning #

To optimize Sink Connector resilience from OutOfMemoryError crashes when facing data surges, we must carefully configure consumption parameters in the connector properties file:

# Maximum number of records fetched in one poll() call
# Default value: 500. Reduce if the per-message payload is very large.
consumer.max.poll.records=200

# Maximum time limit for the task to process data before being considered dead
# If the target database writes slowly and exceeds this limit, 
# the broker considers the Connect Worker dead and triggers a group rebalancing process.
consumer.max.poll.interval.ms=300000

# HTTP/Socket connection wait time to the target system before throwing an Exception timeout
connection.timeout.ms=10000

If the target database often needs batch write times exceeding 5 minutes, we must raise consumer.max.poll.interval.ms so the Connect cluster doesn’t experience a pointless rebalance storm from tasks being falsely considered dead.

Failure Handling Scenarios #

Distributed systems are prone to component failures. The role separation between Source and Sink Connectors determines how fault tolerance is handled.

1. Source Connector Failures #

• Problem: Database connection drops, polling queries time out, or database credentials expire.
• Handling Mechanism:
  • The Source Task tries contacting the database again if configured with connection tolerance. If it fails completely, the task enters FAILED status.
  • Because data in the source database is persistent, no data is lost during task downtime.
  • After the connection problem is fixed, we can send the REST API command POST /connectors/{name}/restart to restart the task. The task reads the last offset from the connect-offsets topic and safely continues data movement.

2. Sink Connector Failures #

• Problem: The Elasticsearch target storage is full (disk space full), API key authentication expires, or data format violates SQL database table schema rules (for example constraint violations).
• Handling Mechanism:
  • If a binary/network error happens, the task throws an exception and stops. The Kafka offset isn’t committed to the broker.
  • If the problem comes from corrupted message formats (poison pills), stopping the entire pipeline harms other valid data. For that, we need to configure special fault tolerance on the Sink side:

    # Continue processing even if corrupted records are found
    errors.tolerance=all
    # Send corrupted records to a Dead Letter Queue (DLQ)
    errors.deadletterqueue.topic.name=orders-sink-dlq
    errors.deadletterqueue.context.headers.enable=true
    

  • This ensures valid data still reaches the target, while problematic data is separated to the DLQ topic for developer team analysis without jamming business processes.

Real Configuration Examples and Parameter Dissection #

To provide comprehensive understanding, let’s dissect example configuration files for each connector type.

1. Source Connector Configuration Example (MySQL JDBC) #

Below is a MySQL Source Connector configuration for pulling transaction data into Kafka:

{
  "name": "mysql-sales-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "2",
    "connection.url": "jdbc:mysql://mysql-server:3306/sales_db",
    "connection.user": "connect_user",
    "connection.password": "Password123!",
    "table.whitelist": "orders,customers",
    
    "//": "Uses the incrementing ID column to detect new rows",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    
    "//": "Each table is written to a separate topic with the mysql-db- prefix",
    "topic.prefix": "mysql-db-",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }
}

• Parameter Dissection:
  • tasks.max: Determines the parallel thread count (maximum 2). Because there are 2 tables (orders and customers), each task reads 1 table in parallel.
  • mode: The incrementing setting tells the connector to track new rows based on primary key (id) value increments. The last largest ID coordinate is continuously recorded in the connect-offsets topic.

2. Sink Connector Configuration Example (Elasticsearch Sink) #

Below is an Elasticsearch Sink Connector configuration for writing data from Kafka into an Elasticsearch index:

{
  "name": "elasticsearch-sales-sink",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "3",
    "topics": "mysql-db-orders",
    "connection.url": "http://elasticsearch-node:9200",
    
    "//": "Uses Kafka offsets to maintain document uniqueness in ES",
    "key.ignore": "false",
    "schema.ignore": "true",
    
    "//": "Sets batching to optimize Elasticsearch I/O performance",
    "batch.size": "1000",
    "flush.timeout.ms": "5000"
  }
}

• Parameter Dissection:
  • tasks.max: Can be set up to 3 if the mysql-db-orders topic has 3 or more partitions. If the topic partition count is only 1, the 2nd and 3rd tasks sit idle.
  • key.ignore: Setting false means the Kafka message key is used as the Elasticsearch document ID. This is important for preventing data duplication if messages are resent from network crashes (idempotent updates).

Comprehensive Source vs Sink Comparison Table #

Summary #

• Flow Direction — Source Connectors act as data producers pulling information from outside into Kafka, while Sink Connectors act as data consumers pushing data from Kafka out.
• Custom vs Standard Offsets — Source Connectors use custom key-value offsets stored in the connect-offsets topic, while Sink Connectors use standard partition offsets in the __consumer_offsets topic.
• Commit Mechanism — Sink Connectors only commit offsets after the target system confirms successful data writes to maintain the At-Least-Once guarantee.
• Backpressure Automation — Sink Connectors use the Kafka Consumer pull model and pause/resume APIs to temporarily stop data flow when the target system is overloaded.
• Parameter Tuning — The consumer.max.poll.interval.ms and consumer.max.poll.records parameters must be adjusted to the target system’s throughput characteristics to prevent unnecessary task restarts.

← Previous: What is Kafka Connect? Next: Standalone vs Distributed Mode →