Topology vs Processor #

Saat kita membangun aplikasi pemrosesan aliran data menggunakan Kafka Streams, semua logika bisnis yang kita tulis pada akhirnya akan diterjemahkan menjadi grafik aliran komputasi yang terstruktur. Struktur ini dikenal sebagai Processor Topology (Topologi Pemroses). Memahami bagaimana topologi ini dirancang, dialokasikan secara internal, dan bagaimana kita berinteraksi dengannya melalui dua antarmuka utama — DSL API (Domain Specific Language) yang berlevel tinggi dan Processor API (PAPI) yang berlevel rendah — adalah kunci untuk membangun sistem pemrosesan aliran data yang andal dan optimal. Melalui artikel ini, kita akan mengupas tuntas struktur internal Processor Topology, perbedaan peran tiga tipe node komputasi, komparasi mendalam antara DSL dan Processor API, serta bagaimana menulis custom processor stateful secara benar di lingkungan produksi.

Struktur Processor Topology: Directed Acyclic Graph (DAG) #

Secara arsitektural, Processor Topology adalah sebuah grafik berarah asiklik (DAG - Directed Acyclic Graph) yang menggambarkan aliran data dari topik input Kafka, melalui serangkaian node logika pemrosesan, hingga berakhir di topik output Kafka.

Di dalam grafik ini, data mengalir dalam bentuk pasangan kunci-nilai (key-value records). Setiap titik atau simpul di dalam grafik disebut sebagai Processor Node, dan setiap garis berarah merepresentasikan jalur aliran data.

Tiga Tipe Utama Node Komputasi #

Di dalam topologi Kafka Streams, terdapat tiga jenis node khusus dengan tanggung jawab yang didefinisikan secara ketat:

1. Source Processor (Node Sumber) #

Source Processor bertindak sebagai pintu masuk aliran data ke dalam topologi kita.

Tugas: Berlangganan (subscribe) ke satu atau beberapa topik Kafka fisik, menerima data secara asinkron dari Broker, mendeserialisasi kunci dan nilai dari format biner (byte[]) menjadi objek Java terstruktur menggunakan serializer/deserializer (Serde), dan meneruskan data tersebut ke node turunan di bawahnya.
Source node tidak memiliki node induk di dalam topologi.

2. Stream Processor (Node Pemroses) #

Stream Processor merepresentasikan tempat di mana transformasi data aktual terjadi.

Tugas: Menerima record dari node induk, menerapkan fungsi transformasi (stateless seperti map dan filter, atau stateful seperti join dan aggregate), dan opsional meneruskan hasil transformasi ke node turunan berikutnya.
Node pemroses dapat mengakses satu atau beberapa State Stores lokal (RocksDB) untuk mempertahankan keadaan sementara.

3. Sink Processor (Node Muara) #

Sink Processor bertindak sebagai pintu keluar aliran data dari topologi kita.

Tugas: Menerima record dari node induk, menserialisasi objek Java kembali menjadi format biner (byte[]), dan menuliskan data hasil akhir tersebut ke topik Kafka tujuan fisik di broker.
Sink node tidak memiliki node turunan di dalam topologi.

flowchart TD
    subgraph TOPOLOGY ["Processor Topology (DAG)"]
        direction TB
        Source["Source Node (Topic: raw-clicks)"]
        ProcFilter["Stateless Processor (Filter Bots)"]
        ProcCount["Stateful Processor (Count Clicks)"]
        Store[("State Store (RocksDB)")]
        Sink["Sink Node (Topic: aggregated-clicks)"]

        Source --> ProcFilter
        ProcFilter --> ProcCount
        ProcCount <--> Store
        ProcCount --> Sink
    end
    
    style Source stroke:#0288d1,stroke-width:2px
    style ProcFilter stroke:#388e3c,stroke-width:2px
    style ProcCount stroke:#388e3c,stroke-width:2px
    style Store stroke:#f57c00,stroke-width:2px
    style Sink stroke:#d32f2f,stroke-width:2px

DSL API vs Processor API (PAPI) #

Kafka Streams menyediakan dua pendekatan berbeda bagi developer untuk mendefinisikan Processor Topology. Memilih di antara keduanya adalah keputusan desain yang krusial.

1. DSL API (Domain Specific Language) #

DSL API adalah antarmuka pemrograman deklaratif fungsional berlevel tinggi. Ini adalah cara paling umum dan direkomendasikan untuk sebagian besar kasus penggunaan biasa.

Deklaratif: Kita menentukan apa yang ingin kita lakukan pada aliran data (misalnya stream.filter(...).map(...).to(...)), dan Kafka Streams secara otomatis menerjemahkan dan mengoptimalkan pernyataan tersebut menjadi DAG topology fisik di latar belakang.
Kaya Fitur Bawaan: Menyediakan operator out-of-the-box seperti pemetaan (map, mapValues), penyaringan (filter), penggabungan (join), pengelompokan (groupBy), windowing, dan penulisan langsung ke topik (to).
Kekurangan: Kustomisasi kontrol tingkat rendah terbatas. Kita tidak memiliki akses langsung ke metadata record (seperti offset, timestamp fisik, atau header Kafka) secara detail, dan kita tidak bisa mengatur jadwal commit state store secara manual per event.

2. Processor API (PAPI) #

Processor API adalah antarmuka pemrograman imperatif berlevel rendah. PAPI memberikan kontrol total kepada developer atas alur komputasi.

Imperatif: Kita harus mendefinisikan topologi secara manual dengan menambahkan source, processor, sink, dan state store ke objek Topology secara eksplisit, lalu menulis logika pemrosesan secara prosedural baris demi baris di dalam metode process().
Kontrol Maksimum: Kita memiliki akses penuh ke ProcessorContext untuk membaca offset, timestamp, header, memancarkan data secara dinamis ke downstream (context.forward()), serta menjadwalkan eksekusi berkala menggunakan Punctuator (misalnya memicu kalkulasi setiap 5 menit waktu sistem atau waktu event).
Kekurangan: Kita harus menulis kode boilerplate yang jauh lebih banyak untuk menangani serialisasi, pengelolaan state store manual, dan penanganan error.

Kode Penerapan: Anti-Pattern vs Solusi Stateful #

Untuk memahami perbedaan konkret di antara kedua API ini dan bagaimana menangani topologi secara aman, mari kita tinjau contoh kasus penggunaan riil.

Kasus Penggunaan #

Kita ingin memantau aliran transaksi kartu kredit (credit-card-events). Jika suatu pengguna melakukan transaksi dengan nilai transaksi yang identik berturut-turut dalam kurun waktu kurang dari 3 detik (indikasi kegagalan sistem kasir atau double tap kartu), kita ingin menyaring event transaksi kedua dan mengirimkannya ke topik khusus alert (suspicious-transactions).

Anti-Pattern: Mencampurkan Logika Stateful secara Ad-Hoc di DSL #

Mencoba membangun logika stateful yang rumit (menyimpan transaksi terakhir per user) menggunakan operator DSL stateless seperti filter dengan variabel HashMap statis eksternal adalah anti-pattern yang berbahaya.

// ANTI-PATTERN: Mencoba mengelola state transaksi terakhir menggunakan variabel statis di filter DSL.
// ✗ Tidak aman dari kegagalan crash, memicu race condition multi-thread, dan merusak scaling horisontal.
public class DangerousDslStateStore {
    // ✗ BAHAYA: HashMap statis tidak terlindungi dari crash dan tidak disinkronkan lintas thread/instance
    private static final Map<String, Transaction> lastTransactions = new ConcurrentHashMap<>();

    public static void buildTopology(StreamsBuilder builder) {
        builder.<String, String>stream("credit-card-events")
            .filter((userId, transactionJson) -> {
                Transaction currentTx = parseJson(transactionJson);
                Transaction lastTx = lastTransactions.get(userId);

                if (lastTx != null) {
                    long diffSeconds = (currentTx.timestamp - lastTx.timestamp) / 1000;
                    // Saringan deteksi transaksi ganda dalam 3 detik
                    if (currentTx.amount == lastTx.amount && diffSeconds < 3) {
                        // ✗ Modifikasi state tanpa backup Changelog
                        lastTransactions.put(userId, currentTx);
                        return true; // Loloskan sebagai transaksi mencurigakan
                    }
                }
                
                lastTransactions.put(userId, currentTx);
                return false;
            })
            .to("suspicious-transactions");
    }

    private static Transaction parseJson(String json) { return new Transaction(); }
    static class Transaction { double amount; long timestamp; }
}

Solusi Praktis 1: Menggunakan Processor API (PAPI) Secara Bersih #

Untuk menyelesaikan tantangan ini secara aman dari kegagalan server dan siap diskalakan, kita wajib menggunakan Processor API dengan state store RocksDB yang terkelola secara resmi.

// BENAR: Menggunakan Processor API untuk mengelola state secara aman dengan RocksDB lokal dan backup Changelog.
// ✓ Aman dari kehilangan data saat container mati, bebas dari masalah konkurensi thread.
public class TransactionDoubleTapProcessor implements Processor<String, String, String, String> {
    private ProcessorContext<String, String> context;
    private KeyValueStore<String, String> stateStore;

    @Override
    public void init(ProcessorContext<String, String> context) {
        this.context = context;
        // ✓ Mengambil referensi ke State Store RocksDB lokal yang terdaftar di topologi
        this.stateStore = context.getStateStore("last-transaction-store");
    }

    @Override
    public void process(Record<String, String> record) {
        String userId = record.key();
        String currentTxJson = record.value();
        
        // Membaca transaksi terakhir pengguna dari RocksDB lokal secara cepat
        String lastTxJson = stateStore.get(userId);

        if (lastTxJson != null) {
            double currentAmount = parseAmount(currentTxJson);
            double lastAmount = parseAmount(lastTxJson);
            long currentTime = record.timestamp();
            long lastTime = parseTimestamp(lastTxJson);

            long diffMs = currentTime - lastTime;
            
            // ✓ Deteksi double tap (transaksi bernilai sama dalam waktu kurang dari 3 detik)
            if (currentAmount == lastAmount && diffMs < 3000) {
                // Pancarkan transaksi mencurigakan ini ke node turunan (downstream)
                context.forward(record);
                
                // Perbarui state transaksi terakhir di database RocksDB
                stateStore.put(userId, currentTxJson);
                return;
            }
        }

        // Simpan transaksi terbaru ke RocksDB
        stateStore.put(userId, currentTxJson);
    }

    @Override
    public void close() {
        // Logika cleanup jika ada resource eksternal open
    }

    private double parseAmount(String json) { return 50.0; }
    private long parseTimestamp(String json) { return System.currentTimeMillis(); }
}

Konfigurasi Topologi Fisik untuk Processor API #

Setelah kita menulis kelas pemroses di atas, kita harus menyusun topologi fisiknya secara eksplisit menggunakan objek Topology bawaan:

// BENAR: Menyusun topologi fisik secara eksplisit untuk mendaftarkan Custom Processor dan State Store.
public class TransactionTopologyBuilder {
    public static Topology build() {
        Topology topology = new Topology();

        // 1. Tambahkan Source Node untuk membaca topik fisik dari Broker
        topology.addSource("TransactionSource", "credit-card-events");

        // 2. Tambahkan Custom Stateful Processor ke dalam topologi
        topology.addProcessor(
            "DoubleTapProcessor",
            TransactionDoubleTapProcessor::new,
            "TransactionSource" // Menentukan induknya (Source Node)
        );

        // 3. Konfigurasikan State Store RocksDB lokal
        StoreBuilder<KeyValueStore<String, String>> storeBuilder = Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore("last-transaction-store"), // RocksDB persisten
            Serdes.String(),
            Serdes.String()
        );

        // 4. Daftarkan State Store ke topologi dan ikat dengan processor yang membutuhkannya
        topology.addStateStore(storeBuilder, "DoubleTapProcessor");

        // 5. Tambahkan Sink Node untuk menulis data yang lolos ke topik keluaran suspicious-transactions
        topology.addSink(
            "AlertSink",
            "suspicious-transactions",
            "DoubleTapProcessor" // Menentukan induknya (Custom Processor Node)
        );

        return topology;
    }
}

Solusi Praktis 2: Pendekatan Hibrida (DSL dengan `process()`) #

Jika kita ingin mempertahankan kenyamanan DSL API namun membutuhkan kekuatan kendali stateful dari Processor API untuk bagian logika tertentu, kita dapat menggunakan pendekatan hibrida melalui operator process() pada KStream.

// BENAR: Pendekatan Hibrida menggunakan KStream.process() di dalam DSL API.
// ✓ Memadukan kemudahan deklaratif DSL dengan fleksibilitas low-level custom processor.
public class HybridTopologyExample {
    public static void buildHybrid(StreamsBuilder builder) {
        // Konfigurasi State Store lokal RocksDB
        StoreBuilder<KeyValueStore<String, String>> storeBuilder = Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore("last-transaction-store"),
            Serdes.String(),
            Serdes.String()
        );
        
        // Daftarkan storeBuilder secara global ke StreamsBuilder
        builder.addStateStore(storeBuilder);

        KStream<String, String> stream = builder.stream("credit-card-events");

        // ✓ Menggunakan operator process() untuk melompat dari DSL ke PAPI secara transparan
        stream
            .process(
                TransactionDoubleTapProcessor::new,
                "last-transaction-store" // Mengaitkan RocksDB store ke instansi processor
            )
            .to("suspicious-transactions");
    }
}

Kapan Memilih DSL dan Kapan Beralih ke PAPI? #

Untuk mempermudah pengambilan keputusan bagi tim pengembang, berikut adalah matriks alur keputusan pemilihan antarmuka:

TETAP GUNAKAN DSL API JIKA:
  ✓ Kebutuhan pemrosesan data kita bersifat umum (seperti melakukan map, filter, atau join sederhana).
  ✓ Kita ingin menulis kode yang ringkas, cepat dibaca, dan mudah dipelihara.
  ✓ Operasi stateful kita dapat dipetakan langsung dengan fungsi aggregate(), count(), atau reduce() bawaan.

BERALIH KE PROCESSOR API (PAPI) JIKA:
  ✗ Kita perlu memancarkan data secara dinamis ke beberapa topik output yang berbeda berdasarkan kondisi logika di runtime (Dynamic Routing).
  ✗ Kita butuh akses langsung ke metadata tingkat rendah seperti Record Headers, Partition metadata, atau Offset.
  ✗ Kita perlu memicu tindakan berkala (seperti menghapus state lama yang kedaluwarsa) berdasarkan interval waktu sistem/event menggunakan Punctuation.
  ✗ Kita ingin memodifikasi perilaku state store lokal RocksDB secara kustom di luar API bawaan.

Ringkasan #

Processor Topology — Grafik berarah asiklik (DAG) yang mendefinisikan aliran data asinkron dari topik input, melalui node pemroses, hingga ke topik output.

Three Node Types — Topologi terdiri dari Source Node (deserialisasi & input), Processor Node (logika transformasi), dan Sink Node (serialisasi & output).

DSL API — Antarmuka deklaratif tingkat tinggi yang kaya fungsi bawaan, sangat cocok untuk pemrosesan aliran standar yang membutuhkan implementasi cepat.

Processor API (PAPI) — Antarmuka imperatif tingkat rendah yang memberikan kontrol penuh atas state management, metadata record, dan penjadwalan eksekusi.

ProcessorContext — Objek penghubung di PAPI yang menyediakan akses ke metadata runtime, state store, dan mekanisme pengiriman data ke downstream (forward()).

Punctuator — Komponen penjadwalan di PAPI untuk mengeksekusi fungsi kustom secara berkala berdasarkan waktu sistem (Wall-clock Time) atau waktu data masuk (Stream Time).

← Sebelumnya: Stream vs Table Berikutnya: State Store →