Schema Registry

Skema yang dikelola sendiri oleh masing-masing tim akan segera saling meleset, karena tidak ada pihak yang menjamin dua tim memegang definisi yang sama. Topik ini membahas Schema Registry, yaitu layanan pusat yang menyimpan skema dan menandai setiap pesan Avro dengan ID skema pada header kecilnya. Pembahasan mencakup peran layanan ini, penamaan subject, format kawat (wire format) yang membawa ID skema, serta alur pesan dari producer sampai consumer.

Peran Schema Registry#

Schema Registry adalah layanan REST terpisah yang menyimpan skema beserta setiap versinya, dengan alamat bawaan http://localhost:8081 pada pemasangan Confluent Platform. Producer mendaftarkan skema sebelum mengirim pesan dan menerima ID skema sebagai pengenalnya, sedangkan consumer membaca ID pada pesan untuk mengambil skema yang sesuai. Skema tidak pernah dikirim ulang bersama pesan, sehingga biaya header hanya lima byte per pesan. Broker tidak mengetahui dan tidak perlu mengetahui skema tersebut, karena serialisasi dan deserialisasi sepenuhnya menjadi pekerjaan klien.

Klien menyimpan hasil permintaan skema di cache lokal, sehingga pemanggilan REST hanya terjadi pada pesan pertama untuk setiap ID skema. Konsekuensinya, beban Registry kecil pada operasi normal, dan layanan ini hanya menjadi sumber kebenaran versi skema. Versi skema yang sudah terdaftar tetap dipertahankan, karena pesan lama pada log masih membawa ID yang merujuk versi tersebut.

Subject dan strategi penamaan#

Skema didaftarkan ke sebuah subject, yaitu unit registrasi pada Registry yang menampung urutan versi skema beserta aturan kompatibilitasnya. Nama subject bukan pilihan bebas, melainkan ditentukan oleh strategi nama subject (subject name strategy) yang diterapkan serializer, yaitu komponen klien yang mengubah objek menjadi byte sebelum dikirim ke broker. Tiga strategi bawaan menempatkan skema record PesananDibuat yang dikirim ke topic pesanan pada nama subject berikut.

TopicNameStrategy menggunakan nama topic beserta akhiran -value untuk nilai dan -key untuk key, sehingga satu topic menampung satu bentuk skema. RecordNameStrategy menempatkan skema berdasarkan nama record, sehingga beberapa bentuk skema dapat berbagi satu topic. TopicRecordNameStrategy menggabungkan nama topic dan nama record. Pemilihan strategi mengikat, karena evaluasi kompatibilitas dilakukan per subject dan strategi yang berbeda mengumpulkan skema pada subject yang berbeda pula. Strategi dipilih lewat konfigurasi subject.name.strategy pada serializer, dan TopicNameStrategy merupakan pilihan bawaannya.

Format kawat pesan Avro#

Setiap pesan Avro yang lolos Registry membawa header lima byte yang dikenal sebagai format kawat. Byte pertama bernilai 0 sebagai penanda versi format, empat byte berikutnya menyimpan ID skema dalam urutan big endian, dan sisanya adalah payload Avro murni. Payload tidak membawa nama field maupun tipe, karena definisinya sudah tersimpan di Registry. Diagram berikut merangkum alur lengkapnya, mulai dari pendaftaran skema hingga pembacaan kembali.

Demo berikut membentuk dan membaca format kawat itu secara manual. Payload Avro untuk string Budi disusun byte demi byte sesuai spesifikasi Avro, tanpa pustaka serialisasi apa pun, sehingga struktur headernya terlihat langsung.

import struct

def zigzag_varint(n):
    # konversi bilangan bulat menjadi varint zigzag sesuai spesifikasi Avro
    zz = (n << 1) ^ (n >> 63) if n < 0 else (n << 1)
    byte = []
    while True:
        if zz & ~0x7F:
            byte.append((zz & 0x7F) | 0x80)
            zz >>= 7
        else:
            byte.append(zz)
            return bytes(byte)

def avro_string(s):
    data = s.encode("utf-8")
    return zigzag_varint(len(data)) + data

def baca_avro_string(payload):
    # baca panjang string (varint zigzag) di depan, lalu kembalikan isinya
    n, shift, i = 0, 0, 0
    while True:
        b = payload[i]
        n |= (b & 0x7F) << shift
        i += 1
        if not (b & 0x80):
            break
        shift += 7
    panjang = (n >> 1) ^ -(n & 1)
    return payload[i:i + panjang].decode("utf-8")

id_skema = 42
payload = avro_string("Budi")

# bentuk pesan sesuai format kawat Confluent: magic 0, ID skema 4 byte, payload Avro
pesan = b"\x00" + struct.pack(">I", id_skema) + payload
print("payload Avro        :", payload)
print("panjang payload     :", len(payload), "byte")
print("hex pesan lengkap   :", pesan.hex(" "))

# baca kembali: pisahkan header dan payload
print("byte pertama (magic):", pesan[0])
print("ID skema terbaca    :", struct.unpack(">I", pesan[1:5])[0])
print("nilai terbaca       :", baca_avro_string(pesan[5:]))

payload Avro        : b'\x08Budi'
panjang payload     : 5 byte
hex pesan lengkap   : 00 00 00 00 2a 08 42 75 64 69
byte pertama (magic): 0
ID skema terbaca    : 42
nilai terbaca       : Budi

Header lima byte pertama adalah 00 00 00 00 2a, yaitu byte magic 0 diikuti ID 42 dalam urutan big endian. Byte 08 merupakan varint zigzag yang menyatakan panjang string 4, yaitu cara Avro menyandikan bilangan bulat tujuh bit per byte dengan tanda yang digeser ke bit terakhir, dan empat byte terakhir adalah UTF-8 dari Budi. Pesan lengkapnya hanya sepuluh byte, padahal skema lengkap yang mendasari pesan itu tercatat di Registry. Inilah bentuk yang sebenarnya digunakan serializer, yaitu header kecil bersama payload yang sudah bebas dari nama field.

Byte yang sama melewati broker#

Broker menyimpan dan meneruskan pesan sebagai byte mentah, sehingga pesan pada demo sebelumnya dapat dikirim ke topic tanpa pengubahan apa pun. Demo berikut menggunakan pustaka kafka-python yang sudah terpasang di sandbox untuk mengirim pesan berformat kawat Confluent ke broker lokal, kemudian membacanya kembali dan memecah header serta payloadnya. Topic yang digunakan dibuat baru pada setiap eksekusi, sehingga pesan yang dibaca pasti berasal dari producer pada eksekusi yang sama.

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

def zigzag_varint(n):
    # konversi bilangan bulat menjadi varint zigzag sesuai spesifikasi Avro
    zz = (n << 1) ^ (n >> 63) if n < 0 else (n << 1)
    byte = []
    while True:
        if zz & ~0x7F:
            byte.append((zz & 0x7F) | 0x80)
            zz >>= 7
        else:
            byte.append(zz)
            return bytes(byte)

def avro_string(s):
    data = s.encode("utf-8")
    return zigzag_varint(len(data)) + data

# pesan dalam format kawat Confluent: magic 0 + ID skema 4 byte + payload Avro
id_skema = 42
pesan = b"\x00" + struct.pack(">I", id_skema) + avro_string("Budi")

topic = f"m3-demo-kawat-{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")
producer.send(topic, pesan)
producer.flush()
producer.close()

consumer = KafkaConsumer(topic, bootstrap_servers="localhost:9092",
                         auto_offset_reset="earliest", consumer_timeout_ms=2000)
for msg in consumer:
    print("byte diterima    :", msg.value.hex(" "))
    print("byte pertama     :", msg.value[0])
    print("ID skema         :", struct.unpack(">I", msg.value[1:5])[0])
    print("payload hex      :", msg.value[5:].hex(" "))
consumer.close()

byte diterima    : 00 00 00 00 2a 08 42 75 64 69
byte pertama     : 0
ID skema         : 42
payload hex      : 08 42 75 64 69

Keluaran menunjukkan byte yang diterima consumer identik dengan byte yang dikirim producer, termasuk byte magic dan ID skema 42. Tidak ada proses di broker yang menafsirkan payload, sehingga seluruh tanggung jawab atas skema berada di klien. Broker hanya mengenal partisi, offset, dan byte, sedangkan makna byte ditetapkan oleh skema yang dipegang Registry. Karena itu pula perubahan pada Registry tidak pernah menyentuh isi log yang sudah tersimpan.

Interaksi lewat REST#

Registry melayani skema lewat REST API dengan permintaan dan respons berformat JSON. Dua operasi paling dasar adalah pendaftaran skema pada sebuah subject dan pengambilan versi terbaru beserta ID-nya. Pendaftaran dilakukan serializer pada pesan pertama, sedangkan pengambilan versi dilakukan deserializer, yaitu komponen klien yang mengubah byte kembali menjadi objek, ketika menemukan ID skema yang belum ada di cache-nya. Pendaftaran skema yang isinya sudah pernah tercatat mengembalikan ID yang sama, sehingga serializer dapat memanggilnya berulang kali tanpa menambah versi baru. Perintah berikut diasumsikan berjalan pada lingkungan tempat Schema Registry aktif di localhost:8081, karena Schema Registry tidak terpasang di sandbox ini.

# tidak-dijalankan
# daftarkan skema pada subject pesanan-value dan terima ID skema
curl -X POST http://localhost:8081/subjects/pesanan-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schemaType": "AVRO", "schema": "{\"type\": \"record\", \"name\": \"PesananDibuat\", \"fields\": [{\"name\": \"id\", \"type\": \"string\"}, {\"name\": \"jumlah\", \"type\": \"int\"}]}"}'

# tidak-dijalankan
# ambil versi skema terbaru pada subject beserta ID-nya
curl http://localhost:8081/subjects/pesanan-value/versions/latest

Ringkasan#

• Schema Registry menyimpan skema secara terpusat, menetapkan ID untuk setiap versi, dan menjadi sumber kebenaran kontrak data.
• Strategi nama subject menentukan tempat skema terdaftar, dengan TopicNameStrategy sebagai pilihan bawaan.
• Format kawat hanya menambahkan lima byte header berupa byte magic 0 dan ID skema 4 byte, sedangkan broker meneruskan byte mentah tanpa menafsirkannya.

Referensi#

• Schema Registry for Confluent Platform — Confluent Documentation — peran Schema Registry, subject, dan REST API-nya.
• Apache Avro Specification — Apache Software Foundation — spesifikasi penyandian biner Avro yang digunakan pada demo pembentukan payload.
• Apache Kafka documentation — Apache Software Foundation — dokumentasi resmi Kafka tentang pemisahan antara broker dan serialisasi klien.