Bagian 3: Error Handling dan Retry

/** Satu baris berkas mentah: nomor urut (mulai dari 1) dan teks aslinya. */
public record BarisData(int nomor, String teks) {}

/** Satu kejadian pada jejak percobaan ulang; diurutkan menurut baris lalu percobaan. */
public record CatatanPercobaan(int nomorBaris, int percobaanKe, String kejadian) {}

import java.util.List;

/**
 * Isi berkas penjualan contoh: 36 baris data kuartal pertama 2026.
 * Dua puluh enam baris normal, empat baris cacat format, dan enam baris
 * berpenanda kondisi layanan untuk menguji percobaan ulang.
 */
public final class DataContoh {

    private DataContoh() {
        // Kelas utilitas: seluruh anggota static.
    }

    /** Isi berkas contoh; pada aplikasi nyata berasal dari Files.readAllLines. */
    public static List<String> isiBerkas() {
        return List.of(
                "2026-01-05;Bandung;Budi;Elektronik;1250000",
                "2026-01-07;Surabaya;Siti;Pakaian;340000",
                "2026-01-09;bandung;Dewi;Buku;95000",
                "2026-01-12;Yogyakarta;Budi;Peralatan;780000",
                "2026-01-15;Medan;Siti;Elektronik;2100000",
                "2026-01-18;Bandung;Dewi;Buku;128000",
                "2026-01-20;Medan;Siti;Pakaian;-150000",
                "2026-01-21;Makassar;Budi;Pakaian;450000",
                "2026-01-24;surabaya;Siti;Peralatan;620000",
                "2026-01-27;Yogyakarta;Dewi;Elektronik;1850000",
                "2026-02-02;Bandung;Siti;Pakaian;295000",
                "2026-02-05;Bandung;Dewi;Buku",
                "2026-02-08;Surabaya;Budi;Elektronik;990000",
                "2026-02-11;Makassar;Siti;Camilan;88000",
                "2026-02-14;Yogyakarta;Dewi;Pakaian;380000",
                "2026-02-17;Medan;Budi;Elektronik;1720000",
                "2026-02-20;bandung;Siti;Buku;112000",
                "2026-02-23;Surabaya;Dewi;Peralatan;690000",
                "2026-02-26;Makassar;Budi;Pakaian;410000",
                "2026-02-28;Yogyakarta;Siti;Elektronik;1580000",
                "2026-03-04;Medan;Dewi;Buku;64000",
                "2026-03-07;Bandung;Budi;Peralatan;830000",
                "2026-03-10;Surabaya;Siti;Elektronik;1350000",
                "2026-03-13;Makassar;Dewi;Buku;88000",
                "2026-03-16;Yogyakarta;Budi;Pakaian;365000",
                "2026-03-19;Bandung;Siti;Peralatan;575000",
                "2026-03-22;medan;Dewi;Elektronik;1980000",
                "2026-03-25;Surabaya;Budi;Buku;102000",
                "2026-03-28;Makassar;Siti;Pakaian;430000",
                "2026-13-31;Yogyakarta;Dewi;Peralatan;715000",
                "2026-03-02;Bandung;Siti;Buku;140000;transien",
                "2026-03-08;Medan;Dewi;Elektronik;870000;transien",
                "2026-03-12;Surabaya;Budi;Peralatan;430000;timeout",
                "2026-03-18;Yogyakarta;Dewi;Buku;95000;timeout",
                "2026-03-22;Makassar;Siti;Pakaian;310000;rusak",
                "2026-03-29;Bandung;Budi;Elektronik;1120000;rusak");
    }
}

import java.util.List;
import java.util.Locale;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/** Bagian 3: DemoRetry — pipeline tahan galat dengan retry, jejak, dan laporan galat. */
public class DemoRetry {

    public static void main(String[] args) {
        List<String> isi = DataContoh.isiBerkas();
        HasilPipeline hasil;
        try (ExecutorService eksekutor = Executors.newVirtualThreadPerTaskExecutor()) {
            hasil = Pipeline.jalankanDenganRetry(isi, eksekutor, KebijakanRetry.baku());
        }

        System.out.println("== Pipeline dengan retry (" + isi.size() + " baris) ==");
        System.out.println("baris sah         : " + hasil.jumlahSah());
        System.out.println("ditolak validasi  : " + hasil.jumlahDitolak());
        System.out.println("gagal tetap       : " + hasil.jumlahGagalTetap());
        System.out.println("unit tunggu total : " + hasil.unitTungguTotal() + " unit");
        System.out.println();
        System.out.println("== Jejak percobaan ulang (terurut baris, lalu percobaan) ==");
        for (CatatanPercobaan c : hasil.jejakPercobaan()) {
            System.out.printf(Locale.US, "  baris %2d percobaan %d: %s%n",
                    c.nomorBaris(), c.percobaanKe(), c.kejadian());
        }
        System.out.println();
        System.out.println("== Laporan galat (terurut nomor baris) ==");
        hasil.laporanGalat().forEach(g -> System.out.println("  " + g));
        System.out.println();
        System.out.println("== Ringkasan agregasi (" + hasil.jumlahSah() + " kejadian) ==");
        System.out.println("kota     : " + hasil.jumlahKota());
        System.out.println("kategori : " + hasil.jumlahKategori());
    }
}

import java.math.BigDecimal;
import java.time.LocalDate;

/**
 * Kontrak keluaran tahap validasi: event penjualan yang lolos pemeriksaan.
 * Komponen penanda (string kosong bila tidak ada) menyimpan simulasi kondisi
 * layanan: "transien", "timeout", atau "rusak".
 */
public record EventPenjualan(int nomorBaris, LocalDate tanggal, String kota,
                             String pelanggan, String kategori, BigDecimal nilai,
                             String penanda) {}

/**
 * Akar hierarki galat pemrosesan baris.
 * Turunan RuntimeException agar task paralel bebas melempar tanpa deklarasi checked.
 */
public class GalatProsesException extends RuntimeException {

    public GalatProsesException(String pesan) {
        super(pesan);
    }

    /** Bentuk berantai: penyebab teknis disimpan agar bisa dirunut kemudian. */
    public GalatProsesException(String pesan, Throwable penyebab) {
        super(pesan, penyebab);
    }
}

/** Galat yang layak dicoba ulang: layanan sedang sibuk atau tidak merespons. */
public class GalatSementaraException extends GalatProsesException {

    public GalatSementaraException(String pesan) {
        super(pesan);
    }

    public GalatSementaraException(String pesan, Throwable penyebab) {
        super(pesan, penyebab);
    }
}

/** Galat yang tidak akan sembuh dicoba ulang; menyimpan penyebab teknisnya. */
public class GalatTetapException extends GalatProsesException {

    public GalatTetapException(String pesan, Throwable penyebab) {
        super(pesan, penyebab);
    }
}

/**
 * Kontrak hasil pemrosesan satu baris: nilai data, bukan exception.
 * Varian terkunci lewat sealed sehingga compiler memaksa setiap switch
 * menangani semua kemungkinan — termasuk varian GagalTetap yang ditambahkan
 * ketika kebijakan percobaan ulang menyerah.
 */
public sealed interface HasilBaris permits HasilBaris.Sah, HasilBaris.TidakSah,
        HasilBaris.GagalTetap {

    /** Nomor baris sumber — dipakai pelaporan galat dan pengurutan hasil. */
    int nomorBaris();

    /** Baris selesai diproses dan membawa kejadian siap agregasi. */
    record Sah(KejadianTerproses kejadian) implements HasilBaris {
        @Override
        public int nomorBaris() {
            return kejadian.nomorBaris();
        }
    }

    /** Baris ditolak validator; alasan ditulis untuk laporan galat. */
    record TidakSah(int nomorBaris, String alasan) implements HasilBaris {
        @Override
        public int nomorBaris() {
            return nomorBaris;
        }
    }

    /** Baris gagal diproses setelah kebijakan retry menyerah; akar masalah dirunut. */
    record GagalTetap(int nomorBaris, int percobaan, String pesan, String penyebabAkar)
            implements HasilBaris {
        @Override
        public int nomorBaris() {
            return nomorBaris;
        }
    }
}

import java.util.List;

/** Hasil pipeline utuh: statistik, laporan utama, laporan galat, dan jejak percobaan. */
public record HasilPipeline(int jumlahBaris,
                            int jumlahSah,
                            int jumlahDitolak,
                            int jumlahGagalTetap,
                            long unitTungguTotal,
                            int jumlahKota,
                            int jumlahKategori,
                            List<String> laporan,
                            List<String> laporanGalat,
                            List<CatatanPercobaan> jejakPercobaan) {}

/**
 * Kebijakan percobaan ulang: batas percobaan dan backoff eksponensial dalam
 * "unit biaya simulasi" — bukan sleep nyata — agar keluaran tetap deterministik.
 */
public record KebijakanRetry(int percobaanMaks, long unitTungguDasar) {

    /** Kebijakan baku: maksimal 3 percobaan, tunggu dasar 20 unit. */
    public static KebijakanRetry baku() {
        return new KebijakanRetry(3, 20);
    }

    /** Panjang tunggu setelah percobaan ke-N gagal: 20, 40, 80, ... unit. */
    public long unitTunggu(int percobaanKe) {
        return unitTungguDasar << (percobaanKe - 1);
    }
}

import java.math.BigDecimal;
import java.time.LocalDate;

/**
 * Kontrak keluaran tahap transformasi: data siap agregasi.
 * Kota sudah dibakukan (huruf kapital) dan nilai berjasa sudah dihitung.
 */
public record KejadianTerproses(int nomorBaris, LocalDate tanggal, String kota,
                                String pelanggan, String kategori, BigDecimal nilai,
                                BigDecimal nilaiBerjasa) {}

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.net.ConnectException;
import java.time.LocalDate;
import java.time.format.DateTimeParseException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.LongAdder;
import java.util.stream.Collectors;

/** Pipeline tahan galat: baca -> validasi -> transformasi (dengan retry) -> agregasi -> tulis. */
public final class Pipeline {

    /** Kategori yang dikenal validator; di luar daftar ini baris ditolak. */
    private static final Set<String> KATEGORI_DIKENAL =
            Set.of("Elektronik", "Pakaian", "Buku", "Peralatan");

    /** Penanda kondisi layanan yang boleh menempel pada baris data. */
    private static final Set<String> PENANDA_DIKENAL =
            Set.of("transien", "timeout", "rusak");

    /** Faktor biaya layanan 2% yang dihitung tahap transformasi. */
    private static final BigDecimal BIAYA_LAYANAN = new BigDecimal("1.02");

    private Pipeline() {
        // Kelas utilitas: seluruh anggota static.
    }

    /** Tahap baca: memberi nomor urut pada tiap baris berkas. */
    public static List<BarisData> baca(List<String> isiBerkas) {
        List<BarisData> baris = new ArrayList<>(isiBerkas.size());
        for (int i = 0; i < isiBerkas.size(); i++) {
            baris.add(new BarisData(i + 1, isiBerkas.get(i)));
        }
        return baris;
    }

    /**
     * Tahap validasi lalu transformasi dengan percobaan ulang untuk satu baris.
     * Setiap percobaan gagal tercatat sebagai jejak; hasil akhir tetap nilai data.
     */
    public static HasilBaris prosesDenganRetry(BarisData baris, KebijakanRetry kebijakan,
            ConcurrentLinkedQueue<CatatanPercobaan> jejak, LongAdder totalTunggu) {
        // --- tahap validasi ---
        String[] segmen = baris.teks().split(";", -1);
        if (segmen.length != 5 && segmen.length != 6) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "jumlah segmen " + segmen.length + ", harus 5 atau 6");
        }
        LocalDate tanggal;
        try {
            tanggal = LocalDate.parse(segmen[0]);
        } catch (DateTimeParseException e) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "tanggal tidak terurai: '" + segmen[0] + "'");
        }
        String kota = segmen[1].trim();
        String pelanggan = segmen[2].trim();
        String kategori = segmen[3].trim();
        if (kota.isEmpty()) {
            return new HasilBaris.TidakSah(baris.nomor(), "kota kosong");
        }
        if (pelanggan.isEmpty()) {
            return new HasilBaris.TidakSah(baris.nomor(), "nama pelanggan kosong");
        }
        if (!KATEGORI_DIKENAL.contains(kategori)) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "kategori tidak dikenal: '" + kategori + "'");
        }
        BigDecimal nilai;
        try {
            nilai = new BigDecimal(segmen[4].trim());
        } catch (NumberFormatException e) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "nilai tidak terurai: '" + segmen[4].trim() + "'");
        }
        if (nilai.signum() <= 0) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "nilai harus positif: " + nilai.toPlainString());
        }
        String penanda = segmen.length == 6 ? segmen[5].trim() : "";
        if (!penanda.isEmpty() && !PENANDA_DIKENAL.contains(penanda)) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "penanda tidak dikenal: '" + penanda + "'");
        }
        EventPenjualan event = new EventPenjualan(baris.nomor(), tanggal, kota,
                pelanggan, kategori, nilai, penanda);

        // --- tahap transformasi dengan percobaan ulang ---
        for (int percobaan = 1; percobaan <= kebijakan.percobaanMaks(); percobaan++) {
            try {
                KejadianTerproses kejadian = prosesDenganLayanan(event, percobaan);
                if (percobaan > 1) {
                    jejak.add(new CatatanPercobaan(baris.nomor(), percobaan,
                            "berhasil setelah percobaan ulang"));
                }
                return new HasilBaris.Sah(kejadian);
            } catch (GalatTetapException e) {
                // Galat tetap tidak layak dicoba ulang: gagal cepat.
                jejak.add(new CatatanPercobaan(baris.nomor(), percobaan,
                        "gagal tetap, tanpa percobaan ulang: " + e.getMessage()));
                return new HasilBaris.GagalTetap(baris.nomor(), percobaan,
                        e.getMessage(), akar(e));
            } catch (GalatSementaraException e) {
                if (percobaan == kebijakan.percobaanMaks()) {
                    jejak.add(new CatatanPercobaan(baris.nomor(), percobaan,
                            "percobaan habis: " + e.getMessage()));
                    return new HasilBaris.GagalTetap(baris.nomor(), percobaan,
                            "percobaan ulang habis: " + e.getMessage(), akar(e));
                }
                long tunggu = kebijakan.unitTunggu(percobaan);
                totalTunggu.add(tunggu);
                jejak.add(new CatatanPercobaan(baris.nomor(), percobaan,
                        "gagal sementara: " + e.getMessage() + "; tunggu " + tunggu + " unit"));
            }
        }
        throw new IllegalStateException("tidak tercapai: loop selalu mengembalikan hasil");
    }

    /** Transformasi satu event; layanan simulasi melempar galat sesuai penanda. */
    private static KejadianTerproses prosesDenganLayanan(EventPenjualan event, int percobaan) {
        String penanda = event.penanda();
        if ("transien".equals(penanda) && percobaan == 1) {
            throw new GalatSementaraException("layanan tarif sedang sibuk");
        }
        if ("timeout".equals(penanda)) {
            throw new GalatSementaraException("layanan tarif tidak merespons",
                    new ConnectException("waktu tunggu koneksi habis"));
        }
        if ("rusak".equals(penanda)) {
            throw new GalatTetapException("sambungan gudang data terputus",
                    new ConnectException("koneksi ditolak oleh gudang data"));
        }
        return ubah(event);
    }

    /** Tahap transformasi: bakukan kota dan hitung nilai berjasa 2%. */
    public static KejadianTerproses ubah(EventPenjualan event) {
        BigDecimal berjasa = event.nilai().multiply(BIAYA_LAYANAN)
                .setScale(2, RoundingMode.HALF_EVEN);
        return new KejadianTerproses(event.nomorBaris(), event.tanggal(),
                event.kota().toUpperCase(Locale.ROOT), event.pelanggan(),
                event.kategori(), event.nilai(), berjasa);
    }

    /** Tahap agregasi: ringkasan terurut dari seluruh kejadian. */
    public static RingkasanAgregasi agregasi(List<KejadianTerproses> kejadian) {
        Map<String, BigDecimal> perKota = kejadian.stream()
                .collect(Collectors.groupingBy(KejadianTerproses::kota,
                        TreeMap::new,
                        Collectors.reducing(BigDecimal.ZERO,
                                KejadianTerproses::nilaiBerjasa, BigDecimal::add)));
        Map<String, Long> perKategori = kejadian.stream()
                .collect(Collectors.groupingBy(KejadianTerproses::kategori,
                        TreeMap::new, Collectors.counting()));
        BigDecimal totalNilai = kejadian.stream()
                .map(KejadianTerproses::nilai)
                .reduce(BigDecimal.ZERO, BigDecimal::add);
        BigDecimal totalBerjasa = kejadian.stream()
                .map(KejadianTerproses::nilaiBerjasa)
                .reduce(BigDecimal.ZERO, BigDecimal::add);
        return new RingkasanAgregasi(kejadian.size(), totalNilai, totalBerjasa,
                perKota, perKategori);
    }

    /** Tahap tulis: susun laporan utama dari ringkasan dan hasil per baris. */
    public static List<String> tulis(RingkasanAgregasi ringkasan, List<HasilBaris> hasil) {
        int sah = 0;
        int ditolak = 0;
        int gagal = 0;
        for (HasilBaris h : hasil) {
            switch (h) {
                case HasilBaris.Sah _ -> sah++;
                case HasilBaris.TidakSah _ -> ditolak++;
                case HasilBaris.GagalTetap _ -> gagal++;
            }
        }
        List<String> laporan = new ArrayList<>();
        laporan.add("=== Laporan pipeline penjualan ===");
        laporan.add("baris dibaca     : " + hasil.size());
        laporan.add("baris sah        : " + sah);
        laporan.add("ditolak validasi : " + ditolak);
        laporan.add("gagal tetap      : " + gagal);
        laporan.add(String.format(Locale.US, "total nilai     : %,14.2f", ringkasan.totalNilai()));
        laporan.add(String.format(Locale.US, "total berjasa   : %,14.2f", ringkasan.totalBerjasa()));
        laporan.add("nilai berjasa per kota:");
        ringkasan.nilaiBerjasaPerKota().forEach((kota, nilai) ->
                laporan.add(String.format(Locale.US, "  %-12s %,14.2f", kota, nilai)));
        laporan.add("jumlah transaksi per kategori:");
        ringkasan.jumlahPerKategori().forEach((kategori, jumlah) ->
                laporan.add(String.format(Locale.US, "  %-12s %6d", kategori, jumlah)));
        return laporan;
    }

    /** Laporan galat terurut nomor baris: penolakan validasi dan kegagalan tetap. */
    public static List<String> tulisGalat(List<HasilBaris> hasil) {
        List<String> galat = new ArrayList<>();
        for (HasilBaris h : hasil) {
            switch (h) {
                case HasilBaris.Sah _ -> { }
                case HasilBaris.TidakSah(int nomor, String alasan) ->
                        galat.add(String.format(Locale.US,
                                "baris %2d  DITOLAK VALIDASI : %s", nomor, alasan));
                case HasilBaris.GagalTetap(int nomor, int percobaan, String pesan,
                        String penyebabAkar) ->
                        galat.add(String.format(Locale.US,
                                "baris %2d  GAGAL TETAP      : %s (percobaan %d; akar: %s)",
                                nomor, pesan, percobaan, penyebabAkar));
            }
        }
        return galat;
    }

    /** Menjalankan pipeline dengan percobaan ulang secara paralel pada eksekutor. */
    public static HasilPipeline jalankanDenganRetry(List<String> isiBerkas,
            ExecutorService eksekutor, KebijakanRetry kebijakan) {
        List<BarisData> baris = baca(isiBerkas);
        Map<Integer, HasilBaris> hasilPerNomor = new ConcurrentHashMap<>();
        // Jejak percobaan ditulis banyak thread sekaligus tanpa penguncian manual.
        ConcurrentLinkedQueue<CatatanPercobaan> jejak = new ConcurrentLinkedQueue<>();
        LongAdder totalTunggu = new LongAdder();
        List<CompletableFuture<Void>> tugas = new ArrayList<>(baris.size());
        for (BarisData b : baris) {
            tugas.add(CompletableFuture.runAsync(() -> hasilPerNomor.put(b.nomor(),
                    prosesDenganRetry(b, kebijakan, jejak, totalTunggu)), eksekutor));
        }
        CompletableFuture.allOf(tugas.toArray(new CompletableFuture<?>[0])).join();
        List<HasilBaris> terurut = new ArrayList<>(new TreeMap<>(hasilPerNomor).values());
        // Urutan kedatangan jejak tidak menentu: urutkan menurut baris lalu percobaan.
        List<CatatanPercobaan> jejakTerurut = jejak.stream()
                .sorted(Comparator.comparingInt(CatatanPercobaan::nomorBaris)
                        .thenComparingInt(CatatanPercobaan::percobaanKe))
                .toList();
        return rakitHasil(baris.size(), terurut, jejakTerurut, totalTunggu.sum());
    }

    /** Merakit hasil per baris menjadi statistik, laporan, dan laporan galat. */
    private static HasilPipeline rakitHasil(int jumlahBaris, List<HasilBaris> hasil,
            List<CatatanPercobaan> jejak, long unitTunggu) {
        List<KejadianTerproses> kejadian = new ArrayList<>(hasil.size());
        int ditolak = 0;
        int gagal = 0;
        for (HasilBaris h : hasil) {
            switch (h) {
                case HasilBaris.Sah(var k) -> kejadian.add(k);
                case HasilBaris.TidakSah _ -> ditolak++;
                case HasilBaris.GagalTetap _ -> gagal++;
            }
        }
        RingkasanAgregasi ringkasan = agregasi(kejadian);
        return new HasilPipeline(jumlahBaris, kejadian.size(), ditolak, gagal, unitTunggu,
                ringkasan.nilaiBerjasaPerKota().size(), ringkasan.jumlahPerKategori().size(),
                tulis(ringkasan, hasil), tulisGalat(hasil), jejak);
    }

    /** Merunut penyebab paling dalam dari rangkaian exception berantai. */
    private static String akar(Throwable galat) {
        Throwable kini = galat;
        while (kini.getCause() != null) {
            kini = kini.getCause();
        }
        return kini.getClass().getSimpleName() + ": " + kini.getMessage();
    }
}