Spring Framework’te Transactional Outbox ile Güvenilir Olay Yayını

Spring Framework’te microservices mimarisi için çift-yazma hatası

Bir microservices mimarisi içinde siparişi PostgreSQL’e yazıp ardından Kafka’ya olay gönderen iki ayrı işlem, atomik değildir: veritabanı commit edildikten sonra süreç ölürse OrderCreated olayı hiç oluşmaz; önce Kafka’ya gönderip DB işlemi rollback olursa hayalet olay oluşur. XA/2PC yerine iş kaydı ile olay kaydını aynı yerel transaction içinde tutmak, PostgreSQL WAL’ın ikisini birlikte kalıcılaştırmasına dayanır. Bu, bir spring framework eğitimi veya java spring eğitimi sırasında yalnızca @Transactional eklemekten daha kritik bir sınırdır: transaction, broker çağrısını değil yalnızca DB değişikliklerini kapsamalıdır.

Outbox tablosunda olayın kimliği, aggregate sırası ve yayın durumu ayrı sütunlar olmalıdır. PostgreSQL migration’ını Flyway ile çalıştırın; aggregate_id, sequence_no benzersizliği aynı aggregate için tekrar eden veya sıra dışı domain olayı üretimini engeller. JSONB payload üzerinde indeks açmayın; publisher’ın sıcak sorgusu durum ve oluşturulma zamanıdır.

-- V12__create_outbox.sql
create table outbox_event (
  id uuid primary key,
  aggregate_type varchar(80) not null,
  aggregate_id uuid not null,
  sequence_no bigint not null,
  event_type varchar(160) not null,
  payload jsonb not null,
  occurred_at timestamptz not null,
  published_at timestamptz,
  attempts integer not null default 0,
  unique (aggregate_id, sequence_no)
);
create index outbox_unpublished_idx
  on outbox_event (occurred_at)
  where published_at is null;

İnce ayrıntı: UUID v4 ile yalnızca occurred_at üzerinden global sıralama varsaymayın; iki node’un saatleri kayabilir ve PostgreSQL transaction commit sırası uygulama saatinden farklı olabilir. İş kuralları sıra gerektiriyorsa sıra anahtarını aggregate_id yapın ve Kafka mesaj anahtarı olarak aynı değeri kullanın. Global total ordering ise hem maliyetli hem de çoğu domain için gereksiz bir beklentidir.

Spring Boot’ta iş verisi ve outbox kaydını tek transaction’da yazmak

Servis metodunda entity kaydı ile outbox kaydını aynı Spring transaction manager altında üretin. Aşağıdaki örnekte TransactionSynchronization ile "commit sonrası Kafka’ya yolla" yaklaşımı özellikle kullanılmıyor: commit sonrası callback JVM çöküşünde yeniden çalışmaz. Kalıcılık katmanı yalnızca outbox satırını yazar; ayrı publisher bu satırı daha sonra bulur.

@Service
@RequiredArgsConstructor
class OrderService {
  private final OrderRepository orders;
  private final OutboxRepository outbox;
  private final ObjectMapper objectMapper;

  @Transactional
  public UUID create(CreateOrder command) throws JsonProcessingException {
    Order order = orders.save(Order.open(command.customerId(), command.lines()));
    long sequence = order.nextEventSequence();

    outbox.save(new OutboxEvent(
        UUID.randomUUID(), "Order", order.getId(), sequence,
        "order.created.v1",
        objectMapper.readTree(objectMapper.writeValueAsBytes(
            new OrderCreated(order.getId(), order.getCustomerId()))),
        Instant.now()));
    return order.getId();
  }
}

@Transactional proxy üzerinden çağrılmadığında uygulanır; aynı sınıftaki this.create(...) self-invocation çağrısı transaction açmaz. Bunu görünür kılmak için testte TransactionSynchronizationManager.isActualTransactionActive() doğrulayın ve integration testini gerçek PostgreSQL ile Testcontainers üzerinde çalıştırın. H2’nin kilit ve JSONB davranışı production PostgreSQL ile eşdeğer değildir.

spring mvc veya bir spring rest api controller’ında HTTP isteğine event publish sorumluluğu vermeyin. Controller yalnızca komutu service’e geçirip 201 döndürmelidir; böylece istemcinin bağlantıyı kapatması veya 5xx retry yapması broker teslim mantığını değiştirmez. İstemci retry’ları için Idempotency-Key değerini ayrı bir request_deduplication tablosunda unique tutmak, aynı REST komutunun iki ayrı order ve iki ayrı outbox olayı üretmesini önler.

Spring Cloud Stream yerine kontrollü polling ve Kafka yayınlama

spring cloud ekosistemindeki Spring Cloud Stream Kafka binder’ı mesajlaşma soyutlaması sağlayabilir; ancak outbox satırını kilitleme, tekrar deneme ve yayın işaretleme kararları uygulamanın sorumluluğundadır. Çok node’lu publisher için PostgreSQL FOR UPDATE SKIP LOCKED kullanın. Her worker birbirinin kilitlediği satırları beklemeden başka batch alır; bunun karşılığında bir batch içindeki olaylar yalnızca seçildiği sırada sıralıdır.

Aşağıdaki sorguyu JdbcTemplate ile kısa bir DB transaction içinde çağırın, satırları IN_FLIGHT benzeri bir durumla sahiplenin ve transaction’ı kapatın. Kafka’ya ağ çağrısı yaparken PostgreSQL row lock tutmak, broker gecikmesinde connection pool’u kilitleyebilir. Şema buna göre lease_until ve status alanları içermelidir; worker ölürse süresi geçen lease yeniden alınır.

with claimed as (
  select id
  from outbox_event
  where published_at is null
    and (lease_until is null or lease_until < now())
  order by occurred_at
  for update skip locked
  limit :batchSize
)
update outbox_event e
set lease_until = now() + interval '30 seconds',
    attempts = attempts + 1
from claimed
where e.id = claimed.id
returning e.id, e.aggregate_id, e.event_type, e.payload;

Kafka producer’da enable.idempotence=true, acks=all ve max.in.flight.requests.per.connection<=5 ayarlayın; idempotent producer, aynı producer session içindeki retry’ın partition’a duplicate append yapmasını önler. Buna rağmen publisher, broker’a başarıyla yazıp published_at güncellemesinden önce çökerse olay yeniden gönderilir. Bu yüzden outbox teslim garantisi at-least-once’dur; "exactly once" iddiası consumer tarafındaki deduplication olmadan doğru değildir.

Spring Security, idempotent consumer ve şema evrimi

Consumer, her mesajı processed_message tablosuna kaydederek idempotent yapmalıdır. Domain değişikliğini ve mesaj kimliği insert’ini aynı DB transaction’da yürütün. Duplicate mesajda PostgreSQL unique violation yerine INSERT ... ON CONFLICT DO NOTHING RETURNING kullanmak exception maliyetini ve log gürültüsünü azaltır.

insert into processed_message (consumer_name, message_id, processed_at)
values (:consumer, :messageId, now())
on conflict (consumer_name, message_id) do nothing
returning message_id;

Bu sorgu satır döndürmüyorsa listener domain handler’ını çağırmadan ACK vermelidir. Dikkat edilmesi gereken edge case şudur: deduplication kaydını ayrı transaction’da commit edip domain işlemini sonra çalıştırırsanız, ikinci işlem çöktüğünde mesaj "işlendi" görünür fakat iş etkisi hiç oluşmaz. Aynı @Transactional sınırında ikisini de commit edin; başarısızlıkta listener exception fırlatsın ve tüketici yeniden denesin.

spring security, Kafka ACL’lerinin yerine geçmez. Uygulamanın yönetimsel yeniden-yayın endpoint’ini OAuth2 resource server ile koruyun; broker tarafında da producer principal’ına yalnızca ilgili topic için WRITE, consumer principal’ına yalnızca READ verin. Örneğin spring-security-oauth2-resource-server ile SecurityFilterChain içinde requestMatchers("/admin/outbox/**").hasAuthority("SCOPE_outbox:replay") kuralı koymak, normal kullanıcı token’ının replay başlatmasını engeller. Payload’a access token, müşteri PII’sı veya ham HTTP header koymayın; event logları ve DLQ’lar çoğu zaman API verisinden daha uzun saklanır.

Event tipini order.created.v1 gibi sürümleyin ve consumer’ı unknown field’ları tolere edecek şekilde Jackson’da yapılandırın. Bir alanı kaldırmak yerine önce nullable/deprecated bırakın; eski consumer grupları geriden okuyabilir. Dead-letter topic, şema uyumsuzluğunu görünmez kılacak bir çöp kutusu değildir: hata sınıfı, original topic/partition/offset ve message id header’larıyla alert üretin.

Outbox gecikmesini ölçmek ve batch boyutunu profil ile ayarlamak

Ölçmeden batchSize artırmak, daha yüksek throughput karşılığında daha uzun lock/lease süresi ve daha yüksek p99 gecikme üretebilir. Micrometer ile outbox_publish_latency_seconds histogramını now - occurred_at üzerinden, ayrıca outbox_backlog_total metriğini published_at is null sorgusundan kaydedin. Prometheus’ta son 10 dakikada p95’i karşılaştırmak için histogram_quantile(0.95, sum(rate(outbox_publish_latency_seconds_bucket[10m])) by (le)) sorgusunu kullanın.

Publisher darboğazını ayırmak için önce PostgreSQL’de pg_stat_statements ile claim sorgusunun ortalama süre, çağrı sayısı ve shared block read değerlerini alın; JVM tarafında Java Flight Recorder (JFR) ile jdk.SocketRead, jdk.JavaMonitorEnter ve allocation örneklerini kaydedin. Örnek deney: 10 dakika sabit 500 event/s yükte batch=100 için p95 gecikme, publish/s ve pg_stat_statements.mean_exec_time değerlerini kaydedin; yalnız batch=500 değiştirip aynı testi tekrarlayın. Sonuçta DB sorgusu ucuzken p95 yükseliyorsa sorun broker flush veya batch bekleme süresidir, indeks değil.

Spring Boot Actuator üzerinden JFR’ı production’da sürekli açık tutmak yerine kontrollü kayıt alın: jcmd <pid> JFR.start name=outbox settings=profile duration=120s filename=/tmp/outbox.jfr. Claim sorgusu Seq Scan gösteriyorsa kısmi indeksin predicate’i ile sorgunun predicate’inin birebir aynı olduğuna bakın; örneğin sorguya lease_until eklendiyse yalnız published_at is null indeksi her zaman yeterli seçiciliği sağlamaz. Bu ölçüm disiplini, bir spring boot eğitimi ya da spring boot kursu içindeki demo ayarlarını production yüküne körlemesine taşımaktan daha güvenilir bir yaklaşımdır.

Sık Sorulan Sorular

Spring Boot uygulamasında transactional outbox Kafka’ya exactly-once teslim sağlar mı?

Hayır. Producer idempotence broker retry duplicate’lerini azaltır; fakat Kafka’ya başarılı gönderim ile outbox satırını published işaretleme arasındaki çöküş yeniden gönderim üretir. Consumer, message id için unique kayıt ve domain değişikliğini aynı DB transaction’ında yaparak iş etkisini idempotent hale getirmelidir.

Spring Cloud Stream ile transactional outbox polling nasıl yapılır?

Binder mesajı publish etmek için kullanılabilir, fakat claim mekanizması yine sizin SQL’inizdir. PostgreSQL’de FOR UPDATE SKIP LOCKED ile kısa transaction’da satırları lease edin, lease dışındayken binder veya KafkaTemplate ile yayınlayın, başarılı sonuçta published_at güncelleyin. Ağ çağrısı boyunca row lock tutmayın.

Spring Security outbox replay endpoint’i nasıl korunmalı?

Endpoint’i OAuth2 resource server olarak yapılandırın ve yalnız SCOPE_outbox:replay yetkisine izin verin; örneğin requestMatchers("/admin/outbox/**").hasAuthority("SCOPE_outbox:replay"). Buna ek olarak Kafka ACL ile replay servisi principal’ını sadece gerekli topic WRITE izniyle sınırlayın ve replay işlemini event id, kullanıcı ve gerekçe ile audit tablosuna yazın.

Spring REST API retry edildiğinde iki outbox olayı oluşması nasıl engellenir?

İstemcinin Idempotency-Key değerini request_deduplication tablosunda unique saklayın. Aynı transaction’da önce anahtarı insert edin, mevcutsa önceki order id ve HTTP sonucunu döndürün; yeni ise order ile outbox kaydını oluşturun. Bu koruma yalnız Kafka consumer deduplication’ının yerine geçmez, HTTP komut tekrarını durdurur.