Saga Pattern Nedir?

Saga Pattern, mikroservis mimarilerinde dağıtık transaction'ları (distributed transactions) yönetmek için kullanılan bir tasarım desenidir. ACID'in aksine, Saga'lar "Eventually Consistent" (Nihai Tutarlılık) prensibiyle çalışır ve uzun süren işlemleri yönetir.

Saga Pattern vs ACID

Saga Pattern Türleri

1. Choreography (Koreografi) - Event Tabanlı

Her servis kendi işini yapar ve event fırlatır. Diğer servisler bu eventleri dinler.

2. Orchestration (Orkestrasyon) - Merkezi Yönetim

Bir orkestratör (saga coordinator) tüm işlemleri yönetir ve sıralar.

Gerçek Hayat Örneği: Sipariş - Ödeme - Stok - Kargo

Senaryo: E-ticaret Sipariş Süreci

1. Sipariş oluştur (Order Service)
2. Ödemeyi al (Payment Service)
3. Stoktan düş (Stock Service)
4. Kargoya ver (Shipping Service)

1. Choreography (Koreografi) Tabanlı Saga

// Domain Events
public class OrderCreatedEvent {
    private String orderId;
    private String customerId;
    private BigDecimal amount;
    private List<OrderItem> items;
    // getters, setters, constructor
}

public class PaymentProcessedEvent {
    private String orderId;
    private String paymentId;
    private PaymentStatus status;
    // getters, setters
}

public class StockReservedEvent {
    private String orderId;
    private List<ReservedItem> reservedItems;
    private boolean success;
    // getters, setters
}

public class ShippingScheduledEvent {
    private String orderId;
    private String trackingNumber;
    private LocalDateTime estimatedDelivery;
    // getters, setters
}

// Compensation Events
public class PaymentFailedEvent {
    private String orderId;
    private String reason;
    // getters, setters
}

public class StockReservationFailedEvent {
    private String orderId;
    private String reason;
    // getters, setters
}

// Order Service
@Service
public class OrderService {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Autowired
    private OrderRepository orderRepository;

    @Transactional
    public Order createOrder(OrderRequest request) {
        // 1. Siparişi kaydet (PENDING durumunda)
        Order order = new Order();
        order.setCustomerId(request.getCustomerId());
        order.setItems(request.getItems());
        order.setTotalAmount(request.getTotalAmount());
        order.setStatus(OrderStatus.PENDING);
        order = orderRepository.save(order);

        // 2. Event fırlat - diğer servisler dinleyecek
        OrderCreatedEvent event = new OrderCreatedEvent();
        event.setOrderId(order.getId());
        event.setCustomerId(order.getCustomerId());
        event.setAmount(order.getTotalAmount());
        event.setItems(order.getItems());

        kafkaTemplate.send("order-created-topic", event);

        return order;
    }

    // Compensation - Payment Failed handler
    @KafkaListener(topics = "payment-failed-topic")
    public void handlePaymentFailed(PaymentFailedEvent event) {
        Order order = orderRepository.findById(event.getOrderId()).get();
        order.setStatus(OrderStatus.PAYMENT_FAILED);
        orderRepository.save(order);

        // İptal edildi bilgilendirmesi
        notifyCustomer(order.getCustomerId(), "Ödeme başarısız, sipariş iptal edildi");
    }
}

// Payment Service
@Service
public class PaymentService {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Autowired
    private PaymentRepository paymentRepository;

    @KafkaListener(topics = "order-created-topic")
    public void handleOrderCreated(OrderCreatedEvent event) {
        try {
            // Ödeme işlemini gerçekleştir
            Payment payment = processPayment(event);

            // Başarılı event fırlat
            PaymentProcessedEvent successEvent = new PaymentProcessedEvent();
            successEvent.setOrderId(event.getOrderId());
            successEvent.setPaymentId(payment.getId());
            successEvent.setStatus(PaymentStatus.SUCCESS);

            kafkaTemplate.send("payment-processed-topic", successEvent);

        } catch (Exception e) {
            // Başarısız event fırlat (compensation)
            PaymentFailedEvent failedEvent = new PaymentFailedEvent();
            failedEvent.setOrderId(event.getOrderId());
            failedEvent.setReason(e.getMessage());

            kafkaTemplate.send("payment-failed-topic", failedEvent);
        }
    }

    private Payment processPayment(OrderCreatedEvent event) {
        // Ödeme gateway entegrasyonu
        Payment payment = new Payment();
        payment.setOrderId(event.getOrderId());
        payment.setAmount(event.getAmount());
        payment.setStatus(PaymentStatus.PROCESSING);
        payment = paymentRepository.save(payment);

        // Ödeme işlemi simülasyonu
        if (event.getAmount().compareTo(new BigDecimal("10000")) > 0) {
            throw new RuntimeException("Yetersiz bakiye");
        }

        payment.setStatus(PaymentStatus.COMPLETED);
        payment.setPaymentDate(LocalDateTime.now());
        return paymentRepository.save(payment);
    }
}

// Stock Service
@Service
public class StockService {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Autowired
    private StockRepository stockRepository;

    @KafkaListener(topics = "payment-processed-topic")
    public void handlePaymentProcessed(PaymentProcessedEvent event) {
        try {
            // Stok rezervasyonu yap
            reserveStock(event);

            // Başarılı event fırlat
            StockReservedEvent successEvent = new StockReservedEvent();
            successEvent.setOrderId(event.getOrderId());
            successEvent.setSuccess(true);

            kafkaTemplate.send("stock-reserved-topic", successEvent);

        } catch (Exception e) {
            // Başarısız event fırlat (compensation)
            StockReservationFailedEvent failedEvent = new StockReservationFailedEvent();
            failedEvent.setOrderId(event.getOrderId());
            failedEvent.setReason(e.getMessage());

            kafkaTemplate.send("stock-failed-topic", failedEvent);
        }
    }

    private void reserveStock(PaymentProcessedEvent event) {
        // Stok kontrolü ve rezervasyon
        // Örnek: Her ürün için stok kontrolü
        List<OrderItem> items = getOrderItems(event.getOrderId());

        for (OrderItem item : items) {
            Stock stock = stockRepository.findByProductId(item.getProductId());
            if (stock.getQuantity() < item.getQuantity()) {
                throw new RuntimeException("Yetersiz stok: " + item.getProductId());
            }
            stock.setQuantity(stock.getQuantity() - item.getQuantity());
            stockRepository.save(stock);
        }
    }

    // Compensation - Payment Failed handler
    @KafkaListener(topics = "payment-failed-topic")
    public void handlePaymentFailed(PaymentFailedEvent event) {
        // Stok rezervasyonu yapılmadı, bir şey yapmaya gerek yok
        log.info("Payment failed for order: {}, no stock reservation needed", event.getOrderId());
    }
}

// Shipping Service
@Service
public class ShippingService {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Autowired
    private ShippingRepository shippingRepository;

    @KafkaListener(topics = "stock-reserved-topic")
    public void handleStockReserved(StockReservedEvent event) {
        try {
            // Kargo planlaması yap
            Shipping shipping = scheduleShipping(event);

            // Başarılı event fırlat
            ShippingScheduledEvent successEvent = new ShippingScheduledEvent();
            successEvent.setOrderId(event.getOrderId());
            successEvent.setTrackingNumber(shipping.getTrackingNumber());
            successEvent.setEstimatedDelivery(shipping.getEstimatedDelivery());

            kafkaTemplate.send("shipping-scheduled-topic", successEvent);

            // Order Service'i güncelle
            updateOrderStatus(event.getOrderId(), OrderStatus.COMPLETED);

        } catch (Exception e) {
            // Kargo planlaması başarısız - compensation başlat
            // Stock'u geri al, ödemeyi iade et
            initiateCompensation(event.getOrderId(), e.getMessage());
        }
    }

    private Shipping scheduleShipping(StockReservedEvent event) {
        Shipping shipping = new Shipping();
        shipping.setOrderId(event.getOrderId());
        shipping.setStatus(ShippingStatus.PLANNED);
        shipping.setEstimatedDelivery(LocalDateTime.now().plusDays(3));
        shipping.setTrackingNumber("TRK" + UUID.randomUUID().toString());
        return shippingRepository.save(shipping);
    }

    private void initiateCompensation(String orderId, String reason) {
        // 1. Stok geri al
        kafkaTemplate.send("stock-revert-topic", orderId);

        // 2. Ödemeyi iade et
        kafkaTemplate.send("payment-refund-topic", orderId);

        // 3. Siparişi iptal et
        kafkaTemplate.send("order-cancel-topic", orderId);
    }
}

2. Orchestration (Orkestrasyon) Tabanlı Saga

// Saga Orchestrator
@Component
public class OrderSagaOrchestrator {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Autowired
    private SagaStateRepository sagaStateRepository;

    @Autowired
    private RestTemplate restTemplate;

    public void startSaga(OrderCreatedEvent event) {
        // Saga state'i kaydet
        SagaState state = new SagaState();
        state.setSagaId(UUID.randomUUID().toString());
        state.setOrderId(event.getOrderId());
        state.setCurrentStep(SagaStep.CREATE_ORDER);
        state.setStatus(SagaStatus.STARTED);
        sagaStateRepository.save(state);

        // İlk adım: Ödeme işle
        processPayment(state);
    }

    private void processPayment(SagaState state) {
        try {
            // Payment Service'e REST çağrısı
            PaymentRequest request = new PaymentRequest();
            request.setOrderId(state.getOrderId());

            ResponseEntity<PaymentResponse> response = restTemplate.postForEntity(
                "http://payment-service/api/payments",
                request,
                PaymentResponse.class
            );

            if (response.getBody().isSuccess()) {
                // Başarılı - Stok adımına geç
                state.setCurrentStep(SagaStep.PROCESS_PAYMENT);
                sagaStateRepository.save(state);

                reserveStock(state, response.getBody().getPaymentId());
            } else {
                // Başarısız - Saga'yı bitir
                failSaga(state, "Ödeme başarısız");
            }

        } catch (Exception e) {
            // Hata durumunda compensation başlat
            compensateSaga(state, "Ödeme hatası: " + e.getMessage());
        }
    }

    private void reserveStock(SagaState state, String paymentId) {
        try {
            // Stock Service'e REST çağrısı
            StockReservationRequest request = new StockReservationRequest();
            request.setOrderId(state.getOrderId());

            ResponseEntity<StockReservationResponse> response = restTemplate.postForEntity(
                "http://stock-service/api/stock/reserve",
                request,
                StockReservationResponse.class
            );

            if (response.getBody().isSuccess()) {
                // Başarılı - Kargo adımına geç
                state.setCurrentStep(SagaStep.RESERVE_STOCK);
                sagaStateRepository.save(state);

                scheduleShipping(state, paymentId, response.getBody().getReservationId());
            } else {
                // Stok yok - Ödemeyi iade et
                compensateSaga(state, "Stok yetersiz");
            }

        } catch (Exception e) {
            compensateSaga(state, "Stok rezervasyon hatası: " + e.getMessage());
        }
    }

    private void scheduleShipping(SagaState state, String paymentId, String reservationId) {
        try {
            // Shipping Service'e REST çağrısı
            ShippingRequest request = new ShippingRequest();
            request.setOrderId(state.getOrderId());

            ResponseEntity<ShippingResponse> response = restTemplate.postForEntity(
                "http://shipping-service/api/shipping/schedule",
                request,
                ShippingResponse.class
            );

            if (response.getBody().isSuccess()) {
                // Başarılı - Saga tamamlandı
                state.setCurrentStep(SagaStep.SCHEDULE_SHIPPING);
                state.setStatus(SagaStatus.COMPLETED);
                sagaStateRepository.save(state);

                // Order Service'i bilgilendir
                completeOrder(state.getOrderId(), response.getBody().getTrackingNumber());

            } else {
                // Kargo planlaması başarısız - Compensation
                compensateSaga(state, "Kargo planlaması başarısız");
            }

        } catch (Exception e) {
            compensateSaga(state, "Kargo planlaması hatası: " + e.getMessage());
        }
    }

    private void compensateSaga(SagaState state, String reason) {
        log.error("Saga compensation başlatılıyor: {} - Reason: {}", state.getSagaId(), reason);

        // Hangi adımdaysak ona göre compensation
        switch (state.getCurrentStep()) {
            case PROCESS_PAYMENT:
                // Sadece ödeme yapıldıysa iade et
                refundPayment(state.getOrderId());
                break;

            case RESERVE_STOCK:
                // Stok rezervasyonu varsa geri al ve ödemeyi iade et
                revertStock(state.getOrderId());
                refundPayment(state.getOrderId());
                break;

            case SCHEDULE_SHIPPING:
                // Kargo planlandıysa iptal et, stoku geri al, ödemeyi iade et
                cancelShipping(state.getOrderId());
                revertStock(state.getOrderId());
                refundPayment(state.getOrderId());
                break;
        }

        // Saga state'i güncelle
        state.setStatus(SagaStatus.COMPENSATED);
        state.setErrorMessage(reason);
        sagaStateRepository.save(state);

        // Order Service'i bilgilendir
        failOrder(state.getOrderId(), reason);
    }

    // Compensation metotları
    private void refundPayment(String orderId) {
        kafkaTemplate.send("payment-refund-topic", orderId);
    }

    private void revertStock(String orderId) {
        kafkaTemplate.send("stock-revert-topic", orderId);
    }

    private void cancelShipping(String orderId) {
        kafkaTemplate.send("shipping-cancel-topic", orderId);
    }

    private void completeOrder(String orderId, String trackingNumber) {
        OrderCompletedEvent event = new OrderCompletedEvent();
        event.setOrderId(orderId);
        event.setTrackingNumber(trackingNumber);
        event.setStatus("COMPLETED");

        kafkaTemplate.send("order-completed-topic", event);
    }

    private void failOrder(String orderId, String reason) {
        OrderFailedEvent event = new OrderFailedEvent();
        event.setOrderId(orderId);
        event.setReason(reason);

        kafkaTemplate.send("order-failed-topic", event);
    }
}

// Saga State Entity
@Entity
public class SagaState {
    @Id
    private String sagaId;
    private String orderId;

    @Enumerated(EnumType.STRING)
    private SagaStep currentStep;

    @Enumerated(EnumType.STRING)
    private SagaStatus status;

    private String errorMessage;
    private LocalDateTime createdAt;
    private LocalDateTime updatedAt;

    // getters, setters
}

public enum SagaStep {
    CREATE_ORDER,
    PROCESS_PAYMENT,
    RESERVE_STOCK,
    SCHEDULE_SHIPPING
}

public enum SagaStatus {
    STARTED,
    COMPLETED,
    FAILED,
    COMPENSATED
}

Önemli Saga Pattern Konuları

1. Idempotency (Tekrarlanabilirlik)

@Service
public class IdempotentPaymentService {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    public PaymentResult processPayment(PaymentRequest request) {
        // Idempotency key kontrolü
        String idempotencyKey = request.getIdempotencyKey();
        String processed = redisTemplate.opsForValue().get(idempotencyKey);

        if (processed != null) {
            // Aynı istek daha önce işlendi
            return getCachedResult(idempotencyKey);
        }

        // İşlemi gerçekleştir
        PaymentResult result = doProcessPayment(request);

        // Sonucu cache'le (TTL ile)
        redisTemplate.opsForValue().set(idempotencyKey, 
            result.getPaymentId(), 
            Duration.ofHours(24));

        return result;
    }
}

2. Compensation Transaction Örnekleri

@Service
public class CompensationManager {

    @Transactional
    public void compensateOrder(String orderId) {
        // 1. Adım: Stok geri al
        compensateStock(orderId);

        // 2. Adım: Ödemeyi iade et
        compensatePayment(orderId);

        // 3. Adım: Sipariş durumunu güncelle
        updateOrderStatus(orderId, OrderStatus.CANCELLED);

        // 4. Adım: Müşteriyi bilgilendir
        notifyCustomer(orderId, "Siparişiniz iptal edildi");
    }

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void compensateStock(String orderId) {
        try {
            // Stok servisine compensation çağrısı
            stockService.revertStock(orderId);
        } catch (Exception e) {
            // Manual intervention gerekebilir
            log.error("Stock compensation failed for order: {}", orderId);
            alertTeam("Stock compensation failed", orderId);
        }
    }
}

3. Saga Monitoring ve Recovery

@Component
public class SagaMonitor {

    @Scheduled(fixedDelay = 60000) // Her dakika
    public void checkStuckSagas() {
        // 10 dakikadan uzun süredir devam eden sagaları bul
        List<SagaState> stuckSagas = sagaStateRepository
            .findByStatusAndLastUpdatedBefore(
                SagaStatus.STARTED,
                LocalDateTime.now().minusMinutes(10)
            );

        for (SagaState saga : stuckSagas) {
            // Manual intervention veya otomatik recovery
            recoverSaga(saga);
        }
    }

    private void recoverSaga(SagaState saga) {
        // Saga'yı yeniden başlat veya compensation uygula
        if (saga.getCurrentStep() == SagaStep.PROCESS_PAYMENT) {
            // Ödeme adımında takıldıysa, payment service'i kontrol et
            PaymentStatus status = checkPaymentStatus(saga.getOrderId());

            if (status == PaymentStatus.SUCCESS) {
                // Ödeme başarılıysa devam et
                orchestrator.reserveStock(saga);
            } else {
                // Ödeme başarısızsa compensation başlat
                orchestrator.compensateSaga(saga, "Timeout - Payment stuck");
            }
        }
    }
}

Saga Pattern Best Practices

1. Event Sıralaması ve Correlation

public class SagaEvent {
    private String sagaId;        // Hangi saga'ya ait
    private String correlationId;  // Hangi işleme ait
    private String eventType;      // Event tipi
    private String step;           // Hangi adım
    private Object data;           // Event verisi
    private LocalDateTime timestamp;

    // getters, setters
}

2. Timeout ve Retry Mekanizması

@Configuration
public class SagaRetryConfig {

    @Bean
    public RetryTemplate retryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();

        // 3 kez dene, her denemede bekleme süresini artır
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(1000);
        backOffPolicy.setMultiplier(2);
        backOffPolicy.setMaxInterval(10000);
        retryTemplate.setBackOffPolicy(backOffPolicy);

        // Sadece belirli hatalarda retry yap
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(3);
        retryTemplate.setRetryPolicy(retryPolicy);

        return retryTemplate;
    }
}

3. Dead Letter Queue (DLQ) Kullanımı

@Component
public class SagaDeadLetterHandler {

    @KafkaListener(topics = "saga-dlq-topic")
    public void handleDeadLetter(SagaDeadLetterMessage message) {
        log.error("Dead letter alındı: {}", message);

        // Manual intervention için alert gönder
        alertTeam("Saga DLQ message received", message);

        // DLQ'daki mesajı veritabanına kaydet
        saveToDeadLetterTable(message);

        // Gerekirse otomatik recovery dene
        if (message.getRetryCount() < 3) {
            retryMessage(message);
        }
    }
}

Saga Pattern Ne Zaman Kullanılmalı?

Kullanılması Gereken Durumlar:

• Mikroservisler arası işlemler: Birden fazla servisi güncelleyen işlemler
• Uzun süren işlemler: Saatler/günler sürebilen business process'ler
• Yüksek skalabilite gereken durumlar: Lock mekanizması istemeyen sistemler
• Event-driven mimariler: Asenkron iletişimin olduğu sistemler

Kullanılmaması Gereken Durumlar:

• Tek veritabanı işlemleri: ACID yeterli ve daha basit
• Kısa süreli kritik işlemler: Finansal işlemlerde ACID daha güvenli
• Basit CRUD operasyonları: Ekstra complexity gereksiz

Saga Pattern, dağıtık sistemlerde veri tutarlılığını sağlamak için güçlü bir araçtır ancak getirdiği complexity'nin farkında olmak ve doğru durumlarda kullanmak önemlidir.