Client Log: Konfigurasi dan Troubleshooting Logging di Sisi Klien #

Ketika kita membangun sistem berbasis event-driven menggunakan Apache Kafka, kestabilan sistem tidak hanya ditentukan oleh kesehatan kluster broker, tetapi juga oleh keandalan aplikasi produser dan konsumen di sisi klien. Saat terjadi kehilangan pesan (message loss), duplikasi data, atau kegagalan koneksi, memeriksa log broker saja sering kali tidak cukup. Kita memerlukan visibilitas penuh terhadap apa yang terjadi di dalam pustaka klien (Kafka Client Library) yang berjalan sebagai bagian dari aplikasi kita.

Sayangnya, banyak tim pengembang yang mengabaikan logging sisi klien. Mereka membiarkan konfigurasi logging bawaan mencampuradukkan pesan debug internal Kafka dengan log bisnis aplikasi, atau bahkan mematikan log klien sama sekali karena dianggap terlalu bising (verbose). Akibatnya, ketika terjadi insiden kritis seperti kegagalan komitmen offset atau pembagian ulang grup (consumer rebalance), mereka kehilangan jejak diagnostik yang krusial.

Dalam panduan ini, kita akan membedah arsitektur logging pustaka klien Kafka berbasis facade SLF4J, menyusun konfigurasi Logback dan Log4j2 yang optimal untuk berbagai lingkungan (development vs produksi), menerjemahkan pesan error kritis klien, serta mengimplementasikan teknik korelasi log menggunakan Mapped Diagnostic Context (MDC) untuk melacak pesan secara end-to-end.

Arsitektur Logging pada Klien Apache Kafka #

Pustaka Java klien Apache Kafka (kafka-clients) dirancang menggunakan SLF4J (Simple Logging Facade for Java) sebagai facade pencatatan log. Pustaka ini tidak mengikatkan dirinya pada framework logging tertentu, melainkan menyerahkan keputusan pemilihan backend logging kepada aplikasi kita yang mengonsumsinya.

flowchart TD
    App["Aplikasi Kita / Kode Bisnis"] --> Facade["SLF4J Facade API (Logger/LoggerFactory)"]
    Lib["Kafka Client Library"] --> Facade
    Facade --> Logback["Logback Binding"]
    Facade --> Log4j2["Log4j2 Binding"]
    Facade --> JUL["JUL Binding"]
    Logback --> LogbackConf["logback.xml / logback-test.xml"]
    Log4j2 --> Log4j2Conf["log4j2.xml"]
    JUL --> JULConf["logging.properties"]

Dengan arsitektur ini, kita dapat menyelaraskan format, rotasi, dan tujuan keluaran log klien Kafka dengan sistem logging aplikasi kita sendiri. Tiga backend logging yang paling populer digunakan di ekosistem Java adalah Logback (default pada Spring Boot), Log4j2, dan Java Util Logging (JUL).

Strategi Konfigurasi Log Klien: Lingkungan Dev vs Produksi #

Tantangan utama dalam mengelola log klien Kafka adalah volume data log yang dihasilkan. Pada tingkat verbosity DEBUG atau TRACE, pustaka klien Kafka akan mencatat setiap detak jantung (heartbeat) konsumen, pembaruan metadata kluster, dan proses polling socket. Untuk itu, kita harus membedakan strategi konfigurasi log klien berdasarkan lingkungan operasional.

1. Lingkungan Pengembangan (Development/Staging) #

Di lingkungan ini, kita membutuhkan detail sebanyak mungkin untuk memahami jabat tangan koneksi, struktur partisi topik, dan perilaku pembagian tugas rebalance. Kita disarankan menyetel level log kategori org.apache.kafka ke DEBUG.

2. Lingkungan Produksi #

Di lingkungan produksi, penulisan log berlebih di tingkat DEBUG akan memicu overhead CPU dan I/O disk yang signifikan pada server aplikasi kita. Oleh karena itu, kita harus membatasi level log default ke INFO atau WARN, dan hanya menaikkan tingkat verbosity pada modul tertentu jika sedang aktif melakukan investigasi masalah.

Berikut adalah contoh konfigurasi Logback (logback.xml) yang seimbang untuk produksi:

<?xml version="1.0" encoding="UTF-8"?>
<configuration>

    <!-- Appender untuk Konsol (Standard Output) -->
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
        </encoder>
    </appender>

    <!-- Appender File untuk Log Klien Kafka Terisolasi -->
    <appender name="KAFKA_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>/var/log/app/kafka-client.log</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
            <fileNamePattern>/var/log/app/kafka-client-%d{yyyy-MM-dd}.%i.log.gz</fileNamePattern>
            <maxFileSize>50MB</maxFileSize>
            <maxHistory>7</maxHistory>
            <totalSizeCap>500MB</totalSizeCap>
        </rollingPolicy>
        <encoder>
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
        </encoder>
    </appender>

    <!-- Setelan Logging Root Aplikasi -->
    <root level="INFO">
        <appender-ref ref="CONSOLE"/>
    </root>

    <!-- Konfigurasi Khusus Pustaka Klien Kafka -->
    <!-- Secara default, setel ke WARN untuk menghindari noise -->
    <logger name="org.apache.kafka" level="WARN" additivity="false">
        <appender-ref ref="KAFKA_FILE"/>
        <appender-ref ref="CONSOLE"/>
    </logger>

    <!-- Kecuali untuk Group Coordinator, gunakan INFO agar kita dapat melacak Rebalance -->
    <logger name="org.apache.kafka.clients.consumer.internals.ConsumerCoordinator" level="INFO" additivity="false">
        <appender-ref ref="KAFKA_FILE"/>
        <appender-ref ref="CONSOLE"/>
    </logger>

    <!-- Pantau proses koneksi awal (Metadata) pada level INFO -->
    <logger name="org.apache.kafka.clients.NetworkClient" level="INFO" additivity="false">
        <appender-ref ref="KAFKA_FILE"/>
    </logger>

</configuration>

Berikut adalah padanannya jika kita menggunakan Log4j2 (log4j2.xml):

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
    <Appenders>
        <Console name="ConsoleAppender" target="SYSTEM_OUT">
            <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
        </Console>
        
        <RollingFile name="KafkaRollingFile" fileName="/var/log/app/kafka-client.log"
                     filePattern="/var/log/app/kafka-client-%d{yyyy-MM-dd}-%i.log.gz">
            <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
            <Policies>
                <SizeBasedTriggeringPolicy size="50 MB"/>
                <TimeBasedTriggeringPolicy/>
            </Policies>
            <DefaultRolloverStrategy max="7"/>
        </RollingFile>
    </Appenders>
    
    <Loggers>
        <Root level="info">
            <AppenderRef ref="ConsoleAppender"/>
        </Root>
        
        <!-- Pustaka Klien Kafka secara umum -->
        <Logger name="org.apache.kafka" level="warn" additivity="false">
            <AppenderRef ref="KafkaRollingFile"/>
            <AppenderRef ref="ConsoleAppender"/>
        </Logger>
        
        <!-- Pantau siklus hidup grup konsumen -->
        <Logger name="org.apache.kafka.clients.consumer.internals.ConsumerCoordinator" level="info" additivity="false">
            <AppenderRef ref="KafkaRollingFile"/>
        </Logger>
    </Loggers>
</Configuration>

Diagram Alur Deteksi Kesalahan pada Pustaka Klien #

Sebelum kita menyelami detail pesan kesalahan spesifik, kita perlu memahami bagaimana pustaka klien memproses kegagalan transmisi dari tingkat socket jaringan hingga sampai ke penanganan eksepsi di kode aplikasi kita.

Berikut adalah visualisasi alur deteksi kesalahan internal pada pustaka klien Java:

flowchart TD
    AppStart["Aplikasi: producer.send() / consumer.poll()"] --> NetworkSend["NetworkClient: Kirim Request via Socket TCP"]
    NetworkSend --> SocketCheck{"Apakah Koneksi Socket OK?"}
    
    SocketCheck -- "Tidak (RST / Timeout)" --> ErrorSocket["Lempar NetworkException / DisconnectException"]
    ErrorSocket --> RetryCheck{"Apakah Mengizinkan Retries?"}
    
    SocketCheck -- "Ya" --> AwaitResponse["Tunggu Response dari Broker"]
    AwaitResponse --> RespCheck{"Apakah Respon Berisi Error Code?"}
    
    RespCheck -- "Ya (Retriable, e.g. NotLeaderForPartition)" --> RetryCheck
    RespCheck -- "Ya (Non-Retriable, e.g. RecordTooLarge)" --> FailImmediate["Bungkus sebagai Exception Fatal"]
    RespCheck -- "Tidak" --> Success["Kembalikan Metadata / Record Sukses"]
    
    RetryCheck -- "Ya (Retries>"0 & Waktu Belum Timeout)\" --> RefreshMetadata[\"Request Refresh Metadata Kluster\""]
    RefreshMetadata --> NetworkSend
    
    RetryCheck -- "Tidak" --> FailImmediate
    
    FailImmediate --> Callback["Picu Callback / Lempar RuntimeException ke Aplikasi"]

Dengan memahami diagram ini, kita tahu bahwa tidak semua peringatan (WARN) di log klien berarti aplikasi kita gagal mengirim atau menerima data. Banyak di antaranya adalah kesalahan yang dapat dipulihkan secara otomatis oleh sistem (retriable errors).

Mendekode dan Menyelesaikan Log Error Kunci Klien #

Ketika terjadi masalah fatal yang tidak dapat ditangani oleh mekanisme retries otomatis klien, pustaka klien akan mencatat error ke log dan melemparkannya ke thread aplikasi kita. Mari kita bedah lima error kunci yang paling sering ditemui di produksi.

1. RecordTooLargeException #

Error ini muncul ketika produser mencoba mengirimkan satu pesan tunggal (atau batch pesan) yang ukurannya melebihi batas penerimaan maksimum yang dikonfigurasi.

Contoh Pesan Log:

org.apache.kafka.common.errors.RecordTooLargeException: The request included a message larger than the max message size the server will accept.

Analisis Akar Masalah: Kafka memiliki beberapa parameter penjaga gerbang ukuran pesan yang harus diselaraskan secara konsisten dari ujung produser hingga broker dan konsumen:

• Di produser: max.request.size membatasi ukuran request maksimum yang dikirim dalam satu panggilan jaringan (default 1 MB).
• Di broker: message.max.bytes (konfigurasi per topik atau global) membatasi ukuran record maksimum yang diizinkan masuk ke disk (default 1 MB).
• Di konsumen: max.partition.fetch.bytes membatasi jumlah data maksimum yang ditarik dari setiap partisi dalam sekali fetch (default 1 MB).

Jika produser mencoba mengirim pesan berukuran 2 MB tanpa mengubah setelan ini, broker akan langsung menolak request tersebut dan melempar RecordTooLargeException ke callback produser.

Langkah Solusi: Jika kita benar-benar harus mengirim pesan berukuran besar (misalnya 5 MB), kita wajib mengubah parameter di ketiga lini:

1. Di broker (server.properties atau dinamis per topik):

   message.max.bytes=5242880
   

2. Di produser:

   max.request.size=5242880
   

3. Di konsumen:

   max.partition.fetch.bytes=5242880
   

Catatan: Sangat disarankan untuk mengaktifkan kompresi data (compression.type=zstd atau lz4) di sisi produser terlebih dahulu sebelum menaikkan batas ukuran ini.

2. CommitFailedException (Offset Commit Failed) #

Error ini eksklusif terjadi di sisi konsumen ketika menggunakan skema manajemen commit offset manual.

Contoh Pesan Log:

org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time processing messages. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.

Analisis Akar Masalah: Pesan log di atas sebenarnya sudah menjelaskan penyebabnya dengan sangat baik. Di dalam arsitektur konsumen Kafka:

• Thread konsumen berkewajiban mengirimkan sinyal detak jantung (heartbeat) secara berkala untuk menyatakan dirinya sehat.
• Selain heartbeat, konsumen harus memanggil fungsi poll() secara rutin. Rentang waktu maksimal antar pemanggilan poll() dibatasi oleh properti max.poll.interval.ms (default 5 menit).
• Jika thread aplikasi kita menghabiskan waktu terlalu lama untuk memproses pesan-pesan yang dikembalikan oleh satu panggilan poll() (misalnya melakukan operasi I/O database yang lambat, integrasi API eksternal yang mengalami timeout), waktu jeda antar poll() berikutnya akan melampaui batas 5 menit.
• Coordinator di sisi broker akan menganggap konsumen tersebut telah mati (hung), mengeluarkan konsumen dari grup, dan memicu rebalance untuk mengalihkan partisi ke anggota grup lainnya.
• Ketika konsumen yang lambat tadi akhirnya selesai memproses datanya dan mencoba memanggil commitSync(), broker menolak komitmen tersebut karena kepemilikan partisi sudah berpindah tangan.

Langkah Solusi:

1. Turunkan Ukuran Batch: Batasi jumlah record yang ditarik dalam sekali poll dengan memperkecil nilai max.poll.records (default 500). Setel menjadi misalnya 50 atau 10 jika pemrosesan setiap record memakan waktu lama.
2. Tingkatkan Batas Waktu: Naikkan nilai max.poll.interval.ms pada konfigurasi konsumen (misalnya menjadi 10 atau 15 menit) jika proses batching tidak bisa dihindari.
3. Delegasikan ke Thread Pool: Implementasikan arsitektur multi-threading di mana thread konsumen utama hanya bertugas melakukan poll() dan memasukkan tugas ke dalam antrean memori (BlockingQueue), sementara eksekusi bisnis dilakukan oleh kumpulan thread pekerja (worker threads).

3. SSLHandshakeException #

Error ini terjadi pada kluster Kafka yang mengaktifkan enkripsi transit (SSL/TLS) ketika pustaka klien gagal membangun koneksi aman dengan broker.

Contoh Pesan Log:

javax.net.ssl.SSLHandshakeException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to neighboring target

Analisis Akar Masalah: JVM tempat aplikasi klien kita berjalan tidak mempercayai otoritas sertifikat (Certificate Authority - CA) yang menandatangani sertifikat SSL milik broker Kafka. Hal ini umum terjadi pada penggunaan sertifikat mandiri (self-signed certificate) atau CA internal perusahaan yang sertifikat root-nya belum didaftarkan ke dalam truststore bawaan Java.

Langkah Solusi: Kita harus menyediakan berkas truststore yang berisi sertifikat CA publik broker ke dalam konfigurasi klien Kafka:

security.protocol=SSL
ssl.truststore.location=/var/private/ssl/client.truststore.jks
ssl.truststore.password=PasswordTruststoreKita123

Jika kita juga mengaktifkan otentikasi dua arah (mTLS), pastikan berkas keystore klien yang berisi sertifikat privat dan kunci publik klien dikonfigurasi dengan benar:

ssl.keystore.location=/var/private/ssl/client.keystore.jks
ssl.keystore.password=PasswordKeystoreKita123
ssl.key.password=PasswordKunciPrivatKita123

4. DisconnectException dan Network Timeouts #

Pesan pemutusan koneksi ini sering kali membingungkan karena biasanya muncul sebagai tingkat peringatan (WARN) namun terjadi berulang kali.

Contoh Pesan Log:

org.apache.kafka.common.errors.DisconnectException: Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.

Analisis Akar Masalah:

1. Broker Mati: Node broker target benar-benar sedang mati atau dalam proses restart.
2. Koneksi Idle Diputus: Secara default, Kafka memiliki properti connections.max.idle.ms (default 9 menit) baik di sisi broker maupun klien. Jika tidak ada transmisi data antara klien dan broker tertentu selama waktu tersebut (misalnya pada topik yang jarang menerima pesan), salah satu pihak akan menutup koneksi TCP secara bersih. Ketika klien membutuhkan koneksi kembali, ia akan mencatat DisconnectException sebelum membangun socket baru. Skenario ini normal dan tidak berbahaya.
3. Masalah DNS/IP Advertisement: Klien berhasil terhubung ke salah satu bootstrap server, tetapi saat meminta metadata, broker mengembalikan alamat host internal (advertised listeners) yang tidak dapat dijangkau oleh jaringan aplikasi klien (masalah umum pada deployment Docker/Kubernetes lintas VPC).

Langkah Solusi:

• Periksa konfigurasi advertised.listeners pada broker. Pastikan host yang diiklankan oleh broker dapat di-resolve secara DNS dan diakses port-nya oleh server tempat aplikasi klien berjalan.
• Gunakan perintah nc -zv [broker-host] [port] dari server aplikasi untuk memastikan tidak ada pemblokiran firewall.

5. GroupAuthorizationException #

Error ini terjadi di sisi konsumen ketika fitur keamanan otorisasi (ACL) diaktifkan di kluster Kafka.

Contoh Pesan Log:

org.apache.kafka.common.errors.GroupAuthorizationException: Not authorized to access group: payment-group

Analisis Akar Masalah: Kredensial otentikasi (SSL Principal atau SASL Username) yang digunakan oleh aplikasi konsumen kita tidak memiliki hak akses (Read) untuk menggunakan nama Consumer Group yang ditentukan (payment-group). Tanpa izin ini, coordinator broker akan menolak request bergabung grup (JoinGroup).

Langkah Solusi: Minta administrator Kafka kita untuk menambahkan ACL Read untuk resource Group yang bersangkutan:

kafka-acls.sh --bootstrap-server localhost:9093 \
  --command-config /etc/kafka/client.properties \
  --add \
  --allow-principal User:CN=payment-app,O=MyCorp \
  --operation Read \
  --group payment-group

Teknik Korelasi Log Aplikasi dengan Metadata Kafka #

Salah satu tantangan terbesar dalam arsitektur microservices adalah melacak perjalanan sebuah pesan dari awal hingga akhir (end-to-end tracing). Ketika ada keluhan bahwa transaksi pembayaran tertentu tidak terproses, kita harus dapat melacak log aplikasi mulai dari API Gateway, ke Produser Kafka, masuk ke Broker, hingga diproses oleh Konsumen.

Untuk melakukan hal ini secara efisien, kita dapat memanfaatkan MDC (Mapped Diagnostic Context) yang didukung oleh Logback dan Log4j2. MDC memungkinkan kita untuk menyisipkan metadata kontekstual (seperti correlation_id, topic, partition, dan offset) ke dalam thread lokal sehingga setiap baris log yang tertulis secara otomatis mengandung informasi tersebut.

Berikut adalah contoh implementasi konsumen Java yang memperkaya MDC pada setiap penarikan pesan:

package com.mycompany.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class TransactionConsumer {

    private static final Logger log = LoggerFactory.getLogger(TransactionConsumer.class);

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "payment-processor");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("enable.auto.commit", "false");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("payment.orders"));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                
                for (ConsumerRecord<String, String> record : records) {
                    // 1. Ekstrak correlation ID dari record header (jika dikirim oleh produser)
                    String correlationId = getCorrelationIdFromHeader(record);
                    
                    // 2. Masukkan metadata Kafka ke dalam Mapped Diagnostic Context (MDC)
                    MDC.put("kafka.topic", record.topic());
                    MDC.put("kafka.partition", String.valueOf(record.partition()));
                    MDC.put("kafka.offset", String.valueOf(record.offset()));
                    MDC.put("correlation.id", correlationId);

                    try {
                        // 3. Jalankan logika bisnis aplikasi kita
                        log.info("Mulai memproses transaksi pembayaran untuk payload: {}", record.value());
                        processPayment(record.value());
                        log.info("Selesai memproses transaksi pembayaran dengan sukses.");
                        
                        // Commit manual setelah berhasil
                        consumer.commitSync();
                    } catch (Exception e) {
                        log.error("Gagal memproses transaksi pada offset {}", record.offset(), e);
                    } finally {
                        // 4. Bersihkan MDC untuk mencegah kebocoran konteks ke thread lain
                        MDC.clear();
                    }
                }
            }
        } finally {
            consumer.close();
        }
    }

    private static String getCorrelationIdFromHeader(ConsumerRecord<String, String> record) {
        if (record.headers() != null) {
            var header = record.headers().lastHeader("X-Correlation-ID");
            if (header != null) {
                return new String(header.value());
            }
        }
        // Fallback ke UUID acak jika tidak disediakan oleh produser
        return java.util.UUID.randomUUID().toString();
    }

    private static void processPayment(String payload) throws Exception {
        // Simulasi logika pemrosesan bisnis
        Thread.sleep(50); 
    }
}

Agar nilai MDC ini tercetak di file log kita, kita harus memperbarui bagian <pattern> di file logback.xml kita dengan menyisipkan variabel %X{key}:

<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - [CorrID: %X{correlation.id}] [Topic: %X{kafka.topic} Part: %X{kafka.partition} Off: %X{kafka.offset}] - %msg%n</pattern>

Ketika aplikasi kita berjalan, output log kita akan terlihat sangat informatif seperti ini:

2026-06-08 15:40:12.512 [main] INFO  c.m.k.c.TransactionConsumer - [CorrID: tx-8829-ac91] [Topic: payment.orders Part: 2 Off: 10421] - Mulai memproses transaksi pembayaran untuk payload: {"order_id": 9921}
2026-06-08 15:40:12.565 [main] INFO  c.m.k.c.TransactionConsumer - [CorrID: tx-8829-ac91] [Topic: payment.orders Part: 2 Off: 10421] - Selesai memproses transaksi pembayaran dengan sukses.

Dengan format log seperti ini, tim operasional kita dapat dengan mudah melakukan pencarian di Elasticsearch/Splunk hanya dengan memfilter ID korelasi atau kombinasi partisi-offset tertentu saat menganalisis kegagalan transaksi.

Kepatuhan Operasional dan Checklist Audit Log Klien #

Lakukan evaluasi berkala terhadap konfigurasi logging aplikasi klien kita menggunakan tabel audit berikut untuk menjamin kepatuhan sistem di produksi:

Ringkasan #

• Gunakan Facade SLF4J — Manfaatkan facade SLF4J untuk mengikat pustaka klien Kafka dengan framework pencatat log utama aplikasi kita (seperti Logback atau Log4j2).
• Batasi Log Level Produksi — Setel level log org.apache.kafka ke WARN di lingkungan produksi untuk meminimalkan overhead I/O disk, namun biarkan modul koordinasi konsumen pada level INFO.
• Dekode Error Secara Cepat — Pahami akar penyebab error kritis seperti RecordTooLargeException (keselarasan batas ukuran) dan CommitFailedException (waktu pemrosesan melebihi batas polling).
• Korelasikan Menggunakan MDC — Implementasikan pengisian metadata Kafka ke dalam Mapped Diagnostic Context (MDC) untuk mempermudah pelacakan pesan secara end-to-end melintasi berbagai layanan.

← Sebelumnya: Broker Log   Berikutnya: Debugging Message Flow →