Apache Kafka
Beginner to Pro

// No key = round-robin distribution
ProducerRecord<String, String> record =
    new ProducerRecord<>(
        "metrics",        // topic
        null,             // no key!
        metricJson        // value
    );
producer.send(record);

public class RegionPartitioner
    implements Partitioner {
  @Override
  public int partition(String topic, Object key,
      byte[] keyBytes, Object value,
      byte[] valueBytes, Cluster cluster) {
    String region = key.toString();
    return switch (region) {
      case "US" -> 0;
      case "EU" -> 1;
      default -> 2;
    };
  }
}

// Custom serializer for domain objects
public class JsonSerializer<T> implements Serializer<T> {
    private final ObjectMapper mapper = new ObjectMapper();

    @Override
    public byte[] serialize(String topic, T data) {
        try {
            return mapper.writeValueAsBytes(data);
        } catch (Exception e) {
            throw new SerializationException(e);
        }
    }
}

// Using it with an Order object
props.put("value.serializer", JsonSerializer.class);
ProducerRecord<String, Order> record =
    new ProducerRecord<>("orders", order.getId(), order);

// Async send with callback — know this for interviews!
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        log.error("Failed to send: {}", exception.getMessage());
        // handle retry or DLQ
    } else {
        log.info("Sent to partition={} offset={}",
            metadata.partition(),
            metadata.offset());
    }
});

// For sync (blocking) confirmation:
RecordMetadata meta = producer.send(record).get();
// .get() blocks until broker acks

@Service
@RequiredArgsConstructor
public class OrderProducerService {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    @Value("${app.kafka.topic.orders}")
    private String ordersTopic;

    public CompletableFuture<SendResult<String, OrderEvent>>
    publishOrderCreated(Order order) {

        OrderEvent event = OrderEvent.builder()
            .orderId(order.getId())
            .userId(order.getUserId())
            .amount(order.getAmount())
            .status("CREATED")
            .timestamp(Instant.now())
            .build();

        // Key = userId ensures ordering per user
        return kafkaTemplate
            .send(ordersTopic, order.getUserId(), event)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Failed to publish order {}: {}",
                        order.getId(), ex.getMessage());
                } else {
                    log.info("Order {} sent to P:{} O:{}",
                        order.getId(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
                }
            });
    }
}

@Component
@Slf4j
public class OrderEventConsumer {

    private final PaymentService paymentService;
    private final MeterRegistry meterRegistry;

    @KafkaListener(
        topics = "${app.kafka.topic.orders}",
        groupId = "${spring.kafka.consumer.group-id}",
        concurrency = "3"  // 3 threads = 3 partitions
    )
    public void handleOrderEvent(
        @Payload OrderEvent event,
        @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
        @Header(KafkaHeaders.OFFSET) long offset,
        Acknowledgment ack
    ) {
        log.info("Processing order {} from P:{} O:{}",
            event.getOrderId(), partition, offset);
        try {
            paymentService.initiatePayment(event);
            ack.acknowledge(); // commit only on success
            meterRegistry.counter("kafka.orders.processed").increment();
        } catch (Exception e) {
            log.error("Failed processing order {}",
                event.getOrderId(), e);
            // Don't ack — message will be retried
            // Or route to DLQ after N retries
        }
    }
}

# docker-compose.yml — Local Kafka dev setup
version: '3.8'
services:
  kafka:
    image: confluentinc/cp-kafka:7.5.0
    ports:
      - "9092:9092"
    environment:
      # KRaft mode (no ZooKeeper!)
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    ports:
      - "8080:8080"
    environment:
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
    depends_on:
      - kafka