Belajar Apache Flink - Custom Connectors & Extensions

#Pendahuluan

Ekosistem connector Flink sangat luas, tetapi tidak mencakup semua sistem. Kadang kalian harus berintegrasi dengan protokol internal, API pihak ketiga, atau format data khusus. Episode 18 ini mengajarkan cara memperluas Flink sendiri — membangun custom source dan sink, menulis serializer, dan memahami lifecycle operator.

Kita akan membuat custom source, mempelajari cara membangun sink, menulis serializer kustom, membedah operator lifecycle dan checkpoint hooks, dan menutup dengan panduan berkontribusi ke ekosistem connector resmi. Setelah episode ini, tidak ada sistem yang benar-benar "tidak bisa" terhubung ke Flink.

#Membuat Custom Source

#RichSourceFunction untuk Source Sederhana

Cara paling mudah membangun source adalah menurunkan dari RichSourceFunction:

import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
 
public class CustomSource extends RichSourceFunction<String> {
    private volatile boolean running = true;
 
    @Override
    public void run(SourceFunction.SourceContext<String> ctx) throws Exception {
        while (running) {
            ctx.collect("hello-" + System.currentTimeMillis());
            Thread.sleep(1000);
        }
    }
 
    @Override
    public void cancel() {
        running = false;
    }
}

ctx.collect mengirim record ke pipeline, dan cancel() dipanggil saat job berhenti — di sinilah kalian menutup koneksi. volatile boolean running menjamin flag terbaca antarthread saat cancel.

#Source yang Lebih Modern

Untuk produksi, gunakan antarmuka Source yang lebih baru yang mendukung checkpoint, pemulihan posisi, dan batching. SourceFunction masih valid dan sering dipakai untuk integrasi cepat, tetapi Source interface adalah arah jangka panjang Flink.

#Membuat Custom Sink

#SinkFunction untuk Menulis Data

Custom sink menurunkan dari RichSinkFunction:

import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 
public class HttpSink extends RichSinkFunction<Order> {
    @Override
    public void invoke(Order value, Context ctx) throws Exception {
        HttpClient.post("/orders", value);
    }
}

invoke dipanggil untuk setiap record. Untuk throughput tinggi, batch record di dalam state dan kirim berkala — memanggil HTTP per record akan membuat sink menjadi bottleneck.

#TwoPhaseCommit untuk Exactly-once

Sink yang mendukung transaksi dua fase mengimplementasikan TwoPhaseCommitSinkFunction — pola yang dipakai Kafka sink untuk exactly-once. Ini mencakup method beginTransaction, preCommit, dan commit. Implementasi penuh membutuhkan dukungan transaksi dari sistem tujuan; jika tidak ada, gunakan sink idempoten sebagai pengganti.

#Serializer dan Codec Kustom

#TypeSerializer untuk Tipe Data Sendiri

Saat Flink belum mengenal tipe data kalian, implementasikan TypeSerializer:

import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
 
public class PointSerializer extends TypeSerializer<Point> {
    @Override
    public void serialize(Point value, DataOutputView target) throws IOException {
        target.writeDouble(value.getX());
        target.writeDouble(value.getY());
    }
 
    @Override
    public Point deserialize(Point reuse, DataInputView source) throws IOException {
        return new Point(source.readDouble(), source.readDouble());
    }
}

serialize menulis objek ke buffer, dan deserialize membacanya kembali. Serializer yang efisien mengurangi overhead transfer data antar operator — salah satu lever performa yang sering diabaikan.

#Codec untuk Format Kustom

Jika kalian memakai format wire yang tidak standar (bukan JSON, Avro, Protobuf), tulis codec sebagai SerializationSchema dan DeserializationSchema. Keduanya dipasang pada connector Kafka dan sejenisnya, sehingga format kustom bisa dipakai tanpa menulis seluruh connector.

#Operator Lifecycle dan Checkpoint Hooks

#Lifecycle Method

Semua function kaya (Rich*) memiliki lifecycle:

• open: inisialisasi, dipanggil sekali sebelum pemrosesan.
• close: pembersihan, dipanggil saat job berhenti.
• snapshotState: menyimpan state ke checkpoint.
• initializeState: memuat state dari checkpoint atau savepoint.

#Checkpoint Hooks

@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
    bufferState.clear();
    for (String item : buffer) {
        bufferState.add(item);
    }
}
 
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
    bufferState = context.getOperatorStateStore()
        .getListState(new ListStateDescriptor<>("buffer", String.class));
    if (context.isRestored()) {
        for (String item : bufferState.get()) {
            buffer.add(item);
        }
    }
}

snapshotState menyimpan buffer ke operator state, dan initializeState memuatnya kembali saat restore. Kombinasi keduanya membuat custom operator kalian setara dengan operator bawaan dalam hal fault tolerance.

#Berkontribusi ke Ekosistem Connector

#Dari Kustom ke Resmi

Jika connector kalian berguna secara umum, pertimbangkan berkontribusi ke apache/flink-connectors. Langkah-langkahnya:

• Ikuti gaya koding dan konvensi proyek.
• Tulis test yang mencakup checkpoint dan restore.
• Dokumentasikan opsi konfigurasi.
• Kirim pull request dan tanggapi review.

mvn clean package -DskipTests
./bin/flink run -d target/custom-connector.jar

Perintah mvn clean package membangun JAR connector, dan ./bin/flink run -d mengujinya di cluster lokal sebelum kalian berbagi dengan komunitas.

#Penutup

Episode 18 mengajarkan kalian memperluas Flink: membangun custom source dan sink, menulis serializer dan codec untuk format khusus, memanfaatkan operator lifecycle dan checkpoint hooks, serta langkah berkontribusi ke ekosistem connector.

Inti yang harus dibawa pulang:

• RichSourceFunction dan RichSinkFunction adalah titik masuk integrasi kustom.
• cancel dan invoke adalah tempat mengelola koneksi di source dan sink.
• TypeSerializer kustom mengurangi overhead transfer data.
• snapshotState dan initializeState membuat custom operator fault-tolerant.
• Connector yang baik dikuji, didokumentasikan, dan dibagikan ke komunitas.

Di episode 19 selanjutnya kita akan membahas operational readiness & runbooks — menyusun runbook untuk kegagalan job, respons terhadap checkpoint failure dan job crash, backup dan disaster recovery, serta chaos testing untuk ketahanan streaming. Kalian akan siap menghadapi malam ketika produksi berteriak.