Join Stream and Table #

Dalam arsitektur mikroservis modern, salah satu pola paling umum yang kita temui adalah kebutuhan untuk memperkaya data aliran event yang masuk dengan informasi kontekstual dari tabel database. Sebagai contoh, di sistem pembayaran e-commerce, kita menerima aliran transaksi pesanan yang hanya memuat ID Pengguna. Kita perlu menggabungkan (join) data transaksi tersebut secara real-time dengan tabel profil pengguna untuk memverifikasi tingkat membership (Platinum, Gold, Silver) sebelum memproses potongan diskon. Apache Kafka Streams menyediakan fitur penggabungan data terdistribusi yang sangat kuat bernama Joins (KStream/KTable Joins). Melalui artikel ini, kita akan membedah secara mendalam matriks join di Kafka Streams, persyaratan keras Co-partitioning, perbedaan semantik Inner/Left/Outer Joins, mekanisme Windowed vs Non-windowed Joins, penanganan data telat (late arrivals), serta implementasi kode Java yang optimal untuk produksi.

Matriks Penggabungan Data (Join Matrix) #

Kafka Streams mendukung berbagai kombinasi penggabungan data di antara abstraksi KStream, KTable, dan GlobalKTable. Setiap kombinasi memiliki perilaku pencocokan temporal dan semantik state yang berbeda:

1. KStream-KStream Join (Stateful & Windowed) #

Menggabungkan dua aliran peristiwa append-only berdasarkan kunci yang sama.

• Semantik: Karena kedua entitas adalah aliran tak terbatas, kita wajib menetapkan jendela waktu (Join Window) menggunakan parameter JoinWindows. KStreams akan mencocokkan event dari Stream A dengan event dari Stream B jika perbedaan timestamp keduanya berada di dalam jendela waktu yang ditentukan.
• Perilaku: Bersifat dua arah. Kedatangan event di Stream A akan memicu pencarian ke cache window Stream B, dan sebaliknya. Di balik layar, Kafka Streams membuat dua state store lokal secara otomatis untuk mencadangkan data jendela waktu masing-masing stream.
• Tipe Join:
  • Inner Join: Menghasilkan record output hanya jika ada pencocokan kunci pada kedua stream di dalam jendela waktu.
  • Left Join: Menghasilkan record output saat ada event di Stream kiri, meskipun tidak ada pencocokan di Stream kanan di dalam jendela (bagian kanan diisi null).
  • Outer Join: Menghasilkan output jika ada event baru di salah satu stream, mengisi sisi kosong dengan null jika tidak ada pencocokan.

2. KStream-KTable Join (Lookup Stateful & Non-windowed) #

Memperkaya aliran peristiwa (KStream) dengan keadaan terbaru dari tabel status (KTable).

• Semantik: Ini adalah operasi pencarian searah (lookup join). Setiap kali record baru tiba di KStream, Kafka Streams akan mencari nilai keadaan terbaru untuk kunci tersebut di dalam KTable lokal (RocksDB).
• Perilaku: Perubahan data di KTable tidak akan memicu pemancaran event output baru (sifatnya pasif). Hanya data transaksi baru di KStream yang memicu output diperkaya.
• Tipe Join: Hanya mendukung Inner Join dan Left Join.

3. KStream-GlobalKTable Join (Lookup Global & Non-windowed) #

Memperkaya aliran peristiwa dengan data tabel replikasi penuh di seluruh instance.

• Semantik: Serupa dengan KStream-KTable, namun karena GlobalKTable memuat data lengkap di setiap host, kita tidak dibatasi oleh syarat penyelarasan partisi. Kita bahkan bisa menggunakan kunci join kustom (Key Transformer) yang berbeda dari kunci asli record KStream.
• Tipe Join: Hanya mendukung Inner Join dan Left Join.

4. KTable-KTable Join (Stateful & Non-windowed) #

Menggabungkan dua tabel keadaan secara real-time.

• Semantik: Operasi ini menghasilkan KTable baru. Setiap kali record baru diperbarui di salah satu KTable (baik Table A maupun Table B), Kafka Streams akan melakukan lookup ke tabel pasangan dan memancarkan status record gabungan terbaru.
• Tipe Join: Mendukung Inner Join, Left Join, dan Outer Join. Jika salah satu entri dihapus (menerima record null/tombstone), join akan memancarkan event pembaruan yang sesuai.

Tabel Perbandingan Tipe Join #

Aturan Keras Co-Partitioning #

Ketika kita melakukan join stateful (kecuali jika menggunakan GlobalKTable), Kafka Streams secara mutlak mengharuskan data kita memenuhi syarat Co-partitioning (Penyelarasan Partisi).

SYARAT WAJIB CO-PARTITIONING:
1. Kedua topik input (misal KStream dan KTable) memiliki JUMLAH PARTISI YANG SAMA.
2. Kedua topik input dipartisi menggunakan ALGORITMA HASHER YANG SAMA (e.g. Murmur2).
3. Kunci record (key) dari kedua topik memiliki TIPE DATA YANG SAMA.

Mengapa Co-Partitioning Sangat Kritis? #

Ingat bahwa Kafka Streams memecah beban kerja berdasarkan partisi. Jika kita melakukan join antara KStream-A dan KTable-B, StreamTask 0 hanya akan memproses KStream-A partisi 0 dan KTable-B partisi 0.

Jika KStream-A memiliki 4 partisi, sedangkan KTable-B memiliki 2 partisi, data transaksi pengguna User_100 pada KStream mendarat di partisi 3, sedangkan data profil User_100 pada KTable mendarat di partisi 1. StreamTask 3 yang memproses KStream partisi 3 tidak akan pernah menemukan data profil User_100 karena RocksDB lokalnya hanya memuat data KTable partisi 3 (yang kosong). Akibatnya, operasi join akan mengembalikan nilai null secara terus-menerus, memicu hilangnya data secara diam-diam.

flowchart TD
    subgraph WRONG ["Skenario SALAH (Jumlah Partisi Berbeda)"]
        direction TB
        S1["KStream (4 Partisi) - Key: UserA (Partisi 3)"]
        T1["KTable (2 Partisi) - Key: UserA (Partisi 1)"]
        Task3["StreamTask 3 (Hanya baca Partisi 3)"]
        S1 --> Task3
        T1 -.->|Tidak Terbaca oleh Task 3!| Task3
        Task3 -->|"Join Result: NULL (Data Hilang!)"| Output1["Topic: enriched-out"]
    end

    subgraph CORRECT ["Skenario BENAR (Co-partitioned)"]
        direction TB
        S2["KStream (4 Partisi) - Key: UserA (Partisi 3)"]
        T2["KTable (4 Partisi) - Key: UserA (Partisi 3)"]
        Task3_Ok["StreamTask 3 (Membaca Partisi 3 dari KStream & KTable)"]
        S2 --> Task3_Ok
        T2 --> Task3_Ok
        Task3_Ok -->|"Join Result: SUCCESS (Profil Tergabung)"| Output2["Topic: enriched-out"]
    end

    style Task3 stroke:#d32f2f,stroke-width:2px
    style Task3_Ok stroke:#388e3c,stroke-width:2px

Menyelesaikan Pelanggaran Co-Partitioning lewat Repartitioning #

Jika kita terpaksa harus menggabungkan dua topik yang memiliki jumlah partisi berbeda, kita wajib melakukan Repartitioning (Partisi Ulang) pada KStream terlebih dahulu menggunakan operator selectKey().

Setiap kali kita memanggil selectKey(), Kafka Streams secara otomatis menandai aliran data tersebut untuk di-repartition. Di balik layar, KStreams akan menulis data dengan key baru tersebut ke topik internal transit tersembunyi (application-id-repartition), mendelegasikan broker untuk menyelaraskan partisi data sebelum dimasukkan ke dalam operator join.

Penanganan Data Telat (Late Arrivals) pada Windowed Joins #

Pada KStream-KStream join yang menggunakan jendela waktu (Windowed Joins), data telat (late arrivals) dikelola menggunakan properti Grace Period yang dideklarasikan pada objek JoinWindows.

• Jika sebuah event telat tiba di Stream A, dan event pasangannya di Stream B berada di dalam rentang waktu jendela yang valid, join akan berhasil dilakukan selama segmen penyimpanan RocksDB untuk jendela waktu tersebut belum dibersihkan (belum melewati masa Grace Period).
• Kita mengonfigurasi toleransi ini melalui metode .grace() pada kelas JoinWindows. Jika diabaikan, Kafka Streams akan menggunakan default grace period yang dapat memicu pembuangan data telat secara dini.

Kode Penerapan: Anti-Pattern vs Solusi Join Terkelola #

Mari kita bandingkan pendekatan anti-pattern dengan solusi terkelola menggunakan API Join resmi di Kafka Streams.

Kasus Penggunaan #

Kita memiliki aliran transaksi pesanan (order-events) dan tabel profil membership pelanggan (member-profiles). Kita ingin menggabungkannya secara real-time untuk menghasilkan data pesanan yang telah diperkaya (enriched-orders).

Anti-Pattern: Melakukan Query Database Eksternal di dalam Map / Filter #

Mencoba melakukan lookup profil membership pelanggan dengan melakukan panggilan HTTP API atau query database SQL langsung dari dalam fungsi .map() KStream adalah kesalahan besar.

// ANTI-PATTERN: Melakukan query database eksternal secara sinkron di dalam thread pemroses.
// ✗ Menghancurkan throughput aplikasi, memicu overhead latensi jaringan parah, dan rawan downtime database.
public class SyncDatabaseLookupProcessor {
    private static final DatabaseClient dbClient = new DatabaseClient("jdbc:postgresql://...");

    public static void buildTopology(StreamsBuilder builder) {
        builder.<String, String>stream("order-events")
            .mapValues(orderJson -> {
                String userId = parseUserId(orderJson);
                
                // ✗ SANGAT BURUK: Panggilan blocking JDBC sinkron lintas jaringan per event!
                String membership = dbClient.queryMembershipSync(userId);
                
                return enrichOrder(orderJson, membership);
            })
            .to("enriched-orders");
    }

    private static String parseUserId(String json) { return "usr_99"; }
    private static String enrichOrder(String json, String member) { return json; }
}

Solusi Praktis 1: KStream-KTable Join (Co-partitioned) #

Di bawah ini adalah cara yang benar dan optimal menggunakan KStream-KTable Join lokal. KTable dikonfigurasi secara persisten dan co-partitioned di broker Kafka.

// BENAR: Menggunakan KStream.join() untuk lookup state store lokal RocksDB secara aman dan cepat.
// ✓ Memanfaatkan zero-network latency, aman dari beban jaringan luar, otomatis ter-scale.
public class ResilientKStreamKTableJoin {
    
    public static void build(StreamsBuilder builder) {
        // 1. Baca KTable profil anggota (Topik profil memiliki 6 partisi)
        KTable<String, String> memberProfiles = builder.table(
            "member-profiles",
            Consumed.with(Serdes.String(), Serdes.String()),
            Materialized.as("member-profile-store")
        );

        // 2. Baca KStream transaksi pesanan (Topik pesanan juga memiliki 6 partisi)
        KStream<String, String> orders = builder.stream(
            "order-events",
            Consumed.with(Serdes.String(), Serdes.String())
        );

        // ✓ 3. Lakukan penggabungan data KStream-KTable secara instan di memori lokal RocksDB
        KStream<String, String> enrichedOrders = orders.join(
            memberProfiles,
            // ValueJoiner: Menggabungkan payload JSON pesanan dengan data membership
            (orderJson, profileJson) -> enrichPayload(orderJson, profileJson),
            // Definisikan Serde untuk join
            Joined.with(Serdes.String(), Serdes.String(), Serdes.String())
        );

        enrichedOrders.to("enriched-orders", Produced.with(Serdes.String(), Serdes.String()));
    }

    private static String enrichPayload(String order, String profile) {
        return String.format("{\"order\":%s,\"profile\":%s}", order, profile);
    }
}

Solusi Praktis 2: KStream-GlobalKTable Join (Non-co-partitioned) #

Jika topik member-profiles memiliki 3 partisi sedangkan topik order-events memiliki 12 partisi (sehingga melanggar co-partitioning), kita wajib menggunakan GlobalKTable.

// BENAR: Menggunakan GlobalKTable untuk join non-co-partitioned.
// ✓ Menghilangkan syarat kesamaan partisi dengan mendistribusikan 100% tabel ke semua instansi.
public class ResilientKStreamGlobalKTableJoin {
    
    public static void buildGlobal(StreamsBuilder builder) {
        // 1. Baca GlobalKTable profil anggota (Jumlah partisi bebas)
        GlobalKTable<String, String> globalProfiles = builder.globalTable(
            "member-profiles",
            Consumed.with(Serdes.String(), Serdes.String()),
            Materialized.as("global-profile-store")
        );

        // 2. Baca KStream pesanan
        KStream<String, String> orders = builder.stream(
            "order-events",
            Consumed.with(Serdes.String(), Serdes.String())
        );

        // ✓ 3. Lakukan join KStream-GlobalKTable
        KStream<String, String> enrichedOrders = orders.join(
            globalProfiles,
            // KeyValueMapper: Menentukan bagaimana mencari key join dari record KStream
            // Kita mengekstrak User ID dari payload JSON pesanan sebagai key pencocokan ke GlobalKTable
            (orderKey, orderValue) -> extractUserId(orderValue),
            // ValueJoiner
            (orderValue, profileValue) -> enrichPayload(orderValue, profileValue)
        );

        enrichedOrders.to("enriched-orders", Produced.with(Serdes.String(), Serdes.String()));
    }

    private static String extractUserId(String orderJson) {
        // Simulasi ekstrak user ID dari JSON
        return "user_123";
    }

    private static String enrichPayload(String order, String profile) {
        return order + "+" + profile;
    }
}

Solusi Praktis 3: KStream-KStream Join (Windowed Outer Join) #

Jika kita ingin menghubungkan dua aliran transaksi yang berbeda (misal, transaksi dari mesin EDC edc-payments dengan laporan kliring bank bank-clearings untuk mendeteksi transaksi yang tidak memiliki pencocokan laporan bank), kita harus menggunakan Outer Join dengan jendela waktu.

// BENAR: Menggunakan KStream.outerJoin() dengan JoinWindows.
// ✓ Melakukan rekonsiliasi data dua arah secara terdistribusi.
public class PaymentReconciliationJoin {
    
    public static void build(StreamsBuilder builder) {
        KStream<String, String> edcPayments = builder.stream(
            "edc-payments",
            Consumed.with(Serdes.String(), Serdes.String())
        );

        KStream<String, String> bankClearings = builder.stream(
            "bank-clearings",
            Consumed.with(Serdes.String(), Serdes.String())
        );

        // ✓ Definisikan jendela waktu pencocokan 1 jam dengan grace period 5 menit
        JoinWindows joinWindow = JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofHours(1))
            .grace(Duration.ofMinutes(5));

        // ✓ Lakukan Outer Join
        KStream<String, String> reconResults = edcPayments.outerJoin(
            bankClearings,
            (edcValue, bankValue) -> reconcile(edcValue, bankValue),
            joinWindow,
            StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String())
        );

        reconResults.to("reconciliation-reports", Produced.with(Serdes.String(), Serdes.String()));
    }

    private static String reconcile(String edc, String bank) {
        if (edc != null && bank != null) {
            return "{\"status\":\"MATCHED\",\"edc\":" + edc + ",\"bank\":" + bank + "}";
        } else if (edc != null) {
            return "{\"status\":\"MISSING_BANK_REPORT\",\"edc\":" + edc + "}";
        } else {
            return "{\"status\":\"MISSING_EDC_RECORD\",\"bank\":" + bank + "}";
        }
    }
}

Ringkasan #

• Join Matrix — Kombinasi penggabungan aliran dan tabel (KStream-KStream, KStream-KTable, KStream-GlobalKTable, KTable-KTable) untuk memfasilitasi enrich data real-time.
• Co-partitioning — Aturan ketat di mana kedua topik input wajib memiliki jumlah partisi yang sama, tipe data key yang sama, dan hasher yang identik untuk mencegah data hilang saat join.
• selectKey() — Operator di KStream untuk mengubah kunci record, yang secara otomatis memicu proses re-partitioning di broker demi memenuhi syarat co-partitioning.
• Lookup Join — Model join searah (seperti KStream-KTable) di mana kedatangan event baru di KStream memicu pencarian pasif ke RocksDB lokal secara instan.
• Join Window — Jendela waktu wajib pada KStream-KStream join untuk membatasi perbedaan waktu timestamp di antara peristiwa yang akan digabungkan.
• GlobalKTable Join — Metode join yang mendistribusikan salinan penuh data tabel ke seluruh instance mikroservis untuk membebaskan tim dari aturan co-partitioning.

← Sebelumnya: Windowing   Berikutnya: Exactly-Once Processing →