Learn Apache Kafka - Kafka Connect: Data Integration

#Introduction

Moving data between Kafka and external systems — databases, object storage, or message queues — doesn't need to be written from scratch every time. Kafka Connect is a framework that runs connectors: components that copy data in (source) from other systems into Kafka, or out (sink) from Kafka to other systems.

Kafka Connect's main value is handling the tricky things that usually eat up time: scaling with parallel tasks, offset management, retry and error handling, and distributing workloads across workers. You just write a configuration file, and the framework takes care of the rest.

Episode 12 takes you through worker architectures, the difference between source and sink connectors, the REST API for managing connectors, deployment techniques with task parallelism, dead letter queues, and Single Message Transforms (SMT) for modifying data in flight.

#Kafka Connect Architecture

#Workers: Standalone vs Distributed

Connectors run inside a process called a worker. Two modes are supported:

• Standalone: a single process, suitable for development and testing. All connectors and tasks run on one node.
• Distributed: many workers form a cluster, connectors are distributed automatically across workers, supporting failover and scaling. This is the production mode.

#Source and Sink Connectors

• Source connector: reads from a source system (database, file, API) and writes to Kafka. Example: JdbcSourceConnector reads database tables into records.
• Sink connector: reads from Kafka and writes to a destination system. Example: S3SinkConnector stores records as files in object storage.

#Converters and Transforms

Converters change the data format between Kafka and external systems: JsonConverter, AvroConverter, or StringConverter. Transforms modify records mid-flight — for example adding a field or changing the key — before they reach the sink. Processing order: source converter → transforms → writer, and the reverse for sinks.

#Connect REST API

Connect workers expose a REST API for managing connectors:

curl -s http://localhost:8083/connectors

curl -s http://localhost:8083/connectors returns the list of connector names as JSON. Other endpoints: GET /connectors/{name} for status, PUT /connectors/{name}/config for updates, and DELETE /connectors/{name} for deletion.

#Built-in and Community Connectors

#File and JDBC Connectors

The Kafka distribution includes two basic connectors: FileStreamSource and FileStreamSink. For database integration, JdbcSourceConnector reads tables or queries into topics, and JdbcSinkConnector writes records to tables:

{
  "name": "jdbc-sink-orders",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "topics": "orders",
    "connection.url": "jdbc:postgresql://localhost:5432/app",
    "auto.create": "true",
    "insert.mode": "upsert"
  }
}

connector.class determines the implementation used, and connection.url points to the destination database. Change the connector name to add more instances.

#Community and the Connector Hub

The Kafka Connect ecosystem is very rich: connectors for MongoDB, Elasticsearch, MySQL, PostgreSQL, AWS S3, Azure Blob, and hundreds more are available on the Confluent Connector Hub. Before writing your own connector, check whether what you need already exists — writing a custom connector is the last thing you want to do.

#Connector Deployment and Management

#Task Parallelism

A single connector divides its work into several tasks that run in parallel. tasks.max determines the maximum number of tasks:

{
  "name": "s3-sink-orders",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "topics": "orders",
    "s3.bucket.name": "data-orders",
    "flush.size": "10000",
    "tasks.max": "4"
  }
}

tasks.max=4 allows the sink to read 4 different partitions in parallel. An important rule: the number of sink tasks must not exceed the topic's partition count, because one partition can only be read by one task in a group.

#Error Handling and Dead Letter Queues

Connectors that fail to process a record can be configured to send problem records to a dead letter queue (DLQ) instead of stopping the pipeline:

{
  "name": "jdbc-sink-orders",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "topics": "orders",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "orders-errors",
    "errors.deadletterqueue.context.headers.enable": "true"
  }
}

With errors.tolerance=all, the connector keeps processing other records and writes failed records to the orders-errors topic along with error context headers — very valuable information for debugging.

#Single Message Transforms (SMT)

#Common Transforms

SMTs modify records one by one before they're written to the sink:

• InsertField: adds a static field, for example a pipeline timestamp.
• ReplaceField: removes or renames fields.
• Flatten: flattens nested structures into flat ones.
• Cast: changes field data types, for example string to integer.
• TimestampRouter: writes records to different topics based on timestamp, useful for time-partitioned logs.

#Example SMT Configuration

{
  "name": "s3-sink-orders",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "topics": "orders",
    "transforms": "insertTs,router",
    "transforms.insertTs.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.insertTs.timestamp.field": "ingested_at",
    "transforms.router.type": "org.apache.kafka.connect.transforms.TimestampRouter",
    "transforms.router.topic.format": "orders-${timestamp}"
  }
}

The transform chain is defined with comma-separated aliases in transforms. SMTs run in sequence: insertTs adds the ingested_at field, then router routes records to the orders-YYYY-MM-DD topic based on timestamp — a common pattern for time-partitioned logs.

#Closing

In this episode 12 you've understood the Kafka Connect architecture, the difference between standalone and distributed workers, source and sink connectors, converters and transforms, the REST API for management, task parallelism, dead letter queues, and Single Message Transforms.

The key takeaways:

• Source connectors write to Kafka; sink connectors read from Kafka.
• Distributed mode is for production: automatic scaling and failover across workers.
• The REST API on port 8083 manages all connectors.
• tasks.max determines parallelism; sink tasks must not exceed the partition count.
• A DLQ with errors.tolerance=all prevents one corrupt record from stopping the pipeline.
• SMTs modify records in flight without writing code.

In the next episode 13 we'll discuss Kafka Streams — a stream processing library for filtering, aggregating, and joining flowing data. You'll learn about KStream, KTable, state stores with RocksDB, windowing, and exactly-once processing guarantees.