Spring Boot'ta Transactional Outbox ile Güvenilir Olay Teslimi

Spring Boot eğitimi: Outbox kaydını domain verisiyle atomik yazmak

Bir sipariş kaydı ile mesaj broker'a yayın çağrısını aynı yerel veritabanı transaction içinde tutmaya çalışmak, iki farklı dayanıklılık sınırını karıştırır. Veritabanı commit edildikten sonra uygulama kapanırsa sipariş vardır fakat olay yoktur; broker'a gönderim başarılı olduktan sonra commit başarısızsa olay vardır fakat sipariş yoktur. Transactional outbox'ta sipariş ve yayınlanacak olay aynı PostgreSQL transaction'ında yazılır. Bu yaklaşım, microservices mimarisi içinde servisler arası veri sahipliğini korurken teslimatı ayrı bir iş akışına taşır.

CREATE TABLE outbox_event (
    id UUID PRIMARY KEY,
    aggregate_type TEXT NOT NULL,
    aggregate_id UUID NOT NULL,
    event_type TEXT NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
    available_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
    locked_until TIMESTAMPTZ,
    lock_token UUID,
    published_at TIMESTAMPTZ,
    attempts INTEGER NOT NULL DEFAULT 0
);

CREATE INDEX outbox_claim_idx
ON outbox_event (available_at, created_at)
WHERE published_at IS NULL;

Kısmi indeks kritik ayrıntıdır: yalnızca yayınlanmamış satırları indeksleyerek tablo büyüdükçe her polling turunda geçmiş olayların taranmasını engeller. UUID'yi uygulamada üretmek de önemlidir. JPA'da IDENTITY stratejisi, kimlik almak için erken INSERT ve flush tetikleyebilir; UUID ile hem aggregate hem outbox olay kimliği transaction başlamadan belirlenir. Bir spring framework eğitimi veya java spring eğitimi içeriğinde bu tasarımı değerlendirirken, outbox payload'ına tam entity snapshot'ı koymak yerine sürümlenmiş bir olay sözleşmesi, örneğin event_type='order.created.v1', koyun.

Spring Framework transaction sınırı: JPA kaydı ve JDBC outbox insert'i

Servis katmanında JPA repository ve JdbcTemplate aynı DataSourceTransactionManager veya JpaTransactionManager altında çalışıyorsa tek fiziksel transaction'a katılır. Aşağıdaki örnekte order ve outbox satırı aynı @Transactional çağrısında oluşur. Olayı ApplicationEventPublisher ile sadece bellek içinde yayınlamak yeterli değildir; uygulama commit sonrası kapanırsa @TransactionalEventListener çalışmadan kaybolabilir.

@Service
@RequiredArgsConstructor
class OrderCommandService {
    private final OrderRepository orders;
    private final NamedParameterJdbcTemplate jdbc;
    private final ObjectMapper mapper;

    @Transactional
    UUID create(CreateOrder command, EventActor actor) throws JsonProcessingException {
        UUID orderId = UUID.randomUUID();
        Order order = Order.create(orderId, command.customerId(), command.lines());
        orders.save(order);

        UUID eventId = UUID.randomUUID();
        String payload = mapper.writeValueAsString(
            new OrderCreatedV1(eventId, orderId, actor.subject(), actor.tenantId())
        );

        jdbc.update("""
            INSERT INTO outbox_event
              (id, aggregate_type, aggregate_id, event_type, payload)
            VALUES (:id, 'order', :aggregateId, 'order.created.v1', CAST(:payload AS jsonb))
            """,
            new MapSqlParameterSource()
                .addValue("id", eventId)
                .addValue("aggregateId", orderId)
                .addValue("payload", payload));
        return orderId;
    }
}

Bu kodda hata alanı bilinçli olarak dardır: mapper.writeValueAsString veya INSERT başarısız olursa transaction rollback olur ve sipariş de kalmaz. Buna karşılık outbox insert'ini REQUIRES_NEW ile ayırmak hatalıdır; dış transaction rollback olduğunda hayali bir OrderCreated olayı üretir. CI testinde bunu doğrulamak için @Transactional olmayan bir entegrasyon testi yazın, geçersiz payload ile insert'i zorlayın ve ardından SELECT count(*) FROM orders ile SELECT count(*) FROM outbox_event sonuçlarının ikisinin de 0 olduğunu doğrulayın.

Spring Cloud ve microservices mimarisi için lease tabanlı yayın

Birden fazla pod aynı outbox tablosunu taradığında satırı SELECT edip sonra UPDATE etmek çift yayına davetiye çıkarır. PostgreSQL'de FOR UPDATE SKIP LOCKED ile satırı kilitleyip aynı statement içinde lease vermek gerekir. Lease süresi, broker'a gönderim için gözlenen p99 süresinden büyük olmalı; örneğin p99 1.2 saniye ise 30 saniye makul bir başlangıçtır, fakat sabit bir sayı olarak kabul edilmemelidir.

WITH candidates AS (
    SELECT id
    FROM outbox_event
    WHERE published_at IS NULL
      AND available_at <= clock_timestamp()
      AND (locked_until IS NULL OR locked_until < clock_timestamp())
    ORDER BY created_at
    FOR UPDATE SKIP LOCKED
    LIMIT :batchSize
)
UPDATE outbox_event o
SET lock_token = :token,
    locked_until = clock_timestamp() + interval '30 seconds',
    attempts = attempts + 1
FROM candidates c
WHERE o.id = c.id
RETURNING o.id, o.event_type, o.aggregate_id, o.payload;

Claim transaction'ını sadece bu SQL ile sınırlayın; KafkaTemplate.send(...).get() veya başka broker I/O'sunu satır kilidi açıkken yapmayın. Yayın onayından sonra aşağıdaki koşullu UPDATE'i ayrı kısa transaction'da çalıştırın: UPDATE outbox_event SET published_at=clock_timestamp(), locked_until=NULL WHERE id=:id AND lock_token=:token. Süreç broker ack'inden sonra bu UPDATE'ten önce ölürse olay yeniden yayınlanır. Bu hata değil, at-least-once semantiğidir. Tüketici tarafında event id için UNIQUE indeksli bir inbox tablosu veya iş tablosunda processed_event_id kontrolü olmadan pratikte etkili-once davranışı elde edilemez.

Spring Cloud kullanımı burada zorunlu değildir; servis keşfi veya merkezi yapılandırma outbox'ın teslim garantisini değiştirmez. Ancak birden fazla deployment replica'sı çalışıyorsa lease sorgusu ortak koordinasyon mekanizmasıdır. spring cloud bileşenleriyle dinamik broker adresi kullanılsa bile event id'yi Kafka key ve header olarak sabit gönderin: key=aggregate_id, header=event-id. Böylece aynı aggregate için partition sırası korunur ve tüketici deduplikasyonu veritabanı anahtarıyla yapılabilir.

Spring Security ve Spring MVC sınırında olay kimliğini taşımak

Spring MVC endpoint'inde doğrulanan kullanıcının SecurityContext'i request thread'ine bağlıdır; @Scheduled ile çalışan outbox publisher'ında bu context yoktur. Bu nedenle olaya kullanıcı kimliğini ve tenant bilgisini command anında yazın. Yetki rollerini tüketiciye körlemesine taşımayın: roller olay işlendiğinde değişmiş olabilir. Tüketici, olaydaki subject ve tenant bilgisini audit için kullanmalı; kendi kaynaklarına erişim kararı için kendi spring security kuralını tekrar uygulamalıdır.

record EventActor(String subject, String tenantId) {}

@RestController
@RequiredArgsConstructor
class OrderController {
    private final OrderCommandService commands;

    @PostMapping("/orders")
    ResponseEntity<Map<String, UUID>> create(
            @RequestBody CreateOrder request,
            @AuthenticationPrincipal Jwt jwt) throws JsonProcessingException {
        String tenantId = jwt.getClaimAsString("tenant_id");
        UUID id = commands.create(request, new EventActor(jwt.getSubject(), tenantId));
        return ResponseEntity.accepted().body(Map.of("id", id));
    }
}

Bu spring rest api örneğinde tenant_id eksikse null ile devam etmeyin. Spring Security JwtAuthenticationConverter içinde tenant claim'ini doğrulayın veya controller'a gelmeden 400/401 üretin. Ayrıca outbox payload'ına access token, e-posta veya tüm JWT claim'lerini koymak log, dead-letter topic ve analitik sistemlerde kişisel veri yayılımı yaratır. EventActor için yalnızca değişmez subject ve uygulamanın tenant anahtarı gibi gerekli alanları tutun.

Spring Boot kursu pratiği: polling maliyetini JFR ve PostgreSQL ile ölçmek

Polling ayarını tahminle değiştirmeyin. Önce tek satır claim eden bir sürümde 10 dakika yük çalıştırın, sonra batchSize=100 kullanan sürümle aynı olay üretim hızında tekrar ölçün. JVM tarafında Java Flight Recorder başlatın:

java -XX:StartFlightRecording=filename=outbox.jfr,settings=profile,dumponexit=true -jar service.jar

JDK Mission Control'de Object Allocation in New TLAB, Java Monitor Blocked ve socket read sürelerini inceleyin. Her olay için yeni ObjectMapper veya büyük String ara nesneleri oluşturmak, broker'dan çok GC baskın maliyet olabilir.

PostgreSQL tarafında pg_stat_statements etkinse iki koşudan önce istatistiği izole ortamda sıfırlayın ve claim sorgusunu karşılaştırın:

SELECT pg_stat_statements_reset();

SELECT calls, rows, mean_exec_time, shared_blks_read, shared_blks_hit
FROM pg_stat_statements
WHERE query LIKE 'WITH candidates AS%';

Önce-sonra karşılaştırmasında yalnızca ortalama süreye bakmayın; calls değeri, çağrı başına dönen rows ve shared_blks_read birlikte batch'in gerçekten indeks kullandığını gösterir. 100'lük batch sonrası calls yaklaşık 100 kat azalmıyor veya shared_blks_read yükseliyorsa, kısmi indeks predicate'i sorgudaki published_at IS NULL koşuluyla birebir eşleşmiyor olabilir.

Uygulama metrikleri için Micrometer ile outbox.claimed, outbox.published, outbox.lease_expired ve outbox.publish.latency sayaçlarını ekleyin. lease_expired / claimed oranı yükselirken broker gecikmesi düşükse, worker'ın fixedDelay değeri değil GC duraklaması, pod CPU throttling'i veya batch içindeki senkron send çağrısı araştırılmalıdır. Bu ölçüm döngüsü, spring boot kursu laboratuvarında konfigürasyon değişikliğinin hangi kaynakta maliyet yarattığını somut olarak ayırır.

Sık Sorulan Sorular

Spring Boot'ta transactional outbox Kafka ile exactly-once sağlar mı?

Hayır. Veritabanına outbox yazımı atomiktir, fakat broker ack'i ile published_at güncellemesi arasında süreç durursa aynı event yeniden gönderilir. Event id'yi Kafka header'ına koyun ve tüketicide INSERT INTO inbox(event_id) ... için PRIMARY KEY veya UNIQUE constraint kullanarak tekrarları reddedin.

Spring REST API içinde outbox olayını ne zaman oluşturmalıyım?

Spring REST API controller'ında değil, @Transactional command service içinde oluşturun. Controller HTTP doğrulama ve Authentication principal çıkarımı yapmalı; service ise aggregate değişikliği ile outbox INSERT'ini aynı transaction'da gerçekleştirmelidir. Böylece HTTP katmanı dışında çalışan bir consumer veya batch job da aynı güvenceyi kullanır.

Java Spring eğitimi için outbox polling batch boyutu nasıl belirlenir?

Başlangıçta batchSize=50 veya 100 ile ölçüm yapın. JFR ile allocation ve GC duraklamalarını, pg_stat_statements ile mean_exec_time ve shared_blks_read değerlerini, Micrometer ile publish latency p95 değerini aynı yükte kaydedin. Lease süresini publish latency p99'un üstünde tutun; batch büyüdükçe lease_expired sayısı artıyorsa batch'i küçültün veya paralel worker sayısını artırın.

Spring Security kullanıcı bilgisini outbox event içine koymalı mı?

Audit için subject ve tenant anahtarı koyabilirsiniz, ancak JWT, access token ve değişken rol listesini koymayın. @Scheduled publisher thread'inde SecurityContext bulunmaz; bu nedenle gerekli minimum kimliği command anında payload'a yazın. Tüketici kendi Spring Security kurallarıyla yeniden yetkilendirme yapmalıdır.