Apache Kafka Interview Guide: Event Streaming for Java Developers

# KRaft mode configuration (Kafka 3.3+)
process.roles=broker,controller
node.id=1
controller.quorum.voters=1@localhost:9093,2@localhost:9094,3@localhost:9095

// Messages with same key go to same partition (ordered)
producer.send(new ProducerRecord<>("orders", "customer-123", order1));
producer.send(new ProducerRecord<>("orders", "customer-123", order2));
// order1 always processed before order2 for customer-123
 
// Messages with different keys may go to different partitions
producer.send(new ProducerRecord<>("orders", "customer-456", order3));
// order3 may be processed before order1 or order2

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 
// acks=0: Fire and forget (fastest, least reliable)
props.put("acks", "0");
 
// acks=1: Leader acknowledgment (balanced)
props.put("acks", "1");
 
// acks=all: All in-sync replicas acknowledge (slowest, most reliable)
props.put("acks", "all");

props.put("enable.idempotence", "true");
// Automatically sets:
// - acks=all
// - retries=Integer.MAX_VALUE
// - max.in.flight.requests.per.connection=5

// Batch size in bytes
props.put("batch.size", 16384);  // 16 KB default
 
// Maximum wait time to fill batch
props.put("linger.ms", 5);  // Wait up to 5ms for more messages
 
// Compression
props.put("compression.type", "lz4");  // Options: none, gzip, snappy, lz4, zstd

// 1. Key-based (default): hash(key) % partitions
producer.send(new ProducerRecord<>("topic", "key", "value"));
 
// 2. Explicit partition
producer.send(new ProducerRecord<>("topic", 2, "key", "value"));
 
// 3. Custom partitioner
public class CustomPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
 
        // Route VIP customers to dedicated partition
        if (key.toString().startsWith("vip-")) {
            return 0;  // VIP partition
        }
 
        // Default: hash-based distribution for others
        return Math.abs(key.hashCode()) % (partitions.size() - 1) + 1;
    }
}

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 
// Auto commit (default, at-least-once)
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "5000");
 
// Manual commit for more control
props.put("enable.auto.commit", "false");
 
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("orders"));
 
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
    }
    // Synchronous commit (blocks until confirmed)
    consumer.commitSync();
 
    // Or async commit (non-blocking)
    consumer.commitAsync((offsets, exception) -> {
        if (exception != null) {
            log.error("Commit failed", exception);
        }
    });
}

// Enable cooperative rebalancing
props.put("partition.assignment.strategy",
    "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");

props.put("transactional.id", "my-transactional-id");
 
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();
 
try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("topic1", "key", "value1"));
    producer.send(new ProducerRecord<>("topic2", "key", "value2"));
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

// Consumer reads with isolation
props.put("isolation.level", "read_committed");
 
// Process and produce atomically
producer.beginTransaction();
for (ConsumerRecord<String, String> record : records) {
    ProducerRecord<String, String> output = process(record);
    producer.send(output);
}
// Commit consumer offsets as part of transaction
producer.sendOffsetsToTransaction(offsets, consumerGroupId);
producer.commitTransaction();

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-processor");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
 
StreamsBuilder builder = new StreamsBuilder();
 
// Read from topic as stream
KStream<String, Order> orders = builder.stream("orders");
 
// Transform: filter, map, flatMap
KStream<String, Order> highValueOrders = orders
    .filter((key, order) -> order.getAmount() > 1000)
    .mapValues(order -> enrichOrder(order));
 
// Write to output topic
highValueOrders.to("high-value-orders");
 
KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

// KStream: Append-only stream of events
// Each record is independent
KStream<String, PageView> pageViews = builder.stream("page-views");
// Records: (user1, view1), (user1, view2), (user2, view1)
 
// KTable: Changelog stream, latest value per key
// Represents current state
KTable<String, UserProfile> users = builder.table("users");
// State: {user1: profile1, user2: profile2}
 
// Join stream with table (enrichment)
KStream<String, EnrichedPageView> enriched = pageViews.join(
    users,
    (pageView, profile) -> new EnrichedPageView(pageView, profile)
);

KStream<String, Transaction> transactions = builder.stream("transactions");
 
// Tumbling window: fixed, non-overlapping
KTable<Windowed<String>, Long> hourlyCounts = transactions
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
    .count();
 
// Hopping window: fixed size, overlapping
KTable<Windowed<String>, Long> slidingCounts = transactions
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(5),
        Duration.ofMinutes(1)
    ).advanceBy(Duration.ofMinutes(1)))
    .count();
 
// Session window: activity-based, variable size
KTable<Windowed<String>, Long> sessionCounts = transactions
    .groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30)))
    .count();

-- Create a stream from a topic
CREATE STREAM orders (
    order_id VARCHAR KEY,
    customer_id VARCHAR,
    amount DECIMAL(10,2),
    order_time TIMESTAMP
) WITH (
    KAFKA_TOPIC='orders',
    VALUE_FORMAT='JSON'
);
 
-- Real-time aggregation
CREATE TABLE hourly_revenue AS
SELECT
    customer_id,
    WINDOWSTART as window_start,
    SUM(amount) as total_revenue,
    COUNT(*) as order_count
FROM orders
WINDOW TUMBLING (SIZE 1 HOUR)
GROUP BY customer_id
EMIT CHANGES;
 
-- Join streams
CREATE STREAM enriched_orders AS
SELECT
    o.order_id,
    o.amount,
    c.name as customer_name
FROM orders o
JOIN customers c ON o.customer_id = c.customer_id
EMIT CHANGES;

# application.yml
spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
      retries: 3
    consumer:
      group-id: my-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      auto-offset-reset: earliest
      properties:
        spring.json.trusted.packages: com.example.dto

@Service
@RequiredArgsConstructor
public class OrderProducer {
 
    private final KafkaTemplate<String, Order> kafkaTemplate;
 
    // Fire and forget
    public void sendOrder(Order order) {
        kafkaTemplate.send("orders", order.getId(), order);
    }
 
    // With callback
    public void sendOrderWithCallback(Order order) {
        CompletableFuture<SendResult<String, Order>> future =
            kafkaTemplate.send("orders", order.getId(), order);
 
        future.whenComplete((result, ex) -> {
            if (ex == null) {
                log.info("Sent order {} to partition {} offset {}",
                    order.getId(),
                    result.getRecordMetadata().partition(),
                    result.getRecordMetadata().offset());
            } else {
                log.error("Failed to send order {}", order.getId(), ex);
            }
        });
    }
 
    // Synchronous (blocking)
    public void sendOrderSync(Order order) throws Exception {
        kafkaTemplate.send("orders", order.getId(), order).get(10, TimeUnit.SECONDS);
    }
}

@Service
@Slf4j
public class OrderConsumer {
 
    // Basic listener
    @KafkaListener(topics = "orders", groupId = "order-processor")
    public void processOrder(Order order) {
        log.info("Received order: {}", order.getId());
        // Process order
    }
 
    // With metadata
    @KafkaListener(topics = "orders", groupId = "order-processor")
    public void processOrderWithMetadata(
            @Payload Order order,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long timestamp) {
        log.info("Received order {} from partition {} at offset {}",
            order.getId(), partition, offset);
    }
 
    // Batch processing
    @KafkaListener(topics = "orders", groupId = "batch-processor",
                   containerFactory = "batchFactory")
    public void processOrderBatch(List<Order> orders) {
        log.info("Received batch of {} orders", orders.size());
        orders.forEach(this::processOrder);
    }
 
    // Manual acknowledgment
    @KafkaListener(topics = "orders", groupId = "manual-ack")
    public void processWithManualAck(Order order, Acknowledgment ack) {
        try {
            processOrder(order);
            ack.acknowledge();  // Commit offset only after successful processing
        } catch (Exception e) {
            // Don't ack - message will be redelivered
            throw e;
        }
    }
}

@Configuration
public class KafkaConfig {
 
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Order> kafkaListenerContainerFactory(
            ConsumerFactory<String, Order> consumerFactory) {
 
        ConcurrentKafkaListenerContainerFactory<String, Order> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
 
        // Retry configuration
        factory.setCommonErrorHandler(new DefaultErrorHandler(
            new DeadLetterPublishingRecoverer(kafkaTemplate),
            new FixedBackOff(1000L, 3)  // 3 retries, 1 second apart
        ));
 
        return factory;
    }
}
 
// Or use retry topics (Spring Kafka 2.7+)
@RetryableTopic(
    attempts = "3",
    backoff = @Backoff(delay = 1000, multiplier = 2),
    dltTopicSuffix = ".DLT",
    autoCreateTopics = "true"
)
@KafkaListener(topics = "orders")
public void processWithRetryTopic(Order order) {
    // Failures automatically sent to orders-retry-0, orders-retry-1, then orders.DLT
    processOrder(order);
}
 
@DltHandler
public void handleDlt(Order order) {
    log.error("Message exhausted retries: {}", order);
    // Alert, store for manual review, etc.
}

# Topic configuration
min.insync.replicas=2

replication.factor=3
min.insync.replicas=2
acks=all

# Larger batches
batch.size=65536
linger.ms=10
 
# Compression
compression.type=lz4
 
# More in-flight requests (with idempotence)
max.in.flight.requests.per.connection=5
 
# Larger buffer
buffer.memory=67108864

# Fetch more data per request
fetch.min.bytes=1048576
fetch.max.wait.ms=500
 
# Larger fetch size
max.partition.fetch.bytes=1048576
 
# Process in batches
max.poll.records=500

# More I/O threads
num.io.threads=16
num.network.threads=8
 
# Socket buffers
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576

# Producer: Send immediately
linger.ms=0
batch.size=16384
 
# Consumer: Fetch frequently
fetch.max.wait.ms=0
fetch.min.bytes=1
 
# Broker: Faster flushing
log.flush.interval.messages=1  # Careful: impacts durability

# Check consumer group lag
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
    --describe --group my-consumer-group
 
# Output:
# TOPIC    PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# orders   0          1000            1500            500
# orders   1          2000            2100            100

// Producer with Avro and Schema Registry
props.put("schema.registry.url", "http://localhost:8081");
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
 
// Consumer
props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
props.put("specific.avro.reader", "true");