Producer Workflow #

Di dalam ekosistem Apache Kafka, produser (producer) bertindak sebagai pintu gerbang utama untuk memasukkan data ke dalam kluster broker. Peran produser tidak sesederhana mengirim data melalui koneksi socket TCP biasa. Di balik pemanggilan fungsi .send() yang tampak sederhana pada aplikasi kita, terdapat arsitektur internal yang sangat kompleks, asinkron, dan sangat dioptimalkan untuk mencapai throughput yang masif dengan latensi yang minimal. Produser Kafka dirancang dengan memisahkan (decoupling) sepenuhnya antara utas aplikasi utama dengan utas jaringan I/O. Memahami bagaimana komponen-komponen internal produser bekerja sama adalah kunci penting bagi setiap arsitek sistem untuk merancang aplikasi pengirim pesan yang tangguh dan efisien.

Siklus Hidup Pengiriman Pesan: Langkah Demi Langkah #

Ketika aplikasi kita memanggil perintah producer.send(record), pesan tidak langsung terbang ke jaringan. Pesan tersebut harus melalui serangkaian pemrosesan internal di dalam memori JVM klien.

Berikut adalah alur perjalanan pesan dari kode aplikasi kita hingga sampai ke broker:

1. Pengambilan Metadata Kluster (Metadata Fetching) #

Sebelum pesan dapat diproses, produser harus mengetahui struktur fisik kluster Kafka saat ini: broker mana yang hidup, topik apa saja yang tersedia, partisi apa saja yang dimiliki topik tersebut, dan broker mana yang bertindak sebagai Leader untuk masing-masing partisi.

  • Mekanisme: Jika metadata belum ada di memori lokal klien (atau sudah kedaluwarsa), produser akan melakukan operasi pemblokiran asinkron untuk meminta metadata terbaru dari salah satu broker aktif (bootstrap servers). Setelah didapatkan, metadata disimpan di memori cache lokal klien.

2. Serialisasi Data (Serialization) #

Kafka adalah sistem penyimpanan byte mentah. Ia tidak mempedulikan struktur objek, kelas, atau tipe data aplikasi kita.

  • Mekanisme: Komponen Serializer mengubah objek kunci (key) dan nilai pesan (value) menjadi array of bytes (byte[]). Kafka menyediakan serializer bawaan untuk tipe data dasar (String, Integer, Double, ByteArray), namun kita dapat menggunakan format kustom seperti Avro atau Protobuf.

3. Penentuan Partisi (Partitioning) #

Setelah data diubah menjadi byte, produser harus memutuskan ke partisi mana pesan tersebut harus dikirim. Tugas ini didelegasikan kepada Partitioner.

  • Mekanisme: Jika pesan memiliki kunci (key), partitioner default menggunakan algoritma hash Murmur2 untuk mengikat kunci tersebut ke nomor partisi tertentu secara konsisten. Jika pesan tidak memiliki kunci, Kafka menggunakan Sticky Partitioner untuk mendistribusikan data secara seimbang dalam bentuk batch.

4. Penyanggaan Memori (Record Accumulator) #

Setelah partisi tujuan ditentukan, pesan masuk ke jantung arsitektur produser Kafka: Record Accumulator. Di sinilah pembagian utas terjadi. Utas aplikasi utama selesai bertugas di sini dan dapat langsung kembali melakukan kalkulasi bisnis, sementara pesan disimpan sementara di memori buffer.

  • Mekanisme: Record Accumulator mengelompokkan pesan-pesan berdasarkan topik dan partisi tujuan mereka. Setiap partisi memiliki antrean batch (double-ended queue of ProducerBatch). Pesan baru akan disematkan di akhir batch aktif saat ini.

5. Utas Pengirim Jaringan (Sender Thread) #

Komponen terakhir adalah Sender Thread, sebuah thread latar belakang (background daemon thread) yang bertugas melakukan operasi I/O jaringan yang sesungguhnya.

  • Mekanisme: Utas ini terus memantau Record Accumulator. Ketika sebuah batch dinilai siap kirim (karena ukurannya penuh atau waktu tunggunya habis), Sender Thread akan mengambil batch tersebut, merakitnya menjadi satu permintaan tulis jaringan (Produce Request), dan mengirimkannya secara asinkron ke broker leader partisi tujuan menggunakan Java NIO Selector (epoll atau kqueue di tingkat OS).

Arsitektur Internal Utas Produser #

Visualisasi arsitektur internal produser Kafka memperlihatkan pemisahan tegas antara utas pengirim aplikasi (App Thread) dengan utas pengirim jaringan (Sender Thread):

flowchart TD
    subgraph App_Execution ["1. Utas Aplikasi Utama (App Thread)"]
        direction TB
        SendCall["producer.send(record)"] --> Metadata["1. Metadata Fetcher"]
        Metadata --> Serializer["2. Serializer (Object -> byte[])"]
        Serializer --> Partitioner["3. Partitioner (Pilih Partisi)"]
    end

    subgraph Memory_Accumulator ["2. Buffer Penyangga (Record Accumulator)"]
        direction TB
        subgraph Partition_Queues ["Antrean Batch Per Partisi"]
            P0_Q["Queue Partisi 0 <br> [Batch A] -> [Batch B]"]
            P1_Q["Queue Partisi 1 <br> [Batch C]"]
        end
        BP["Buffer Pool Manager (RAM)"]
    end

    subgraph Network_Execution ["3. Utas Jaringan (Sender Thread)"]
        direction TB
        Sender["Sender I/O Thread"] --> Selector["Java NIO Selector"]
    end

    Partitioner -->|"Kirim ke Buffer"| Partition_Queues
    BP -.->|"Alokasikan Memori"| Partition_Queues
    Partition_Queues -.->|"Ambil Batch Siap Kirim"| Sender
    Selector -->|"Kirim Paket ke Broker via Socket"| Broker["Kafka Broker Leader"]

    style App_Execution stroke:#e5e7eb
    style Memory_Accumulator stroke:#e5e7eb
    style Network_Execution stroke:#e5e7eb
    style P0_Q stroke:#0288d1,stroke-width:2px
    style P1_Q stroke:#2e7d32,stroke-width:2px

Struktur Data Record Accumulator & Concurrency Control #

Di tingkat kode Java, Record Accumulator dikelola menggunakan struktur data yang sangat optimal untuk akses konkuren multi-thread.

Antrean per partisi disimpan di dalam sebuah map thread-safe:

// Struktur data internal Record Accumulator
ConcurrentMap<TopicPartition, ArrayDeque<ProducerBatch>> batches;

Mengapa Desain ini Sangat Cepat? #

  1. Thread-Safe Map: ConcurrentMap memungkinkan beberapa thread aplikasi memanggil .send() secara paralel untuk partisi yang berbeda tanpa memblokir satu sama lain.
  2. Locking yang Sempit (Fine-Grained Locking): Ketika beberapa thread mencoba menulis pesan ke partisi yang sama, Kafka tidak mengunci seluruh Record Accumulator. Penguncian (locking) disekat hanya pada tingkat antrean ArrayDeque dari partisi spesifik tersebut.
  3. ProducerBatch: Alih-alih menyalin pesan satu per satu, pesan baru langsung ditransisikan ke dalam array byte fisik di dalam buffer dari ProducerBatch aktif menggunakan system call salin memori cepat (System.arraycopy).

Siklus Kerja Utas Pengirim (Sender Thread Mechanics) #

Sender Thread adalah mesin I/O non-blocking yang mengeksekusi Selector Loop Java NIO. Utas ini terus-menerus berjalan melakukan siklus evaluasi berikut:

1. Pemindaian Accumulator (Ready Check) #

Sender Thread memeriksa seluruh antrean partisi untuk menentukan batch mana saja yang sudah “siap kirim”. Sebuah batch dinyatakan siap jika:

  • Ukuran batch telah penuh menyentuh parameter batch.size.
  • Batch tersebut telah tertahan di accumulator selama lebih dari parameter linger.ms.
  • Terjadi pemanggilan paksa flush (producer.flush()).

2. Pengelompokan Berdasarkan Node Broker (Node Grouping) #

Pesan-pesan di accumulator diatur berdasarkan partisi. Namun, jaringan TCP beroperasi berdasarkan server (Node Broker).

  • Konsolidasi: Sender Thread melakukan pengelompokan ulang (regrouping). Jika ada batch untuk TopicA-Partition0 dan TopicB-Partition2 yang kebetulan memiliki leader broker yang sama (misal Broker 2), Sender Thread akan menyatukan kedua batch ini ke dalam satu ClientRequest tunggal.
  • Efisiensi: Langkah ini memangkas overhead TCP header dan mengoptimalkan ukuran paket jaringan.

3. Kontrol In-Flight Requests #

Sebelum mengirimkan request ke socket, Sender Thread memverifikasi batas in-flight request:

  • Properti max.in.flight.requests.per.connection (default: 5) mengontrol jumlah request tertunda (belum menerima ACK) yang boleh dikirimkan produser ke satu broker secara bersamaan.
  • Jika batas ini tercapai, Sender Thread akan menahan request baru untuk broker tersebut demi mencegah terjadinya kemacetan bus jaringan (network congestion).

Simulasi Timeline Alokasi Memori Buffer Pool #

Mari kita simulasikan bagaimana memori BufferPool dialokasikan dan dikembalikan saat produser asinkron berjalan dengan spesifikasi batch.size=16KB:

  • Detik 0.000: Utas Aplikasi memanggil .send() untuk pesan berukuran 10 KB di Partisi 0.
    • Record Accumulator melihat belum ada batch aktif untuk Partisi 0.
    • Accumulator meminta memori 16 KB ke Buffer Pool.
    • Buffer Pool memberikan satu ByteBuffer kosong (16 KB). Sisa RAM Buffer Pool tersisa 32 MB - 16 KB.
    • Pesan 10 KB ditulis ke ByteBuffer tersebut. Sisa ruang kosong di batch tersebut adalah 6 KB.
  • Detik 0.002: Utas Aplikasi mengirim pesan kedua berukuran 5 KB ke Partisi 0.
    • Accumulator melihat batch aktif Partisi 0 masih memiliki ruang kosong (6 KB > 5 KB).
    • Pesan 5 KB langsung ditulis ke ByteBuffer yang sama. Sisa ruang kosong tinggal 1 KB.
  • Detik 0.004: Utas Aplikasi mengirim pesan ketiga berukuran 5 KB ke Partisi 0.
    • Accumulator mendeteksi sisa ruang kosong tidak cukup (1 KB < 5 KB).
    • Batch pertama (15 KB) dinyatakan CLOSED dan siap diambil oleh Sender Thread.
    • Accumulator meminta ByteBuffer 16 KB baru ke Buffer Pool untuk menampung pesan 5 KB tersebut.
  • Detik 0.005: Sender Thread mengambil batch pertama yang closed, mengirimkannya ke broker, dan menerima ACK sukses.
    • Memori ByteBuffer batch pertama dibersihkan secara total.
    • Memori tersebut dikembalikan kembali ke Buffer Pool. RAM Buffer Pool bertambah kembali sebesar 16 KB.

Pola daur ulang memori ini mengeliminasi pembuatan objek byte array dinamis yang dapat memicu Garbage Collection pause.

Perbandingan: Pengiriman Asinkron vs Sinkron #

Sebagai pengembang aplikasi, kita memiliki kendali penuh untuk mengeksekusi pengiriman data secara asinkron (menggunakan callback) atau sinkron (memblokir thread menunggu hasil).

1. Skenario Sinkron (ANTI-PATTERN untuk Throughput Tinggi) #

Pengiriman sinkron dilakukan dengan memanggil fungsi .get() langsung pada objek Future hasil dari .send().

// ANTI-PATTERN: Memblokir thread aplikasi, melumpuhkan throughput
public class SynchronousProducer {
    public void sendData(KafkaProducer<String, String> producer) {
        for (int i = 0; i < 10000; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("orders", "Key-" + i, "Data-" + i);
            try {
                // Memanggil .get() memaksa thread menunggu ACK dari broker sebelum mengirim pesan berikutnya
                RecordMetadata metadata = producer.send(record).get(); // ✗ Throughput anjlok!
                System.out.printf("Pesan sukses ditulis ke offset %d\n", metadata.offset());
            } catch (Exception e) {
                System.err.println("Gagal mengirim data: " + e.getMessage());
            }
        }
    }
}
  • Dampak Buruk: Throughput hancur total karena siklus pengiriman dibatasi oleh latensi jaringan RTT (Round Trip Time). Jika RTT jaringan adalah 10 ms, maka maksimal throughput kita hanya 100 pesan per detik per thread produser.

2. Skenario Asinkron (BENAR & Direkomendasikan) #

Pengiriman asinkron sejati memanfaatkan callback untuk menangani hasil tanpa pernah menahan eksekusi thread bisnis utama.

// BENAR: Menggunakan Callback asinkron, throughput optimal
public class AsynchronousProducer {
    public void sendData(KafkaProducer<String, String> producer) {
        for (int i = 0; i < 10000; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("orders", "Key-" + i, "Data-" + i);
            
            // Pengiriman non-blocking, langsung lanjut ke iterasi berikutnya
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        // ✓ Tangani error di thread terpisah (Sender Thread)
                        System.err.printf("Gagal mengirim pesan: %s\n", exception.getMessage());
                    } else {
                        // ✓ Sukses mencatat data
                        System.out.printf("Pesan sukses ditulis ke partisi %d, offset %d\n", 
                            metadata.partition(), metadata.offset());
                    }
                }
            });
        }
    }
}

Keamanan Memori dan Mekanisme Backpressure #

Apa yang terjadi jika produser kita menghasilkan data dengan kecepatan 100 MB/detik, sedangkan jaringan internet kita hanya mampu mengirimkan data ke broker dengan kecepatan 20 MB/detik?

Di sinilah sistem pertahanan Backpressure Kafka aktif bekerja:

  1. Pesan baru terus masuk ke Record Accumulator.
  2. Karena data tidak kunjung terkirim, Buffer Pool kehabisan ByteBuffer kosong. Memori 32 MB terisi penuh.
  3. Ketika pemanggilan .send() berikutnya dipanggil oleh aplikasi, fungsi tersebut tidak akan mengalokasikan RAM baru (yang bisa memicu OOM).
  4. Fungsi .send() akan ditahan secara sinkron (blocked) selama parameter durasi waktu max.block.ms (default: 60.000 ms / 1 menit).
  5. Thread aplikasi kita akan melambat secara alami mengikuti kecepatan jaringan. Ini adalah mekanisme backpressure yang sangat aman.
  6. Jika dalam waktu 1 menit broker belum pulih atau memori buffer belum dibebaskan, produser akhirnya menyerah dan melempar exception TimeoutException.

Menampilkan Metrik Pemantauan Buffer JMX #

Untuk mendeteksi gejala bottleneck atau backpressure di produser, kita wajib memantau metrik JMX berikut:

# Mengukur waktu rata-rata utas aplikasi terblokir menunggu memori buffer pool
kafka.producer:type=producer-metrics,client-id=[clientId],name=bufferpool-wait-time-ns

# Mengukur waktu rata-rata record pesan tertahan di dalam Record Accumulator sebelum dikirim
kafka.producer:type=producer-metrics,client-id=[clientId],name=record-queue-time-avg

Mekanisme Pembaruan Metadata Kluster secara Otomatis #

Metadata kluster adalah peta jalan bagi produser. Tanpa metadata yang akurat, produser tidak akan tahu broker mana yang bertindak sebagai Leader partisi untuk mengirimkan pesan kita. Oleh karena itu, klien Kafka memiliki mekanisme internal untuk selalu memperbarui metadata ini secara otomatis dan asinkron.

1. Pembaruan Periodik (Periodic Refresh) #

Produser akan melakukan pembaruan metadata secara berkala di latar belakang setiap kali waktu tunggu mencapai parameter metadata.max.age.ms (secara default disetel ke 300.000 ms atau 5 menit). Hal ini dilakukan untuk memastikan bahwa produser mendeteksi jika ada penambahan partisi baru pada topik, atau jika ada broker baru yang ditambahkan ke dalam kluster.

2. Pembaruan Reaktif (Reactive Refresh) #

Jika terjadi kegagalan saat mengirimkan data (misalnya broker leader tiba-tiba mati), broker yang tersisa akan menolak permintaan tulis dari produser dan melempar exception seperti NotLeaderOrFollowerException atau KafkaStorageException. Ketika produser menerima galat (error) semacam ini, ia menyadari bahwa peta metadata lokalnya sudah usang. Klien produser akan segera menandai metadata lokal sebagai stale dan memicu pembaruan metadata reaktif sesegera mungkin sebelum mencoba mengirim ulang (retry) pesan tersebut.

Kustomisasi Partitioner: Menulis Algoritma Pembagian Sendiri #

Secara default, produser Kafka menggunakan hashing Murmur2 jika kita menyertakan kunci (key) pada pesan, atau menggunakan Sticky Partitioner jika tidak ada kunci. Namun, terkadang kebutuhan bisnis menuntut kontrol penuh atas penempatan pesan di dalam partisi. Untuk memenuhi kebutuhan tersebut, kita dapat mengimplementasikan komponen Partitioner kustom.

Sebagai contoh, bayangkan kita memiliki sistem pemrosesan pesanan e-commerce. Kita ingin menempatkan semua transaksi pelanggan VIP di partisi 0 agar mendapatkan prioritas resource pemrosesan yang lebih tinggi dan isolasi ketat, sementara transaksi pelanggan reguler disebar secara acak di partisi lainnya.

Berikut adalah perbandingan antara cara penanganan pembagian partisi yang salah di tingkat aplikasi (anti-pattern) dengan cara yang benar menggunakan interface Partitioner kustom:

// ANTI-PATTERN: Menentukan partisi secara manual di dalam kode utama aplikasi
public class OrderService {
    public void sendOrder(KafkaProducer<String, String> producer, Order order) {
        int targetPartition = 0;
        if (order.isVip()) {
            targetPartition = 0; // Mengirim semua VIP ke partisi 0 secara manual
        } else {
            // JANGAN LAKUKAN INI: Logika hashing manual mengotori kode bisnis dan sulit dipelihara
            targetPartition = Math.abs(order.getCustomerId().hashCode()) % 3 + 1;
        }
        
        // ✗ Menyebutkan nomor partisi secara eksplisit saat membuat objek Record
        ProducerRecord<String, String> record = new ProducerRecord<>(
            "orders-topic", targetPartition, order.getId(), order.toJson()
        );
        producer.send(record);
    }
}

// BENAR: Membuat implementasi Partitioner terpisah dan meregistrasikannya via properti konfigurasi
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;

public class CustomerPriorityPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        // Ambil total partisi yang tersedia untuk topik ini di broker
        int numPartitions = cluster.partitionCountForTopic(topic);
        
        if (key == null) {
            return 0; // Pesan tanpa key diarahkan ke partisi 0 sebagai fallback
        }
        
        String keyString = (String) key;
        if (keyString.startsWith("VIP_")) {
            // ✓ Porsi khusus: Semua pelanggan VIP masuk ke partisi 0
            return 0;
        }
        
        // ✓ Hashing standard Murmur2 untuk pelanggan reguler
        // Sebarkan data di partisi selain partisi 0 (yaitu partisi 1 hingga numPartitions - 1)
        int hash = org.apache.kafka.common.utils.Utils.toPositive(
            org.apache.kafka.common.utils.Utils.murmur2(keyBytes)
        );
        return (hash % (numPartitions - 1)) + 1;
    }

    @Override
    public void close() {
        // Melakukan pembersihan resource jika diperlukan saat produser dimatikan
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // Membaca konfigurasi tambahan yang dilewatkan ke produser jika ada
    }
}

Setelah kelas partitioner kustom kita selesai dibuat, kita cukup meregistrasikannya ke dalam properti konfigurasi produser:

import org.apache.kafka.clients.producer.ProducerConfig;
import java.util.Properties;

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

// ✓ Daftarkan kelas partitioner kustom kita di sini
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, CustomerPriorityPartitioner.class.getName());

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

Dengan memindahkan logika pembagian data ke kelas Partitioner khusus, kode bisnis utama kita tetap bersih dari detail infrastruktur partisi Kafka, dan kita dapat dengan mudah mengganti strategi pembagian partisi di masa depan tanpa mengubah kode pemrosesan pesanan utama kita.

Ringkasan #

  • Decoupled Architecture: Produser memisahkan proses pengiriman antara thread aplikasi (App Thread) dengan thread jaringan (Sender Thread) secara asinkron.
  • Record Accumulator: Berfungsi sebagai wadah penyangga memori tempat pesan dikelompokkan menjadi batch per partisi sebelum dikirim.
  • Buffer Pool Manager: Optimasi memori JVM yang mendaur ulang ByteBuffer tetap berukuran batch.size untuk menghindari Garbage Collection overhead.
  • Node Grouping: Sender Thread menyatukan beberapa batch partisi dengan leader broker yang sama ke dalam satu Produce Request untuk efisiensi koneksi.
  • Skenario Asinkron: Selalu gunakan Callback asinkron pada .send() di lingkungan produksi untuk menghindari bottleneck latensi jaringan.
  • Backpressure: Diatur oleh parameter buffer.memory dan max.block.ms, menahan secara aman laju utas aplikasi jika kapasitas buffer penuh.

← Sebelumnya: Log Compaction   Berikutnya: Serialization →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact