Kafka

60-Day Kafka 4 Learning Plan · Week 6 — Day 36 of 60

Editorial Team

60-Day Kafka 4 Learning Plan · Week 6 — Kafka Connect & ksqlDB Sources: Kafka: The Definitive Guide Ch.9 · kafka.apache.org/documentation/#connect · confluent.io/hub

Goal

Understand what Kafka Connect is, how source and sink connectors work, how workers and tasks scale the pipeline, how converters and Single Message Transforms shape the data in flight, how Connect handles errors and exactly-once delivery, and how to deploy, monitor, and secure connectors via the REST API.

1. What is Kafka Connect?

Kafka Connect is a framework for streaming data between Kafka and external systems without writing custom producer/consumer code. You configure connectors via JSON, deploy them to a Connect cluster, and they handle polling, offset tracking, error handling, and scaling automatically.

External System (DB / S3 / API / File)
  → Source Connector → Kafka Topic → Sink Connector → External System

Why Connect instead of custom code?

  • Offset management handled automatically (at-least-once by default, exactly-once in distributed mode)
  • Fault-tolerant: task failures are restarted by the framework
  • Horizontally scalable: add workers to increase throughput
  • 200+ pre-built connectors on Confluent Hub — no reinventing the wheel

2. Source vs Sink connectors

Source connector — External → Kafka

Pulls data from an external system and publishes it to a Kafka topic.

Sink connector — Kafka → External

Reads from a Kafka topic and writes to an external system.

3. Workers & tasks — how Connect scales

A Connect cluster is one or more JVM processes called workers. Each worker runs one or more tasks (threads that do the actual work). The framework distributes tasks across workers automatically.

Connect cluster
├── Worker 1
│   ├── Source Task 1  (partition 0–2 of source)
│   └── Source Task 2  (partition 3–5 of source)
├── Worker 2
│   ├── Source Task 3  (partition 6–8 of source)
│   └── Source Task 4  (partition 9–11 of source)
└── Worker 3
    ├── Sink Task 1    (topic partition 0–5)
    └── Sink Task 2    (topic partition 6–11)

Scaling: increase tasks.max in connector config → framework spawns more tasks and rebalances. Add more workers → tasks redistribute.

Fault tolerance: if a worker dies, its tasks are reassigned to remaining workers (distributed mode only).

4. Docker Compose — run Connect locally

# docker-compose.yml — Kafka 4 KRaft + Kafka Connect (distributed mode)
version: "3.8"
services:
  broker:
    image: confluentinc/cp-kafka:7.6.0
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:9092
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:9093
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

  connect:
    image: confluentinc/cp-kafka-connect:7.6.0
    depends_on: [broker]
    ports:
      - "8083:8083"
    environment:
      CONNECT_BOOTSTRAP_SERVERS: broker:9092
      CONNECT_GROUP_ID: labs-connect-group            # consumer group for the cluster
      CONNECT_CONFIG_STORAGE_TOPIC: _connect-configs  # stores connector configs
      CONNECT_OFFSET_STORAGE_TOPIC: _connect-offsets  # stores source connector offsets
      CONNECT_STATUS_STORAGE_TOPIC: _connect-status   # stores connector/task status
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_REST_PORT: 8083
      CONNECT_REST_ADVERTISED_HOST_NAME: connect
      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter
      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_PLUGIN_PATH: /usr/share/confluent-hub-components

Dev-only setting to flag now, fix in production: all three internal topics above use REPLICATION_FACTOR: 1 — same single-point-of-failure issue covered for Schema Registry’s _schemas topic in Day 23 §8. Production Connect clusters need replication.factor=3 on _connect-configs/_connect-offsets/_connect-status, or losing one broker loses connector configuration and offset tracking cluster-wide.

Installing connector plugins

# Install JDBC connector plugin into the Connect image
docker exec connect confluent-hub install confluentinc/kafka-connect-jdbc:10.7.4 --no-prompt

# Install Debezium MySQL connector
docker exec connect confluent-hub install debezium/debezium-connector-mysql:2.4.2 --no-prompt

# Restart Connect after installing plugins
docker-compose restart connect

5. REST API — deploy & manage connectors

Connect exposes a full REST API on port 8083. No code changes, no restarts — manage everything at runtime.

# List all deployed connectors
curl http://localhost:8083/connectors

# Get connector plugins available
curl http://localhost:8083/connector-plugins

# Deploy a FileStream source connector (for testing)
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "labs-file-source",
    "config": {
      "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
      "file": "/data/orders.txt",
      "topic": "labs.events",
      "tasks.max": "1"
    }
  }'

# Check connector status
curl http://localhost:8083/connectors/labs-file-source/status

# Pause a connector
curl -X PUT http://localhost:8083/connectors/labs-file-source/pause

# Resume a connector
curl -X PUT http://localhost:8083/connectors/labs-file-source/resume

# Delete a connector
curl -X DELETE http://localhost:8083/connectors/labs-file-source

# Get connector config
curl http://localhost:8083/connectors/labs-file-source/config

# Restart a failed task
curl -X POST http://localhost:8083/connectors/labs-file-source/tasks/0/restart

6. Standalone vs Distributed mode

The three internal Kafka topics

These are regular Kafka topics — they survive Connect restarts automatically.

7. Converters — the serialization layer connectors sit on top of

Every connector reads/writes data through a converter, which handles the translation between Connect’s internal data format and what actually gets stored on the Kafka topic — this is the same serialization concern from Week 4 (Day 22–25), just configured at the Connect-worker level instead of in application code.

CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081

The schemas.enable=true trap: JsonConverter‘s default embeds a full schema definition in every single message — for a small payload, this schema envelope can be larger than the actual data, defeating the compactness JSON already lacks (Day 22 §1’s JSON pain points, made worse). Either set schemas.enable=false for schema-less JSON, or — better for anything production-grade — use AvroConverter/ProtobufConverter with Schema Registry, getting the same compatibility enforcement benefits from Week 4 applied to Connect pipelines.

8. Single Message Transforms (SMTs) — lightweight in-flight editing

SMTs let a connector modify each record as it passes through, without needing a separate Kafka Streams app for simple transformations — mask a field, rename it, route to a different topic, or drop records matching a condition.

{
  "name": "labs-file-source",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "file": "/data/orders.txt",
    "topic": "labs.events",
    "transforms": "maskSSN,addPrefix",
    "transforms.maskSSN.type": "org.apache.kafka.connect.transforms.MaskField$Value",
    "transforms.maskSSN.fields": "ssn",
    "transforms.addPrefix.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.addPrefix.static.field": "source",
    "transforms.addPrefix.static.value": "labs-file-connector"
  }
}

SMTs are chained in the order listed in transforms — each one’s output feeds the next.

When to reach for an SMT vs a real Kafka Streams topology: SMTs are for lightweight, stateless, single-record edits (renaming a field, masking PII, adding metadata) — they have no access to state, can’t join against other data, and can’t aggregate. The moment a transformation needs anything from Week 5’s stateful toolkit (joins, aggregations, windowing), that’s a Kafka Streams job, not an SMT. Using an SMT chain to approximate stateful logic is a common anti-pattern that produces a fragile, hard-to-test pipeline.

9. Error handling — errors.tolerance and dead letter queues

By default, a single bad record (malformed data, a sink write that violates a DB constraint) stops the task entirely — the same “one poison pill blocks everything” problem covered for consumers in Day 18, but at the Connect framework level.

{
  "config": {
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "labs.connect.dlq",
    "errors.deadletterqueue.topic.replication.factor": 3,
    "errors.deadletterqueue.context.headers.enable": "true",
    "errors.log.enable": "true",
    "errors.log.include.messages": "true"
  }
}

This mirrors Day 18’s DefaultErrorHandler + DLT pattern almost exactly — same underlying idea (don’t let one bad record halt an entire pipeline, but make failures visible and recoverable), just configured declaratively instead of in application code. The same operational discipline applies: always monitor the DLQ topic (Day 18 §8), because a Connect DLQ with no consumer is exactly as invisible a failure mode as an unmonitored application DLT.

10. Exactly-once source connectors

Since Kafka 3.3+ (carried forward into Kafka 4), source connectors can opt into exactly-once semantics using the same transactional producer machinery from Day 11 — worth knowing exists, since Connect’s default is at-least-once.

{
  "config": {
    "exactly.once.support": "required"
  }
}
# Worker-level config — must be enabled cluster-wide
CONNECT_EXACTLY_ONCE_SOURCE_SUPPORT: enabled

Not every source connector supports this — it requires the connector implementation to participate correctly in Connect’s transactional offset-commit protocol (conceptually similar to sendOffsetsToTransaction from Day 11 §11, applied to a connector’s source offsets instead of a consumer’s). Check the specific connector’s documentation before relying on exactly.once.support: required — setting it on a connector that doesn’t actually implement the support will fail to start rather than silently downgrading to at-least-once, which is the safer failure mode but still worth knowing about upfront rather than discovering at deploy time.

11. Monitoring and securing the REST API

Monitoring — beyond GET /status polling, Connect exposes JMX metrics per connector/task:

# Sink connectors run as a consumer group named after the connector
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group connect-labs-jdbc-sink

Securing — the REST API in §5/§4 has no authentication by default, meaning anyone reaching port 8083 can deploy, reconfigure, or delete connectors (including ones touching production databases):

CONNECT_REST_EXTENSION_CLASSES: org.apache.kafka.connect.rest.basic.auth.extension.BasicAuthSecurityRestExtension

Minimum bar for anything beyond local dev, same as Schema Registry (Day 23 §9): authentication on the REST API plus network-level restriction (not exposed publicly) — a connector’s config often embeds real credentials (DB passwords, API keys) for the external system it talks to, making an unauthenticated Connect REST API a genuine credential-exposure risk, not just a “someone could pause my connector” inconvenience.

12 Common pitfalls

  • Leaving internal Connect topics at replication.factor=1 — the same single-point-of-failure risk as any other under-replicated critical topic (Day 10 §3), here specifically threatening connector config and offset durability
  • Using JsonConverter with schemas.enable=true (the default) for high-volume topics — the embedded schema envelope can dwarf the actual payload; either disable it or switch to Avro/Protobuf with Schema Registry
  • Reaching for SMT chains to implement logic that’s actually stateful — joins, aggregations, and windowing belong in Kafka Streams, not a fragile chain of single-record transforms
  • Leaving errors.tolerance at the default none for any pipeline that will encounter real-world messy data — one bad record halts the entire task with no automatic recovery path, mirroring the pre-DLT consumer problem from Day 18 §1
  • No authentication on the Connect REST API — connector configs frequently embed real external-system credentials, making this a credential-exposure risk, not just an operational inconvenience

Key Takeaways

  • Connect = integration without custom producer/consumer code — just JSON config
  • Source connectors pull data into Kafka; sink connectors push data out of Kafka
  • Workers are JVM processes; tasks are threads inside workers — both scale horizontally
  • Converters (JSON, Avro, Protobuf) determine the wire format — prefer Avro/Protobuf + Schema Registry over JSON’s schema-envelope overhead for production volume
  • SMTs handle lightweight, stateless, single-record transforms — stateful logic belongs in Kafka Streams instead
  • errors.tolerance: all + a DLQ topic mirrors Day 18’s consumer error-handling pattern, applied at the Connect framework level — and needs the same monitoring discipline
  • Exactly-once source connectors exist (exactly.once.support: required) but require both connector- and worker-level support — check compatibility before relying on it
  • REST API on :8083 needs authentication in anything beyond local dev — connector configs often embed real credentials
  • Distributed mode stores config in Kafka topics — survives worker restarts, but only if those topics are properly replicated
  • 200+ connectors on Confluent Hub — JDBC, Debezium, S3, Redis, Elasticsearch…

Support me through GitHub Sponsors.

Thank you for Reading !! See you in the next post.

Next

➡️ Day 37: JDBC connector — stream DB changes into Kafka 4

Resources

Link to Medium blog

, ,

Related Posts

Kafka

60-Day Kafka 4 Learning Plan · Week 7 — Day 46 of 60

Editorial Team
Kafka

60-Day Kafka 4 Learning Plan · Week 7 — Day 45 of 60

Editorial Team