Definition
Operating Kafka Connect in production involves three concerns: lifecycle management, monitoring, and failure handling.
- REST API – the single interface for creating, reading, updating, deleting, pausing, resuming, and restarting connectors and their tasks in distributed mode.
- JMX metrics – the telemetry workers expose for task state, record throughput, error rates, and consumer lag, typically scraped into a metrics system.
- Error tolerance – a setting (
errors.tolerance=all) that tells the runtime to skip unprocessable records instead of failing the entire task. - Dead letter queue – a Kafka topic (
errors.deadletterqueue.topic.name) that captures skipped records with diagnostic context headers so nothing is silently lost. - Task restart – a REST endpoint that recovers a FAILED task after the underlying issue is resolved.
At National Bank of India, managing Kafka Connect is like running a courier dispatch office. The REST API is the control desk: you phone in to start a delivery route, pause one during a storm, resume it later, or cancel it entirely. JMX metrics are the dashboards showing which vans are moving, how fast, and how many parcels were damaged. The dead letter queue is the damaged-parcel shelf: when a single parcel cannot be delivered, the van sets it aside with a note explaining what went wrong and keeps delivering the rest, instead of the whole route stopping.
Understanding the Concepts
What is the REST API for managing connectors?
In distributed mode, every worker exposes a REST API on port 8083. Changes propagate across the cluster through the internal config topic, so you can call any worker. The key endpoints handle the full lifecycle of connectors.
At National Bank of India, Rajesh manages the transaction ingestion pipeline entirely through these API calls.
# List all running connectors
curl http://localhost:8083/connectors
# Create a new connector
curl -X POST http://localhost:8083/connectors \
-H "Content-Type: application/json" \
-d '{
"name": "bank-txn-source",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"tasks.max": "3",
"connection.url": "jdbc:postgresql://bankdb:5432/nbi",
"table.whitelist": "transactions",
"mode": "incrementing",
"incrementing.column.name": "txn_id",
"topic.prefix": "bank."
}
}'
# Check status of a connector and its tasks
curl http://localhost:8083/connectors/bank-txn-source/status
# Pause a connector (stops all its tasks)
curl -X PUT http://localhost:8083/connectors/bank-txn-source/pause
# Resume a paused connector
curl -X PUT http://localhost:8083/connectors/bank-txn-source/resume
# Restart a failed task (task 0)
curl -X POST http://localhost:8083/connectors/bank-txn-source/tasks/0/restart
# Update connector configuration
curl -X PUT http://localhost:8083/connectors/bank-txn-source/config \
-H "Content-Type: application/json" \
-d '{
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"tasks.max": "5",
"connection.url": "jdbc:postgresql://bankdb:5432/nbi",
"table.whitelist": "transactions,accounts",
"mode": "incrementing",
"incrementing.column.name": "txn_id",
"topic.prefix": "bank."
}'
# Delete a connector
curl -X DELETE http://localhost:8083/connectors/bank-txn-source
Expected output (status check):
{
"name": "bank-txn-source",
"connector": { "state": "RUNNING", "worker_id": "worker-1:8083" },
"tasks": [
{ "id": 0, "state": "RUNNING", "worker_id": "worker-1:8083" },
{ "id": 1, "state": "RUNNING", "worker_id": "worker-2:8083" },
{ "id": 2, "state": "RUNNING", "worker_id": "worker-3:8083" }
]
}
Why should connector configurations be version-controlled?
Connector configs submitted via curl are ad-hoc and unreproducible. If the cluster fails and you rebuild it, you need to know exactly what connectors were running with what settings. Storing configs in Git and applying them through CI/CD pipelines makes the cluster state reproducible.
At National Bank of India, Pankaj keeps all connector JSON files in a Git repository. A CI/CD pipeline applies them to the Connect cluster on every merge to main.
package com.bank.connect;
import java.io.IOException;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.file.Files;
import java.nio.file.Path;
/**
* Deploys connector configs from version-controlled JSON files.
* Used in CI/CD pipeline at National Bank of India.
*/
public class ConnectorDeployer {
private static final String CONNECT_URL = "http://connect-cluster:8083";
public static void main(String[] args) throws IOException, InterruptedException {
if (args.length == 0) {
System.out.println("Usage: java ConnectorDeployer <config-file.json>");
return;
}
String configJson = Files.readString(Path.of(args[0]));
HttpClient client = HttpClient.newHttpClient();
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(CONNECT_URL + "/connectors"))
.header("Content-Type", "application/json")
.POST(HttpRequest.BodyPublishers.ofString(configJson))
.build();
HttpResponse<String> response = client.send(request,
HttpResponse.BodyHandlers.ofString());
System.out.println("Status: " + response.statusCode());
System.out.println("Response: " + response.body());
}
}
Expected output:
Status: 201
Response: {"name":"bank-txn-source","config":{...},"tasks":[],"type":"source"}
How do you monitor Kafka Connect with JMX metrics?
Workers expose metrics over JMX that cover task state, throughput, and errors. In production, you export these with a JMX exporter into Prometheus, Grafana, or a similar system and build dashboards and alerts.
The critical metrics to monitor are:
- connector-status and task-status tell you if anything is FAILED or PAUSED.
- source-record-poll-rate and sink-record-read-rate show throughput.
- error-rate and total-errors-logged show how many records are failing.
- dead-letter-queue-produce-rate shows how fast bad records accumulate.
- consumer lag for sink connectors shows if the sink is falling behind.
# Enable JMX on the Connect worker
export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote \
-Dcom.sun.management.jmxremote.port=9999 \
-Dcom.sun.management.jmxremote.authenticate=false \
-Dcom.sun.management.jmxremote.ssl=false"
bin/connect-distributed.sh config/distributed-worker.properties
package com.bank.connect;
import javax.management.MBeanServerConnection;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;
/**
* Reads Kafka Connect JMX metrics for monitoring.
* Rajesh uses this to build custom alerts for the banking pipeline.
*/
public class ConnectMetricsReader {
public static void main(String[] args) throws Exception {
String jmxUrl = "service:jmx:rmi:///jndi/rmi://localhost:9999/jmxrmi";
JMXServiceURL url = new JMXServiceURL(jmxUrl);
try (JMXConnector connector = JMXConnectorFactory.connect(url)) {
MBeanServerConnection mbean = connector.getMBeanServerConnection();
// Check source record poll rate
ObjectName sourceRate = new ObjectName(
"kafka.connect:type=source-task-metrics,connector=bank-txn-source,task=0");
Double pollRate = (Double) mbean.getAttribute(sourceRate, "source-record-poll-rate");
System.out.println("Source poll rate: " + pollRate + " records/sec");
// Check total errors
ObjectName errorMetric = new ObjectName(
"kafka.connect:type=task-error-metrics,connector=bank-txn-source,task=0");
Double errorRate = (Double) mbean.getAttribute(errorMetric, "total-errors-logged");
System.out.println("Total errors logged: " + errorRate);
}
}
}
Expected output:
Source poll rate: 1250.0 records/sec
Total errors logged: 3.0
What is error tolerance and how does it work?
By default, errors.tolerance is set to none. This means the first unprocessable record fails the task and the pipeline stops. This is safe but brittle in production, because a single malformed transaction record can halt the entire ingestion.
Setting errors.tolerance=all tells the runtime to skip records that fail conversion, transformation, or writing, and continue processing. On its own, this silently drops bad records. You always pair it with a dead letter queue to capture the skipped records.
# Connector config with error tolerance and dead letter queue
{
"name": "bank-txn-sink",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"tasks.max": "3",
"topics": "bank.transactions",
"connection.url": "jdbc:postgresql://warehouse:5432/analytics",
"auto.create": "true",
"errors.tolerance": "all",
"errors.deadletterqueue.topic.name": "bank.txn.dlq",
"errors.deadletterqueue.topic.replication.factor": 3,
"errors.deadletterqueue.context.headers.enable": true,
"errors.log.enable": true,
"errors.log.include.messages": true
}
}
package com.bank.connect;
public class ErrorToleranceDemo {
public static void main(String[] args) {
System.out.println("=== errors.tolerance=none (default) ===");
System.out.println("Record 1: OK -> written to warehouse");
System.out.println("Record 2: OK -> written to warehouse");
System.out.println("Record 3: MALFORMED -> task FAILS, pipeline STOPS");
System.out.println("Record 4: never processed");
System.out.println();
System.out.println("=== errors.tolerance=all + DLQ ===");
System.out.println("Record 1: OK -> written to warehouse");
System.out.println("Record 2: OK -> written to warehouse");
System.out.println("Record 3: MALFORMED -> sent to bank.txn.dlq with headers");
System.out.println("Record 4: OK -> written to warehouse (pipeline continues)");
}
}
Expected output:
=== errors.tolerance=none (default) ===
Record 1: OK -> written to warehouse
Record 2: OK -> written to warehouse
Record 3: MALFORMED -> task FAILS, pipeline STOPS
Record 4: never processed
=== errors.tolerance=all + DLQ ===
Record 1: OK -> written to warehouse
Record 2: OK -> written to warehouse
Record 3: MALFORMED -> sent to bank.txn.dlq with headers
Record 4: OK -> written to warehouse (pipeline continues)
How does the dead letter queue work?
The dead letter queue (DLQ) is a Kafka topic that captures records the sink connector could not process. Each dead-lettered record carries diagnostic headers that tell you where it came from and why it failed.
When errors.deadletterqueue.context.headers.enable=true, each record in the DLQ includes headers for the original topic, partition, offset, the exception class, and the exception message. This lets you build a triage and replay process.
package com.bank.connect;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.header.Header;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
/**
* Reads dead letter queue records at National Bank of India.
* Priya uses this to triage failed transaction records.
*/
public class DlqConsumer {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "kafka-1:9092");
props.put("group.id", "bank-dlq-triage");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest");
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Collections.singletonList("bank.txn.dlq"));
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));
for (ConsumerRecord<String, String> record : records) {
System.out.println("--- Dead Letter Record ---");
System.out.println("Key: " + record.key());
System.out.println("Value: " + record.value());
// Read diagnostic headers
for (Header header : record.headers()) {
System.out.println("Header: " + header.key()
+ " = " + new String(header.value()));
}
System.out.println();
}
}
}
}
Expected output:
--- Dead Letter Record ---
Key: SAV2001
Value: {malformed json here
Header: __connect.errors.topic = bank.transactions
Header: __connect.errors.partition = 2
Header: __connect.errors.offset = 48231
Header: __connect.errors.exception.class = org.apache.kafka.connect.errors.DataException
Header: __connect.errors.exception.message = Failed to deserialize value
What are common mistakes with error handling?
The most frequent mistake is enabling errors.tolerance=all without a dead letter queue. This silently drops bad records, and you never know data is missing.
Wrong: error tolerance without a DLQ.
"errors.tolerance": "all"
# No DLQ configured -- bad records are silently dropped!
# At National Bank of India, this means lost transaction records
# that nobody knows about until an audit finds gaps.
Correct: error tolerance with DLQ and context headers.
"errors.tolerance": "all",
"errors.deadletterqueue.topic.name": "bank.txn.dlq",
"errors.deadletterqueue.topic.replication.factor": 3,
"errors.deadletterqueue.context.headers.enable": true,
"errors.log.enable": true,
"errors.log.include.messages": true
Another mistake is confusing record-level errors with infrastructure errors. The DLQ handles record-level failures like deserialization errors. If the target database is down, that is an infrastructure failure, and the task itself will fail and need a restart.
package com.bank.connect;
public class ErrorTypesDemo {
public static void main(String[] args) {
System.out.println("=== Record-level error (handled by DLQ) ===");
System.out.println("Cause: malformed JSON, schema mismatch, bad data type");
System.out.println("Result: record sent to DLQ, task keeps running");
System.out.println();
System.out.println("=== Infrastructure error (NOT handled by DLQ) ===");
System.out.println("Cause: database down, network timeout, auth failure");
System.out.println("Result: task FAILS, needs manual restart via REST API");
System.out.println("Fix: curl -X POST .../tasks/0/restart");
}
}
Expected output:
=== Record-level error (handled by DLQ) ===
Cause: malformed JSON, schema mismatch, bad data type
Result: record sent to DLQ, task keeps running
=== Infrastructure error (NOT handled by DLQ) ===
Cause: database down, network timeout, auth failure
Result: task FAILS, needs manual restart via REST API
Fix: curl -X POST .../tasks/0/restart
How do you build a triage and replay process for the DLQ?
The dead letter queue is not the end of the story. You need a process that reads DLQ records, diagnoses the problem, fixes the data, and replays them into the main pipeline.
At National Bank of India, Priya reviews the DLQ daily. She reads the diagnostic headers to understand why each record failed, fixes the upstream data or the connector configuration, and then replays the corrected records back to the original topic.
package com.bank.connect;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
/**
* Replays fixed records from the DLQ back to the original topic.
*/
public class DlqReplayService {
public static void main(String[] args) {
Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "kafka-1:9092");
consumerProps.put("group.id", "bank-dlq-replay");
consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "kafka-1:9092");
producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps)) {
consumer.subscribe(Collections.singletonList("bank.txn.dlq"));
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));
int replayed = 0;
for (ConsumerRecord<String, String> record : records) {
String fixedValue = fixRecord(record.value());
if (fixedValue != null) {
producer.send(new ProducerRecord<>("bank.transactions",
record.key(), fixedValue));
replayed++;
}
}
System.out.println("Replayed " + replayed + " fixed records to bank.transactions");
}
}
private static String fixRecord(String brokenValue) {
// Fix the record: correct JSON, fill missing fields, etc.
// Return null if the record cannot be fixed
if (brokenValue == null || brokenValue.isEmpty()) {
return null;
}
return brokenValue.replace("{malformed", "{\"fixed\":true,");
}
}
Expected output:
Replayed 5 fixed records to bank.transactions
What about securing the REST API?
The Connect REST API can create, modify, and delete connectors across the entire cluster. In production, this must be secured. An unsecured API means anyone on the network can deploy a connector that reads your database.
# Secure the REST API with TLS
listeners=https://0.0.0.0:8083
rest.advertised.listener=https
listeners.https.ssl.keystore.location=/etc/kafka/connect.keystore.jks
listeners.https.ssl.keystore.password=changeit
listeners.https.ssl.key.password=changeit
listeners.https.ssl.truststore.location=/etc/kafka/connect.truststore.jks
listeners.https.ssl.truststore.password=changeit
package com.bank.connect;
public class RestApiSecurity {
public static void main(String[] args) {
System.out.println("=== REST API Security Checklist ===");
System.out.println("1. Enable TLS on the REST listener");
System.out.println("2. Use mutual TLS (mTLS) for client authentication");
System.out.println("3. Restrict network access with firewall rules");
System.out.println("4. Use RBAC if running Confluent Platform");
System.out.println("5. Never expose port 8083 to the public internet");
}
}
Expected output:
=== REST API Security Checklist ===
1. Enable TLS on the REST listener
2. Use mutual TLS (mTLS) for client authentication
3. Restrict network access with firewall rules
4. Use RBAC if running Confluent Platform
5. Never expose port 8083 to the public internet
What should a senior engineer know?
A senior engineer understands the operational nuances beyond the happy path.
- The DLQ handles record-level errors only. Infrastructure failures (database down, network partition) cause task failures that need a restart via the REST API.
errors.tolerance=allplus a DLQ converts a hard stop into a soft, observable skip. But you must monitor the DLQ topic or it silently accumulates lost records.- Single Message Transforms (SMTs) can mask sensitive fields like Priya's account number before the data reaches the warehouse.
- Connector configs should go through CI/CD. Ad-hoc
curlcalls create drift between what is running and what is documented. - In a bank, the DLQ topic must have the same retention and access controls as the main topic, because it contains production data.
package com.bank.connect;
public class SeniorChecklist {
public static void main(String[] args) {
System.out.println("Senior Connect Ops Checklist:");
System.out.println("1. DLQ = record-level errors only");
System.out.println("2. Infrastructure errors = task restart");
System.out.println("3. Monitor DLQ produce rate (alert if climbing)");
System.out.println("4. DLQ topic needs same ACLs as main topic");
System.out.println("5. Use SMTs for field masking (PII, account numbers)");
System.out.println("6. Version-control all connector configs");
System.out.println("7. Secure REST API with TLS + auth");
}
}
Expected output:
Senior Connect Ops Checklist:
1. DLQ = record-level errors only
2. Infrastructure errors = task restart
3. Monitor DLQ produce rate (alert if climbing)
4. DLQ topic needs same ACLs as main topic
5. Use SMTs for field masking (PII, account numbers)
6. Version-control all connector configs
7. Secure REST API with TLS + auth
Visual Overview
┌───────────────────────────────────────────────────┐
│ Kafka Connect Management Flow │
│ │
│ Developer / CI/CD │
│ │ │
│ │ POST /connectors (create) │
│ │ PUT /connectors/{name}/config (update) │
│ │ PUT /connectors/{name}/pause │
│ │ PUT /connectors/{name}/resume │
│ │ POST /connectors/{name}/tasks/0/restart │
│ ▼ │
│ ┌──────────────────┐ │
│ │ REST API :8083 │◄─── Any worker in cluster │
│ └────────┬─────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────┐ │
│ │ Internal Config │ (Kafka topic, RF=3) │
│ │ Topic │ │
│ └────────┬─────────┘ │
│ │ propagates to all workers │
│ ▼ │
│ ┌────────┐ ┌────────┐ ┌────────┐ │
│ │Worker 1│ │Worker 2│ │Worker 3│ │
│ │ Task 0 │ │ Task 1 │ │ Task 2 │ │
│ └───┬────┘ └───┬────┘ └───┬────┘ │
│ │ │ │ │
│ └───────────┼───────────┘ │
│ ▼ │
│ JMX Metrics ──► Prometheus / Grafana │
└───────────────────────────────────────────────────┘
Error Handling Flow:
Source ──► Kafka Topic ──► Sink Task
│
Record OK? ──yes──► Target DB
│
no
│
errors.tolerance=all?
│ │
no yes
│ │
Task FAILS ┌───▼────────────┐
│ Dead Letter │
│ Queue Topic │
│ + context hdrs │
└────────────────┘
Key Takeaway
Managing Kafka Connect in production revolves around three concerns. The REST API is your single control plane for creating, updating, pausing, resuming, and deleting connectors and their tasks in distributed mode. At National Bank of India, Pankaj keeps all connector configurations in Git and applies them through a CI/CD pipeline, so the cluster state is always reproducible and never depends on ad-hoc commands.
Monitoring relies on JMX metrics that every worker exposes. The critical metrics are task state (to catch FAILED or PAUSED tasks), source and sink record rates (to track throughput), consumer lag (to detect a sink falling behind), and the dead letter queue produce rate (to catch accumulating bad records). These metrics feed into dashboards and alerts so the team knows about problems before users report missing data.
The failure handling is what makes Connect production-ready. Setting errors.tolerance=all with a dead letter queue topic and context headers converts a hard pipeline stop into a soft, observable skip. A single malformed transaction record goes to the DLQ with diagnostic headers instead of halting the entire ingestion, and the rest of the pipeline continues. The DLQ handles record-level errors; infrastructure failures like a database outage cause the task itself to fail and need a restart via the REST API.
Remember one sentence: manage connectors through the REST API, monitor with JMX metrics, and pair errors.tolerance=all with a dead letter queue so bad records are captured with diagnostics instead of crashing the pipeline or being silently lost.
---