import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.List;

/** Broker mini: partisi per kunci (urutan per kunci terjaga) + dead letter queue. */
public class AntreanPesan {

    /** Satu pesan: kunci penentu partisi + isi. */
    record Pesan(String kunci, String isi) {}

    /** Broker dengan jumlah partisi tetap dan dead letter queue (DLQ). */
    static class Broker {
        private final List<Deque<Pesan>> partisi;
        private final Deque<Pesan> deadLetter = new ArrayDeque<>();

        Broker(int jumlahPartisi) {
            partisi = new ArrayList<>();
            for (int i = 0; i < jumlahPartisi; i++) {
                partisi.add(new ArrayDeque<>());
            }
        }

        /** Kunci yang sama selalu jatuh ke partisi yang sama. */
        int kirim(Pesan p) {
            int idx = Math.floorMod(p.kunci().hashCode(), partisi.size());
            partisi.get(idx).offer(p); // FIFO: offer di ekor
            return idx;
        }

        /** Konsumsi satu partisi; pesan bermasalah di-retry lalu dibuang ke DLQ. */
        void konsumsi(int idx) {
            Deque<Pesan> antrean = partisi.get(idx);
            while (!antrean.isEmpty()) {
                Pesan p = antrean.poll(); // FIFO: poll dari kepala
                int percobaan = 0;
                boolean sukses = false;
                while (percobaan < 3 && !sukses) {
                    percobaan++;
                    sukses = !p.isi().contains("RACUN");
                }
                if (sukses) {
                    System.out.println("  partisi " + idx + ": [" + p.kunci() + "] "
                            + p.isi() + " -> diproses (percobaan ke-" + percobaan + ")");
                } else {
                    deadLetter.add(p);
                    System.out.println("  partisi " + idx + ": [" + p.kunci() + "] "
                            + p.isi() + " -> gagal " + percobaan
                            + " percobaan, dipindah ke DLQ");
                }
            }
        }

        Deque<Pesan> deadLetter() {
            return deadLetter;
        }
    }

    public static void main(String[] args) {
        Broker broker = new Broker(2); // dua partisi, seperti topik Kafka

        System.out.println("== Penempatan partisi: kunci sama -> partisi sama ==");
        List<Pesan> pesanan = List.of(
                new Pesan("pelanggan-7", "perbarui alamat"),
                new Pesan("pelanggan-7", "aktifkan langganan"),
                new Pesan("pelanggan-7", "kirim faktur"),
                new Pesan("pelanggan-2", "perbarui alamat"),
                new Pesan("pelanggan-7", "RACUN: isi tidak bisa diproses"),
                new Pesan("pelanggan-2", "naikkan paket"));

        for (Pesan p : pesanan) {
            int idx = broker.kirim(p);
            System.out.println("  [" + p.kunci() + "] " + p.isi() + " -> partisi " + idx);
        }

        System.out.println();
        System.out.println("== Konsumsi per partisi (urutan FIFO terjaga) ==");
        broker.konsumsi(0);
        broker.konsumsi(1);

        System.out.println();
        System.out.println("== Dead letter queue: pesan racun terisolasi ==");
        for (Pesan p : broker.deadLetter()) {
            System.out.println("  [" + p.kunci() + "] " + p.isi());
        }
    }
}