Kafka Deep Dive

graph TD
    subgraph Kafka Cluster
        B1[Broker 1]
        B2[Broker 2]
        B3[Broker 3]
    end
    P1[Producer 1] --> B1
    P2[Producer 2] --> B2
    C1[Consumer Group A] --> B1
    C1 --> B2
    C2[Consumer Group B] --> B2
    C2 --> B3
    ZK[Zookeeper / KRaft] --> B1
    ZK --> B2
    ZK --> B3

graph TD
    T[Topic: orders]
    T --> P0[Partition 0]
    T --> P1[Partition 1]
    T --> P2[Partition 2]
    P0 --> M0A[Offset 0]
    P0 --> M0B[Offset 1]
    P0 --> M0C[Offset 2]
    P1 --> M1A[Offset 0]
    P1 --> M1B[Offset 1]

// Same orderId always goes to same partition
producer.send(new ProducerRecord<>(
    "orders",           // topic
    order.getId(),      // key (partition key)
    orderEvent          // value
));

Properties props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
props.put("acks", "all");
props.put("enable.idempotence", "true");
props.put("retries", 10);
props.put("key.serializer", StringSerializer.class);
props.put("value.serializer", KafkaAvroSerializer.class);

KafkaProducer<String, OrderEvent> producer = new KafkaProducer<>(props);

graph TD
    subgraph Topic: orders - 4 partitions
        P0[Partition 0]
        P1[Partition 1]
        P2[Partition 2]
        P3[Partition 3]
    end
    subgraph Consumer Group A - 2 consumers
        C1[Consumer 1]
        C2[Consumer 2]
    end
    P0 --> C1
    P1 --> C1
    P2 --> C2
    P3 --> C2

sequenceDiagram
    participant C as Consumer
    participant K as Kafka
    participant DB as Application DB

    C->>K: Poll messages
    K->>C: Messages (offset 100-109)
    C->>DB: Process and save
    C->>K: Commit offset 110
    Note over C,K: If crash before commit, reprocess from 100

while (true) {
    ConsumerRecords<String, OrderEvent> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, OrderEvent> record : records) {
        processEvent(record.value());  // Process
    }
    consumer.commitSync();  // Commit after processing
}

// Producer side: enable transactions
props.put("enable.idempotence", "true");
props.put("transactional.id", "order-processor");

// Consumer side: read committed only
props.put("isolation.level", "read_committed");

public void processEvent(OrderEvent event) {
    // Check if already processed
    if (processedEventRepository.exists(event.getEventId())) {
        return;  // Skip duplicate
    }

    // Process
    orderService.processOrder(event);

    // Mark as processed
    processedEventRepository.save(event.getEventId());
}

graph LR
    subgraph Partition 0
        L[Leader - Broker 1]
        F1[Follower - Broker 2]
        F2[Follower - Broker 3]
    end
    L --> F1
    L --> F2

# Topic configuration
replication.factor=3
min.insync.replicas=2

# Producer
acks=all

graph LR
    P[Producer] -->|Register schema| SR[Schema Registry]
    P -->|Send with schema ID| K[Kafka]
    K --> C[Consumer]
    C -->|Fetch schema| SR

graph LR
    T[orders topic] --> C[Consumer]
    C -->|Process| App[Application]
    C -->|Retry exhausted| DLQ[orders.dlq topic]
    DLQ --> Alert[Alerting]
    DLQ --> Reprocess[Manual reprocess]

public void processWithRetry(ConsumerRecord<String, OrderEvent> record) {
    int retries = 0;
    while (retries < MAX_RETRIES) {
        try {
            process(record.value());
            return;
        } catch (RetryableException e) {
            retries++;
            backoff(retries);
        }
    }
    // Send to DLQ
    dlqProducer.send(new ProducerRecord<>("orders.dlq", record.key(), record.value()));
}