Cheat SheetsApache KafkaKafka Connect

Kafka Connect — Cheat Sheet

Apache Kafka · 5 topics. Download the PDF or the Instagram carousel and share it.

Cheat Sheet · AiCanCode.org

Kafka Connect

Apache Kafka5 topicsQuick revision reference

Kafka Connect

Kafka Connect is a scalable framework for streaming data between Kafka and external systems (databases, file systems, cloud services) without writing custom producer/consumer code.

✓Connect eliminates custom producer/consumer code for standard integrations — use it before writing code.
✓Distributed mode is production-standard; standalone mode is for single-worker development only.
✓Connectors run tasks that are distributed and rebalanced across workers automatically on failure.
✓Always configure a DLQ (errors.deadletterqueue.topic.name) so a bad record does not halt the connector.
✓Internal topics (connect-configs, connect-offsets, connect-status) must have replication.factor=3 for HA.
✓Use Confluent Hub to find pre-built connectors (jdbc, s3, elasticsearch, debezium) before writing custom plugins.

Properties + Shell — Connect distributed config + REST

# connect-distributed.properties
bootstrap.servers=kafka-1:9092,kafka-2:9092,kafka-3:9092
group.id=connect-cluster-1

# Internal topics for offset, config, and status storage
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status
config.storage.replication.factor=3
offset.storage.replication.factor=3

# Converters — Avro with Schema Registry
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://schema-registry:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://schema-registry:8081

plugin.path=/usr/share/confluent-hub-components

# Start worker
bin/connect-distributed.sh config/connect-distributed.properties

# REST API — list running connectors
curl http://connect:8083/connectors

# REST API — check connector status
curl http://connect:8083/connectors/my-connector/status

Source Connectors & Debezium CDC

Source connectors pull data into Kafka from external systems; the Debezium CDC connector captures database changes as Kafka events in real time using DB transaction logs.

✓Debezium reads the binary log directly — zero polling overhead, sub-second latency, captures DELETEs.
✓JDBC Source polling requires an updated_at or auto-increment column and cannot capture hard DELETEs.
✓Debezium events carry full before/after row state — ideal for audit, cache invalidation, and search index sync.
✓The Outbox Pattern uses Debezium to reliably publish domain events: write to outbox table → Debezium delivers to Kafka.
✓Debezium captures an initial snapshot of existing data before streaming new changes from the binlog.
✓Each Debezium MySQL connector is single-task (tasks.max=1); scale by running more connectors per table subset.

SQL + Shell — MySQL CDC connector setup

-- MySQL prerequisites
-- 1. Enable binlog in my.cnf
-- [mysqld]
-- server-id=1
-- log_bin=/var/log/mysql/mysql-bin.log
-- binlog_format=ROW
-- binlog_row_image=FULL

-- 2. Create replication user
CREATE USER 'debezium'@'%' IDENTIFIED BY 'dbz_password';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE,
      REPLICATION CLIENT ON *.* TO 'debezium'@'%';
FLUSH PRIVILEGES;

# Deploy Debezium MySQL connector
curl -X POST http://connect:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "mysql-cdc",
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "database.hostname": "mysql",
      "database.port": "3306",
      "database.user": "debezium",
      "database.password": "dbz_password",
      "database.server.id": "184054",
      "topic.prefix": "dbserver1",
      "database.include.list": "orders",
      "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
      "schema.history.internal.kafka.topic": "dbhistory.orders",
      "tasks.max": "1"
    }
  }'

Sink Connectors

Sink connectors push Kafka records to external systems (Elasticsearch, S3, JDBC); they support upsert semantics and dead-letter queuing for failed records.

✓Sink connectors read from Kafka and write to external targets — offsets are committed by the Connect framework, not the connector
✓JDBC Sink supports INSERT/UPSERT modes; pk.mode=record_key uses the Kafka record key as the primary key
✓S3 Sink batches records into Parquet/Avro files; flush.size and rotate.interval.ms control file roll-over
✓Time-based partitioning creates Hive-compatible S3 paths (year/month/day/hour) for efficient Athena/Spark queries
✓errors.tolerance=all + DLQ prevents a single bad record from halting the connector — DLQ holds failed records with error headers
✓Monitor DLQ topic depth as a data quality metric — growing DLQ indicates schema mismatches or target availability issues

JSON — JDBC Sink Connector with upsert and DLQ configuration

# JDBC Sink Connector configuration
curl -X POST http://connect:8083/connectors -H 'Content-Type: application/json' -d '{
  "name": "orders-jdbc-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "3",
    "topics": "orders",
    "connection.url": "jdbc:mysql://mysql:3306/analytics",
    "connection.user": "sink_user",
    "connection.password": "secret",
    "insert.mode": "upsert",          // INSERT or upsert (requires pk.mode)
    "pk.mode": "record_key",          // use Kafka record key as PK
    "pk.fields": "order_id",
    "auto.create": "true",            // create table if not exists
    "auto.evolve": "true",            // add new columns on schema change
    "table.name.format": "${topic}",  // table name = topic name
    // Dead-letter queue for failed records
    "errors.tolerance": "all",        // continue on error (vs fail fast)
    "errors.deadletterqueue.topic.name": "orders-sink-dlq",
    "errors.deadletterqueue.context.headers.enable": "true"
  }
}'

Debezium CDC Connector

Debezium is a log-based CDC connector that streams database row-level changes into Kafka topics with sub-second latency and zero impact on the source database.

✓Debezium reads DB transaction logs — zero polling overhead on the source DB
✓Every INSERT/UPDATE/DELETE becomes a Kafka event with before/after row image
✓op field: c=create, u=update, d=delete, r=snapshot read
✓ExtractNewRecordState SMT flattens the envelope to just the after image
✓Requires binlog_format=ROW for MySQL; wal_level=logical for PostgreSQL

Debezium — MySQL CDC connector

curl -X POST http://kafka-connect:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "mysql-orders-connector",
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "database.hostname": "mysql",
      "database.port": "3306",
      "database.user": "debezium",
      "database.password": "secret",
      "database.server.id": "1",
      "topic.prefix": "myapp",
      "database.include.list": "orders_db",
      "table.include.list": "orders_db.orders",
      "transforms": "unwrap",
      "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState"
    }
  }'
# Events: op=c (insert), op=u (update), op=d (delete)

Single Message Transforms (SMT)

SMTs are lightweight, stateless transformations applied to each message in a Kafka Connect pipeline — routing, masking PII, renaming topics, or flattening nested structures without custom code.

✓SMTs run in-process inside the Connect worker — zero network overhead
✓SMTs are stateless — cannot join or aggregate across messages
✓Chain multiple SMTs with a comma-separated "transforms" list
✓$Key and $Value suffixes select key or value transformation
✓For stateful logic, use Kafka Streams or ksqlDB, not SMT chains

Kafka Connect — SMT chaining

{
  "transforms": "maskPII,addTimestamp,routeByTable",

  "transforms.maskPII.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.maskPII.fields": "email,phone_number",
  "transforms.maskPII.replacement": "****",

  "transforms.addTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.addTimestamp.timestamp.field": "ingested_at",

  "transforms.routeByTable.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.routeByTable.regex": "myapp\.public\.(.*)",
  "transforms.routeByTable.replacement": "users-$1"
}

Learn this free with Aria, your AI tutor → AiCanCode.org/learn/kafka

Cheat Sheet

Kafka Connect

Apache Kafka

5 topics · swipe →

AiCanCode.org

Kafka Connect

✓Connect eliminates custom producer/consumer code for standard integrations — use it before writing code.
✓Distributed mode is production-standard; standalone mode is for single-worker development only.
✓Connectors run tasks that are distributed and rebalanced across workers automatically on failure.
✓Always configure a DLQ (errors.deadletterqueue.topic.name) so a bad record does not halt the connector.
✓Internal topics (connect-configs, connect-offsets, connect-status) must have replication.factor=3 for HA.

# connect-distributed.properties
bootstrap.servers=kafka-1:9092,kafka-2:9092,kafka-3:9092
group.id=connect-cluster-1

# Internal topics for offset, config, and status storage
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status
config.storage.replication.factor=3
…

AiCanCode.org

Kafka Connect

Source Connectors & Debezium CDC

✓Debezium reads the binary log directly — zero polling overhead, sub-second latency, captures DELETEs.
✓JDBC Source polling requires an updated_at or auto-increment column and cannot capture hard DELETEs.
✓Debezium events carry full before/after row state — ideal for audit, cache invalidation, and search index sync.
✓The Outbox Pattern uses Debezium to reliably publish domain events: write to outbox table → Debezium delivers to Kafka.
✓Debezium captures an initial snapshot of existing data before streaming new changes from the binlog.

-- MySQL prerequisites
-- 1. Enable binlog in my.cnf
-- [mysqld]
-- server-id=1
-- log_bin=/var/log/mysql/mysql-bin.log
-- binlog_format=ROW
-- binlog_row_image=FULL

-- 2. Create replication user
…

AiCanCode.org

Kafka Connect

Sink Connectors

✓Sink connectors read from Kafka and write to external targets — offsets are committed by the Connect framework, not the connector
✓JDBC Sink supports INSERT/UPSERT modes; pk.mode=record_key uses the Kafka record key as the primary key
✓S3 Sink batches records into Parquet/Avro files; flush.size and rotate.interval.ms control file roll-over
✓Time-based partitioning creates Hive-compatible S3 paths (year/month/day/hour) for efficient Athena/Spark queries
✓errors.tolerance=all + DLQ prevents a single bad record from halting the connector — DLQ holds failed records with error headers

# JDBC Sink Connector configuration
curl -X POST http://connect:8083/connectors -H 'Content-Type: application/json' -d '{
  "name": "orders-jdbc-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "3",
    "topics": "orders",
    "connection.url": "jdbc:mysql://mysql:3306/analytics",
    "connection.user": "sink_user",
…

AiCanCode.org

Kafka Connect

Debezium CDC Connector

✓Debezium reads DB transaction logs — zero polling overhead on the source DB
✓Every INSERT/UPDATE/DELETE becomes a Kafka event with before/after row image
✓op field: c=create, u=update, d=delete, r=snapshot read
✓ExtractNewRecordState SMT flattens the envelope to just the after image
✓Requires binlog_format=ROW for MySQL; wal_level=logical for PostgreSQL

curl -X POST http://kafka-connect:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "mysql-orders-connector",
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "database.hostname": "mysql",
      "database.port": "3306",
      "database.user": "debezium",
…

AiCanCode.org

Kafka Connect

Single Message Transforms (SMT)

✓SMTs run in-process inside the Connect worker — zero network overhead
✓SMTs are stateless — cannot join or aggregate across messages
✓Chain multiple SMTs with a comma-separated "transforms" list
✓$Key and $Value suffixes select key or value transformation
✓For stateful logic, use Kafka Streams or ksqlDB, not SMT chains

{
  "transforms": "maskPII,addTimestamp,routeByTable",

  "transforms.maskPII.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.maskPII.fields": "email,phone_number",
  "transforms.maskPII.replacement": "****",

  "transforms.addTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.addTimestamp.timestamp.field": "ingested_at",
…

AiCanCode.org

Master Kafka Connect

Full lessons + an AI tutor, free.

AiCanCode.org

/learn/kafka