Kenapa penting

Hampir setiap sistem e-commerce punya alur "setelah X terjadi, kabari Y": pesanan dibayar → kirim email, stok menipis → kabari gudang, refund disetujui → kabari akuntansi. Alur ini terlihat sepele sampai kamu menemui kasus nyata: email terkirim untuk pembayaran yang ternyata gagal di-commit, event hilang saat broker restart, atau pelanggan menerima lima email yang sama karena consumer di-retry. Proyek mini ini merangkai semua bab M11 — RestClient, resilience, Kafka, idempotensi + outbox, dan scheduler — menjadi satu alur yang tahan gangguan. Polanya persis seperti yang dipakai tim backend di production, hanya ukurannya diperkecil.

Konsep inti

1. Kebutuhan

Kode	Kebutuhan
K-1	Saat pesanan berubah PENDING → PAID, event OrderPaid pasti tercatat — tidak boleh hilang.
K-2	Event tidak boleh tercatat bila perubahan status gagal di-commit.
K-3	Pelanggan menerima satu email per pembayaran, walau pesan Kafka terkirim lebih dari sekali.
K-4	Penyedia email yang lambat atau mati tidak boleh membuat consumer macet tanpa batas.
K-5	Pesan yang gagal permanen disimpan di DLT untuk diperiksa dan dikirim ulang.

2. Diagram alur

Untuk proyek mini, layanan pesanan dan layanan notifikasi ditulis sebagai dua package dalam satu aplikasi Spring Boot agar mudah dijalankan dan dites. Di production keduanya biasanya aplikasi terpisah dengan database masing-masing — kodenya tidak berubah, hanya cara deploy-nya.

[Klien] --POST /orders/{id}/pay--> [OrderPaymentService]
                                      |  satu transaksi DB:
                                      |   1) UPDATE orders SET status='PAID'
                                      |   2) INSERT INTO outbox_event (...)
                                      v
                               PostgreSQL: outbox_event (published_at IS NULL)
                                      |
                       @Scheduled OutboxRelay (tiap 1 detik, SKIP LOCKED)
                                      |  send(key=orderId) lalu set published_at
                                      v
                              Kafka topic: order.paid (3 partisi)
                                      |
                  @KafkaListener OrderPaidListener (group notification-service)
                                      |  INSERT processed_event ON CONFLICT DO NOTHING
                                      |  baris baru? -> EmailClient
                                      v
                 RestClient (timeout) -> Retry -> CircuitBreaker -> Penyedia Email
                                      |
                       gagal setelah retry / error permanen
                                      v
                           Kafka topic: order.paid.DLT

3. Keputusan desain dan alasannya

• Outbox, bukan kirim langsung. Commit database dan send ke Kafka adalah dua sistem berbeda; tidak ada transaksi yang membungkus keduanya. Outbox membuat event ikut transaksi database (K-1, K-2), lalu relay memindahkannya ke Kafka belakangan.
• Key = ID pesanan. Kafka menjamin urutan hanya di dalam satu partisi. Dengan key orderId, semua event untuk pesanan yang sama masuk partisi yang sama dan dibaca berurutan.
• At-least-once + consumer idempoten. Relay bisa mengirim ulang (misalnya crash di antara send dan update published_at). Daripada mengejar exactly-once yang mahal, consumer mencatat event_id yang sudah diproses (K-3).
• Retry berlapis yang sengaja dibatasi. Resilience4j melakukan retry cepat untuk gangguan singkat; DefaultErrorHandler Spring Kafka melakukan retry dengan jeda lebih panjang; setelah itu pesan ke DLT (K-4, K-5). Jumlah total percobaan = perkalian keduanya, jadi angka kecil saja.
• Idempotency key ke penyedia email. Tabel processed_event tidak bisa menutup celah "email sudah terkirim, tapi commit gagal". Header Idempotency-Key berisi event_id membuat penyedia (yang mendukungnya) menolak duplikat di sisinya.

4. Simulasi at-least-once tanpa framework

Sebelum masuk Spring, lihat inti idempotensi dalam Java murni. Set.add mengembalikan false bila elemen sudah ada — perannya sama dengan INSERT ... ON CONFLICT DO NOTHING yang mengembalikan 0 baris.

import java.util.*;

public class AtLeastOnceDemo {
    record OrderPaid(String eventId, long orderId) {}

    static final Set<String> processed = new HashSet<>();
    static int emailsSent = 0;

    static void consume(OrderPaid event) {
        if (!processed.add(event.eventId())) {
            System.out.println("skip duplikat " + event.eventId());
            return;
        }
        emailsSent++;
        System.out.println("email untuk pesanan #" + event.orderId());
    }

    public static void main(String[] args) {
        var e1 = new OrderPaid("evt-1", 1001);
        var e2 = new OrderPaid("evt-2", 1002);
        // Relay crash setelah kirim e1 tapi sebelum menandai published -> e1 terkirim ulang.
        List<OrderPaid> delivered = List.of(e1, e1, e2);
        delivered.forEach(AtLeastOnceDemo::consume);
        System.out.println("pesan diterima=" + delivered.size() + ", email terkirim=" + emailsSent);
    }
}

Output:

email untuk pesanan #1001
skip duplikat evt-1
email untuk pesanan #1002
pesan diterima=3, email terkirim=2

Di aplikasi nyata, Set di memori diganti tabel database karena consumer bisa restart dan berjalan di beberapa instance.

5. Langkah implementasi bertahap

1. Jalankan infrastruktur lokal dengan compose.yaml.
2. Buat skema lewat migrasi Flyway: orders, outbox_event, processed_event.
3. Tulis OrderPaymentService.markPaid — update status + insert outbox dalam satu transaksi.
4. Tulis OutboxRelay dengan @Scheduled dan aktifkan @EnableScheduling.
5. Tulis EmailClient (timeout, retry, circuit breaker) dan OrderPaidListener yang idempoten.
6. Konfigurasikan DefaultErrorHandler + DLT.
7. Tulis test integrasi dengan Testcontainers, lalu uji manual skenario gangguan (matikan penyedia email, restart Kafka).

Latihan kode menyusul

Starter project dan test untuk proyek mini ini akan tersedia di folder exercises/. Untuk sekarang, ikuti langkah di atas di project Spring Boot milikmu sendiri.

Contoh di dunia backend

Infrastruktur lokal. Image resmi apache/kafka berjalan dalam mode KRaft (tanpa ZooKeeper) dengan konfigurasi bawaan satu node yang listener-nya localhost:9092.

# compose.yaml
services:
  postgres:
    image: postgres:17-alpine
    environment:
      POSTGRES_DB: shop
      POSTGRES_USER: shop
      POSTGRES_PASSWORD: shop-local-only
    ports:
      - "5432:5432"
  kafka:
    image: apache/kafka:4.1.0
    ports:
      - "9092:9092"

# src/main/resources/application.yaml — Potongan
spring:
  datasource:
    url: jdbc:postgresql://localhost:5432/shop
    username: shop
    password: shop-local-only
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      acks: all
    consumer:
      auto-offset-reset: earliest
email:
  base-url: http://localhost:8089
outbox:
  relay:
    delay: 1000

Skema. Index parsial membuat query relay tetap cepat walau tabel outbox berisi jutaan baris yang sudah terkirim.

-- V1__order_notification.sql
create table orders (
    id             bigserial primary key,
    customer_email text   not null,
    total_amount   bigint not null,
    status         text   not null
);

create table outbox_event (
    id             uuid primary key,
    aggregate_type text        not null,
    aggregate_id   text        not null,
    event_type     text        not null,
    payload        jsonb       not null,
    created_at     timestamptz not null default now(),
    published_at   timestamptz
);
create index idx_outbox_unpublished on outbox_event (created_at) where published_at is null;

-- Milik layanan notifikasi
create table processed_event (
    event_id     uuid primary key,
    processed_at timestamptz not null default now()
);

Layanan pesanan: tulis status + outbox dalam satu transaksi. Spring Boot 4 memakai Jackson 3 (package tools.jackson), dan JsonMapper sudah tersedia sebagai bean.

// Potongan — package id.jalurjava.shop.order
public record OrderPaidEvent(UUID eventId, long orderId, String customerEmail,
                             long totalAmount, Instant paidAt) {}

record OrderRow(long id, String customerEmail, long totalAmount) {}

@Service
public class OrderPaymentService {
    private final JdbcClient jdbc;
    private final JsonMapper json;

    public OrderPaymentService(JdbcClient jdbc, JsonMapper json) {
        this.jdbc = jdbc;
        this.json = json;
    }

    @Transactional
    public void markPaid(long orderId) {
        int updated = jdbc.sql("update orders set status = 'PAID' where id = :id and status = 'PENDING'")
                .param("id", orderId)
                .update();
        if (updated == 0) {
            throw new IllegalStateException("Pesanan " + orderId + " tidak dalam status PENDING");
        }
        OrderRow order = jdbc.sql("select id, customer_email, total_amount from orders where id = :id")
                .param("id", orderId)
                .query(OrderRow.class)
                .single();
        var event = new OrderPaidEvent(UUID.randomUUID(), order.id(), order.customerEmail(),
                order.totalAmount(), Instant.now());
        jdbc.sql("""
                insert into outbox_event (id, aggregate_type, aggregate_id, event_type, payload)
                values (:id, 'order', :aggregateId, 'OrderPaid', cast(:payload as jsonb))
                """)
                .param("id", event.eventId())
                .param("aggregateId", String.valueOf(order.id()))
                .param("payload", json.writeValueAsString(event))
                .update();
    }
}

Relay terjadwal. Jika satu pengiriman gagal, exception membatalkan transaksi sehingga published_at baris sebelumnya ikut di-rollback dan dikirim ulang pada putaran berikutnya — duplikat yang memang sudah kita antisipasi.

// Potongan — aktifkan @EnableScheduling di class konfigurasi
record OutboxRow(UUID id, String aggregateId, String payload) {}

@Component
public class OutboxRelay {
    private final JdbcClient jdbc;
    private final KafkaTemplate<String, String> kafka;

    public OutboxRelay(JdbcClient jdbc, KafkaTemplate<String, String> kafka) {
        this.jdbc = jdbc;
        this.kafka = kafka;
    }

    @Scheduled(fixedDelayString = "${outbox.relay.delay:1000}")
    @Transactional
    public void publishPending() {
        List<OutboxRow> batch = jdbc.sql("""
                select id, aggregate_id, payload::text as payload
                from outbox_event
                where published_at is null
                order by created_at
                limit 100
                for update skip locked
                """)
                .query(OutboxRow.class)
                .list();
        for (OutboxRow row : batch) {
            try {
                kafka.send("order.paid", row.aggregateId(), row.payload()).get(5, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new IllegalStateException("Relay diinterupsi", e);
            } catch (ExecutionException | TimeoutException e) {
                throw new IllegalStateException("Gagal mengirim outbox " + row.id(), e);
            }
            jdbc.sql("update outbox_event set published_at = now() where id = :id")
                    .param("id", row.id())
                    .update();
        }
    }
}

Klien email dengan timeout, retry, dan circuit breaker. Timeout dipasang di request factory; Resilience4j dipakai lewat API inti agar tidak bergantung pada starter tertentu.

// Potongan — package id.jalurjava.shop.notification
record EmailRequest(String to, String subject, String body) {}

@Component
public class EmailClient {
    private final RestClient restClient;
    private final CircuitBreaker circuitBreaker;
    private final Retry retry;

    public EmailClient(RestClient.Builder builder, @Value("${email.base-url}") String baseUrl) {
        HttpClient http = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(2)).build();
        var factory = new JdkClientHttpRequestFactory(http);
        factory.setReadTimeout(Duration.ofSeconds(3));
        this.restClient = builder.baseUrl(baseUrl).requestFactory(factory).build();

        this.circuitBreaker = CircuitBreaker.of("email", CircuitBreakerConfig.custom()
                .slidingWindowSize(20)
                .minimumNumberOfCalls(10)
                .failureRateThreshold(50)
                .waitDurationInOpenState(Duration.ofSeconds(30))
                .build());
        this.retry = Retry.of("email", RetryConfig.custom()
                .maxAttempts(3)
                .waitDuration(Duration.ofMillis(300))
                .retryExceptions(ResourceAccessException.class, HttpServerErrorException.class)
                .build());
    }

    public void sendOrderPaid(OrderPaidEvent event) {
        Runnable call = () -> restClient.post()
                .uri("/v1/messages")
                .header("Idempotency-Key", event.eventId().toString())
                .body(new EmailRequest(event.customerEmail(), "Pembayaran diterima",
                        "Pesanan #" + event.orderId() + " sudah dibayar. Terima kasih!"))
                .retrieve()
                .toBodilessEntity();
        Retry.decorateRunnable(retry, CircuitBreaker.decorateRunnable(circuitBreaker, call)).run();
    }
}

Consumer idempoten. Insert ke processed_event dan pemanggilan email ada dalam satu transaksi: jika email gagal, insert ikut di-rollback sehingga retry berikutnya mencoba lagi.

// Potongan
@Component
public class OrderPaidListener {
    private static final Logger log = LoggerFactory.getLogger(OrderPaidListener.class);
    private final JdbcClient jdbc;
    private final JsonMapper json;
    private final EmailClient emailClient;

    public OrderPaidListener(JdbcClient jdbc, JsonMapper json, EmailClient emailClient) {
        this.jdbc = jdbc;
        this.json = json;
        this.emailClient = emailClient;
    }

    @KafkaListener(topics = "order.paid", groupId = "notification-service")
    @Transactional
    public void onOrderPaid(String payload) {
        OrderPaidEvent event = json.readValue(payload, OrderPaidEvent.class);
        int inserted = jdbc.sql("insert into processed_event (event_id) values (:id) on conflict do nothing")
                .param("id", event.eventId())
                .update();
        if (inserted == 0) {
            log.info("Event {} sudah diproses, dilewati", event.eventId());
            return;
        }
        emailClient.sendOrderPaid(event);
    }
}

Topic, error handler, dan DLT. Spring Boot otomatis memasang bean CommonErrorHandler ke container factory listener.

// Potongan
@Configuration
@EnableScheduling
public class MessagingConfig {

    @Bean
    NewTopic orderPaidTopic() {
        return TopicBuilder.name("order.paid").partitions(3).replicas(1).build();
    }

    @Bean
    NewTopic orderPaidDlt() {
        return TopicBuilder.name("order.paid.DLT").partitions(3).replicas(1).build();
    }

    @Bean
    DefaultErrorHandler kafkaErrorHandler(KafkaTemplate<String, String> template) {
        var recoverer = new DeadLetterPublishingRecoverer(template,
                (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition()));
        var handler = new DefaultErrorHandler(recoverer, new FixedBackOff(2_000L, 4));
        handler.addNotRetryableExceptions(JacksonException.class, HttpClientErrorException.class);
        return handler;
    }
}

Test integrasi. @ServiceConnection mengisi URL datasource dan bootstrap server dari container; EmailClient diganti mock agar test tidak memanggil layanan luar.

// Potongan — Testcontainers 2.0, JUnit Jupiter
@SpringBootTest
@Testcontainers
class OrderNotificationFlowTest {

    @Container
    @ServiceConnection
    static PostgreSQLContainer postgres = new PostgreSQLContainer("postgres:17-alpine");

    @Container
    @ServiceConnection
    static KafkaContainer kafka = new KafkaContainer("apache/kafka:4.1.0");

    @MockitoBean
    EmailClient emailClient;

    @Autowired
    OrderPaymentService payments;

    @Autowired
    JdbcClient jdbc;

    @Autowired
    KafkaTemplate<String, String> kafkaTemplate;

    @Test
    void pesananDibayarMengirimSatuEmail() {
        long orderId = insertPendingOrder();

        payments.markPaid(orderId);

        verify(emailClient, timeout(15_000)).sendOrderPaid(argThat(e -> e.orderId() == orderId));
    }

    @Test
    void eventDuplikatHanyaMengirimSatuEmail() {
        String payload = """
                {"eventId":"8f8a2a52-1c1e-4f2e-9d55-0c6a3c1b7e01","orderId":77,
                 "customerEmail":"budi@example.com","totalAmount":99000,
                 "paidAt":"2026-10-01T10:00:00Z"}
                """;
        kafkaTemplate.send("order.paid", "77", payload);
        kafkaTemplate.send("order.paid", "77", payload);

        verify(emailClient, after(5_000).times(1)).sendOrderPaid(argThat(e -> e.orderId() == 77));
    }

    private long insertPendingOrder() {
        return jdbc.sql("""
                insert into orders (customer_email, total_amount, status)
                values ('ani@example.com', 150000, 'PENDING') returning id
                """)
                .query(Long.class)
                .single();
    }
}

Checklist sebelum menyebut proyek selesai:

• markPaid pada pesanan yang bukan PENDING melempar exception dan tidak menulis outbox.
• Relay mengirim dengan key orderId dan hanya menandai published_at setelah broker mengonfirmasi.
• Dua instance relay tidak mengirim baris yang sama secara bersamaan (SKIP LOCKED).
• Event duplikat menghasilkan satu email (test lulus).
• Penyedia email dimatikan: consumer tidak macet lebih dari total backoff, pesan masuk order.paid.DLT.
• Error 4xx langsung ke DLT tanpa retry; error 5xx/timeout di-retry.
• Ada cara mengirim ulang pesan dari DLT, dan baris outbox lama dibersihkan berkala.

Pertanyaan interview

1. Kenapa tidak cukup memanggil kafkaTemplate.send setelah repository.save di method yang sama?

Karena itu dual write: database dan Kafka tidak berbagi transaksi. Urutan apa pun punya celah — kirim dulu lalu commit gagal menghasilkan event "hantu", commit dulu lalu kirim gagal menghilangkan event. Transactional outbox menyimpan event di database dalam transaksi yang sama dengan perubahan bisnis, lalu proses terpisah memindahkannya ke broker dengan jaminan at-least-once.

2. Bagaimana kamu memastikan email tidak terkirim dua kali?

Pertama, consumer idempoten: event_id unik dicatat di tabel processed_event dengan INSERT ... ON CONFLICT DO NOTHING; jika tidak ada baris yang masuk, event dilewati. Kedua, karena pemanggilan HTTP tidak bisa ikut di-rollback, event_id juga dikirim sebagai Idempotency-Key ke penyedia email. Tanpa lapisan kedua masih ada celah kecil saat email sukses tetapi commit database gagal.

3. Apa risiko retry berlapis (klien HTTP + Kafka error handler)?

Jumlah percobaan berlipat: 3 percobaan Resilience4j dikali 5 percobaan error handler berarti sampai 15 panggilan per pesan, dan selama retry blocking partisi itu tertahan. Karena itu angka retry dibuat kecil, error permanen (4xx, payload rusak) ditandai not retryable, circuit breaker menghentikan beban ke penyedia yang mati, dan sisanya diserahkan ke DLT untuk replay.

Jebakan umum

• Mengirim ke Kafka di dalam transaksi bisnis dan menganggapnya atomik — itu dual write, bukan outbox.
• Menandai published_at sebelum broker mengonfirmasi (send tanpa menunggu hasil) — event bisa hilang diam-diam.
• Consumer tanpa idempotensi; duplikat pasti terjadi di at-least-once, bukan "jarang".
• Me-retry semua exception, termasuk payload rusak dan 4xx — pesan beracun memblokir partisi lalu tetap berakhir di DLT.
• Membuat topic DLT dengan partisi lebih sedikit daripada topic utama padahal resolver memakai record.partition().
• Lupa membersihkan baris outbox yang sudah terkirim sehingga tabel tumbuh tanpa batas.

Ringkasan

• Outbox menyelesaikan dual write: event ditulis ke tabel dalam transaksi yang sama dengan perubahan status pesanan.
• Relay @Scheduled dengan FOR UPDATE SKIP LOCKED memindahkan outbox ke Kafka secara at-least-once dan aman dijalankan beberapa instance.
• Consumer idempoten (processed_event + Idempotency-Key) mengubah at-least-once menjadi efek "sekali" dari sudut pandang pelanggan.
• RestClient dengan timeout, retry untuk error sementara, dan circuit breaker melindungi consumer dari penyedia yang lambat atau mati.
• DefaultErrorHandler + DeadLetterPublishingRecoverer menampung pesan gagal di DLT; Testcontainers membuktikan seluruh alur dengan Kafka dan PostgreSQL sungguhan.

Bacaan lanjutan

• Spring for Apache Kafka — Handling Exceptions & Dead Letter
• Spring Framework — REST Clients
• Spring Boot — Testcontainers