Connect, Retry & Streams

Solution and explanation

What it controls

session.timeout.ms is how long the group coordinator waits **without receiving a heartbeat** from a consumer before declaring it dead, removing it from the group and triggering a rebalance. Default: **45 seconds** (Kafka 3.0+; 10 s before).

Heartbeats are sent by a background thread, so this detects **process crashes, network partitions and frozen JVMs**, not slow processing (that is max.poll.interval.ms).

Trade-off

The value must lie between the broker's group.min.session.timeout.ms (6 s) and group.max.session.timeout.ms (30 min).

Static membership

With group.instance.id set, a consumer that restarts (e.g. a Kubernetes pod during a rolling deploy) and rejoins **within** session.timeout.ms gets its old partitions back **without a rebalance**. Teams often raise the session timeout to 1–5 minutes with static membership.

group.instance.id=payment-consumer-0     # stable per pod, e.g. StatefulSet name
session.timeout.ms=120000
heartbeat.interval.ms=10000

Clean shutdown

consumer.close() sends a LeaveGroup request so the rebalance starts immediately rather than after the timeout (static members do not send it, by design).

Summary

session.timeout.ms is the heartbeat silence after which a consumer is considered dead; balance fast failure detection against false alarms, and use static membership to avoid rebalances on restarts.

Solution and explanation

What it controls

heartbeat.interval.ms is **how often the consumer's background thread sends heartbeats** to the group coordinator. Default: **3 seconds**.

Heartbeats serve two purposes:

1. Prove the consumer is alive (checked against session.timeout.ms).
2. Learn quickly that a **rebalance** has started: the coordinator answers a heartbeat with "rebalance in progress", so the consumer rejoins.

Rule of thumb

Set it to no more than **one third** of session.timeout.ms, so several heartbeats can be missed before the session expires:

session.timeout.ms=45000
heartbeat.interval.ms=3000       # up to 15000 is valid

Effects of the value

• **Lower:** faster reaction to rebalances, slightly more network traffic to the coordinator (negligible for most clusters).
• **Higher:** fewer requests but slower rebalance participation; if close to the session timeout, a couple of lost packets can expel the consumer.

Heartbeats do not prove progress

Because heartbeats come from a separate thread, a consumer stuck in an infinite loop or a blocked database call keeps heartbeating. That case is caught by max.poll.interval.ms, which requires the main loop to call poll().

Kafka 4.0 note

With the new consumer group protocol (KIP-848), the heartbeat interval and session timeout are controlled by the broker (group.consumer.heartbeat.interval.ms, group.consumer.session.timeout.ms).

Summary

heartbeat.interval.ms sets how often a consumer reports liveness and checks for rebalances; keep it at about a third of the session timeout or less.

Solution and explanation

What it controls

max.poll.records is the **maximum number of records a single poll() returns**. Default: **500**. It does not change how much data is fetched from brokers (that is fetch.max.bytes, max.partition.fetch.bytes); extra fetched records wait in the consumer's buffer for the next poll.

Why it matters

Processing time per poll ≈ max.poll.records × time per record. That total must stay below max.poll.interval.ms, otherwise the consumer is kicked out of the group.

Tuning

• **Raise it** when processing is fast and you batch work (bulk inserts) to cut per-poll overhead.
• **Lower it** when each record is expensive or variable (calls to payment gateways, PDF generation) to avoid poll timeouts and to commit progress more often (smaller reprocessing window after a crash).

max.poll.records=100
max.poll.interval.ms=300000
fetch.min.bytes=1

Spring Kafka

spring.kafka.consumer.max-poll-records: 100
spring.kafka.listener.type: batch     # receive the whole poll as a List

Summary

max.poll.records caps records per poll; size it so a batch is processed well within max.poll.interval.ms, larger for cheap batched work and smaller for slow per-record work.

Solution and explanation

What it is

A dead-letter topic (DLT, or DLQ) is a separate topic where a consumer sends records it **cannot process** after a limited number of attempts, so they do not block the partition or get lost. The main consumer commits past the bad record and continues.

Typical reasons records go to a DLT

• Deserialisation errors (corrupt JSON, schema mismatch): **poison pills**.
• Validation failures (missing customer id, negative amount).
• Business rule violations that will never succeed on retry.
• Persistent downstream errors after all retries.

What to store

The original key and value unchanged, plus headers describing the failure:

headers: original-topic=payments, original-partition=3, original-offset=884213,
         exception-class=..., exception-message=..., failed-at=2026-10-15T10:20:00Z, attempts=4

Spring Kafka example

@Bean
DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
    var recoverer = new DeadLetterPublishingRecoverer(template,
        (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition()));
    var handler = new DefaultErrorHandler(recoverer, new ExponentialBackOff(1000, 2.0) {{ setMaxElapsedTime(30_000); }});
    handler.addNotRetryableExceptions(DeserializationException.class, ValidationException.class);
    return handler;
}

Kafka Connect has the same idea: errors.tolerance=all, errors.deadletterqueue.topic.name=....

Operating a DLT

• **Alert** when records arrive; a DLT that nobody watches is silent data loss.
• Provide tooling to inspect, fix and **replay** records back to the original topic.
• Give the DLT long retention.
• Remember ordering: a record parked in the DLT means later events for the same key were processed before it.

Summary

A dead-letter topic isolates records that repeatedly fail, with failure metadata in headers, so the main flow keeps moving; monitor it and provide replay.

Solution and explanation

The problem with retrying in place

Retrying a failed record inside the consumer blocks the whole partition: if the SMS provider is down for 10 minutes, every later record waits. Retrying immediately also hammers the failing system.

Non-blocking retries with retry topics

Failed records are published to **delay topics**, each consumed after a waiting period, and finally to a DLT:

orders  ──fail──▶ orders-retry-1m ──fail──▶ orders-retry-10m ──fail──▶ orders-retry-1h ──fail──▶ orders-dlt
   ▲ main consumer keeps going        (each retry consumer waits until the record's due time)

A retry consumer reads a record, checks a header like retry-at, and if the time has not come, pauses that partition (consumer.pause) or sleeps until due, then processes it. Success at any level ends the chain.

Spring Kafka support

@RetryableTopic(
    attempts = "4",
    backoff = @Backoff(delay = 60_000, multiplier = 10.0, maxDelay = 3_600_000),
    exclude = { ValidationException.class },
    dltTopicSuffix = ".dlt")
@KafkaListener(topics = "orders", groupId = "notifier")
public void onOrder(OrderEvent e) { notifier.send(e); }

@DltHandler
public void onDlt(OrderEvent e, @Header(KafkaHeaders.EXCEPTION_MESSAGE) String error) { alerts.raise(e, error); }

Spring creates the retry topics, headers and consumers automatically.

Design points

• Retry only **transient** errors (timeouts, 503s); send permanent errors (bad data) straight to the DLT.
• **Ordering is lost** for retried records: later events for the same key are processed first. If order matters, use blocking retries for that flow or park all events of the key.
• Make processing idempotent: a retried record may have partly succeeded.
• Monitor every retry topic's lag and the DLT.

Summary

Retry topics move failed records into delayed topics with increasing back-off and finally a DLT, keeping the main partition flowing at the cost of per-key ordering.

Solution and explanation

What it is

Kafka Connect is Kafka's framework for **moving data between Kafka and other systems without writing custom producer or consumer code**. You configure ready-made **connectors** with JSON; Connect workers run them, scale them and track their progress.

Architecture

• **Workers:** JVM processes running connectors, in **standalone** (one process, dev) or **distributed** mode (a cluster sharing work, storing configs, offsets and status in Kafka topics).
• **Connectors:** plugins for a specific system (MySQL, S3, Elasticsearch, Salesforce...).
• **Tasks:** units of parallel work a connector is split into.
• **Converters:** turn data into bytes (JSON, Avro, Protobuf with Schema Registry).
• **SMTs (Single Message Transforms):** light per-record changes: rename or drop fields, mask values, route by field, add timestamps.

Example: MySQL changes to Kafka with Debezium

{
  "name": "jobs-db-cdc",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql", "database.port": "3306",
    "database.user": "debezium", "database.password": "${file:/secrets/db.properties:password}",
    "database.server.id": "5501", "topic.prefix": "portal",
    "table.include.list": "portal.jobs,portal.applications",
    "schema.history.internal.kafka.topic": "schema-history.portal",
    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092"
  }
}

curl -X POST -H "Content-Type: application/json" --data @jobs-db-cdc.json http://connect:8083/connectors

Benefits

• No custom code for common integrations; fault tolerance and offset tracking built in.
• Scales horizontally by adding workers and tasks.
• REST API for management; error handling with dead-letter queues.

Summary

Kafka Connect runs configurable source and sink connectors in a scalable worker cluster to stream data in and out of Kafka without writing integration code.

Solution and explanation

What it is

A source connector **reads data from an external system and writes it into Kafka topics**. Connect tracks the source position (e.g. database binlog position, file offset, last timestamp) so it resumes correctly after restarts.

Common source connectors

CDC versus polling

• **CDC (Debezium):** captures every insert, update and delete in commit order with low latency and little database load; needs log access permissions.
• **JDBC polling:** simple, but misses deletes, can miss updates between polls, and adds query load.

{
  "name": "orders-jdbc-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:mysql://db:3306/shop",
    "mode": "timestamp+incrementing",
    "timestamp.column.name": "updated_at",
    "incrementing.column.name": "id",
    "table.whitelist": "orders",
    "topic.prefix": "shop.",
    "poll.interval.ms": "5000"
  }
}

Use cases

• Stream database changes to a search index or cache.
• Feed a data lake or warehouse in near real time.
• Build the **outbox pattern**: Debezium reads an outbox table and publishes events reliably.

Summary

A source connector pulls data from an external system into Kafka, ideally via change data capture, with Connect managing source offsets and fault tolerance.

Solution and explanation

What it is

A sink connector **reads records from Kafka topics and writes them to an external system**. Internally it is a consumer group: it commits Kafka offsets after the data is written, and scales by running several tasks (up to the number of partitions).

Common sink connectors

Example: index job postings in Elasticsearch

{
  "name": "jobs-es-sink",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "topics": "portal.jobs",
    "connection.url": "http://elasticsearch:9200",
    "key.ignore": "false",
    "behavior.on.null.values": "delete",
    "tasks.max": "4",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "dlq.jobs-es-sink",
    "errors.deadletterqueue.context.headers.enable": "true"
  }
}

Using the record key as the document id makes writes **idempotent** (a redelivered record overwrites the same document), and tombstones delete documents.

Delivery semantics

Sinks are usually at-least-once; idempotent writes (upsert by key, deterministic file names) give effectively exactly-once results.

Summary

A sink connector consumes Kafka topics and writes to external systems such as search engines, databases and data lakes, with offset tracking, scaling and dead-letter handling built in.

Solution and explanation

What it is

Kafka Streams is a **Java library for building stream-processing applications** that read from Kafka, transform, aggregate or join data, and write results back to Kafka. It is not a separate cluster: you embed it in a normal application (e.g. a Spring Boot service) and scale by running more instances.

Capabilities

• Stateless operations: filter, map, flatMap, branch.
• Stateful operations: groupByKey, count, aggregate, reduce, with local **state stores** (RocksDB) backed by changelog topics.
• **Windowing:** tumbling, hopping, sliding and session windows on event time.
• **Joins:** stream-stream, stream-table, table-table.
• **Exactly-once processing:** processing.guarantee=exactly_once_v2.
• Interactive queries of state stores.

Example: count job applications per company per hour

StreamsBuilder builder = new StreamsBuilder();
builder.stream("applications", Consumed.with(Serdes.String(), applicationSerde))
       .selectKey((k, app) -> app.getCompanyId())
       .groupByKey(Grouped.with(Serdes.String(), applicationSerde))
       .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
       .count(Materialized.as("applications-per-company-hourly"))
       .toStream()
       .map((window, count) -> KeyValue.pair(window.key(), count))
       .to("applications-per-company", Produced.with(Serdes.String(), Serdes.Long()));

KafkaStreams streams = new KafkaStreams(builder.build(), props);   // application.id, bootstrap.servers
streams.start();

How it scales and recovers

Work is split into **tasks**, one per input partition. Instances with the same application.id share tasks like a consumer group. If an instance dies, its tasks move to others, which rebuild state from changelog topics (standby replicas make this faster).

Alternatives

ksqlDB (SQL over Kafka Streams), Apache Flink (separate cluster, very powerful for large stateful jobs), Spark Structured Streaming.

Summary

Kafka Streams is an embeddable Java library for stateful, fault-tolerant stream processing with windows, joins and exactly-once guarantees, scaled simply by running more instances.

Solution and explanation

Different jobs

• **Kafka Connect** moves data **between Kafka and external systems** (databases, S3, Elasticsearch) using configured connectors.
• **Kafka Streams** **processes data inside Kafka**: reads topics, computes, writes topics, using code you write.

Comparison

Using them together

MySQL ──(Debezium source)──▶ portal.applications
                                   │
                           Kafka Streams app: enrich with job data,
                           count per company, detect spam
                                   │
                         applications.enriched ──(Elasticsearch sink)──▶ search index
                                                └─(S3 sink)──▶ data lake

Connect brings data in and out; Streams does the business logic in between.

When to choose which

• "Copy table X into Kafka" or "write topic Y into Snowflake": Connect.
• "Compute a running total", "join orders with customers", "flag 5 failed logins in a minute": Streams (or ksqlDB/Flink).

Summary

Connect is for moving data in and out of Kafka with configuration; Streams is for processing data within Kafka with code; real pipelines often combine both.