Spring Boot'ta Transactional Outbox ile Güvenilir Olay Yayınlama

Java microservices için Transactional Outbox sınırı

Bir sipariş kaydı ile Kafka'ya OrderCreated olayı göndermek iki farklı kaynak yöneticisine yazmaktır: PostgreSQL commit olurken broker'a gönderim zaman aşımına uğrayabilir; tersinde ise olay yayınlanıp transaction rollback olabilir. Java microservices içinde bunu uygulama seviyesinde iki aşamalı commit ile çözmeye çalışmak, broker ve veritabanının XA davranışına, recovery log'larına ve operasyonel uyumluluğa bağlanır. Transactional Outbox yaklaşımında iş verisi ve yayın niyetini tek yerel transaction içinde aynı PostgreSQL veritabanına yazın. Flyway ile şemayı açıkça sürümleyin:

CREATE TABLE outbox_event (
  id UUID PRIMARY KEY,
  aggregate_type VARCHAR(80) NOT NULL,
  aggregate_id UUID NOT NULL,
  event_type VARCHAR(120) NOT NULL,
  payload JSONB NOT NULL,
  occurred_at TIMESTAMPTZ NOT NULL,
  published_at TIMESTAMPTZ,
  attempts INTEGER NOT NULL DEFAULT 0,
  lease_until TIMESTAMPTZ,
  last_error TEXT
);

CREATE INDEX outbox_unpublished_idx
  ON outbox_event (occurred_at)
  WHERE published_at IS NULL;

Partial index'in published_at IS NULL koşulu önemlidir: milyonlarca tamamlanmış satırın bulunduğu tabloda poller sadece yayınlanmamış çalışma kümesine iner. EXPLAIN (ANALYZE, BUFFERS) çıktısında sıralı tarama görüyorsanız bu indeksin predicate'i ile sorgunuzun predicate'i birebir aynı değildir.

Bu tasarımda atomiklik yalnızca veritabanı commit'i için garanti edilir; Kafka'ya tam olarak bir kez teslim garantisi vermez. Bu ayrımı java eğitimi ve java programlama eğitimi içeriklerinde özellikle belirtmek gerekir: pratik hedef, en az bir kez yayınlama ve tüketicide idempotency'dir. Her olayın değişmez bir id değeri, olay şemasının event_type alanı ve aggregate başına monoton bir sıra numarası olmalıdır. Sadece occurred_at ile sıralama yapmak güvenilir değildir; aynı timestamp çözünürlüğünde iki commit farklı bağlantılardan gelebilir.

Spring Framework transaction senkronizasyonu yerine aynı transaction'da kayıt

Spring Framework içindeki @TransactionalEventListener(phase = AFTER_COMMIT), işlem başarıyla bittiğinde uygulama içi callback çalıştırmak için yararlıdır; fakat süreç callback'ten önce ölürse olay kaybolur. Outbox satırını callback içinde değil, domain değişikliğiyle aynı @Transactional metodunda oluşturun. Hibernate ORM persistence context'i transaction sonunda flush ettiği için iki save çağrısı aynı fiziksel transaction'a katılır:

@Service
@RequiredArgsConstructor
class OrderService {
  private final OrderRepository orders;
  private final OutboxEventRepository outbox;

  @Transactional
  public UUID create(CreateOrderCommand cmd) {
    Order order = orders.save(Order.create(cmd.customerId(), cmd.lines()));

    OutboxEvent event = OutboxEvent.pending(
        UUID.randomUUID(),
        "Order",
        order.getId(),
        "orders.v1.created",
        new OrderCreated(order.getId(), order.getCustomerId())
    );
    outbox.save(event);
    return order.getId();
  }
}

Bu örnekte OrderCreated nesnesini JSON'a çevirmek için Jackson ObjectMapper kullanılabilir, ancak serializer hatasının commit sırasında değil iş mantığı çalışırken görünmesi için payload'u entity oluşturulmadan önce serialize edin.

Spring Data JPA repository metodu ile outbox eklemek kolaydır, fakat aggregate kimliği GenerationType.IDENTITY ile üretiliyorsa Hibernate ORM kimliği alabilmek için erken INSERT çalıştırabilir. Bu normaldir, ancak SQL sırasının iş kuralınız olduğunu varsaymayın. Ayrıca save() dönüşünden sonra Kafka'ya gönderim yapmayın: transaction henüz commit edilmemiştir ve consumer olay geldiğinde veriyi okuyamayabilir. Bir spring boot eğitimi laboratuvarında bunu kanıtlamak için Testcontainers PostgreSQL ile transaction rollback testi yazın ve hem orders hem outbox_event satır sayısının sıfır kaldığını doğrulayın:

@Test
void rollbackRemovesOrderAndOutboxRow() {
  assertThatThrownBy(() -> service.createThenFail(command))
      .isInstanceOf(IllegalStateException.class);

  assertThat(jdbc.queryForObject("select count(*) from orders", Long.class)).isZero();
  assertThat(jdbc.queryForObject("select count(*) from outbox_event", Long.class)).isZero();
}

Spring Data JPA poller ile lease alma ve Kafka'ya yayınlama

Birden fazla uygulama pod'u aynı outbox tablosunu tarıyorsa SELECT ... FOR UPDATE SKIP LOCKED kullanın. Ancak Kafka'ya gönderimi veritabanı kilidi tutulurken yapmayın: broker gecikmesi transaction'ı uzatır, VACUUM'u geciktirir ve diğer poller'ları bekletir. Bunun yerine kısa bir transaction içinde satırlara süreli lease verin; ardından lease sahibi olayları yayınlasın. PostgreSQL native sorgusu Spring Data JPA üzerinden şu şekilde tanımlanabilir:

@Query(value = """
  WITH claimed AS (
    SELECT id
    FROM outbox_event
    WHERE published_at IS NULL
      AND (lease_until IS NULL OR lease_until < clock_timestamp())
    ORDER BY occurred_at
    FOR UPDATE SKIP LOCKED
    LIMIT :batchSize
  )
  UPDATE outbox_event e
  SET lease_until = clock_timestamp() + (:leaseSeconds * interval '1 second'),
      attempts = attempts + 1
  FROM claimed
  WHERE e.id = claimed.id
  RETURNING e.*
  """, nativeQuery = true)
List<OutboxEvent> claimBatch(int batchSize, int leaseSeconds);

clock_timestamp() burada uygulama JVM saatinden daha güvenilirdir; pod saatinin geri gitmesi lease'in beklenenden uzun görünmesine neden olmaz. Lease süresi, Kafka gönderiminin p99 süresinden ve retry budget'tan büyük, fakat bir pod çökünce kabul edilecek yeniden teslim gecikmesinden küçük seçilmelidir.

Yayınlama sonrasında published_at güncellemesini olay kimliği ve lease ile koşullandırın. Bu koşul, lease süresi dolmuş bir olayın başka pod tarafından yeniden alınması halinde eski worker'ın yanlışlıkla yeni sahibin işini tamamlamasını engeller. Kafka producer'da enable.idempotence=true, acks=all ve key=aggregateId kullanın; key, aynı partition içinde aggregate sırasını korur. Buna rağmen producer acknowledgement kaybolursa aynı olay yeniden gönderilebilir. Tüketici tarafında PostgreSQL inbox tablosuna unique constraint koymak zorunludur:

CREATE TABLE consumed_event (
  event_id UUID PRIMARY KEY,
  consumed_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

INSERT INTO consumed_event(event_id) VALUES (:eventId)
ON CONFLICT DO NOTHING;

INSERT sonucu 0 satırsa iş etkisini çalıştırmayın. Redis SETNX ile deduplication, TTL bittikten sonra eski olay yeniden gelirse etkisiz kalır; kalıcı iş etkileri için veritabanı unique constraint'i daha doğru sınırdır.

Java backend geliştirme için throughput ve gecikme ölçümü

Poller kapasitesini tahminle ayarlamak yerine iki dağılımı ölçün: outbox_oldest_unpublished_seconds ve publish latency histogramı. Micrometer ile ilk metriği PostgreSQL'den her 10 saniyede bir okuyup Prometheus'a verin. Alarm eşiğini örneğin sabit bir sayı olarak değil, ürünün olay görünürlük SLO'suna göre belirleyin: 60 saniyelik hedefiniz varsa p99 yaşının 45 saniyeyi geçmesi erken uyarıdır. Ayrıca attempts değerini event type etiketinde değil, düşük kardinaliteli sonuç etiketiyle ölçün; event id'yi metric etiketi yapmak Prometheus time series sayısını patlatır.

Batch boyutunu değiştirmeden önce ve sonra aynı yükte karşılaştırma yapın. k6 ile saniyede sabit sipariş oluşturun, Prometheus'tan outbox yaşının p50/p95/p99 değerlerini alın, PostgreSQL'de pg_stat_statements ile claim sorgusunun mean execution time ve shared block reads değerlerini kaydedin. JVM tarafında Java Flight Recorder kaydı alın:

java -XX:StartFlightRecording=filename=outbox-before.jfr,dumponexit=true   -jar app.jar

jfr summary outbox-before.jfr
jfr print --events jdk.ExecutionSample outbox-before.jfr | head -50

Örneğin batch'i 100'den 500'e çıkarıp sadece ortalama throughput'a bakmak yanıltır: daha büyük batch, tek worker'ın lease ettiği iş miktarını artırarak pod ölümü sonrası yeniden görünme süresini yükseltebilir. Karşılaştırmada broker hata oranı, p99 outbox yaşı, PostgreSQL lock wait ve CPU örneklerini birlikte değerlendirin.

Outbox temizliği için doğrudan DELETE FROM outbox_event WHERE published_at < now() - interval '7 days' çalıştırmak büyük tabloda uzun transaction ve WAL sıçraması üretir. PostgreSQL'de küçük parçalar halinde silin ve her turdan sonra commit edin. Örneğin bakım job'ı her dakika 5.000 satır silebilir; pg_stat_user_tables.n_dead_tup ve autovacuum gecikmesini izleyerek sayıyı ayarlayın. Çok yüksek hacimde aylık partition ve eski partition için DROP TABLE tercih edin; bu, satır satır delete yerine metadata işlemi yapar.

Spring AI, Spring MCP ve Model Context Protocol ile operasyonel erişim

Spring AI ve spring mcp, outbox yayıncısının güvenilirlik mekanizmasının yerine geçmez; fakat model context protocol kullanan bir operasyon asistanına kontrollü teşhis araçları sunabilir. Aracın doğrudan SQL çalıştırmasına izin vermek yerine, salt-okunur ve sınırları belli bir endpoint tanımlayın: son 15 dakikadaki yayınlanmamış olay sayısı, en eski olay yaşı ve event type bazında hata sayısı. Araç çağrısına tenant, event payload veya kişisel veri döndürmeyin; model bağlamına giren veri log saklama ve erişim sınırlarını da etkiler.

Örnek olarak Spring Boot Actuator üzerinden özel bir health indicator yayınlayın ve Spring MCP tool'u yalnızca bu özetle besleyin. Bu yaklaşım, bir java fullstack eğitimi projesindeki yönetim panelinin de aynı metrikleri güvenle göstermesini sağlar:

@Component
@RequiredArgsConstructor
class OutboxHealthIndicator implements HealthIndicator {
  private final JdbcTemplate jdbc;

  @Override
  public Health health() {
    Long oldestSeconds = jdbc.queryForObject("""
      select coalesce(extract(epoch from now() - min(occurred_at)), 0)::bigint
      from outbox_event where published_at is null
      """, Long.class);

    return oldestSeconds < 60
        ? Health.up().withDetail("oldestUnpublishedSeconds", oldestSeconds).build()
        : Health.down().withDetail("oldestUnpublishedSeconds", oldestSeconds).build();
  }
}

Model context protocol aracını çağıran kimliğin Actuator endpoint'ine erişimi mTLS veya OAuth2 resource server ile sınırlandırılmalıdır. Bir java kursu içinde bu ayrımın öğretilmesi değerlidir: LLM'nin yorum üretmesi ile üretim verisine yetkili erişim vermek aynı tasarım kararı değildir.

Sık Sorulan Sorular

Spring Boot eğitimi kapsamında @TransactionalEventListener yerine Transactional Outbox ne zaman kullanılmalı?

Olay başka bir süreçte kalıcı yan etki oluşturuyorsa, örneğin Kafka consumer stok ayıracak veya e-posta sağlayıcısına iş gönderecekse Outbox kullanın. AFTER_COMMIT callback'i süreç çöküşü ile callback çalışması arasındaki boşluğu kapatmaz. Aynı transaction içinde outbox satırı yazın, ayrı worker ile yayınlayın ve consumer'da event_id için unique constraint kurun.

Hibernate ORM ve Spring Data JPA ile Outbox satırı neden aynı transaction'da yazılmalı?

Aynı @Transactional sınırında Order ve OutboxEvent insert'leri tek PostgreSQL commit kaydına girer: rollback olursa ikisi de görünmez, commit olursa ikisi de kalıcıdır. Kafka send işlemini bu sınıra koymak atomiklik sağlamaz; broker acknowledgement kaybında duplicate olasılığı devam eder. Bu nedenle producer idempotence ve consumer inbox deduplication birlikte gerekir.

Java microservices ortamında birden fazla outbox poller aynı olayı nasıl paylaşmadan işler?

PostgreSQL'de FOR UPDATE SKIP LOCKED ile aday satırları seçip kısa transaction içinde lease_until güncelleyin. Worker Kafka'ya gönderdikten sonra UPDATE ... WHERE id = :id AND lease_until = :lease ile published_at yazmalıdır. Lease süresi dolarsa başka worker olayı yeniden alabilir; bu tasarımın sonucu beklenen en az bir kez teslimdir.

Spring AI ve Spring MCP Transactional Outbox sistemine doğrudan Kafka producer olarak eklenmeli mi?

Hayır. Spring AI veya Spring MCP, olay yayınlama yolunun içine konursa model yanıt süresi ve dış servis hataları iş transaction'ına bağımlılık ekler. Model context protocol üzerinden sadece outbox yaşı, hata sayısı ve backlog gibi düşük kardinaliteli, maskelenmiş operasyon özetlerini read-only tool olarak sunun.