Large Message Problem #

Apache Kafka dirancang dan dioptimalkan secara mendalam untuk menangani lalu lintas aliran data streaming (stream data) berukuran kecil secara masif — biasanya berkisar antara beberapa ratus byte hingga beberapa puluh kilobyte per pesan (seperti pesan log, event klik, atau transaksi keuangan). Namun, dalam implementasi di dunia nyata, sering kali kita dihadapkan pada kebutuhan untuk mengirimkan file berukuran besar (misalnya dokumen PDF hasil pindaian, file gambar resolusi tinggi, rekaman audio, atau arsip zip berukuran di atas 1 MB). Memaksa mengirimkan file besar ini secara langsung ke Kafka tanpa strategi khusus adalah salah satu kesalahan arsitektural (anti-pattern) terburuk. Hal ini tidak hanya memicu kegagalan instan saat pengiriman, namun juga dapat menurunkan kestabilan dan kecepatan kluster broker secara keseluruhan. Oleh karena itu, kita harus memahami konsekuensi fisik dari pesan besar, bagaimana melakukan penyelarasan parameter secara ketat jika terpaksa, serta bagaimana menerapkan solusi arsitektur Claim Check Pattern sebagai jalan keluar terbaik.

Mengapa Mengirim File Besar Adalah Anti-Pattern di Kafka #

Banyak pengembang menganggap Kafka sama seperti sistem penyimpanan berkas atau FTP server yang dapat menampung payload biner berukuran apa pun. Anggapan ini keliru karena Kafka beroperasi di tingkat memori dan I/O disk yang sangat ketat.

Berikut adalah dampak buruk pengiriman pesan besar langsung ke Kafka:

1. Masalah Pemblokiran Antrean (Head-of-Line Blocking) #

Broker Kafka melayani tulis-baca partisi menggunakan satu utas I/O per soket. Jika produser mengirimkan satu pesan berukuran 10 MB ke partisi topik, broker akan menghabiskan waktu beberapa detik untuk membaca paket biner tersebut dari socket jaringan dan menuliskannya ke disk. Selama proses penulisan file besar ini berlangsung, broker akan menunda (block) antrean ratusan ribu pesan kecil lainnya yang mengantre di belakang pesan besar tersebut. Latensi kluster akan melonjak drastis.

2. Tekanan Memori Heap dan Garbage Collection Overhead #

Ketika broker menerima pesan besar, ia harus mengalokasikan ruang memori heap JVM untuk menampung pesan tersebut sebelum ditulis ke disk. Jika ada beberapa produser yang mengirimkan pesan besar secara paralel, penggunaan memori heap broker akan melonjak secara mendadak. Kondisi ini memaksa Java Garbage Collector melakukan siklus pembersihan memori secara menyeluruh (Full GC pause). Selama jeda GC ini berjalan, broker akan membeku (freeze), berhenti merespons detak jantung (heartbeat) kluster, sehingga dicurigai mati oleh controller dan memicu proses pemilihan leader ulang (unnecessary leader election) yang mengacaukan kluster.

3. Pemborosan Disk yang Mahal #

Disk yang digunakan untuk broker Kafka biasanya menggunakan media penyimpanan cepat (SSD/NVMe) dengan konfigurasi RAID yang mahal untuk mendukung kecepatan I/O tinggi. Menyimpan file gambar atau dokumen PDF berkapasitas gigabyte di dalam disk biner Kafka sangat tidak efisien secara biaya, terlebih data tersebut akan direplikasi sebanyak Replication Factor kita (misal dikalikan 3 kali lipat di seluruh broker).

Penyelarasan Parameter Ukuran Payload (The Golden Triangle) #

Secara default, Kafka membatasi ukuran pesan maksimum sebesar 1 MB (1.048.576 byte). Jika payload kita berukuran sedikit di atas 1 MB (misalnya 1,5 MB) dan kita benar-benar harus mengirimkannya langsung ke Kafka, kita wajib menaikkan parameter batas ukuran.

Penyetelan ini tidak boleh dilakukan secara sembarangan. Kita harus menyelaraskan tiga parameter di sisi produser, broker, dan konsumen agar tidak memicu kegagalan transmisi data. Aturan penyelarasan ini sering disebut sebagai The Golden Triangle:

$$\text{max.request.size (Produser)} \le \text{message.max.bytes (Broker)} \le \text{max.partition.fetch.bytes (Konsumen)}$$

1. Konfigurasi Sisi Produser: max.request.size #

Menentukan batas ukuran maksimum dari satu Produce Request (permintaan tulis) yang boleh dikirim oleh produser ke broker. Jika objek ProducerRecord kita melebihi batas ini, klien produser akan langsung melempar exception RecordTooLargeException secara lokal sebelum data dikirim ke soket.

2. Konfigurasi Sisi Broker: message.max.bytes #

Menentukan batas ukuran maksimum pesan biner yang diizinkan ditulis ke log partisi broker. Parameter ini dapat disetel secara global di berkas server.properties atau disetel khusus per topik (melalui parameter properti topik max.message.bytes). Jika produser mengirim data di atas batas ini, broker akan langsung menolak dan mengembalikan error RecordTooLargeException.

3. Konfigurasi Sisi Konsumen: max.partition.fetch.bytes #

Menentukan batas ukuran memori maksimum per partisi yang akan diambil oleh konsumen dalam satu kali permintaan ambil data (fetch request). Parameter ini wajib disetel lebih besar atau sama dengan nilai message.max.bytes di broker. Jika tidak, konsumen kita tidak akan pernah bisa membaca pesan besar tersebut dari broker dan proses konsumsi akan terhenti selamanya (stuck) karena konsumen kekurangan kapasitas memori untuk mendeserialisasi pesan tersebut.

[!CAUTION] Kita juga harus menyesuaikan properti konsumen fetch.max.bytes (default: 52.428.800 atau 50 MB) untuk membatasi ukuran total pesan gabungan dari seluruh partisi yang dibaca dalam satu fetch agar tidak memicu OutOfMemory di sisi aplikasi konsumen kita.

Solusi Arsitektur Modern: Claim Check Pattern #

Untuk menghindari seluruh masalah di atas, arsitektur microservices modern menerapkan pola Claim Check Pattern (juga dikenal sebagai Reference-Based Messaging). Pola ini memisahkan payload biner yang berat dengan jalur pesan event ringkas Kafka.

Cara Kerja Claim Check Pattern: #

  1. Unggah Payload: Utas aplikasi produser mendeteksi file besar. Alih-alih mengirimkannya ke Kafka, produser mengunggah file tersebut ke sistem penyimpanan objek eksternal (Cloud Object Storage seperti AWS S3, Google Cloud Storage, atau MinIO lokal).
  2. Ambil Pointer: Object storage memproses unggahan dan mengembalikan URL pointer unik yang merujuk pada file tersebut (misalnya: s3://bucket-transaksi/dokumen-12345.pdf).
  3. Kirim Metadata Ringan: Produser merakit pesan JSON metadata ringan yang berisi URL pointer tersebut, lalu mengirimkannya ke topik Kafka. Pesan ini hanya berukuran beberapa ratus byte.
  4. Baca Metadata: Konsumen membaca pesan metadata dari topik Kafka secara cepat tanpa membebani disk broker.
  5. Unduh Payload: Konsumen mengekstrak URL pointer dari pesan metadata, lalu mengunduh file biner fisik langsung dari AWS S3 secara asinkron menggunakan client SDK penyimpanan objek terkait.
flowchart TD
    subgraph Direct ["Pendekatan 1: Tulis Langsung (Anti-Pattern)"]
        direction TB
        App1["Utas Aplikasi (File Besar 10MB)"] -->|"Kirim Payload Fisik"| Prod1["Kafka Producer Client"]
        Prod1 -->|"ProduceRequest (10MB)"| Broker1["Kafka Broker (orders topic)"]
        Broker1 -->|"Menulis 10MB ke Disk (Sangat Lambat)"| Disk1["Disk Broker"]
    end
    
    subgraph ClaimCheck ["Pendekatan 2: Claim Check Pattern (Rekomendasi)"]
        direction TB
        App2["Utas Aplikasi (File Besar 10MB)"] -->|"1. Unggah Payload Fisik"| S3["Object Storage (AWS S3 / GCS)"]
        S3 -->|"2. Kembalikan URL Pointer (e.g., s3://bucket/file-uuid)"| App2
        App2 -->|"3. Kirim Metadata Ringan (JSON)"| Prod2["Kafka Producer Client"]
        Prod2 -->|"4. ProduceRequest (1 KB)"| Broker2["Kafka Broker (orders topic)"]
        Broker2 -->|"Tulis 1 KB ke Disk (Instan)"| Disk2["Disk Broker"]
    end
    
    style Direct stroke:#e5e7eb
    style ClaimCheck stroke:#e5e7eb
    style S3 stroke:#2e7d32,stroke-width:2px
    style Disk1 stroke:#c62828,stroke-width:2px
    style Disk2 stroke:#2e7d32,stroke-width:2px

Keuntungan Claim Check Pattern: #

  • Performa Kafka Tetap Stabil: Broker hanya melayani pesan-pesan metadata super cepat, meminimalkan latensi dan GC overhead.
  • Skalabilitas Tanpa Batas: AWS S3 atau Google Cloud Storage dirancang khusus untuk menyimpan file berukuran besar secara masif dengan biaya yang sangat murah dibandingkan SSD broker.
  • Isolasi Kegagalan: Kegagalan jaringan saat mengunduh file besar di sisi konsumen tidak mengganggu alur pesan di partisi Kafka lainnya.

Implementasi Claim Check Pattern di Java #

Berikut adalah contoh kode Java yang membandingkan penanganan pengiriman file biner besar secara mentah (anti-pattern) dengan implementasi Claim Check Pattern yang aman menggunakan penyimpanan AWS S3 Mock:

// ANTI-PATTERN: Mengirim file PDF biner mentah berukuran besar langsung ke Kafka
// Memicu penolakan broker atau memblokir antrean pesan lainnya
public class NaiveImageProducer {
    public void sendLargeFile(KafkaProducer<String, byte[]> producer, String orderId, byte[] rawPdfBytes) {
        // ✗ Payload rawPdfBytes berukuran 15MB dikirimkan langsung
        ProducerRecord<String, byte[]> record = new ProducerRecord<>("orders-pdf-topic", orderId, rawPdfBytes);
        producer.send(record); // Memicu Exception RecordTooLargeException!
    }
}

// BENAR: Menggunakan Claim Check Pattern (Reference-Based Messaging)
// Menyimpan file fisik di S3, dan hanya mengirimkan URL pointer di Kafka
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.model.ObjectMetadata;
import java.io.ByteArrayInputStream;
import java.util.UUID;

public class ClaimCheckOrderProducer {
    private final AmazonS3 s3Client;
    private final String s3BucketName = "my-large-orders-bucket";

    public ClaimCheckOrderProducer(AmazonS3 s3Client) {
        this.s3Client = s3Client;
    }

    public void sendOrderWithLargeFile(KafkaProducer<String, String> kafkaProducer, String orderId, byte[] rawPdfBytes) {
        // 1. Buat UUID unik untuk nama file di S3
        String fileKey = "orders/" + orderId + "-" + UUID.randomUUID().toString() + ".pdf";

        // 2. Unggah file biner fisik langsung ke AWS S3
        ObjectMetadata metadata = new ObjectMetadata();
        metadata.setContentLength(rawPdfBytes.length);
        metadata.setContentType("application/pdf");

        System.out.printf("Mengunggah berkas %s ke S3 (%d bytes)...\n", fileKey, rawPdfBytes.length);
        s3Client.putObject(s3BucketName, fileKey, new ByteArrayInputStream(rawPdfBytes), metadata);

        // 3. Ambil URL referensi pointer (Claim Check)
        String s3PointerUrl = String.format("s3://%s/%s", s3BucketName, fileKey);

        // 4. Buat metadata JSON ringan untuk dikirim ke Kafka
        String kafkaJsonPayload = String.format(
            "{\"orderId\":\"%s\",\"claimCheckUrl\":\"%s\",\"sizeBytes\":%d}",
            orderId, s3PointerUrl, rawPdfBytes.length
        );

        // 5. Kirim metadata ringan (1 KB) ke Kafka
        // ✓ BENAR: Beban I/O disk broker sangat ringan, performa kluster tetap stabil
        ProducerRecord<String, String> record = new ProducerRecord<>("orders-metadata-topic", orderId, kafkaJsonPayload);
        kafkaProducer.send(record, (recMetadata, exception) -> {
            if (exception != null) {
                System.err.println("Gagal mengirim metadata ke Kafka: " + exception.getMessage());
            } else {
                System.out.printf("Metadata sukses dikirim ke partisi %d, offset %d\n", 
                    recMetadata.partition(), recMetadata.offset());
            }
        });
    }
}

Keamanan dan Pembersihan Data pada Claim Check Pattern #

Ketika mengimplementasikan Claim Check Pattern, ada dua aspek operasional non-fungsional yang wajib kita perhatikan secara serius di lingkungan produksi: Keamanan Akses Data dan Siklus Pembersihan Berkas (Data Cleanup).

1. Mengamankan Akses Payload Menggunakan Presigned URL #

Mengirimkan URL statis yang menunjuk langsung ke berkas sensitif di object storage (seperti s3://my-bucket/orders/pdf-1.pdf) dapat memicu celah keamanan. Setiap layanan konsumen yang membaca pesan Kafka tersebut dapat mengakses berkas tersebut tanpa batas waktu.

  • Solusi Keamanan: Produser dapat dikonfigurasi untuk menghasilkan Presigned URL yang memiliki batas waktu kedaluwarsa singkat (misalnya 15 menit). Konsumen harus segera mengunduh berkas tersebut sebelum masa kedaluwarsa habis. Hal ini menjamin bahwa meskipun data event di Kafka bocor atau dibaca oleh pihak lain di kemudian hari, URL unduhan fisik tersebut sudah tidak dapat digunakan kembali.

Berikut contoh generator Presigned URL menggunakan AWS SDK Java:

// ✓ BENAR: Menggunakan AWS SDK untuk menghasilkan Presigned URL dengan batas waktu kedaluwarsa singkat
import com.amazonaws.HttpMethod;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.model.GeneratePresignedUrlRequest;
import java.net.URL;
import java.util.Date;

public class SecurePointerGenerator {
    public String generatePresignedUrl(AmazonS3 s3Client, String bucketName, String fileKey) {
        Date expiration = new Date();
        long expTimeMillis = expiration.getTime() + 1000 * 60 * 15; // URL valid selama 15 menit
        expiration.setTime(expTimeMillis);
        
        GeneratePresignedUrlRequest request = new GeneratePresignedUrlRequest(bucketName, fileKey)
            .withMethod(HttpMethod.GET)
            .withExpiration(expiration);
        
        URL url = s3Client.generatePresignedUrl(request);
        return url.toString(); // Mengembalikan URL bertoken unik
    }
}

2. Otomatisasi Pembersihan Berkas di Object Storage #

Berbeda dengan broker Kafka yang memiliki kebijakan penghapusan log otomatis berbasis waktu (retention.ms), sistem penyimpanan objek eksternal (AWS S3/MinIO) tidak akan menghapus berkas yang kita unggah secara otomatis hanya karena pesan terkait di Kafka sudah kedaluwarsa. Tanpa adanya penanganan, ukuran data di S3 akan terus membengkak tanpa batas dan menimbun biaya operasional.

  • Solusi Pembersihan: Kita harus mengonfigurasi S3 Lifecycle Policy (Kebijakan Siklus Hidup) pada bucket penyimpanan terkait. Kita dapat membuat aturan otomatis di AWS S3 untuk mendeteksi berkas di bawah prefix orders/ dan menghapusnya secara permanen setelah 7 atau 14 hari, diselaraskan dengan parameter retensi topik Kafka kita.

Implementasi Claim Check Pattern di Sisi Konsumen (Downloader) #

Di sisi konsumen, aplikasi kita harus melakukan deserialisasi JSON metadata ringan dari Kafka, mengekstrak URL pointer, lalu menggunakan client SDK object storage untuk mengunduh payload fisik asli.

Berikut adalah contoh implementasi konsumen Java:

// BENAR: Konsumen membaca metadata ringan, lalu mengunduh file biner fisik dari S3 secara terpisah
import org.apache.kafka.clients.consumer.ConsumerRecord;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.model.S3Object;
import java.io.InputStream;

public class ClaimCheckConsumer {
    private final AmazonS3 s3Client;
    private final ObjectMapper objectMapper = new ObjectMapper();

    public ClaimCheckConsumer(AmazonS3 s3Client) {
        this.s3Client = s3Client;
    }

    public void processRecord(ConsumerRecord<String, String> record) {
        try {
            // 1. Deserialisasikan metadata JSON ringan dari Kafka
            OrderMetadata metadata = objectMapper.readValue(record.value(), OrderMetadata.class);
            String s3Url = metadata.getClaimCheckUrl(); // Contoh: "s3://my-large-orders-bucket/orders/pdf-1.pdf"
            
            // 2. Ekstrak nama bucket dan key dari URL referensi
            String bucketName = s3Url.substring(5, s3Url.indexOf("/", 5));
            String fileKey = s3Url.substring(s3Url.indexOf("/", 5) + 1);
            
            // 3. Unduh berkas fisik secara terpisah dari S3
            System.out.printf("Mengunduh payload dari S3: %s...\n", fileKey);
            S3Object s3Object = s3Client.getObject(bucketName, fileKey);
            
            try (InputStream in = s3Object.getObjectContent()) {
                byte[] fileBytes = in.readAllBytes();
                // ✓ Lakukan pemrosesan file PDF biner secara aman di tingkat aplikasi
                processPdf(metadata.getOrderId(), fileBytes);
            }
        } catch (Exception e) {
            System.err.println("Gagal memproses claim check record: " + e.getMessage());
        }
    }

    private void processPdf(String orderId, byte[] pdfBytes) {
        System.out.printf("Sukses memproses berkas PDF untuk Order ID %s\n", orderId);
    }
}

Alternatif Lain: Pemecahan Pesan (Chunking/Message Splitting) #

Jika kita tidak memiliki akses ke Cloud Object Storage eksternal, alternatif kedua untuk menangani file besar adalah dengan melakukan Chunking (Message Splitting).

Konsep Chunking #

  • Sisi Produser: File berukuran 10 MB dipecah menjadi 10 bagian kecil berukuran masing-masing 1 MB di dalam memori JVM produser. Setiap bagian dibungkus dengan metadata khusus:
    • File_UUID: ID unik file utuh.
    • Chunk_Index: Indeks potongan (0 sampai 9).
    • Total_Chunks: Jumlah total potongan (10).
    • Setiap potongan dikirimkan sebagai pesan mandiri ke Kafka.
  • Sisi Konsumen: Konsumen harus memiliki area memori penyangga (assembly buffer) lokal. Konsumen membaca potongan-potongan tersebut, menahannya di memori, menyusunnya kembali berdasarkan Chunk_Index, dan mengeksekusi file setelah seluruh 10 bagian lengkap terkumpul.

Kekurangan Fatal Chunking: #

  1. Masalah Kehilangan Urutan: Jika potongan dikirim ke partisi yang berbeda (atau urutannya berubah karena kegagalan retry jaringan tanpa idempotensi), merakit kembali file tersebut di konsumen akan sangat rumit dan rawan korup.
  2. Kebocoran Memori Konsumen: Konsumen harus mengalokasikan RAM yang besar untuk menampung potongan-potongan file yang belum lengkap dari ribuan pengirim berbeda. Jika salah satu potongan hilang di jalan, konsumen akan menahan sisa potongan lainnya di memori selamanya, memicu Memory Leak.
  3. Kompleksitas Kode: Membebani tim pengembang untuk menulis kode manajemen state perakitan file yang sangat rentan bug.

Oleh karena itu, Claim Check Pattern tetap merupakan standar industri teraman untuk menangani masalah pesan besar.

Ringkasan #

  • Large Message Danger: Pengiriman pesan di atas 1 MB secara langsung memicu masalah head-of-line blocking dan GC overhead yang dapat melumpuhkan kluster.
  • The Golden Triangle: Penyelarasan mutlak antara properti max.request.size (produser), message.max.bytes (broker), dan max.partition.fetch.bytes (konsumen).
  • Claim Check Pattern: Strategi menyimpan berkas besar di penyimpanan objek eksternal (AWS S3) dan hanya mengirimkan URL pointer referensi ringan ke Kafka.
  • Cost Efficiency: Menghindari penyimpanan biner mahal di disk broker Kafka yang direplikasi berulang kali.
  • Chunking Alternative: Teknik pemecahan berkas besar menjadi bagian-bagian kecil yang dikirim terpisah, namun menyimpan risiko kebocoran memori konsumen.
  • Metadata Separation: Menjaga jalur transmisi pesan Kafka tetap ramping dan cepat hanya untuk data event terstruktur ringan.

← Sebelumnya: Over-Partitioning   Berikutnya: Producer Misconfiguration →

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