Applies toApache Kafka Connect (distributed mode), JDBC/S3/Elasticsearch/Snowflake sink connectors, Debezium source connectors, Amazon MSK Connect, Confluent connectors
1. Measure
| Metric | Meaning |
|---|---|
source-record-poll-rate, source-record-write-rate | Source connector throughput |
sink-record-read-rate, sink-record-send-rate | Sink throughput |
Consumer group lag of connect-<name> | Sink falling behind |
put-batch-avg-time-ms | Time the sink spends writing each batch |
| Task status / restarts | Failing tasks |
curl -s http://connect:8083/connectors/orders-sink/status | jq
curl -s http://connect:8083/connectors?expand=status | jq 'to_entries[] | {name: .key, tasks: [.value.status.tasks[].state]}'
2. Parallelism: tasks.max
- Sink connectors: tasks up to the number of partitions of the input topics.
- Source connectors: depends on the connector (JDBC per table/query; Debezium for most databases is single-task by design).
curl -s -X PUT http://connect:8083/connectors/orders-sink/config -H 'Content-Type: application/json' \
-d @orders-sink.json # with "tasks.max": "8"
3. Batching
Connector-specific settings (examples):
| Connector | Settings |
|---|---|
| JDBC sink | batch.size, insert.mode=upsert with proper keys |
| S3 sink | flush.size, rotate.interval.ms, s3.part.size; larger files = fewer objects |
| Elasticsearch sink | batch.size, max.buffered.records, linger.ms |
| Snowflake sink | buffer.count.records, buffer.flush.time, buffer.size.bytes (or Snowpipe Streaming mode) |
Client overrides per connector (requires connector.client.config.override.policy=All on workers):
"consumer.override.max.poll.records": "2000",
"consumer.override.fetch.min.bytes": "65536",
"producer.override.linger.ms": "20",
"producer.override.compression.type": "zstd"
4. Converters
| Converter | Notes |
|---|---|
JSON with schemas.enable=true | Embeds schema in every message, which is large and slow |
| JSON without schemas | Smaller, but sinks needing schemas can't infer types |
| Avro / Protobuf / JSON Schema + Schema Registry | Compact, typed, schema evolution; usually the best for throughput |
5. Bad records: DLQ instead of stopping
"errors.tolerance": "all",
"errors.deadletterqueue.topic.name": "dlq.orders-sink",
"errors.deadletterqueue.context.headers.enable": "true",
"errors.log.enable": "true"
DLQs apply to sink connectors (conversion/transformation errors and, depending on connector, write errors). Monitor the DLQ topic and have a replay process, so tolerating errors doesn't silently lose data.
6. Worker cluster sizing
- Enough workers that tasks spread out (CPU and memory per worker), plus headroom for a worker failure.
- Separate Connect clusters for critical CDC vs bulk sinks, so a heavy sink can't starve CDC.
- Heap sized for buffers of all tasks on the worker; watch GC pauses.
- Internal topics (
config,offset,status) with RF 3.
7. Single Message Transforms
SMTs run per record. Heavy transformation chains slow everything; move complex logic to stream processing or downstream (dbt/Spark).
Tip
Most "Connect is slow" cases are one task and small batches. Before adding workers, check
tasks.max, the connector's batch/flush settings and the converter.
Checklist
- Throughput, lag and batch timings measured
-
tasks.maxmatched to partitions/sources - Connector and client batch settings tuned
- Schema-based converters with Schema Registry
- DLQ with monitoring and replay
- Worker cluster sized and separated by criticality