Topics, Partitions, and Offsets

Topics are Kafka's logical channels for related messages. Each topic splits into partitions — ordered, immutable sequences of records. Partitions enable parallelism: multiple consumers in a group each handle one or more partitions. Messages within a partition have offsets — sequential IDs that allow consumers to track their position and replay from any point.

Choose partition count based on throughput requirements and consumer parallelism. A good starting rule: partitions = max(expected consumers) * 2. More partitions increase throughput but also add ZooKeeper overhead and rebalancing time. Keys determine partitioning — messages with the same key go to the same partition.

# Create topic with CLI
kafka-topics.bat --create --topic orders \
  --bootstrap-server localhost:9092 \
  --partitions 6 \
  --replication-factor 3

# Describe topic
kafka-topics.bat --describe --topic orders --bootstrap-server localhost:9092
# Topic: orders PartitionCount: 6 ReplicationFactor: 3

Producers: Writing Events

Kafka producers publish messages to topics. Configure key serializer, value serializer, acks, and retries. The acks setting controls durability: acks=0 (fire-and-forget), acks=1 (leader writes), acks=all (all replicas acknowledge). For critical data, use acks=all with min.insync.replicas=2.

Enable idempotence (enable.idempotence=true) to prevent duplicate messages on producer retries. The producer automatically retries transient errors. Use callback-based sends for async with error handling rather than the blocking get().

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.RETRIES_CONFIG, 10);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

KafkaProducer producer = new KafkaProducer<>(props);
OrderEvent event = new OrderEvent(orderId, customerId, amount);
ProducerRecord record =
    new ProducerRecord<>("orders", orderId.toString(), event);

producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        log.error("Failed to send order event: {}", exception.getMessage());
    } else {
        log.info("Sent to partition {} offset {}", metadata.partition(), metadata.offset());
    }
});
producer.flush();

Consumers and Consumer Groups

Consumers read messages from topics. Consumer groups enable horizontal scaling: each partition is assigned to exactly one consumer in the group. If a consumer fails, its partitions are reassigned (rebalance). Subscribe to topics with a group ID, poll for records, process them, and commit offsets.

Offset commits can be automatic (every poll interval) or manual. Manual commits give you control: commitSync blocks until acknowledged (slow but safe), commitAsync triggers async commit (fast but may skip on shutdown). Use commitAsync with a callback in the main loop and commitSync during graceful shutdown.

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processor");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);

KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders"));

try {
    while (true) {
        ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord record : records) {
            processOrder(record.value());
        }
        consumer.commitAsync((offsets, exception) -> {
            if (exception != null) log.error("Commit failed", exception);
        });
    }
} finally {
    try { consumer.commitSync(); } catch (Exception e) {}
    consumer.close();
}

Exactly-Once Semantics

Kafka guarantees message delivery semantics at three levels: at-most-once (auto-commit before processing), at-least-once (commit after processing — messages may replay), and exactly-once (idempotent producers plus transactional API). Exactly-once requires enable.idempotence=true, transactional.id set on producer, and isolation.level=read_committed on consumer.

// Exactly-once producer
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-producer-1");

KafkaProducer producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("orders", key, event));
    producer.send(new ProducerRecord<>("audit-log", key, event));
    producer.commitTransaction();
} catch (ProducerFencedException e) {
    producer.abortTransaction();
}

// Exactly-once consumer
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

Kafka Streams for Stream Processing

Kafka Streams is a lightweight stream processing library that runs in your application (no separate cluster). It processes data from Kafka topics, applies transformations, and writes results to other topics. Key abstractions: KStream (record stream), KTable (changelog, keyed by record key), GlobalKTable (fully replicated across nodes).

Use State Stores for stateful operations like aggregations and joins. Kafka Streams handles fault tolerance by backing up state stores to Kafka topics. The exactly-once semantics guarantee that each record processes exactly once, even on failure and restart.

StreamsBuilder builder = new StreamsBuilder();

KStream orders = builder.stream("orders",
    Consumed.with(Serdes.String(), orderEventSerde));

KTable totalsByCustomer = orders
    .groupBy((key, order) -> order.customerId())
    .aggregate(
        () -> BigDecimal.ZERO,
        (customerId, order, total) -> total.add(order.amount()),
        Materialized.with(Serdes.String(), bigDecimalSerde)
    );

totalsByCustomer.toStream().to("customer-totals",
    Produced.with(Serdes.String(), bigDecimalSerde));

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

Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

Schema Registry and Avro Serialization

Schema Registry stores and retrieves Avro, Protobuf, or JSON schemas for topic data. Producers and consumers agree on schema versions, preventing serialization errors when schemas evolve. Configure the URL in producer/consumer properties and use specific Avro serdes for type-safe record handling.

Schema evolution rules (backward, forward, full compatibility) control which schema changes are allowed without breaking existing consumers. Backward compatibility is the safest default — new schemas can read old data, so consumers upgrade before producers.

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(CommonClientConfigs.SCHEMA_REGISTRY_URL_CONFIG,
    "http://schema-registry:8081");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
    KafkaAvroSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    KafkaAvroSerializer.class);

KafkaProducer producer = new KafkaProducer<>(props);

// Avro-generated specific record
OrderEvent event = OrderEvent.newBuilder()
    .setOrderId("ORD-12345")
    .setCustomerId("CUST-678")
    .setAmount(299.99f)
    .setStatus(OrderStatus.PENDING)
    .build();

ProducerRecord record =
    new ProducerRecord<>("orders", event.getOrderId(), event);
producer.send(record);