Header dan metadata

Topik ini membahas metadata yang menyertai setiap pesan Kafka, yaitu header untuk pasangan kunci-nilai tambahan dan timestamp sebagai penanda waktu pesan. Pembahasan ditutup dengan interceptor, yaitu hook Java yang dipanggil producer di tengah jalur pengiriman. Demo Python dijalankan pada broker nyata dan keluarannya ditempel apa adanya, sedangkan contoh Java diperlihatkan sebagai potongan konseptual. Setelah membacanya, metadata tambahan dapat ditempatkan pada tempat yang benar antara header, payload, dan timestamp.

Header pesan#

Header pesan adalah pasangan kunci-nilai berformat bytes yang menyertai pesan tanpa menjadi bagian payload-nya, yaitu isi kejadian yang dikirim producer. Satu pesan dapat membawa beberapa header sekaligus, dan broker meneruskan kunci maupun nilai header tanpa menafsirkan isinya, sehingga formatnya sepenuhnya menjadi kesepakatan antara pengirim dan pembaca. Isi header cocok untuk metadata kecil yang dibutuhkan sebelum payload dibuka, misalnya id pelacakan, tipe payload, nama sumber data, atau id skema.

Penempatan metadata di header memisahkan dua peran yang berbeda. Payload membawa data kejadian itu sendiri, sedangkan header membawa informasi tentang cara payload dibaca dan dilacak. Aplikasi yang mengarahkan pesan berdasarkan tipenya dapat membaca header tanpa membongkar payload, dan sistem pelacakan dapat mengikuti id yang sama lintas layanan. Metadata yang besarnya setara payload justru membebani setiap pembacaan log, sehingga isi header dijaga tetap kecil.

Pada klien kafka-python, header dikirim lewat argumen headers pada send() dalam bentuk daftar pasangan bytes, dan pesan yang diterima membawa atribut headers berisi pasangan yang sama. Kunci maupun nilai tetap berupa bytes tanpa penafsiran apa pun, sehingga pembaca mengubahnya sendiri dengan decode() apabila isinya berupa teks. Bentuk mentah inilah yang membuat header tahan terhadap perubahan format, karena broker tidak pernah mengasumsikan tipe apa pun di dalamnya.

Demo header pada producer dan consumer#

Demo berikut mengirim tiga pesan bacaan sensor dengan dua header per pesan dan timestamp tetap, lalu membacanya kembali dan mencetak seluruh metadata yang tiba. Topic dibuat terlebih dahulu lewat KafkaAdminClient dengan satu partisi, dan timestamp dikunci pada satu nilai agar keluarannya dapat dibandingkan antar eksekusi. Nama topic diberi akhiran waktu agar setiap eksekusi memperoleh log yang kosong.

from kafka import KafkaProducer, KafkaConsumer, KafkaAdminClient
from kafka.admin import NewTopic
from kafka.errors import TopicAlreadyExistsError
import time

topic = f"m1-header-{int(time.time())}"  # nama unik per eksekusi

admin = KafkaAdminClient(bootstrap_servers="localhost:9092")
try:
    admin.create_topics([NewTopic(name=topic, num_partitions=1, replication_factor=1)])
except TopicAlreadyExistsError:
    pass
admin.close()

producer = KafkaProducer(bootstrap_servers="localhost:9092")
for i in range(3):
    producer.send(
        topic,
        value=f"bacaan-{i}".encode(),
        headers=[("sumber", b"sensor-suhu"), ("satuan", b"celsius")],
        timestamp_ms=1700000000000,  # timestamp tetap agar keluaran stabil
    )
producer.flush()
producer.close()

consumer = KafkaConsumer(
    topic,
    bootstrap_servers="localhost:9092",
    auto_offset_reset="earliest",
    consumer_timeout_ms=2000,
)
for msg in consumer:
    header = ", ".join(f"{k}={v.decode()}" for k, v in msg.headers)
    print(f"offset={msg.offset} timestamp={msg.timestamp} header[{header}] nilai={msg.value.decode()}")
consumer.close()

offset=0 timestamp=1700000000000 header[sumber=sensor-suhu, satuan=celsius] nilai=bacaan-0
offset=1 timestamp=1700000000000 header[sumber=sensor-suhu, satuan=celsius] nilai=bacaan-1
offset=2 timestamp=1700000000000 header[sumber=sensor-suhu, satuan=celsius] nilai=bacaan-2

Setiap baris keluaran memuat offset, timestamp, dua header, dan nilai pesan dengan urutan yang sama seperti saat dikirim. Header sumber dan satuan tiba utuh tanpa menyentuh payload bacaan-N, sehingga pembaca dapat menafsirkan nilai tanpa menambah atribut baru ke dalam payload. Nilai header diset pada setiap panggilan send(), bukan pada pembuatan producer, sehingga pesan-pesan dari satu producer boleh membawa header yang berbeda satu sama lain. Timestamp yang dikirim producer kembali tanpa perubahan karena tipe timestamp topic berada pada nilai default CreateTime, perilaku yang dibahas pada bagian berikutnya.

Tipe timestamp pesan#

Setiap pesan menyimpan satu timestamp, dan tipe timestamp menentukan siapa yang menetapkan nilainya. Nilai CreateTime menyatakan timestamp ditetapkan producer saat pesan dibuat, sedangkan LogAppendTime menyatakan broker menimpanya dengan waktu pesan ditambahkan ke log. Tipe ini diatur per topic lewat konfigurasi message.timestamp.type dengan nilai default CreateTime, dan nilainya tidak berubah sepanjang usia pesan di dalam log.

Validasi hanya berlaku pada CreateTime. Broker menolak pesan yang timestampnya menjauh dari waktu broker hingga melewati ambang yang dikonfigurasikan, dengan message.timestamp.before.max.ms untuk pesan yang lebih tua dan message.timestamp.after.max.ms untuk pesan yang lebih baru. Nilai bawaan pertama praktis tanpa batas, sedangkan nilai bawaan kedua berupa 3600000 milidetik atau satu jam, sehingga pesan dengan waktu di masa depan yang mencurigakan tertangkap sebelum masuk log. Dokumentasi Kafka 4.4 tidak lagi mencantumkan konfigurasi lama message.timestamp.difference.max.ms yang menggabungkan kedua arah dalam satu nilai, karena perannya sudah digantikan dua varian tersebut.

Pilihan tipe memengaruhi makna waktu yang dibaca consumer. LogAppendTime menyamakan titik acuan semua pesan dengan waktu broker sehingga perbedaan jam antar mesin pengirim hilang, tetapi waktu kejadian yang sebenarnya ikut terhapus. CreateTime mempertahankan waktu sumber sebagaimana demo di bagian sebelumnya, dengan konsekuensi validasi yang harus dikonfigurasikan agar pesan dengan waktu yang tidak masuk akal tertangkap lebih awal.

Interceptor pada producer#

Interceptor adalah hook Java yang dipanggil producer di tengah jalur pengiriman, diaktifkan lewat konfigurasi interceptor.classes dengan daftar kelas yang mengimplementasikan antarmuka ProducerInterceptor. Metode onSend() dipanggil sebelum pesan diserialisasi dan dikirim, sehingga record yang dikembalikannya menjadi record yang benar-benar dikirim, sedangkan onAcknowledgement() dipanggil ketika broker mengirim konfirmasi atau error. Keduanya berjalan pada thread pengiriman utama producer.

Potongan berikut memperlihatkan bentuk antarmukanya, yaitu interceptor yang menyuntikkan header id pelacakan pada setiap pesan sebelum pengiriman. Contoh ini tidak dieksekusi pada sandbox karena ditulis dalam Java dengan klien asli Kafka, dan cukup dibaca sebagai pola.

// tidak-dijalankan
public class PenyuntikHeader implements ProducerInterceptor<String, String> {
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        record.headers().add("id-pelacakan", UUID.randomUUID().toString().getBytes());
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        // pencacahan pengiriman yang berhasil atau gagal ditulis di sini
    }

    @Override
    public void close() {
    }

    @Override
    public void configure(Map<String, ?> configs) {
    }
}

Interceptor mewadahi kebutuhan yang berlaku untuk semua pesan, yaitu pencacahan metrik pengiriman, penyuntikan header standar, dan pemberian tanda pada pesan tanpa menyunting setiap pemanggilan send(). Record yang dikembalikan onSend() menjadi pesan yang benar-benar dikirim, sehingga perubahan pada record itu ikut terkirim ke broker. Interceptor yang lambat menurunkan kapasitas kirim producer karena dijalankan pada jalur pengiriman utama, sehingga isinya dijaga seminimal mungkin dan pekerjaan berat tetap berada di aplikasi.

Kesalahan umum#

1. Nilai header langsung dibandingkan dengan teks karena dianggap berformat string, padahal kunci dan nilainya berupa bytes sehingga perbandingan selalu gagal. Ubah nilai header dengan decode() di sisi pembaca, atau kirim nilai yang sudah konsisten formatnya antara pengirim dan pembaca.
2. Metadata berukuran besar dipindahkan ke header sehingga ukuran pesan membengkak dan setiap pembacaan log mengangkut beban yang sama. Simpan data besar di payload dan batasi header pada penunjuk kecil, misalnya id yang merujuk ke skema atau arsip.
3. Topic dibuat dengan LogAppendTime lalu logika bisnis membaca timestamp sebagai waktu kejadian, padahal nilainya adalah waktu broker saat pesan masuk log. Tetapkan tipe timestamp saat topic dirancang, atau sertakan waktu kejadian di dalam payload apabila waktu sumber benar-benar dibutuhkan.

Ringkasan#

• Header menyimpan pasangan kunci-nilai bytes di luar payload, dan broker meneruskannya tanpa menafsirkan isinya.
• message.timestamp.type memilih antara CreateTime dari producer dan LogAppendTime dari broker, dengan nilai default CreateTime.
• Pesan CreateTime yang menjauh dari waktu broker ditolak sesuai ambang message.timestamp.before.max.ms dan message.timestamp.after.max.ms.
• ProducerInterceptor menyuntikkan header dan pencacahan metrik di tengah jalur pengiriman lewat onSend() dan onAcknowledgement().

Referensi#

• Producer configs — Apache Software Foundation — konfigurasi interceptor.classes yang mengaktifkan ProducerInterceptor pada producer.
• ProducerInterceptor — Apache Software Foundation — dokumentasi antarmuka onSend() dan onAcknowledgement() yang digunakan potongan Java.
• Topic configs — Apache Software Foundation — konfigurasi message.timestamp.type beserta ambang validasi timestamp pada topic.