Kafka engineer coding interview questions

As asked

Write a Java method that creates a KafkaConsumer, subscribes to a topic called 'orders', polls in a loop, processes each record, commits offsets manually, and shuts down cleanly when a shutdown flag is set. Handle the case where commitSync may fail.

Sample answer outline

The solution should configure consumer with bootstrap.servers, group.id, key/value deserializers, and enable.auto.commit=false. The poll loop should call commitSync() inside a try-catch that logs errors without crashing the loop. A volatile boolean or CountDownLatch should signal shutdown, and the shutdown path should call consumer.wakeup() from another thread then call commitSync one final time in a finally block. The consumer must be closed in a finally block.

Reference implementation (java)

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "order-consumer");
props.put("enable.auto.commit", "false");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("orders"));

volatile boolean running = true;

try {
    while (running) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            process(record);
        }
        // TODO: commitSync here
    }
} finally {
    consumer.close();
}

Expect these follow-ups

• How would you add per-partition offset tracking to commit only successfully processed offsets?
• What is the difference between commitSync and commitAsync in a shutdown path?

As asked

Given a map from consumer member ID to a list of assigned partition numbers, write a function that returns true if any consumer has more than ceiling(total_partitions / member_count) + 1 partitions assigned, indicating a skewed assignment.

Sample answer outline

Count total partitions (sum of all list sizes), compute the ceiling, then check each member's count against that threshold. The edge case is when total_partitions is not divisible by member_count: some members get one extra partition, so the threshold is ceiling + 1. The function should handle empty input gracefully.

Reference implementation (python)

import math
from typing import Dict, List

def is_skewed(assignment: Dict[str, List[int]]) -> bool:
    if not assignment:
        return False
    member_count = len(assignment)
    total_partitions = sum(len(v) for v in assignment.values())
    threshold = math.ceil(total_partitions / member_count) + 1
    return any(len(partitions) > threshold for partitions in assignment.values())

Expect these follow-ups

• How would you extend this to detect key hot spots across partitions?
• What Kafka metric would you use to detect this skew in production without this code?

As asked

Implement a Kafka Partitioner in Java that routes any message whose key starts with 'VIP:' to partition 0, and distributes all other messages using a murmur2 hash of the key across the remaining partitions. Handle a null key gracefully.

Sample answer outline

The class implements org.apache.kafka.clients.producer.Partitioner and overrides partition(). For VIP keys return 0. For all others, compute Math.abs(Utils.murmur2(keyBytes)) % (numPartitions - 1) + 1 to avoid partition 0. Null keys should fall back to a random partition in the 1 to numPartitions-1 range. The configure() and close() methods should be implemented even if empty.

Reference implementation (java)

public class VipPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        int numPartitions = cluster.partitionCountForTopic(topic);
        if (keyBytes != null && new String(keyBytes).startsWith("VIP:")) {
            return 0;
        }
        if (keyBytes == null) {
            return (int)(Math.random() * (numPartitions - 1)) + 1;
        }
        return Math.abs(Utils.murmur2(keyBytes)) % (numPartitions - 1) + 1;
    }
    @Override public void configure(Map<String, ?> configs) {}
    @Override public void close() {}
}

Expect these follow-ups

• What is a drawback of routing VIP traffic to a fixed partition in terms of throughput?
• How would you test this partitioner without a real Kafka cluster?

As asked

Using the Kafka Streams DSL in Java, write a topology that reads strings from topic 'text-input', splits each value into words, lowercases them, counts occurrences in a tumbling window of 1 minute, and writes the windowed counts to topic 'word-counts'.

Sample answer outline

Use StreamsBuilder.stream() to create a KStream, flatMapValues to split and lowercase, then groupBy(word), windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1))), count(), then toStream() to convert KTable back to stream. The to() sink must use a WindowedSerde for the key. Strong answers note that using ofSizeWithNoGrace drops late arrivals immediately and mention the trade-off with grace periods.

Reference implementation (java)

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> textLines = builder.stream("text-input");

textLines
    .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
    .groupBy((key, word) -> word)
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)))
    .count()
    .toStream()
    .to("word-counts", Produced.with(
        WindowedSerdes.timeWindowedSerdeFrom(String.class, 60000L),
        Serdes.Long()
    ));

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

Expect these follow-ups

• How would you change this to a hopping window refreshed every 30 seconds?
• Where does the count state get stored and how is it recovered after a failure?

As asked

Write a Java method that takes a KafkaConsumer and a target Instant, then seeks all currently assigned partitions to the earliest offset with a timestamp greater than or equal to that instant. Use the offsetsForTimes API.

Sample answer outline

Build a Map from TopicPartition to the target epoch millis for all assigned partitions, call consumer.offsetsForTimes(timestampsToSearch), then iterate the result and call consumer.seek(tp, offsetAndTimestamp.offset()) for each partition where the result is non-null. If a partition returns null (no messages at or after the timestamp), seek to the end of that partition. Error handling should cover the case where the consumer has no assignment yet.

Reference implementation (java)

public void seekToTimestamp(KafkaConsumer<?, ?> consumer, Instant target) {
    long targetMs = target.toEpochMilli();
    Set<TopicPartition> assignment = consumer.assignment();
    Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();
    for (TopicPartition tp : assignment) {
        timestampsToSearch.put(tp, targetMs);
    }
    Map<TopicPartition, OffsetAndTimestamp> offsets =
        consumer.offsetsForTimes(timestampsToSearch);
    for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : offsets.entrySet()) {
        if (entry.getValue() != null) {
            consumer.seek(entry.getKey(), entry.getValue().offset());
        } else {
            consumer.seekToEnd(Collections.singleton(entry.getKey()));
        }
    }
}

Expect these follow-ups

• What does offsetsForTimes return if the topic has no messages after the target time?
• How would you use this to implement a replay feature for a consumer group?

As asked

Write a Java snippet that produces the same order event atomically to both 'orders-raw' and 'orders-audit' topics using the Kafka transactional producer API. Show the correct error handling and how to abort on failure.

Sample answer outline

Set transactional.id in producer config and call producer.initTransactions() once at startup. In the send path: beginTransaction(), send to both topics, then commitTransaction(). Catch ProducerFencedException (another instance stole the transactional.id, must restart) and OutOfOrderSequenceException separately from retriable errors. On KafkaException call abortTransaction() then retry. Strong answers note that initTransactions fences previous instances of the same transactional.id.

Reference implementation (java)

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("orders-raw", key, value));
    producer.send(new ProducerRecord<>("orders-audit", key, auditValue));
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException e) {
    producer.close(); // Fatal - must recreate producer
    throw e;
} catch (KafkaException e) {
    producer.abortTransaction();
    // Retry logic here
}

Expect these follow-ups

• What is the difference between ProducerFencedException and OutOfOrderSequenceException?
• Why must you call initTransactions exactly once and not inside the retry loop?

As asked

Using the Kafka AdminClient Java API, write a method that takes a group ID and returns a map from TopicPartition to lag (long) for each assigned partition. Use listConsumerGroupOffsets and describeTopics or listOffsets.

Sample answer outline

Call adminClient.listConsumerGroupOffsets(groupId).partitionsToOffsetAndMetadata().get() to get committed offsets. Then call adminClient.listOffsets(partitionToSpec with OffsetSpec.latest()) to get current end offsets. Lag per partition = endOffset minus committedOffset. Handle missing committed offsets (partition not yet consumed) by returning endOffset as lag. Properly close AdminClient resources.

Reference implementation (java)

public Map<TopicPartition, Long> getConsumerLag(AdminClient admin, String groupId)
        throws ExecutionException, InterruptedException {
    Map<TopicPartition, OffsetAndMetadata> committed =
        admin.listConsumerGroupOffsets(groupId)
             .partitionsToOffsetAndMetadata().get();
    Map<TopicPartition, OffsetSpec> latestRequest = new HashMap<>();
    committed.keySet().forEach(tp -> latestRequest.put(tp, OffsetSpec.latest()));
    Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> endOffsets =
        admin.listOffsets(latestRequest).all().get();
    Map<TopicPartition, Long> lag = new HashMap<>();
    for (Map.Entry<TopicPartition, OffsetAndMetadata> e : committed.entrySet()) {
        long end = endOffsets.get(e.getKey()).offset();
        lag.put(e.getKey(), end - e.getValue().offset());
    }
    return lag;
}

Expect these follow-ups

• How would you extend this to alert when lag exceeds a threshold?
• What is the difference between listOffsets with LATEST vs EARLIEST spec?

As asked

Write a utility in Python using the confluent-kafka library that injects an OpenTelemetry trace context as a Kafka message header on produce, and extracts it on consume to continue the trace span. Use the W3C TraceContext format.

Sample answer outline

On produce, use the opentelemetry propagate API to inject into a dict carrier, then convert each carrier key-value to a Kafka header tuple (key, value.encode()). On consume, read headers as a list of tuples, build a dict carrier from them, and call propagate.extract(carrier) to restore the parent context. Wrap the consumer processing in a span created from that context. Strong answers handle the case where headers are None.

Reference implementation (python)

from opentelemetry import propagate, trace
from confluent_kafka import Producer, Consumer

def produce_with_trace(producer: Producer, topic: str, key: str, value: str):
    carrier = {}
    propagate.inject(carrier)
    headers = [(k, v.encode()) for k, v in carrier.items()]
    producer.produce(topic, key=key, value=value, headers=headers)

def consume_with_trace(msg):
    headers = dict((k, v.decode()) for k, v in (msg.headers() or []))
    ctx = propagate.extract(headers)
    tracer = trace.get_tracer(__name__)
    with tracer.start_as_current_span("consume", context=ctx):
        process(msg.value())

Expect these follow-ups

• How does this approach interact with exactly-once semantics if the message is retried?
• What happens to the trace if the consumer commits the offset before the downstream call completes?

As asked

Write a Kafka Connect REST API payload (JSON) to configure a JDBC Sink connector that reads from topic 'user_profiles', upserts rows into a Postgres table by primary key 'user_id', uses the Avro value converter with Schema Registry, and handles 3 retries on connection failure.

Sample answer outline

The connector config must include connector.class=io.confluent.connect.jdbc.JdbcSinkConnector, connection.url with the Postgres JDBC URL, insert.mode=upsert, pk.mode=record_value, pk.fields=user_id, value.converter=io.confluent.connect.avro.AvroConverter with schema.registry.url, topics=user_profiles, and max.retries=3 with retry.backoff.ms. Strong answers note the table.name.format setting and that the Avro schema must match the Postgres column types.

Reference implementation (json)

{
  "name": "user-profiles-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "3",
    "topics": "user_profiles",
    "connection.url": "jdbc:postgresql://db:5432/mydb",
    "connection.user": "kafkauser",
    "connection.password": "${file:/opt/kafka/secrets/creds:db_password}",
    "insert.mode": "upsert",
    "pk.mode": "record_value",
    "pk.fields": "user_id",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "max.retries": "3",
    "retry.backoff.ms": "3000"
  }
}

Expect these follow-ups

• How does the JDBC Sink handle schema evolution when a new column is added to the Avro schema?
• What is the difference between insert, upsert, and update modes in the JDBC Sink?

As asked

Messages on topic 'payments' may be duplicated within a 10-minute window due to upstream retries. Write a Kafka Streams topology in Java that deduplicates by a 'payment_id' field in the message value, allowing exactly one message per ID within a 10-minute window to pass through.

Sample answer outline

Use a Transformer with a WindowStore keyed by payment_id. On each record, check if the payment_id exists in the store within the current window. If not, add it and forward the record. If it exists, drop the record. The WindowStore should have a retention of 10 minutes. Register the store with StoreBuilder and add it to the topology via addStateStore. Strong answers mention that the store is backed by a changelog topic and is restored on restart.

Reference implementation (java)

StoreBuilder<WindowStore<String, Boolean>> storeBuilder =
    Stores.windowStoreBuilder(
        Stores.persistentWindowStore("dedup-store",
            Duration.ofMinutes(10), Duration.ofMinutes(10), false),
        Serdes.String(), Serdes.Boolean());

builder.addStateStore(storeBuilder);

KStream<String, Payment> payments = builder.stream("payments");
payments
    .transform(() -> new DedupTransformer("dedup-store"), "dedup-store")
    .to("payments-deduped");

// DedupTransformer.transform():
// WindowStore store = context.getStateStore("dedup-store");
// if (store.fetch(paymentId, now-window, now).hasNext()) return null;
// store.put(paymentId, true);
// return KeyValue.pair(key, value);

Expect these follow-ups

• Why is a WindowStore preferable to a KeyValueStore for this deduplication use case?
• What happens to the deduplication window when the Streams instance restarts?

As asked

Using the requests library in Python, write code that registers an initial Avro schema for subject 'Order-value', then attempts to register an evolved version that adds a nullable 'discount_percent' field with a default. Show how you would first check compatibility before registering.

Sample answer outline

POST to /subjects/Order-value/versions to register the first schema. For the evolved schema, POST to /compatibility/subjects/Order-value/versions/latest with the new schema to get the compatibility check result. If is_compatible is true, POST to /subjects/Order-value/versions to register it. The new field must be a union ['null', 'float'] with a default of null to maintain BACKWARD compatibility. Strong answers note that null must be the first type in the union for Avro to recognize null as the default.

Reference implementation (python)

import requests, json

SR_URL = "http://localhost:8081"
SUBJECT = "Order-value"

v1_schema = json.dumps({"type": "record", "name": "Order",
    "fields": [{"name": "order_id", "type": "string"}]})

v2_schema = json.dumps({"type": "record", "name": "Order",
    "fields": [
        {"name": "order_id", "type": "string"},
        {"name": "discount_percent", "type": ["null", "float"], "default": None}
    ]})

requests.post(f"{SR_URL}/subjects/{SUBJECT}/versions",
    json={"schema": v1_schema})

compat = requests.post(
    f"{SR_URL}/compatibility/subjects/{SUBJECT}/versions/latest",
    json={"schema": v2_schema}).json()

if compat.get("is_compatible"):
    requests.post(f"{SR_URL}/subjects/{SUBJECT}/versions", json={"schema": v2_schema})

Expect these follow-ups

• Why must null appear first in the Avro union type for a default of null to work?
• What HTTP status code does Schema Registry return for an incompatible schema registration attempt?

As asked

Using the Kafka Streams DSL in Java, write a topology that reads from topic 'events', then routes each message to one of three output topics based on the value's 'eventType' field: 'orders' goes to 'events-orders', 'returns' goes to 'events-returns', and everything else goes to 'events-other'. Parse the value as JSON.

Sample answer outline

Use KStream.split().branch() (Kafka Streams 2.8+ API) with three predicates. Each branch writes to its target topic with stream.to(). A fallback branch (always true) handles the else case. For the JSON parsing, use a custom Serde or map() to extract eventType before branching. Strong answers note that the older selectKey/filter approach is more verbose and that split().branch() guarantees each record lands in exactly one branch.

Reference implementation (java)

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> events = builder.stream("events");

Map<String, KStream<String, String>> branches = events
    .split(Named.as("branch-"))
    .branch((key, value) -> getEventType(value).equals("order"),
            Branched.as("orders"))
    .branch((key, value) -> getEventType(value).equals("return"),
            Branched.as("returns"))
    .defaultBranch(Branched.as("other"));

branches.get("branch-orders").to("events-orders");
branches.get("branch-returns").to("events-returns");
branches.get("branch-other").to("events-other");

// helper:
// static String getEventType(String json) {
//     return new ObjectMapper().readTree(json).get("eventType").asText();
// }

Expect these follow-ups

• What happens to a message if it matches no branch and there is no default branch?
• How would you unit test this topology using Kafka Streams TopologyTestDriver?

As asked

Implement a bounded worker pool in Go that processes jobs, propagates cancellation and returns when all workers have stopped.

Sample answer outline

The core is a fixed number of workers reading from a jobs channel and a WaitGroup that represents worker lifetime. The producer should stop enqueueing when the context is cancelled and close the jobs channel when no more work will be sent. Workers should select on ctx.Done as well as the jobs channel so shutdown does not depend on new jobs arriving. Results and errors need their own bounded path or a clear aggregation strategy. Candidates often create leaks by returning early without closing channels or waiting for workers.

Reference implementation (go)

func RunPool(ctx context.Context, jobs <-chan Job, workers int, handle func(context.Context, Job) error) error {
	g, ctx := errgroup.WithContext(ctx)
	for i := 0; i < workers; i++ {
		g.Go(func() error {
			for {
				select {
				case <-ctx.Done():
					return ctx.Err()
				case job, ok := <-jobs:
					if !ok {
						return nil
					}
					if err := handle(ctx, job); err != nil {
						return err
					}
				}
			}
		})
	}
	return g.Wait()
}

Expect these follow-ups

• How would you return the first error and cancel the rest?
• What happens if processing blocks forever inside a worker?
• When would errgroup be a better fit?