Kafka Connect: a Kafka & WebSockets guide for developers

Introduction

By day 38, you've built WebSocket pipelines and learned how Kafka brokers handle millions of messages. Now comes the question every platform team faces: how do we get data into Kafka from legacy databases, and how do we push processed events out to data warehouses, search engines, and analytical systems? Kafka Connect: a Kafka & WebSockets guide for developers answers that challenge by letting us move data without writing custom producers and consumers. This distributed framework handles connectors—pluggable integrations that extract from sources, transform payloads on the fly, and load into sinks—all while managing failures, scaling horizontally, and keeping your team's codebase lean.

In this guide, we'll explore how to architect data pipelines using Kafka Connect, from pulling rows out of PostgreSQL into topics to streaming enriched events into Elasticsearch. You'll see why distributed Connect clusters matter for production reliability, and where single message transforms fit into your topology.

Source connectors — Databases, files and SaaS APIs into Kafka

A source connector's job is straightforward: poll or listen to an external system, capture changes or new records, and write them to Kafka topics. The most common pattern in e-commerce and SaaS is capturing database changes.

Database polling with JDBC source connector

The Confluent JDBC source connector queries a database table on a schedule and publishes rows as events. This works well for relational databases where you need a periodic snapshot or incremental read based on a timestamp or ID column.

Consider an order management system: every 30 seconds, you want new orders from your PostgreSQL orders table streamed into a Kafka topic so downstream services can process shipments or trigger fulfillment workflows.

// Example JDBC connector configuration (JSON)
{
  "name": "postgres-orders-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.source.JdbcSourceConnector",
    "connection.url": "jdbc:postgresql://db.prod:5432/orders_db",
    "connection.user": "kafka_user",
    "connection.password": "${file:/secrets/db-password.txt}",
    "table.whitelist": "orders",
    "mode": "incrementing",
    "incrementing.column.name": "order_id",
    "topic.prefix": "db.",
    "poll.interval.ms": 30000,
    "tasks.max": 2
  }
}

The connector tracks the last order_id processed. On each poll, it fetches rows where order_id > last_seen_id, writes them to db.orders topic, and updates its state. If the connector restarts, it resumes from the checkpoint—no duplicates or gaps.

Change Data Capture (CDC) for real-time sync

Polling works, but there's latency. For real-time order pipelines, CDC connectors like Debezium listen to database transaction logs and emit a change event every time a row is inserted, updated, or deleted.

In a ride-hailing platform, the moment a rider cancels a trip, the database writes that update. A Debezium PostgreSQL connector tails the WAL (write-ahead log), captures the DELETE or UPDATE, and publishes a change event within milliseconds. Your cancellation handler subscribes and processes refunds instantly.

The payload looks like this:

{
  "before": { "trip_id": 12345, "status": "active", "rider_id": 999 },
  "after": { "trip_id": 12345, "status": "cancelled", "rider_id": 999 },
  "source": { "lsn": 1234567890, "txId": 5678 },
  "op": "u",  // 'c' = create, 'u' = update, 'd' = delete
  "ts_ms": 1672531200000
}

This structure lets consumers understand exactly what changed and react accordingly—reroute nearby drivers if a trip was cancelled, trigger a notification if a passenger was reassigned.

SaaS APIs and file ingestion

Not all sources are databases. A travel platform might ingest flight inventory updates from a supplier API every hour, or load user feedback CSV files from an S3 bucket daily. Custom connectors or community connectors (S3, HTTP, Salesforce) poll these sources and publish events.

The key advantage: the connector framework handles retries, rate limiting, batching, and offset management. Your team doesn't write polling loops or handle partial failures.

Sink connectors — Kafka into S3, Elasticsearch, warehouses and databases

A sink connector consumes Kafka topics and writes events to external systems. This is where enriched, deduplicated, or aggregated data lands for analytics, search, or downstream operational systems.

S3 for data lake and compliance

The S3 sink connector groups messages from a topic into Parquet or Avro files and uploads them periodically. A travel platform streams booking events into an S3 bucket, organized by date and source. Data engineers then query those files with Athena or load them into a data warehouse for revenue analysis and customer segmentation.

{
  "name": "kafka-bookings-s3-sink",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "topics": "travel.bookings",
    "s3.bucket.name": "company-data-lake",
    "s3.prefix": "bookings/",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "s3.region": "us-west-2",
    "flush.size": 10000,
    "rotate.interval.ms": 3600000,
    "tasks.max": 4
  }
}

This connector batches 10,000 records or waits 1 hour (whichever comes first), then writes a Parquet file. Compliance teams get immutable event archives; analysts query fresh data hourly without any custom ETL scripts.

Elasticsearch for search and dashboards

Real-time search and observability often require Elasticsearch. An e-commerce platform streams product catalog updates, searches, and user behavior events into ES indices. Your frontend search autocomplete, admin dashboards, and fraud detection queries all read from ES, staying fresh as events arrive.

The Elasticsearch sink connector handles mapping Kafka message schemas to ES document types, batching writes, and managing index rotations.

Data warehouse and analytics

Snowflake, BigQuery, and Redshift connectors load Kafka topics into warehoused tables. In a CRM system, every customer interaction (email, chat, call) flows into Kafka. The Snowflake sink writes them into a facts table. Analysts then join with customer attributes, run cohort analysis, and feed predictions back to the CRM for personalized campaigns.

The warehouse connector often handles schema evolution: if your event payload adds a new field, the connector detects it and alters the destination table automatically.

Operational databases

Sometimes processed data needs to flow back into your operational database. An order management system calculates fulfillment costs and fees, publishes them to Kafka, and a sink connector writes them back to PostgreSQL order_fulfillment table. Your API servers read the updated records with no latency.

Single message transforms — Light changes like renaming fields or routing

Not every event is perfect as it arrives. A source connector might extract fields with database column names (order_created_at), but your downstream topic schema expects camelCase (orderCreatedAt). SMTs (Single Message Transforms) apply lightweight field transformations without spinning up a separate stream processor.

Common transform patterns

Kafka Connect ships with built-in SMTs for renaming, casting, filtering, and routing messages. Chain them together to build flexible pipelines.

{
  "name": "postgres-orders-transform",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.source.JdbcSourceConnector",
    "connection.url": "jdbc:postgresql://db:5432/orders",
    "table.whitelist": "orders",
    "topic.prefix": "db.",
    "transforms": "RenameFields,RouteByStatus,MaskEmail",
    "transforms.RenameFields.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
    "transforms.RenameFields.renames": "order_created_at:orderCreatedAt,customer_email:email",
    "transforms.RouteByStatus.type": "org.apache.kafka.connect.transforms.Router",
    "transforms.RouteByStatus.regex": "([^.]+)\\.([^.]+)",
    "transforms.RouteByStatus.replacement": "$1.$2.${status}",
    "transforms.MaskEmail.type": "com.example.transforms.MaskPII",
    "transforms.MaskEmail.fields": "email"
  }
}

Here, we rename fields to match downstream schema, route messages by the status field (so pending orders go to db.orders.pending, completed to db.orders.completed), and mask email addresses for privacy compliance.

SMTs run in the connector process itself, so there's no network hop and minimal latency. For complex transformations—joins, aggregations, stateful logic—use Kafka Streams or ksqlDB instead. SMTs excel at simple field mappings and enrichments.

Run it distributed — Connect clusters scale and recover tasks automatically

Standalone Connect workers are fine for development, but production demands high availability. A Connect cluster runs multiple worker processes that coordinate via the internal topic __connect-offsets, __connect-configs, and __connect-status. If a worker crashes, another adopts its tasks.

Cluster deployment and task distribution

Deploy three Connect workers on separate machines. Submit a connector configuration (like the S3 sink above) via the REST API. The cluster leader assigns tasks to workers based on available capacity. A sink connector with tasks.max=4 might assign 2 tasks to worker A and 2 to worker B.

Each task independently consumes from assigned partitions and writes to the sink. If worker A dies, the cluster rebalances—worker B adopts A's tasks and resumes from the last committed offset. No manual intervention, no data loss.

// Deploy a connector to the cluster via REST API
curl -X POST http://connect-worker:8083/connectors \
  -H "Content-Type: application/json" \
  -d @orders-s3-sink.json

// Check status
curl http://connect-worker:8083/connectors/kafka-bookings-s3-sink/status

// Scale up by increasing tasks
curl -X PUT http://connect-worker:8083/connectors/kafka-bookings-s3-sink/config \
  -H "Content-Type: application/json" \
  -d '{"tasks.max": 6}'

Offsets and exactly-once semantics

Connect maintains source offsets (where it last read from a database or API) and sink offsets (which Kafka messages it's processed). The framework commits offsets only after the sink confirms the write succeeded. If a process crashes mid-write, offsets aren't committed, and recovery replays that batch—preventing data loss at the cost of occasional duplicates.

For truly exactly-once semantics, configure idempotent sink connectors (write with unique keys, so retries don't duplicate) and transactional producers/consumers (supported in newer Kafka versions).

Monitoring and alerting

Connect exposes metrics via JMX: task lag, connector rate, error counts. Wire these into your observability stack (Prometheus, Datadog). If a sink falls behind, data accumulates in Kafka; if a source stalls, you're not capturing changes. Lag alerts keep you ahead of issues before your team notices data missing in dashboards or reports.

Common mistakes

• Ignoring schema evolution. Adding a nullable field to your order table seems harmless, but if the sink connector doesn't handle it, writes fail. Use schema registries (Confluent Schema Registry) and configure connectors to support schema evolution.
• Not setting tasks.max correctly. A single-task JDBC source connector polling a massive table every 30 seconds becomes a bottleneck. Underprovisioning tasks wastes cluster capacity; overprovisioning wastes resources and can overload the source system.
• Forgetting connector offset checkpoints. If you use standalone mode and restart the worker without preserving the offset storage (usually a file or Kafka topic), the connector restarts from the beginning and replays all data. Always use Kafka-based offset storage in production.
• Applying expensive transforms in SMTs. If you're calling an external API or doing complex JSON parsing per message, move that logic to a stream processor. SMTs are meant to be fast; slow SMTs block the entire connector.
• Mixing concerns in a single connector config. Avoid daisy-chaining too many transforms or trying to source from multiple tables in one connector. Split into separate connectors for clarity, independent scaling, and easier debugging.

Key takeaways

• Source connectors pull data from databases (JDBC, Debezium), APIs, and files into Kafka topics—no custom polling code needed.
• Sink connectors push Kafka topics to data lakes (S3), search engines (Elasticsearch), warehouses (Snowflake), and operational databases—enabling real-time analytics and downstream workflows.
• Single message transforms handle lightweight field mappings and routing without the overhead of a stream processor.
• Distributed Connect clusters scale horizontally, recover from failures automatically, and maintain exactly-once delivery semantics via offset checkpoints.
• Always use schema registries, monitor lag metrics, and test failover scenarios before trusting Connect in production.