Kafka Connect in Production — Connectors, SMTs, Schema Registry, and Fault Tolerance

Why Kafka Connect Instead of Custom Code

Every data platform eventually needs to move data between systems: from a relational database into Kafka, from Kafka into a data warehouse, from an object store into a search index. The naive approach is custom ETL scripts — a Python script that reads from Postgres and publishes to Kafka, another that consumes and writes to S3. These scripts work until they don't: they restart from scratch on failure, they don't handle schema changes, they require separate monitoring, and each new integration multiplies the maintenance burden.

Kafka Connect is Apache Kafka's built-in integration framework. It runs connectors — plugins that know how to read from or write to a specific system — as distributed, fault-tolerant tasks managed by a cluster of workers. Connect handles offset tracking, task rebalancing on worker failure, parallelism, error handling with dead letter queues, and schema evolution via Schema Registry — the same registry that manages Avro schemas for event-driven architectures across your entire data platform. What would be thousands of lines of custom code across dozens of integrations becomes connector configuration files.

Architecture: Workers, Tasks, and the REST API

A Kafka Connect cluster is a group of workers — JVM processes that load connector plugins and execute tasks. Workers coordinate through three internal Kafka topics: connect-configs stores connector and task configurations, connect-offsets stores source connector read positions, and connect-status tracks connector and task state. These topics must exist before the cluster starts; Create them with aggressive replication and compaction.

Each worker exposes a REST API on port 8083. You create, update, pause, and delete connectors by POSTing JSON to this API. In a multi-worker cluster, any worker can serve REST requests and will propagate the change to all peers via the internal Kafka topics. There is no leader election among workers; coordination is Kafka-native.

A connector is a configuration object that tells Connect what to integrate and how. A connector spawns one or more tasks — the actual threads that do the work. Source tasks poll the source system and write records to Kafka; sink tasks consume from Kafka and write to the destination. If a task fails, the worker restarts it. If a worker fails, its tasks are reassigned to surviving workers within seconds.

Setting Up a Distributed Connect Cluster

Standalone mode — a single worker process — is suitable only for development. Production requires distributed mode with at least three workers for fault tolerance. The essential configuration lives in connect-distributed.properties:

# connect-distributed.properties — production configuration

# Kafka bootstrap servers
bootstrap.servers=kafka-1:9092,kafka-2:9092,kafka-3:9092

# All workers in the same cluster must share the same group.id
group.id=connect-cluster-production

# Internal topics for coordination — pre-create these with high replication
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
status.storage.replication.factor=3

# Offset flush — how often source connector offsets are committed
offset.flush.interval.ms=10000

# REST API
rest.host.name=0.0.0.0
rest.port=8083

# Advertised hostname — must be reachable by other workers (use pod IP in K8s)
rest.advertised.host.name=connect-worker-1
rest.advertised.port=8083

# Plugin path — where connector JARs are located
plugin.path=/opt/kafka/plugins

# Converter defaults — use Schema Registry + Avro for production
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

# Allow connector configuration overrides of converters
connector.client.config.override.policy=All

# Worker-level performance tuning
task.shutdown.graceful.timeout.ms=5000
offset.storage.partitions=25
status.storage.partitions=5

# Create the internal coordination topics before starting workers
# These must be compacted and highly replicated
kafka-topics.sh --bootstrap-server kafka-1:9092   --create --topic connect-configs   --replication-factor 3 --partitions 1   --config cleanup.policy=compact

kafka-topics.sh --bootstrap-server kafka-1:9092   --create --topic connect-offsets   --replication-factor 3 --partitions 25   --config cleanup.policy=compact

kafka-topics.sh --bootstrap-server kafka-1:9092   --create --topic connect-status   --replication-factor 3 --partitions 5   --config cleanup.policy=compact

# Start the distributed worker
connect-distributed.sh /opt/kafka/config/connect-distributed.properties

# Kubernetes Deployment for a Connect cluster
apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafka-connect
  namespace: data-platform
spec:
  replicas: 3
  selector:
    matchLabels:
      app: kafka-connect
  template:
    metadata:
      labels:
        app: kafka-connect
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "9404"
    spec:
      containers:
        - name: kafka-connect
          image: confluentinc/cp-kafka-connect:7.6.0
          ports:
            - containerPort: 8083   # REST API
            - containerPort: 9404   # JMX Prometheus exporter
          env:
            - name: CONNECT_BOOTSTRAP_SERVERS
              value: "kafka-1:9092,kafka-2:9092,kafka-3:9092"
            - name: CONNECT_GROUP_ID
              value: "connect-cluster-production"
            - name: CONNECT_CONFIG_STORAGE_TOPIC
              value: "connect-configs"
            - name: CONNECT_OFFSET_STORAGE_TOPIC
              value: "connect-offsets"
            - name: CONNECT_STATUS_STORAGE_TOPIC
              value: "connect-status"
            - name: CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR
              value: "3"
            - name: CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR
              value: "3"
            - name: CONNECT_STATUS_STORAGE_REPLICATION_FACTOR
              value: "3"
            - name: CONNECT_KEY_CONVERTER
              value: "io.confluent.connect.avro.AvroConverter"
            - name: CONNECT_VALUE_CONVERTER
              value: "io.confluent.connect.avro.AvroConverter"
            - name: CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL
              value: "http://schema-registry:8081"
            - name: CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL
              value: "http://schema-registry:8081"
            - name: CONNECT_REST_ADVERTISED_HOST_NAME
              valueFrom:
                fieldRef:
                  fieldPath: status.podIP
            - name: CONNECT_PLUGIN_PATH
              value: "/usr/share/java,/usr/share/confluent-hub-components"
            - name: KAFKA_JMX_PORT
              value: "9999"
            - name: KAFKA_OPTS
              value: "-javaagent:/opt/jmx-exporter/jmx_prometheus_javaagent.jar=9404:/opt/jmx-exporter/kafka-connect.yml"
          resources:
            requests:
              memory: "2Gi"
              cpu: "500m"
            limits:
              memory: "4Gi"
              cpu: "2"
          livenessProbe:
            httpGet:
              path: /connectors
              port: 8083
            initialDelaySeconds: 60
            periodSeconds: 30
          readinessProbe:
            httpGet:
              path: /connectors
              port: 8083
            initialDelaySeconds: 30
            periodSeconds: 10

Source Connectors: JDBC and Debezium CDC

The JDBC Source Connector polls a relational database on a configurable interval and publishes new or updated rows to Kafka topics. It supports two modes: incrementing (using an auto-increment primary key) and timestamp+incrementing (using a timestamp column plus a primary key for updates). Neither mode captures deletes — for full CDC including deletes, use Debezium.

# Deploy JDBC Source Connector via REST API
curl -X POST http://connect-worker-1:8083/connectors   -H "Content-Type: application/json"   -d '{
    "name": "jdbc-source-orders",
    "config": {
      "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
      "connection.url": "jdbc:postgresql://prod-db:5432/ecommerce",
      "connection.user": "connect_reader",
      "connection.password": "${file:/opt/secrets/db-credentials.properties:db.password}",

      "mode": "timestamp+incrementing",
      "timestamp.column.name": "updated_at",
      "incrementing.column.name": "order_id",
      "table.whitelist": "orders,order_items,customers",

      "topic.prefix": "postgres.ecommerce.",
      "poll.interval.ms": "5000",
      "batch.max.rows": "500",
      "db.timezone": "UTC",

      "transforms": "AddNamespace",
      "transforms.AddNamespace.type": "org.apache.kafka.connect.transforms.SetSchemaMetadata$Value",
      "transforms.AddNamespace.schema.name": "com.example.ecommerce.Order",

      "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",

      "tasks.max": "3"
    }
  }'

For full CDC — capturing inserts, updates, and deletes — Debezium's PostgreSQL connector reads the database's write-ahead log (WAL) rather than polling tables. This gives sub-second latency and full change history, including delete events with the row's before-state. The CDC guide covers Debezium connector setup in depth, including logical replication slot configuration, snapshot modes, and wiring sink connectors to downstream databases for replication pipelines.

# Debezium PostgreSQL CDC Source Connector
curl -X POST http://connect-worker-1:8083/connectors   -H "Content-Type: application/json"   -d '{
    "name": "debezium-source-ecommerce",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "database.hostname": "prod-db",
      "database.port": "5432",
      "database.user": "debezium",
      "database.password": "${file:/opt/secrets/db-credentials.properties:debezium.password}",
      "database.dbname": "ecommerce",
      "database.server.name": "ecommerce-prod",

      "plugin.name": "pgoutput",
      "slot.name": "debezium_connect",
      "publication.name": "dbz_publication",

      "table.include.list": "public.orders,public.order_items,public.customers",
      "column.exclude.list": "public.customers.password_hash",

      "snapshot.mode": "initial",
      "snapshot.locking.mode": "none",

      "topic.prefix": "cdc.ecommerce",
      "topic.creation.enable": "true",
      "topic.creation.default.replication.factor": "3",
      "topic.creation.default.partitions": "6",
      "topic.creation.default.cleanup.policy": "delete",
      "topic.creation.default.retention.ms": "604800000",

      "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",

      "heartbeat.interval.ms": "10000",
      "heartbeat.topics.prefix": "__debezium-heartbeat",

      "tasks.max": "1"
    }
  }'

Sink Connectors: S3 and JDBC Sink

Sink connectors consume from Kafka topics and write to external systems. The S3 Sink Connector writes records to S3 (or any S3-compatible store) as Parquet, Avro, JSON, or CSV files, partitioned by time or field value. It buffers records in memory until it has accumulated enough to flush a file — configurable by record count, file size, or time interval.

# S3 Sink Connector — writes CDC events to S3 as Parquet, partitioned by date
curl -X POST http://connect-worker-1:8083/connectors   -H "Content-Type: application/json"   -d '{
    "name": "s3-sink-ecommerce-cdc",
    "config": {
      "connector.class": "io.confluent.connect.s3.S3SinkConnector",
      "tasks.max": "6",

      "topics": "cdc.ecommerce.public.orders,cdc.ecommerce.public.order_items",
      "s3.region": "eu-west-1",
      "s3.bucket.name": "my-data-lake",
      "s3.part.size": "67108864",

      "storage.class": "io.confluent.connect.s3.storage.S3Storage",
      "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
      "parquet.codec": "snappy",

      "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
      "path.format": "'''year'''=YYYY/'''month'''=MM/'''day'''=dd/'''hour'''=HH",
      "locale": "en_US",
      "timezone": "UTC",
      "timestamp.extractor": "RecordField",
      "timestamp.field": "created_at",

      "flush.size": "10000",
      "rotate.interval.ms": "600000",
      "rotate.schedule.interval.ms": "600000",

      "schema.compatibility": "FULL",
      "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",

      "errors.tolerance": "all",
      "errors.deadletterqueue.topic.name": "dlq.s3-sink-ecommerce-cdc",
      "errors.deadletterqueue.topic.replication.factor": "3",
      "errors.deadletterqueue.context.headers.enable": "true",
      "errors.log.enable": "true",
      "errors.log.include.messages": "true"
    }
  }'

# JDBC Sink Connector — write Kafka topics back to a PostgreSQL analytics database
curl -X POST http://connect-worker-1:8083/connectors   -H "Content-Type: application/json"   -d '{
    "name": "jdbc-sink-analytics",
    "config": {
      "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
      "tasks.max": "3",

      "topics": "postgres.ecommerce.orders",
      "connection.url": "jdbc:postgresql://analytics-db:5432/reporting",
      "connection.user": "connect_writer",
      "connection.password": "${file:/opt/secrets/db-credentials.properties:analytics.password}",

      "auto.create": "false",
      "auto.evolve": "true",
      "insert.mode": "upsert",
      "pk.mode": "record_key",
      "pk.fields": "order_id",

      "table.name.format": "staging.${topic}",

      "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",

      "errors.tolerance": "all",
      "errors.deadletterqueue.topic.name": "dlq.jdbc-sink-analytics",
      "errors.deadletterqueue.topic.replication.factor": "3",
      "errors.deadletterqueue.context.headers.enable": "true"
    }
  }'

Single Message Transforms: Routing, Filtering, and Field Manipulation

Single Message Transforms (SMTs) are connector-level plugins that transform each record in the pipeline — before it's written to Kafka by a source connector, or before it's written to the destination by a sink connector. SMTs chain: you apply them in order, each receiving the output of the previous. Common built-in transforms cover field renaming, dropping fields, routing to different topics based on field values, flattening nested structs, and casting field types.

# SMT examples in connector configuration

# 1. Drop a field (e.g. remove internal _db_metadata from CDC events)
"transforms": "DropMetadata",
"transforms.DropMetadata.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
"transforms.DropMetadata.blacklist": "_db_metadata,_row_version",

# 2. Rename a field
"transforms": "RenameId",
"transforms.RenameId.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
"transforms.RenameId.renames": "id:order_id,ts:created_at",

# 3. Route records to different topics based on a field value
# (high-value orders go to a separate priority topic)
"transforms": "RouteByValue",
"transforms.RouteByValue.type": "org.apache.kafka.connect.transforms.ContentBasedRouter$Value",
"transforms.RouteByValue.language": "jsr223.groovy",
"transforms.RouteByValue.topic.expression": "value.total_amount > 10000 ? 'orders.high-value' : null",

# 4. Flatten a nested Struct into top-level fields
"transforms": "Flatten",
"transforms.Flatten.type": "org.apache.kafka.connect.transforms.Flatten$Value",
"transforms.Flatten.delimiter": "_",

# 5. Add a static field to every record
"transforms": "AddSource",
"transforms.AddSource.type": "org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.AddSource.static.field": "source_system",
"transforms.AddSource.static.value": "ecommerce-prod",

# 6. Extract the nested Debezium "after" field (for CDC events)
"transforms": "Unwrap",
"transforms.Unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.Unwrap.drop.tombstones": "false",
"transforms.Unwrap.delete.handling.mode": "rewrite",
"transforms.Unwrap.add.fields": "op,table,lsn,source.ts_ms",
"transforms.Unwrap.add.headers": "db"

# Chaining multiple transforms
"transforms": "Unwrap,DropMetadata,AddSource",
"transforms.Unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.Unwrap.drop.tombstones": "false",
"transforms.DropMetadata.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
"transforms.DropMetadata.blacklist": "_row_version",
"transforms.AddSource.type": "org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.AddSource.static.field": "source_system",
"transforms.AddSource.static.value": "ecommerce-prod"

Schema Registry and Avro: Enforcing Contracts at the Wire Level

Using JSON as the wire format for Connect is convenient but fragile: a column type change in Postgres silently changes the JSON, and consumers crash when they encounter an unexpected type. Avro with Schema Registry enforces a schema contract at every producer and consumer boundary — a schema change that would break consumers is rejected at write time, before the message enters the topic.

The Confluent Schema Registry stores Avro, JSON Schema, and Protobuf schemas under subject names that correspond to Kafka topic names. When a connector serializes a record with AvroConverter, it registers the schema under {topic}-value (or {topic}-key) and writes the schema ID as the first 5 bytes of the payload. Consumers look up the schema by ID to deserialize without embedding the full schema in every message. For complete event streaming architecture patterns with Schema Registry, the event-driven architecture guide covers schema evolution strategies, compatibility modes (BACKWARD, FORWARD, FULL), and subject naming conventions across producer and consumer teams.

# Schema Registry configuration for Connect workers
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://schema-registry:8081
key.converter.auto.register.schemas=true
key.converter.use.latest.version=false

value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://schema-registry:8081
value.converter.auto.register.schemas=true
value.converter.use.latest.version=false

# Set compatibility globally at the registry level
# curl -X PUT http://schema-registry:8081/config #   -H "Content-Type: application/json" #   -d '{"compatibility": "FULL_TRANSITIVE"}'

# Per-subject compatibility override (for topics that need looser rules)
# curl -X PUT http://schema-registry:8081/config/cdc.ecommerce.public.orders-value #   -H "Content-Type: application/json" #   -d '{"compatibility": "BACKWARD"}'

# Python consumer verifying schema compatibility with confluent-kafka
from confluent_kafka import Consumer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer
from confluent_kafka.serialization import SerializationContext, MessageField

schema_registry_client = SchemaRegistryClient({"url": "http://schema-registry:8081"})
avro_deserializer = AvroDeserializer(schema_registry_client)

consumer = Consumer({
    "bootstrap.servers": "kafka-1:9092,kafka-2:9092,kafka-3:9092",
    "group.id": "analytics-consumer",
    "auto.offset.reset": "earliest",
})
consumer.subscribe(["cdc.ecommerce.public.orders"])

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        print(f"Consumer error: {msg.error()}")
        continue

    order = avro_deserializer(
        msg.value(),
        SerializationContext(msg.topic(), MessageField.VALUE)
    )
    print(f"order_id={order['order_id']} total={order['total_amount']} op={order.get('__op', 'r')}")

Error Handling and Dead Letter Queues

By default, a single bad record causes the entire task to fail and halt. The task enters FAILED state and stops processing until you manually restart it. For production, configure errors.tolerance=all to skip bad records and route them to a dead letter queue (DLQ) topic — a Kafka topic that receives the raw bytes of every failed record along with headers explaining the error. The pipeline continues; you inspect and replay DLQ messages separately.

# Error handling configuration for any connector
"errors.tolerance": "all",
"errors.retry.delay.max.ms": "60000",
"errors.retry.timeout": "300000",

# Dead Letter Queue
"errors.deadletterqueue.topic.name": "dlq.my-connector",
"errors.deadletterqueue.topic.replication.factor": "3",
"errors.deadletterqueue.context.headers.enable": "true",

# Log errors to Connect worker logs as well
"errors.log.enable": "true",
"errors.log.include.messages": "true"

# Inspect DLQ messages — headers carry error context
from confluent_kafka import Consumer

consumer = Consumer({
    "bootstrap.servers": "kafka-1:9092",
    "group.id": "dlq-inspector",
    "auto.offset.reset": "earliest",
})
consumer.subscribe(["dlq.s3-sink-ecommerce-cdc"])

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        break
    if msg.error():
        continue

    # DLQ headers contain error details
    headers = {k: v.decode("utf-8") for k, v in (msg.headers() or [])}
    print("=== DLQ Record ===")
    print(f"  Topic:         {headers.get('__connect.errors.topic', '-')}")
    print(f"  Partition:     {headers.get('__connect.errors.partition', '-')}")
    print(f"  Offset:        {headers.get('__connect.errors.offset', '-')}")
    print(f"  Error class:   {headers.get('__connect.errors.exception.class.name', '-')}")
    print(f"  Error message: {headers.get('__connect.errors.exception.message', '-')}")
    print(f"  Stage:         {headers.get('__connect.errors.stage', '-')}")
    print(f"  Connector:     {headers.get('__connect.errors.connector.name', '-')}")
    print(f"  Raw bytes:     {msg.value()[:100]}...")
    print()

# Replay DLQ messages — copy from DLQ back to the original topic
# After fixing the root cause, use MirrorMaker2 or a script to replay

from confluent_kafka import Consumer, Producer

consumer = Consumer({
    "bootstrap.servers": "kafka-1:9092",
    "group.id": "dlq-replayer",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": "false",
})
producer = Producer({"bootstrap.servers": "kafka-1:9092"})

consumer.subscribe(["dlq.s3-sink-ecommerce-cdc"])

replayed = 0
while True:
    msg = consumer.poll(2.0)
    if msg is None:
        break

    headers = {k: v.decode("utf-8") for k, v in (msg.headers() or [])}
    original_topic = headers.get("__connect.errors.topic")
    if not original_topic:
        continue

    # Republish original bytes to the original topic
    producer.produce(
        topic=original_topic,
        key=msg.key(),
        value=msg.value(),
    )
    replayed += 1
    consumer.commit(msg)

producer.flush()
print(f"Replayed {replayed} records from DLQ")

Monitoring with JMX and Prometheus

Kafka Connect workers expose operational metrics through JMX. The JMX Prometheus Exporter translates these into Prometheus metrics, making them scrapable without any additional instrumentation. Key metrics to alert on: connector task state (any FAILED tasks), source connector poll latency, sink connector put latency, offset commit success rate, and DLQ write rate (a non-zero DLQ rate indicates active data quality or schema issues).

# jmx-exporter config for Kafka Connect — kafka-connect.yml
rules:
  # Connector status — alert if any task is not RUNNING
  - pattern: "kafka.connect<type=connector-task-metrics, connector=(.+), task=(.+)><>status"
    name: kafka_connect_task_status
    type: UNTYPED
    labels:
      connector: "$1"
      task: "$2"

  # Source connector records produced per second
  - pattern: "kafka.connect<type=source-task-metrics, connector=(.+), task=(.+)><>source-record-write-rate"
    name: kafka_connect_source_record_write_rate
    type: GAUGE
    labels:
      connector: "$1"
      task: "$2"

  # Sink connector records consumed per second
  - pattern: "kafka.connect<type=sink-task-metrics, connector=(.+), task=(.+)><>sink-record-read-rate"
    name: kafka_connect_sink_record_read_rate
    type: GAUGE
    labels:
      connector: "$1"
      task: "$2"

  # Sink put latency percentiles
  - pattern: "kafka.connect<type=sink-task-metrics, connector=(.+), task=(.+)><>put-batch-max-time-ms"
    name: kafka_connect_sink_put_batch_max_ms
    type: GAUGE
    labels:
      connector: "$1"
      task: "$2"

  # DLQ write rate — non-zero means records are being dropped
  - pattern: "kafka.connect<type=task-error-metrics, connector=(.+), task=(.+)><>deadletterqueue-produce-requests"
    name: kafka_connect_dlq_produce_requests_total
    type: COUNTER
    labels:
      connector: "$1"
      task: "$2"

# Prometheus alerting rules for Kafka Connect
groups:
  - name: kafka-connect
    rules:
      - alert: KafkaConnectTaskFailed
        expr: kafka_connect_task_status{status="FAILED"} == 1
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "Kafka Connect task {{ $labels.connector }}/{{ $labels.task }} is FAILED"
          description: "Task has been in FAILED state for 2 minutes. Check logs and restart if needed."

      - alert: KafkaConnectDLQActiveWrites
        expr: rate(kafka_connect_dlq_produce_requests_total[5m]) > 0
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "DLQ receiving messages for connector {{ $labels.connector }}"
          description: "Records are being dropped to the DLQ — indicates schema or data quality issues."

      - alert: KafkaConnectSinkLagHigh
        expr: kafka_connect_sink_put_batch_max_ms > 30000
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "Sink connector {{ $labels.connector }} has high put latency"
          description: "Put batch max latency exceeds 30s — destination may be under pressure."

# Useful REST API operations for day-to-day management

# List all connectors
curl http://connect-worker-1:8083/connectors | jq .

# Get connector status (including individual task states)
curl http://connect-worker-1:8083/connectors/debezium-source-ecommerce/status | jq .

# Restart a failed task
curl -X POST   http://connect-worker-1:8083/connectors/debezium-source-ecommerce/tasks/0/restart

# Restart all tasks for a connector
curl -X POST   "http://connect-worker-1:8083/connectors/debezium-source-ecommerce/restart?includeTasks=true&onlyFailed=true"

# Pause a connector (stops tasks but preserves offsets)
curl -X PUT http://connect-worker-1:8083/connectors/s3-sink-ecommerce-cdc/pause

# Resume a paused connector
curl -X PUT http://connect-worker-1:8083/connectors/s3-sink-ecommerce-cdc/resume

# Update connector configuration without deleting (triggers rebalance)
curl -X PUT http://connect-worker-1:8083/connectors/jdbc-source-orders/config   -H "Content-Type: application/json"   -d '{"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "tasks.max": "5", ...}'

# Delete a connector
curl -X DELETE http://connect-worker-1:8083/connectors/jdbc-source-orders

Kafka Connect is optimized for connector-level data movement — getting data in and out of Kafka. For stateful stream processing within Kafka — joins, windowed aggregations, real-time enrichment — pair it with Kafka Streams, which provides join semantics across streams and changelog tables, exactly-once processing semantics, and interactive queries against materialized state stores without leaving the Kafka ecosystem.

Production Checklist