Bagian 2: Coroutines dan Concurrency

Pada Bagian 1 kita mendesain struktur pipeline data multi-stage. Sekarang kita implementasikan dengan coroutines: fetch paralel dengan async, transformasi dengan Flow, dan sink sebagai terminal operasi. Pendekatan ini memanfaatkan structured concurrency untuk mengelola lifecycle coroutine dan menyediakan mekanisme cancel yang otomatis.

Tiga Pola Concurrency Utama #

Pipeline asinkron di Kotlin biasanya menggabungkan tiga pola:

Pola	API	Cocok untuk
Parallel fetch	`async { }.awaitAll()`	Memanggil banyak sumber data sekaligus
Streaming	`Flow<T>`	Pipeline dengan banyak item, satu per satu
Retry	custom suspend	Operasi yang mungkin gagal sementara (network glitch)

Implementasi Pipeline #

Berikut implementasi pipeline 3-stage dengan async untuk fetch paralel dan Flow untuk transformasi:

kotlin

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

// === Pipeline data asinkron: 3 stage ===
// 1. Fetch: ambil batch data dari sumber (simulasi network)
// 2. Transform: enrich data
// 3. Sink: tulis ke output (simulasi database)

data class DataMentah(val id: Int, val raw: String)
data class DataKaya(val id: Int, val raw: String, val enriched: String, val processedAt: Long)

// === Retry dengan exponential backoff ===
suspend fun <T> retryWithBackoff(
    maxAttempts: Int = 3,
    initialDelayMs: Long = 10,
    factor: Double = 2.0,
    block: suspend () -> T
): T {
    var currentDelay = initialDelayMs
    var lastError: Throwable? = null
    repeat(maxAttempts) { attempt ->
        try {
            return block()
        } catch (e: Throwable) {
            lastError = e
            println("  [retry] attempt ${attempt + 1} gagal: ${e.message}, tunggu ${currentDelay}ms")
            kotlinx.coroutines.delay(currentDelay)
            currentDelay = (currentDelay * factor).toLong()
        }
    }
    throw lastError ?: IllegalStateException("Tidak ada percobaan")
}

// === Pipeline dengan Flow ===
fun pipelineFlow(items: List<DataMentah>): Flow<DataKaya> = flow {
    for (item in items) {
        // Simulasi: skip item bermasalah, log saja
        if (item.id == 3) {
            println("  [skip] id=${item.id} gagal, lanjut ke item berikutnya")
            continue
        }
        val enriched = item.raw.uppercase() + "-ENRICHED"
        emit(DataKaya(item.id, item.raw, enriched, System.currentTimeMillis()))
    }
}

fun main() = runBlocking {
    val dataRaw = (1..5).map { DataMentah(it, "data-$it") }

    println("=== Stage 1: Fetch (paralel) ===")
    val fetched = coroutineScope {
        dataRaw.map { item ->
            async(Dispatchers.IO) {
                println("  [fetch] id=${item.id}")
                kotlinx.coroutines.delay(50)
                item
            }
        }.awaitAll()
    }
    println("Fetched ${fetched.size} items")

    println("\n=== Stage 2: Transform dengan retry ===")
    val results = mutableListOf<DataKaya>()
    pipelineFlow(fetched).collect { data ->
        results.add(data)
        println("  [transform] id=${data.id}: ${data.enriched}")
    }

    println("\n=== Stage 3: Sink ===")
    results.forEach { d ->
        println("  [sink] id=${d.id} → db (enriched: ${d.enriched})")
    }

    println("\n=== Test retry dengan backoff ===")
    var attemptCount = 0
    val result = retryWithBackoff(maxAttempts = 4, initialDelayMs = 5) {
        attemptCount++
        println("  [attempt $attemptCount]")
        if (attemptCount < 3) throw RuntimeException("sengaja gagal")
        "OK pada attempt ke-$attemptCount"
    }
    println("Hasil retry: $result")

    println("\nPipeline asinkron selesai.")
}

Output yang Dihasilkan:

text
=== Stage 1: Fetch (paralel) ===
  [fetch] id=1
  [fetch] id=3
  [fetch] id=4
  [fetch] id=5
  [fetch] id=2
Fetched 5 items

=== Stage 2: Transform dengan retry ===
  [transform] id=1: DATA-1-ENRICHED
  [transform] id=2: DATA-2-ENRICHED
  [skip] id=3 gagal, lanjut ke item berikutnya
  [transform] id=4: DATA-4-ENRICHED
  [transform] id=5: DATA-5-ENRICHED

=== Stage 3: Sink ===
  [sink] id=1 → db (enriched: DATA-1-ENRICHED)
  [sink] id=2 → db (enriched: DATA-2-ENRICHED)
  [sink] id=4 → db (enriched: DATA-4-ENRICHED)
  [sink] id=5 → db (enriched: DATA-5-ENRICHED)

=== Test retry dengan backoff ===
  [attempt 1]
  [retry] attempt 1 gagal: sengaja gagal, tunggu 5ms
  [attempt 2]
  [retry] attempt 2 gagal: sengaja gagal, tunggu 10ms
  [attempt 3]
Hasil retry: OK pada attempt ke-3

Pipeline asinkron selesai.

Mari kita bedah baris-baris penting di atas:

coroutineScope { ... }: scoped builder yang membatalkan semua child jika ada yang gagal. Fetch paralel di-bungkus dalam scope ini agar jika satu async gagal, semua di-cancel.
async(Dispatchers.IO) { ... }.awaitAll(): meluncurkan 5 fetch secara paralel pada dispatcher IO (khusus operasi blocking). awaitAll menunggu semua selesai dan mengembalikan List<DataMentah>.
flow { emit(...) }: cold stream — baru dieksekusi saat ada .collect. Setiap collector mendapatkan eksekusi independen, cocok untuk retry.
continue di dalam flow { }: skip item bermasalah tanpa menggagalkan seluruh pipeline. Strategi graceful degradation.
retryWithBackoff { block() }: suspend function generic yang mencoba block hingga maxAttempts. Setiap percobaan gagal diikuti delay yang meningkat eksponensial (5ms → 10ms → 20ms).
System.currentTimeMillis(): timestamp untuk metadata. Catatan: nilai ini non-deterministik sehingga tidak boleh dimasukkan ke output yang diverifikasi — di sini hanya untuk demonstrasi, bukan di-assert.

Backpressure dan Buffering #

Flow bersifat synchronous antar stage — emit menunggu collect selesai. Untuk stage yang lambat, gunakan .buffer() untuk mendecouple producer/consumer:

kotlin
pipelineFlow(data)
    .buffer(10) // buffer 10 item antar stage
    .collect { ... }

.conflate() hanya menyimpan item terakhir. .flatMapLatest() membatalkan flow sebelumnya saat item baru tiba.

Cancellation Propagation #

Structured concurrency memastikan cancellation menyebar ke child:

kotlin
coroutineScope {
    val job = launch { while (isActive) { /* long work */ } }
    delay(1000)
    job.cancel() // anak ikut dibatalkan
}

Coroutine yang sedang suspend (delay, await, receive) akan memeriksa status dan melempar CancellationException.

Referensi Lanjutan #

Kotlin coroutines design — dokumentasi desain coroutines untuk koordinasi launch dan async lintas stage
Kotlin documentation home — halaman utama dokumentasi Kotlin sebagai rujukan fitur pendukung konkurensi

Selanjutnya #

Pada Bagian 3, kita akan membungkus pipeline ini dengan error handling yang lebih lengkap: try-catch per stage, SupervisorJob untuk isolasi failure, dan strategi retryWhen untuk kondisi spesifik.