Overview
A Kafka data pipeline moves data from sources to destinations reliably, at scale, and with well-defined delivery semantics. The three core concerns: throughput (partition count, batch sizes, compression), reliability (offset management, DLQ for poison pills, idempotent consumers), and observability (consumer lag, error rates, schema health). This page draws from real patterns used in high-volume data movement at scale.
Architecture
END-TO-END DATA PIPELINE
════════════════════════════════════════════════════════
Source Systems Kafka Cluster Target Systems
────────────────────── ────────────────────────── ─────────────────────
Postgres (CDC) raw-events Snowflake (analytics)
Application events ───▶ (high-volume, raw) ───▶ Elasticsearch (search)
Click stream enriched-events S3 (data lake)
IoT sensors (after transformation) Redis (cache warm-up)
KAFKA STREAMS / CONSUMER APP
raw-events → filter → enrich → validate → enriched-events
CONSUMER LAG: THE HEALTH METRIC
════════════════════════════════════════════════════════
Consumer Lag = Latest Offset - Current Committed Offset
Partition 0: log end offset = 10,000
committed offset = 9,800
LAG = 200 ← consumer 200 messages behind
Acceptable lag: depends on SLA (real-time vs batch)
Lag growing over time: consumer slower than producer → scale!
Lag spikes then recovers: rebalance, GC pause, burst traffic
Alert on: lag > threshold AND rate-of-change > 0 (not catching up)
Tools: Kafka Consumer Groups CLI, Burrow (LinkedIn), Prometheus JMX Exporter
DEAD LETTER QUEUE PATTERN
════════════════════════════════════════════════════════
Record fails processing (deserialization error, business logic failure)
│
Retry N times with backoff
│
Still failing → publish to DLT (Dead Letter Topic): orders.DLT
│
DLT consumer: alert, inspect, manual replay after fix
DLT message includes:
- Original key/value
- Exception message + stack trace
- Original topic, partition, offset
- Retry count
Technical Implementation
High-Throughput Pipeline Consumer
@Service
public class DataPipelineConsumer {
@KafkaListener(
topics = "raw-events",
groupId = "pipeline-enricher",
concurrency = "6", // 6 parallel consumer threads
containerFactory = "batchFactory"
)
public void processBatch(
List<ConsumerRecord<String, RawEvent>> records,
Acknowledgment ack
) {
// Batch enrichment — one external call for the whole batch
List<String> userIds = records.stream()
.map(r -> r.value().getUserId())
.distinct()
.collect(toList());
Map<String, UserProfile> profiles = userService.batchFetch(userIds);
List<EnrichedEvent> enriched = records.stream()
.map(record -> {
UserProfile profile = profiles.getOrDefault(
record.value().getUserId(), UserProfile.unknown());
return EnrichedEvent.from(record.value(), profile);
})
.collect(toList());
// Bulk publish to enriched topic
enriched.forEach(event ->
kafkaTemplate.send("enriched-events", event.getUserId(), event)
);
ack.acknowledge(); // commit after successful enrichment + publish
}
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, RawEvent> batchFactory(
ConsumerFactory<String, RawEvent> cf) {
var factory = new ConcurrentKafkaListenerContainerFactory<String, RawEvent>();
factory.setConsumerFactory(cf);
factory.setBatchListener(true);
factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
factory.setCommonErrorHandler(new DefaultErrorHandler(
new DeadLetterPublishingRecoverer(kafkaTemplate,
(r, e) -> new TopicPartition(r.topic() + ".DLT", r.partition())),
new ExponentialBackOffWithMaxRetries(3)
));
return factory;
}
Consumer Lag Monitoring with Micrometer
@Component
public class KafkaLagMonitor {
private final AdminClient adminClient;
private final MeterRegistry meterRegistry;
@Scheduled(fixedDelay = 30_000) // check every 30 seconds
public void checkConsumerLag() {
Map<String, ConsumerGroupDescription> groups =
adminClient.describeConsumerGroups(List.of("pipeline-enricher")).all().get();
Map<TopicPartition, OffsetAndMetadata> committed =
adminClient.listConsumerGroupOffsets("pipeline-enricher").partitionsToOffsetAndMetadata().get();
Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> endOffsets =
adminClient.listOffsets(
committed.keySet().stream()
.collect(toMap(tp -> tp, tp -> OffsetSpec.latest()))
).all().get();
committed.forEach((tp, offsetMeta) -> {
long lag = endOffsets.get(tp).offset() - offsetMeta.offset();
// Publish lag as metric
Gauge.builder("kafka.consumer.lag", lag, d -> d)
.tag("topic", tp.topic())
.tag("partition", String.valueOf(tp.partition()))
.tag("consumer_group", "pipeline-enricher")
.register(meterRegistry);
if (lag > 10_000) {
log.warn("High consumer lag on {}-{}: {}", tp.topic(), tp.partition(), lag);
}
});
}
}
Schema Evolution Pipeline
// Handling schema evolution in a long-running pipeline
// Old messages (v1 schema) must still be processable after v2 schema is deployed
@KafkaListener(topics = "events")
public void handleEvent(ConsumerRecord<String, GenericRecord> record) {
GenericRecord event = record.value();
String schemaName = event.getSchema().getName();
int schemaVersion = getSchemaVersion(record); // from Schema Registry metadata
// Version-aware handling: process both old and new schema versions
switch (schemaVersion) {
case 1 -> processV1(event);
case 2 -> processV2(event);
default -> throw new UnknownSchemaVersionException(schemaVersion);
}
}
// OR: use schema evolution properly — add fields with defaults
// V2 schema adds optional "deviceType" field with default "unknown"
// V1 messages: deviceType = "unknown" (default applied by Avro)
// V2 messages: deviceType = "mobile" (explicit value)
// Single processV2 handles both:
private void processV2(GenericRecord event) {
String deviceType = event.get("deviceType") != null
? event.get("deviceType").toString()
: "unknown"; // fallback for V1 messages without this field
}
Interview Preparation
Q: How do you design a Kafka pipeline for exactly-once ETL (database to data warehouse)?
Use the transactional outbox + idempotent sink pattern: (1) Source: Debezium CDC captures every row change from the source DB without duplicates (each change has a unique LSN/offset). (2) Pipeline: Kafka Streams processor with processing.guarantee=exactly_once_v2 — reads, transforms, and writes to output topic atomically. (3) Sink: use insert.mode=upsert in JDBC Sink Connector with the primary key — idempotent writes. If the same row arrives twice (due to at-least-once delivery upstream), the upsert overwrites with the same data — no duplicate rows. Combine: CDC (unique source events) + exactly-once Streams (no duplicated processing) + upsert sink (idempotent write) = exactly-once ETL end-to-end.
Q: What is consumer lag and how do you respond to it?
Consumer lag is the number of messages a consumer group is behind the end of the partition log. Rising lag means the consumer is processing slower than the producer is producing. Response strategy: (1) First check if lag is due to a rebalance (temporary, self-corrects). (2) If persistent: check consumer throughput — are individual records slow? Profile the processing logic. (3) Scale consumers: increase concurrency (more threads) or add more partitions + consumers. Partitions can only increase, so plan partition count in advance (rule of thumb: 3× your expected max consumer count). (4) For batch processing consumers, increase max.poll.records and batch size. Alert on lag when: lag exceeds SLA threshold AND is not decreasing.
Q: What is a Dead Letter Topic and how do you handle poison pills?
A poison pill is a message that always fails processing — invalid schema, corrupt data, or a business rule violation that can't be resolved. Without a DLT, the consumer retries forever and the entire partition stalls. With a DLT: after N retries, the message is published to a .DLT topic with the original message + exception context. The main consumer acknowledges the offset and moves on. A separate DLT consumer sends alerts, logs to an audit table, and waits for human review. After the underlying bug is fixed, the DLT messages can be replayed to the original topic for reprocessing. The DLT pattern prevents one bad message from blocking all subsequent messages in a partition.