Kenapa penting

Kafka adalah broker yang paling sering kamu temui di lowongan backend Java untuk integrasi antar-layanan: event pesanan, mutasi saldo, log audit, sampai sinkronisasi data ke search engine. Di bab sebelumnya kamu sudah memahami konsepnya — pub/sub, at-least-once, DLQ, ordering per entitas. Bab ini menerjemahkan konsep itu ke Kafka dan Spring: bagaimana data disusun di topic dan partition, bagaimana consumer group membagi beban, bagaimana menulis producer dan consumer dengan Spring for Apache Kafka, serta bagaimana menangani pesan gagal tanpa kehilangan data. Kesalahan kecil seperti lupa key atau salah mengatur commit offset adalah sumber bug produksi yang sulit dilacak.

Konsep inti

Topic, partition, dan offset

Kafka menyimpan pesan sebagai log append-only. Satu pesan disebut record, berisi key (opsional), value, timestamp, dan header.

• Topic — nama aliran record, mis. orders.
• Partition — topic dipecah menjadi beberapa partition. Tiap partition adalah log berurutan tersendiri yang bisa berada di broker berbeda; inilah yang membuat Kafka bisa diskalakan.
• Offset — nomor urut record di dalam satu partition (0, 1, 2, …). Offset tidak berlaku lintas partition.

Topic: orders (3 partition)
partition 0: [0: ORD-1 placed] [1: ORD-4 placed] [2: ORD-1 paid] ...
partition 1: [0: ORD-2 placed] [1: ORD-2 paid]   ...
partition 2: [0: ORD-3 placed] [1: ORD-5 placed] ...

Urutan hanya dijamin di dalam satu partition. Tiap partition juga direplikasi ke beberapa broker (replication factor, biasanya 3 di produksi) sehingga data tetap aman bila satu broker mati.

Key dan ordering

Partition tujuan ditentukan oleh key: producer meng-hash key, sehingga key yang sama selalu masuk partition yang sama (selama jumlah partition tidak berubah). Pakai ID entitas sebagai key — orderId, accountId — agar semua event untuk entitas itu berurutan. Record tanpa key disebar ke berbagai partition, sehingga urutannya tidak bisa diandalkan.

Menambah partition

Menambah partition pada topic yang sudah dipakai mengubah pemetaan key → partition. Event lama ORD-1 ada di partition 0, event baru bisa masuk partition lain, dan urutan per key sempat rusak. Tentukan jumlah partition dengan ruang tumbuh sejak awal.

Consumer group dan rebalance

Consumer bergabung dalam consumer group lewat group.id. Aturannya:

• Setiap group menerima semua record topic (pub/sub antar-group).
• Di dalam satu group, satu partition hanya dibaca oleh satu consumer (queue di dalam group).
• Paralelisme maksimum satu group = jumlah partition. Consumer berlebih akan menganggur.

Ketika consumer bergabung, keluar, atau mati, Kafka melakukan rebalance: partition dibagi ulang. Selama rebalance, pemrosesan bisa tertunda sejenak, dan record yang sudah diproses tetapi offset-nya belum di-commit akan diproses ulang oleh consumer pengganti — alasan lain consumer harus idempoten.

Retention dan KRaft

Berbeda dengan queue klasik, Kafka tidak menghapus record setelah dibaca. Record dihapus berdasarkan retention: waktu (retention.ms, default 7 hari) atau ukuran (retention.bytes). Karena itu consumer group baru bisa membaca data lama, dan consumer yang bug bisa "diputar ulang" dengan memundurkan offset. Ada juga log compaction (cleanup.policy=compact) yang hanya menyimpan record terakhir per key — cocok untuk data "status terkini".

Sejak Kafka 4.0, metadata cluster dikelola sepenuhnya oleh KRaft (protokol konsensus berbasis Raft di dalam Kafka sendiri). ZooKeeper sudah dihapus, jadi kamu cukup menjalankan proses Kafka saja. Untuk belajar lokal, image resmi apache/kafka sudah berjalan dalam mode KRaft single-node:

docker run -d --name kafka -p 9092:9092 apache/kafka:4.1.0

# Buat topic dengan 3 partition
docker exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic orders --partitions 3 --replication-factor 1

# Lihat detail topic
docker exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --describe --topic orders

Spring for Apache Kafka: konfigurasi

Di Spring Boot 4, tambahkan starter Kafka (versinya dikelola Boot):

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-kafka</artifactId>
</dependency>

Konfigurasi lewat properti spring.kafka.*. Spring for Apache Kafka 4 memakai Jackson 3 lewat JacksonJsonSerializer/JacksonJsonDeserializer:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JacksonJsonSerializer
      acks: all
    consumer:
      group-id: inventory-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JacksonJsonDeserializer
      properties:
        spring.json.trusted.packages: "id.jalurjava.shop.events"

• acks: all — producer menunggu semua replika sinkron menerima record; paling aman untuk data bisnis.
• auto-offset-reset — dipakai hanya bila group belum punya committed offset: earliest mulai dari awal log, latest dari ujung.
• spring.json.trusted.packages — deserializer hanya mau membuat object dari package yang dipercaya. Jangan isi * di produksi.

Producer dengan KafkaTemplate

// Potongan — producer event pesanan
@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);
    private final KafkaTemplate<String, OrderPlacedEvent> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, OrderPlacedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publish(OrderPlacedEvent event) {
        kafkaTemplate.send("orders", event.orderId(), event) // key = orderId → urutan per pesanan
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("Gagal kirim event {}", event.eventId(), ex);
                    } else {
                        log.info("Terkirim ke partition {} offset {}",
                                result.getRecordMetadata().partition(),
                                result.getRecordMetadata().offset());
                    }
                });
    }
}

send bersifat asinkron dan mengembalikan CompletableFuture<SendResult<K, V>>. Jangan anggap record pasti terkirim hanya karena send tidak melempar exception — periksa hasilnya. Masalah "simpan ke database lalu kirim ke Kafka" yang bisa gagal di tengah jalan diselesaikan dengan pola outbox di bab berikutnya.

Consumer dengan @KafkaListener dan commit offset

// Potongan — consumer di inventory-service
@Component
public class OrderPlacedListener {

    private final StockService stockService;

    public OrderPlacedListener(StockService stockService) {
        this.stockService = stockService;
    }

    @KafkaListener(topics = "orders", groupId = "inventory-service")
    public void onOrderPlaced(OrderPlacedEvent event,
                              @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                              @Header(KafkaHeaders.OFFSET) long offset) {
        stockService.reserve(event.orderId(), event.items());
    }
}

Spring menjalankan listener di container yang melakukan poll, memanggil method-mu, lalu meng-commit offset setelah method selesai tanpa exception. Secara default Spring mematikan enable.auto.commit milik Kafka dan memakai AckMode.BATCH (commit setelah semua record hasil satu poll diproses). Ini memberi at-least-once: crash di tengah pemrosesan → record dibaca ulang setelah restart. Jangan menyalakan enable.auto.commit=true, karena commit berbasis timer bisa terjadi sebelum record selesai diproses (pesan bisa hilang). Bila butuh kontrol penuh, set spring.kafka.listener.ack-mode: manual dan panggil Acknowledgment.acknowledge() sendiri.

Retry dan dead letter topic

Bila listener melempar exception, error handler yang memutuskan nasib record. Ada dua pendekatan:

1. Blocking retry dengan DefaultErrorHandler. Record di-retry di tempat (partition menunggu), lalu diserahkan ke recoverer bila retry habis:

// Potongan — konfigurasi error handler; Boot otomatis memasangnya ke listener container factory
@Configuration
public class KafkaErrorConfig {

    @Bean
    DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
        var recoverer = new DeadLetterPublishingRecoverer(template); // kirim ke "<topic>-dlt"
        var handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 2L)); // 1 + 2 retry
        handler.addNotRetryableExceptions(IllegalArgumentException.class); // error permanen → langsung DLT
        return handler;
    }
}

Cocok untuk gangguan singkat dan bila urutan per partition harus dijaga. Kelemahannya: selama retry, record lain di partition yang sama ikut tertahan.

2. Non-blocking retry dengan @RetryableTopic. Record gagal dipindah ke topic retry (orders-retry-0, …) dengan jeda, sehingga partition utama tetap jalan:

// Potongan — retry non-blocking, cocok bila urutan tidak kritis
@RetryableTopic(attempts = "4", backOff = @BackOff(delay = 1000, multiplier = 2.0))
@KafkaListener(topics = "orders", groupId = "email-service")
public void sendConfirmation(OrderPlacedEvent event) {
    emailClient.sendOrderConfirmation(event.customerId(), event.orderId());
}

@DltHandler
public void onDeadLetter(OrderPlacedEvent event) {
    log.error("Email konfirmasi gagal permanen untuk {}", event.orderId());
}

Trade-off-nya: record yang di-retry diproses setelah record yang datang belakangan, jadi urutan dikorbankan. Pakai untuk pekerjaan yang tidak bergantung urutan seperti email atau notifikasi. Apa pun pendekatannya, topic -dlt wajib punya monitoring dan prosedur untuk memproses ulang.

Consumer lag

Consumer lag = offset terakhir di partition dikurangi offset yang sudah di-commit group. Lag stabil mendekati nol berarti consumer mengikuti; lag yang terus naik berarti pemrosesan kalah cepat dari laju record masuk. Contoh output (kolom CONSUMER-ID, HOST, dan CLIENT-ID disingkat):

docker exec kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group inventory-service

GROUP              TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
inventory-service  orders  0          1520            1522            2
inventory-service  orders  1          1498            1498            0
inventory-service  orders  2          1210            4380            3170

Lag besar di satu partition saja biasanya tanda hot key (satu key terlalu ramai) atau satu record yang terus di-retry. Di produksi, lag diekspos sebagai metrik dan diberi alert, bukan dicek manual.

Testcontainers Kafka sekilas

Untuk integration test, jalankan Kafka sungguhan di container. Dengan Testcontainers 2.x (artefak org.testcontainers:testcontainers-kafka) dan @ServiceConnection, Boot otomatis mengisi bootstrap-servers:

// Potongan — integration test producer → consumer
@SpringBootTest
@Testcontainers
class OrderFlowIT {

    @Container
    @ServiceConnection
    static KafkaContainer kafka = new KafkaContainer("apache/kafka:4.1.0"); // org.testcontainers.kafka

    @Autowired OrderEventPublisher publisher;
    @Autowired StockRepository stockRepository;

    @Test
    void reservesStockWhenOrderPlaced() {
        publisher.publish(OrderPlacedEvent.of("ORD-1001", "CUS-77", List.of(new Item("SKU-1", 2))));

        await().atMost(Duration.ofSeconds(10))
                .untilAsserted(() -> assertThat(stockRepository.reservedFor("ORD-1001")).isEqualTo(2));
    }
}

Karena consumer berjalan asinkron, test menunggu dengan Awaitility (await()), bukan Thread.sleep.

Contoh di dunia backend

Layanan pembayaran menerbitkan PaymentCaptured; layanan ledger mencatat mutasi ke buku besar. Kebutuhannya: urutan per akun terjaga, pesan tidak boleh hilang, dan pesan rusak tidak boleh memblokir.

# application.yml — ledger-service
spring:
  kafka:
    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS}
    consumer:
      group-id: ledger-service
      auto-offset-reset: earliest
      max-poll-records: 100
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JacksonJsonDeserializer
      properties:
        spring.json.trusted.packages: "id.jalurjava.payment.events"
    listener:
      concurrency: 3 # sama dengan jumlah partition topic payments

// Potongan — consumer ledger
@Component
public class PaymentCapturedListener {

    private final LedgerService ledgerService;

    public PaymentCapturedListener(LedgerService ledgerService) {
        this.ledgerService = ledgerService;
    }

    // Producer memakai accountId sebagai key → mutasi satu akun selalu berurutan di satu partition
    @KafkaListener(topics = "payments")
    public void onPaymentCaptured(PaymentCapturedEvent event) {
        if (event.amount().signum() <= 0) {
            throw new IllegalArgumentException("amount harus positif: " + event.paymentId());
        }
        ledgerService.recordOnce(event.eventId(), event.accountId(), event.amount()); // idempoten via eventId
    }
}

Keputusan desainnya:

• Key accountId menjaga urutan mutasi per akun; concurrency: 3 cocok dengan 3 partition.
• Blocking retry (DefaultErrorHandler dari bagian sebelumnya) karena urutan mutasi saldo penting; database yang timeout sesaat akan pulih dalam beberapa detik.
• IllegalArgumentException tidak di-retry — payload salah tidak akan sembuh, langsung ke payments-dlt.
• recordOnce idempoten berdasarkan eventId, karena rebalance atau crash bisa membuat record diproses ulang.
• Alert bila lag group ledger-service > 1.000 selama 5 menit atau payments-dlt menerima record.

Pertanyaan interview

1. Bagaimana Kafka menjamin urutan pesan?

Kafka hanya menjamin urutan di dalam satu partition. Untuk mendapatkan urutan per entitas, producer mengirim record dengan key berupa ID entitas sehingga semua record entitas itu masuk ke partition yang sama. Di sisi consumer, satu partition dibaca satu consumer dalam group, dan retry non-blocking perlu dihindari bila urutan penting.

2. Apa hubungan jumlah partition dengan jumlah consumer?

Dalam satu consumer group, satu partition dibaca oleh tepat satu consumer, sehingga paralelisme maksimum sama dengan jumlah partition. Consumer lebih banyak dari partition akan menganggur. Group berbeda membaca topic secara independen dengan offset masing-masing.

3. Kapan offset di-commit di Spring Kafka, dan kenapa itu penting?

Secara default, listener container meng-commit offset setelah listener selesai memproses record tanpa exception (AckMode.BATCH), dan auto-commit Kafka dimatikan. Hasilnya at-least-once: bila crash sebelum commit, record diproses ulang. Auto-commit berbasis timer bisa meng-commit sebelum pemrosesan selesai sehingga pesan bisa hilang.

Jebakan umum

• Mengirim tanpa key padahal urutan per entitas penting — event bisa diproses tidak berurutan.
• Consumer lebih banyak dari partition dengan harapan makin cepat — kelebihannya hanya menganggur.
• trusted.packages: "*" di produksi — deserializer mau membuat class apa pun dari header; batasi ke package event.
• enable.auto.commit=true — offset bisa di-commit sebelum record selesai diproses, pesan hilang saat crash.
• Retry tanpa DLT atau DLT tanpa alert — poison message memblokir partition atau data "hilang" diam-diam.
• Menambah partition sembarangan — pemetaan key berubah dan urutan per key sempat rusak.

Ringkasan

• Topic dipecah menjadi partition; offset adalah urutan record di satu partition, dan urutan hanya dijamin per partition.
• Key yang sama masuk partition yang sama — pakai ID entitas sebagai key untuk ordering per entitas.
• Satu partition dibaca satu consumer per group; paralelisme maksimum = jumlah partition.
• Kafka 4.x berjalan dengan KRaft tanpa ZooKeeper; record disimpan sesuai retention dan bisa diputar ulang.
• Spring Kafka: KafkaTemplate untuk kirim, @KafkaListener untuk terima, commit offset setelah sukses (at-least-once).
• Gagal → DefaultErrorHandler (blocking, menjaga urutan) atau @RetryableTopic (non-blocking), berakhir di -dlt; pantau consumer lag.

Bacaan lanjutan

• Apache Kafka — Quickstart
• Spring for Apache Kafka — Reference
• Spring Boot Reference — Apache Kafka Support