2.1 – Connector Model: Source vs Sink

Overview

This module is the authoritative explanation of Source and Sink connector semantics. By the end, you will understand:

• How Source connectors pull data into Kafka.
• How Sink connectors push data out of Kafka.
• How offset management differs between Source and Sink.
• Pull vs. push integration patterns.
• Delivery guarantees and exactly-once limitations.

No other module re-explains these concepts. Subsequent modules will reference this module when discussing Source/Sink behavior.

The Connector Model: Source vs. Sink

Kafka Connect uses a bidirectional model for data integration:

• Source Connectors: Import data into Kafka from external systems.
• Sink Connectors: Export data from Kafka to external systems.

External System ──[Source Connector]──► Kafka ──[Sink Connector]──► External System

Key Insight: Connectors are not bidirectional. A single connector is either a Source or a Sink, never both. If you need data to flow in both directions, you deploy two separate connectors.

Source Connectors

Purpose

Source connectors continuously poll external systems for new or updated data and produce that data into Kafka topics.

Typical Use Cases

• Database CDC (Change Data Capture): Stream database changes (inserts, updates, deletes) into Kafka.
  • Example: Debezium connectors for MySQL, PostgreSQL, MongoDB.
• Log File Ingestion: Tail log files and produce each line as a Kafka record.
  • Example: Syslog connector, Filebeat.
• Cloud Storage: Read files from S3, GCS, or Azure Blob Storage.
  • Example: S3 Source Connector.
• Message Queue Integration: Pull messages from RabbitMQ, ActiveMQ, etc.
  • Example: JMS Source Connector.

Data Flow

External System (e.g., Database)
    │
    ├─ SELECT * FROM users WHERE id > last_offset
    │
    ▼
Source Task (poll loop)
    │
    ├─ Convert rows to SourceRecords
    │
    ▼
Kafka Producer (internal to Connect)
    │
    ▼
Kafka Topic

Key Responsibilities

1. Polling: Source tasks periodically query the external system for new data.
2. Transformation: Convert external data formats (e.g., database rows, JSON objects) into Connect's SourceRecord format.
3. Producing: Use an internal Kafka producer to write records to Kafka.
4. Offset Tracking: Store source offsets (e.g., database row ID, file position) to resume from the correct point after restarts.

Offset Handling in Source Connectors

What is a Source Offset?

A source offset is a marker that tracks the position in the external system where the connector last read data.

Examples:

• Database (Incrementing Mode): {"table": "users", "id": 12345}
• Database (Timestamp Mode): {"table": "orders", "last_modified": "2023-10-27T10:30:00Z"}
• File: {"file": "/var/log/app.log", "position": 56789}
• Kafka Topic (MirrorMaker 2): {"topic": "source-topic", "partition": 0, "offset": 1000}

Where Are Offsets Stored?

In Distributed Mode, source offsets are stored in an internal Kafka topic:

• Topic Name: connect-offsets (configurable via offset.storage.topic)
• Format: Key-value pairs where:
  • Key: Connector name + partition (logical partition within the source).
  • Value: Offset data (e.g., last row ID, file position).

In Standalone Mode, offsets are stored in a local file (offset.storage.file.filename).

Offset Commit Behavior

• When Offsets Are Committed: After records are successfully produced to Kafka and acknowledged by the broker.
• Frequency: Controlled by offset.flush.interval.ms (default: 60 seconds) and offset.flush.timeout.ms.

Critical Point: If a source task crashes before offsets are committed, it will reprocess records from the last committed offset, leading to at-least-once delivery.

Source Connector Example: JDBC Source

Scenario

Stream data from a PostgreSQL users table into Kafka topic postgres-users.

Configuration

{
  "name": "jdbc-source-users",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:postgresql://localhost:5432/mydb",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "table.whitelist": "users",
    "topic.prefix": "postgres-"
  }
}

Behavior

1. Polling: The task queries: SELECT * FROM users WHERE id > <last_offset>.
2. Producing: Each row is converted to a SourceRecord and produced to postgres-users.
3. Offset Commit: After records are acknowledged, the offset {"table": "users", "id": <max_id>} is committed.

Restart Scenario: If the connector restarts, it reads the last committed offset from connect-offsets and resumes from id > <last_offset>.

Sink Connectors

Purpose

Sink connectors continuously consume records from Kafka topics and write them to external systems.

Typical Use Cases

• Database Writes: Insert/update records in PostgreSQL, MySQL, Cassandra, etc.
  • Example: JDBC Sink Connector.
• Search Indexing: Index documents in Elasticsearch, Solr.
  • Example: Elasticsearch Sink Connector.
• Cloud Storage: Write records as files in S3, GCS, Azure Blob Storage.
  • Example: S3 Sink Connector.
• Data Warehouses: Stream data into Snowflake, BigQuery, Redshift.
  • Example: Snowflake Sink Connector.

Data Flow

Kafka Topic
    │
    ▼
Kafka Consumer (internal to Connect)
    │
    ├─ Consume records (batches)
    │
    ▼
Sink Task (put loop)
    │
    ├─ Convert SinkRecords to external format
    │
    ▼
External System (e.g., Database)
    │
    ├─ INSERT INTO orders ...

Key Responsibilities

1. Consuming: Sink tasks consume records from assigned Kafka partitions.
2. Batching: Tasks accumulate records into batches before writing (configurable via consumer.max.poll.records).
3. Transformation: Convert Connect's SinkRecord format into the external system's format.
4. Writing: Execute batch writes to the external system (e.g., database transaction, S3 PUT).
5. Offset Commit: Commit Kafka consumer offsets after successful writes.

Offset Handling in Sink Connectors

What is a Sink Offset?

A sink offset is simply a Kafka consumer offset—the position in the Kafka topic partition where the sink last successfully consumed and processed records.

Example: {"topic": "orders", "partition": 0, "offset": 5000}

Where Are Offsets Stored?

In Distributed Mode, consumer offsets are stored in Kafka's internal __consumer_offsets topic (same as regular Kafka consumers).

In Standalone Mode, offsets are stored in a local file (same as source offsets).

Offset Commit Behavior

• When Offsets Are Committed: After records are successfully written to the external system.
• Commit Strategy:
  • Automatic (Default): Connect commits offsets after put() completes without exceptions.
  • Manual (Advanced): Connectors can override preCommit() to implement custom commit logic.

Critical Point: If a sink task crashes before offsets are committed, it will reprocess records from the last committed offset, leading to at-least-once delivery.

Sink Connector Example: JDBC Sink

Scenario

Stream records from Kafka topic orders into a PostgreSQL orders table.

Configuration

{
  "name": "jdbc-sink-orders",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "3",
    "topics": "orders",
    "connection.url": "jdbc:postgresql://localhost:5432/mydb",
    "insert.mode": "upsert",
    "pk.mode": "record_key",
    "pk.fields": "order_id",
    "auto.create": "true"
  }
}

Behavior

1. Consuming: Each task consumes records from assigned partitions of orders.
2. Batching: Tasks batch records (default: 3000 per poll, controlled by consumer.max.poll.records).
3. Writing: Execute INSERT ... ON CONFLICT UPDATE for each batch.
4. Offset Commit: After successful writes, commit Kafka consumer offsets.

Restart Scenario: If a task crashes before offsets are committed, it will re-consume records from the last committed offset and re-write them (idempotency in the database prevents duplicates).

Pull vs. Push Integration Patterns

Pull (Source Connectors)

Definition: The connector actively polls the external system for data.

Characteristics:

• Control: Connect controls polling frequency and batch size.
• Backpressure: Connect can throttle polling if Kafka is overloaded.
• Stateful: Requires offset tracking to know where to resume.

Examples:

• JDBC Source (polls database tables)
• File Source (tails log files)
• HTTP Source (polls REST API endpoints)

Challenge: Polling frequency must balance latency (faster polling = lower latency) and load (too-frequent polling = high database load).

Push (Sink Connectors)

Definition: The connector passively consumes data pushed from Kafka.

Characteristics:

• Reactive: Data arrival triggers processing.
• Backpressure: Connect can slow consumption if the external system is overloaded (via put() blocking).
• Stateless (Partition Assignment): Kafka's consumer group protocol handles partition assignment.

Examples:

• JDBC Sink (writes to database)
• S3 Sink (uploads files)
• Elasticsearch Sink (indexes documents)

Challenge: Sink connectors must handle external system throttling and temporary failures (e.g., database connection timeout).

Data Movement Patterns

1. Continuous Replication

• Pattern: Source connector continuously polls for new data; sink connector continuously writes.
• Use Case: Real-time database replication (e.g., MySQL → Kafka → PostgreSQL).
• Delivery: At-least-once.

2. Event-Driven Integration

• Pattern: Source connector reacts to external events (e.g., database triggers, file uploads).
• Use Case: CDC (Change Data Capture) from databases.
• Delivery: At-least-once (or exactly-once with transactional sources like Debezium).

3. Batch Ingestion

• Pattern: Source connector periodically ingests large batches of data.
• Use Case: Daily ETL jobs (e.g., load daily sales data from S3).
• Delivery: At-least-once.

4. Fan-Out

• Pattern: Single source connector → Kafka → multiple sink connectors.
• Use Case: Ingest once, distribute to multiple destinations (e.g., Kafka → Elasticsearch + S3 + Snowflake).
• Benefit: Decouples data producers from consumers.

Delivery Guarantees

At-Least-Once Delivery (Default)

Both Source and Sink connectors provide at-least-once delivery by default:

• Source: If a task crashes before committing offsets, it re-reads and re-produces records.
• Sink: If a task crashes before committing offsets, it re-consumes and re-writes records.

Implication: Downstream systems must handle duplicate records (idempotency is critical).

Exactly-Once Delivery (Limitations)

Kafka Connect does not natively provide end-to-end exactly-once semantics, but you can achieve it in specific scenarios:

For Source Connectors

Approach: Use a transactional source system + idempotent producer.

Example: Debezium (CDC connector) can guarantee exactly-once by:

1. Reading database transaction logs (ensures every change is captured once).
2. Using Kafka's idempotent producer (prevents duplicate writes to Kafka).

Limitation: Most source connectors cannot guarantee exactly-once because external systems don't support transactional reads.

For Sink Connectors

Approach: Use idempotent writes in the external system.

Example: JDBC Sink with INSERT ... ON CONFLICT UPDATE ensures duplicate writes are harmless.

Limitation: Not all systems support idempotent writes (e.g., append-only file systems, message queues).

End-to-End Exactly-Once

Scenario: Kafka → Kafka (via MirrorMaker 2, which uses Connect framework).

How: MirrorMaker 2 uses Kafka Streams' transactional capabilities to provide exactly-once semantics for topic replication.

General Rule: For non-Kafka sources/sinks, exactly-once requires:

1. Idempotent or transactional external systems.
2. Careful offset management and retry logic.

Source vs. Sink: Key Differences

Common Misconceptions

Misconception 1: "Connect is Bidirectional"

Reality: A single connector is unidirectional. If you need bidirectional data flow, deploy both a Source and a Sink connector.

Misconception 2: "Connect Guarantees Exactly-Once"

Reality: Connect provides at-least-once by default. Exactly-once requires external system support (idempotency, transactions).

Misconception 3: "Source Connectors Read All Historical Data"

Reality: Most source connectors start from a specific offset (e.g., latest database row, current file position). To ingest historical data, use bulk import modes or snapshot features (e.g., Debezium's initial snapshot).

Misconception 4: "Sink Connectors Always Write in Order"

Reality: If a Kafka topic has multiple partitions, and a sink connector uses multiple tasks, writes to the external system may be out of order across partitions. To maintain order, use a single partition or a single task.

Key Takeaways

1. Source connectors pull data into Kafka; Sink connectors push data out.
2. Source offsets are custom (e.g., row ID, file position) and stored in connect-offsets.
3. Sink offsets are Kafka consumer offsets stored in __consumer_offsets.
4. Default delivery guarantee is at-least-once for both Source and Sink.
5. Exactly-once requires external system support (idempotency, transactions).
6. Pull (Source) vs. Push (Sink) determines how data flows through Connect.

What's Next?

• Module 3 covers connector configuration, deployment modes, and task scaling.
• Module 6 explains error handling, retries, and Dead Letter Queues.
• Module 7 teaches how to implement custom Source and Sink connectors.

Further Reading

• Kafka Connect Deep Dive (Confluent Blog)
• Debezium Documentation (CDC source connector)
• KIP-618: Exactly-Once Support for Source Connectors

End of Module 2.1