// ANTI-PATTERN: Two separate writes with no atomicity guarantee
@Service
public class OrderServiceBroken {
 
    @Transactional
    public void createOrder(CreateOrderRequest request) {
        Order order = new Order(request);
        orderRepository.save(order);     // Write 1: DB succeeds
 
        // If this fails: DB has the order, Kafka does not
        // Downstream services are never notified -- silent data inconsistency
        kafkaTemplate.send("orders", order.getId().toString(),
            new OrderCreatedEvent(order));
    }
}

// CORRECT: Atomic write to DB + outbox in one transaction
@Service
@Transactional
public void createOrder(CreateOrderRequest request) {
    Order order = orderRepository.save(new Order(request));
    // Event written to DB in same atomic transaction
    outboxRepository.save(OutboxEvent.from(order, "ORDER_CREATED"));
    // Separate async publisher reads outbox and publishes to Kafka
}

// ANTI-PATTERN: Using eventually consistent reads for account balance
@Transactional(readOnly = true)  // Routes to read replica -- may be stale!
public BigDecimal getBalanceBeforeTransfer(Long accountId) {
    // If replica is lagging 2 seconds, the balance here could be $0
    // even though the actual balance is $1000
    // Leading to INCORRECT "insufficient funds" rejections
    return accountRepository.findBalance(accountId);
}

public class AccountService {
 
    // Money-critical: always read from primary
    @Transactional(isolation = Isolation.READ_COMMITTED)
    public BigDecimal getBalanceForTransfer(Long accountId) {
        // This MUST use the primary -- routes via write datasource
        DataSourceContextHolder.setDataSourceType(DataSourceType.WRITE);
        try {
            return accountRepository.findBalance(accountId);
        } finally {
            DataSourceContextHolder.clearDataSourceType();
        }
    }
 
    // Non-critical display: eventually consistent is fine
    @Transactional(readOnly = true)
    public AccountSummary getDashboardSummary(Long accountId) {
        // Showing $999.90 instead of $1000 for 50ms is acceptable in a dashboard
        return accountRepository.findSummary(accountId);
    }
}

// ANTI-PATTERN: Retry without idempotency
@Retryable(retryFor = RuntimeException.class, maxAttempts = 3)
public void chargeCustomer(Long customerId, BigDecimal amount) {
    // If first attempt charges successfully but response times out,
    // retry will charge the customer AGAIN
    paymentGateway.charge(customerId, amount);
}

// CORRECT: Idempotency key prevents duplicate charges
@Retryable(retryFor = RuntimeException.class, maxAttempts = 3,
           backoff = @Backoff(delay = 1000, multiplier = 2))
public PaymentResult chargeCustomer(String idempotencyKey,
                                    Long customerId, BigDecimal amount) {
    // Check for existing charge first
    Optional<Payment> existing = paymentRepository.findByIdempotencyKey(idempotencyKey);
    if (existing.isPresent()) {
        log.info("Returning existing charge for idempotency key {}", idempotencyKey);
        return PaymentResult.from(existing.get());
    }
 
    // Create payment record BEFORE calling external gateway
    // This ensures we track the intent even if gateway call is slow
    Payment payment = Payment.builder()
        .idempotencyKey(idempotencyKey)
        .customerId(customerId)
        .amount(amount)
        .status(PaymentStatus.PROCESSING)
        .build();
    paymentRepository.save(payment);
 
    try {
        GatewayResult result = paymentGateway.charge(customerId, amount);
        payment.setStatus(PaymentStatus.SUCCESS);
        payment.setGatewayTransactionId(result.getTransactionId());
    } catch (Exception e) {
        payment.setStatus(PaymentStatus.FAILED);
        payment.setFailureReason(e.getMessage());
        throw e;
    } finally {
        paymentRepository.save(payment);
    }
 
    return PaymentResult.from(payment);
}

// Strategy 1: Force primary read after write
@Service
public class ProfileService {
 
    @Transactional
    public UserProfile updateProfile(Long userId, ProfileUpdateRequest request) {
        User user = userRepository.findById(userId).orElseThrow();
        user.update(request);
        userRepository.save(user);
 
        // Mark session so next reads go to primary
        UserWriteTracker.markRecentWrite(userId);
 
        return UserProfile.from(user);
    }
 
    @Transactional(readOnly = true)
    public UserProfile getProfile(Long userId) {
        // If user recently wrote, read from primary
        if (UserWriteTracker.hasRecentWrite(userId, Duration.ofSeconds(5))) {
            DataSourceContextHolder.setDataSourceType(DataSourceType.WRITE);
        }
        try {
            return userRepository.findById(userId).map(UserProfile::from).orElseThrow();
        } finally {
            DataSourceContextHolder.clearDataSourceType();
        }
    }
}
 
// Strategy 2: Optimistic response -- return updated data from the write response
// Don't round-trip to read; return the object you just saved
@Transactional
public UserProfile updateProfile(Long userId, ProfileUpdateRequest request) {
    User user = userRepository.findById(userId).orElseThrow();
    user.update(request);
    User saved = userRepository.save(user);
    return UserProfile.from(saved);  // Return immediately, no secondary read
}

// WRONG: SERIALIZABLE for a simple lookup
@Transactional(isolation = Isolation.SERIALIZABLE)
public Product getProduct(Long id) {
    return productRepository.findById(id).orElseThrow();
}
 
// CORRECT: READ_COMMITTED for simple reads (or readOnly = true)
@Transactional(readOnly = true)  // Uses READ_COMMITTED snapshot
public Product getProduct(Long id) {
    return productRepository.findById(id).orElseThrow();
}
 
// Only use SERIALIZABLE for critical operations requiring full isolation
@Transactional(isolation = Isolation.SERIALIZABLE)
public void onCallScheduleUpdate(Long departmentId, Long doctorId) {
    // Write skew prevention needed: checking on-call count then updating
    long onCallCount = doctorRepository.countOnCall(departmentId);
    if (onCallCount <= 1) {
        throw new ScheduleViolationException("Cannot remove last on-call doctor");
    }
    doctorRepository.setOffCall(doctorId);
}

// ANTI-PATTERN: Transaction held open during external calls
@Transactional
public void processOrderWithExternalCalls(Long orderId) {
    Order order = orderRepository.findById(orderId).orElseThrow();
    order.setStatus(OrderStatus.PROCESSING);
    orderRepository.save(order);
 
    // DANGER: DB transaction is OPEN during this HTTP call (could take 2-10 seconds)
    PaymentResult result = paymentGateway.charge(order.getTotalAmount());  // network call!
 
    // DANGER: DB connection held, locks held, undo log growing
    InventoryResult inv = inventoryService.reserve(order.getItems());  // another network call!
 
    order.setStatus(OrderStatus.CONFIRMED);
    orderRepository.save(order);
}

// CORRECT: Narrow transactions, external calls outside transactions
@Service
public class OrderProcessingService {
 
    public void processOrder(Long orderId) {
        // 1. Short transaction: load and mark processing
        Order order = markOrderProcessing(orderId);
 
        // 2. External calls OUTSIDE transaction
        PaymentResult paymentResult = paymentGateway.charge(order.getTotalAmount());
        InventoryResult inventoryResult = inventoryService.reserve(order.getItems());
 
        // 3. Short transaction: update final state
        finalizeOrder(orderId, paymentResult, inventoryResult);
    }
 
    @Transactional(timeout = 5)
    private Order markOrderProcessing(Long orderId) {
        Order order = orderRepository.findByIdWithLock(orderId).orElseThrow();
        order.setStatus(OrderStatus.PROCESSING);
        return orderRepository.save(order);
    }  // Transaction closes here
 
    @Transactional(timeout = 5)
    private void finalizeOrder(Long orderId, PaymentResult payment,
                                InventoryResult inventory) {
        Order order = orderRepository.findById(orderId).orElseThrow();
        order.setStatus(payment.isSuccess() ? OrderStatus.CONFIRMED : OrderStatus.FAILED);
        order.setPaymentTransactionId(payment.getTransactionId());
        orderRepository.save(order);
    }  // Transaction closes here
}

// ANTI-PATTERN: Cache with no expiry
redisTemplate.opsForValue().set("user:123:preferences", preferences);
// Never expires! If user changes preferences, cache always returns old value.
 
// ANTI-PATTERN: Very long TTL with no invalidation
redisTemplate.opsForValue().set("product:456:price", price, Duration.ofDays(7));
// Price might change. Customer pays wrong price for up to 7 days.

// CORRECT: Appropriate TTL + active invalidation
@Service
public class UserPreferenceService {
 
    private static final Duration PREF_TTL = Duration.ofHours(1);
 
    @Transactional
    public void updatePreferences(Long userId, PreferenceUpdate update) {
        // Update DB
        userPreferenceRepository.update(userId, update);
 
        // Actively invalidate cache
        String cacheKey = "user:" + userId + ":preferences";
        redisTemplate.delete(cacheKey);
 
        log.debug("Invalidated preferences cache for user {}", userId);
    }
 
    public UserPreferences getPreferences(Long userId) {
        String cacheKey = "user:" + userId + ":preferences";
        UserPreferences cached = redisTemplate.opsForValue().get(cacheKey);
        if (cached != null) return cached;
 
        UserPreferences prefs = userPreferenceRepository.findByUserId(userId);
        redisTemplate.opsForValue().set(cacheKey, prefs, PREF_TTL);  // TTL as safety net
        return prefs;
    }
}

// CORRECT: Distributed lock prevents stampede
public List<Product> getTrendingProducts() {
    String cacheKey = "trending:products";
 
    // Try cache first (fast path)
    List<Product> cached = (List<Product>) redisTemplate.opsForValue().get(cacheKey);
    if (cached != null) return cached;
 
    // Stampede prevention: only ONE instance rebuilds the cache
    RLock lock = redissonClient.getLock("lock:rebuild:" + cacheKey);
    try {
        if (lock.tryLock(5, 10, TimeUnit.SECONDS)) {
            // Double-check after acquiring lock
            cached = (List<Product>) redisTemplate.opsForValue().get(cacheKey);
            if (cached != null) return cached;
 
            // Only one thread/instance reaches here
            List<Product> fresh = productRepository.findTrending();
            redisTemplate.opsForValue().set(cacheKey, fresh, Duration.ofMinutes(10));
            return fresh;
        } else {
            // Could not get lock -- serve stale if possible, otherwise wait and retry
            return getTrendingProductsWithStaleFallback(cacheKey);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new RuntimeException("Interrupted waiting for cache rebuild lock", e);
    } finally {
        if (lock.isHeldByCurrentThread()) lock.unlock();
    }
}
 
// Better: Use "soft expiry" -- cache stores expiry time internally
// Data appears valid longer than it is; probabilistic early refresh
public List<Product> getTrendingProductsProactive() {
    SoftCachedData<List<Product>> softCached = softCache.get("trending:products");
 
    if (softCached != null) {
        // Schedule background refresh if within 20% of expiry
        if (softCached.isNearExpiry(0.20)) {
            cacheRefreshExecutor.submit(() -> refreshTrendingCache());
        }
        return softCached.getData();  // Return current data while refreshing in background
    }
    return refreshTrendingCache();
}

@Service
public class SafeLockService {
 
    private final RedissonClient redissonClient;
    private final ResourceRepository resourceRepository;
 
    public void processWithFencing(String resourceId) {
        RLock lock = redissonClient.getLock("lock:" + resourceId);
 
        try {
            // Get fencing token BEFORE acquiring lock
            // Token is monotonically increasing -- if we see an older token, we're a zombie
            long fencingToken = lock.tryLockAndGetFencingToken(10, 30, TimeUnit.SECONDS);
 
            if (fencingToken < 0) {
                throw new LockAcquisitionException("Could not acquire lock for " + resourceId);
            }
 
            // Include fencing token in every write operation
            doWorkWithFencingToken(resourceId, fencingToken);
 
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
 
    private void doWorkWithFencingToken(String resourceId, long fencingToken) {
        // Store fencing token in DB alongside the data
        // If token in DB > our token: we're a zombie, abort
        resourceRepository.updateWithFencingToken(resourceId, fencingToken, newData);
    }
}
 
// In repository: reject writes with stale fencing tokens
@Transactional
public void updateWithFencingToken(String resourceId, long fencingToken, Data newData) {
    Resource resource = resourceRepository.findById(resourceId).orElseThrow();
 
    if (resource.getLastFencingToken() >= fencingToken) {
        throw new StaleWriteException("Stale write detected: our token " + fencingToken +
            " is not newer than stored " + resource.getLastFencingToken());
    }
 
    resource.setData(newData);
    resource.setLastFencingToken(fencingToken);
    resourceRepository.save(resource);
}

// Redisson watchdog: automatically extends lock TTL if owner is still alive
// Enabled by default -- lock TTL is renewed every 10 seconds if holder is running
RLock lock = redissonClient.getLock("lock:resource");
lock.lock();  // No explicit TTL -- watchdog auto-renews until unlock()
// If holder dies: watchdog stops, TTL expires naturally, lock released

// Option 1: Use SQS FIFO with message group per order
// All events for orderId=123 go to the same group, processed in order
 
// Option 2: Kafka -- all events for same orderId go to same partition (same consumer)
kafkaTemplate.send("order-events",
    orderId.toString(),  // partition key -- same orderId = same partition = ordered
    event);
 
// Option 3: Application-level ordering enforcement
@KafkaListener(topics = "order-events", groupId = "order-consumer")
@Transactional
public void handleOrderEvent(OrderEvent event) {
    Order order = orderRepository.findById(event.getOrderId()).orElseThrow();
 
    // State machine: only apply event if current state allows it
    if (!order.canTransitionTo(event.getTargetState())) {
        log.warn("Out-of-order event: order {} cannot transition from {} to {}. Event: {}",
            event.getOrderId(), order.getStatus(), event.getTargetState(), event.getType());
 
        // Option A: Dead-letter the event (it may arrive again after in-order events)
        // Option B: Store as "pending" and reprocess after state catches up
        outOfOrderEventStore.save(event);
        return;
    }
 
    order.applyEvent(event);
    orderRepository.save(order);
}
 
// Reprocess stored out-of-order events periodically
@Scheduled(fixedDelay = 5000)
@Transactional
public void reprocessOutOfOrderEvents() {
    List<OrderEvent> pending = outOfOrderEventStore.findPending();
    for (OrderEvent event : pending) {
        Order order = orderRepository.findById(event.getOrderId()).orElse(null);
        if (order != null && order.canTransitionTo(event.getTargetState())) {
            order.applyEvent(event);
            orderRepository.save(order);
            outOfOrderEventStore.delete(event);
        }
    }
}

// ANTI-PATTERN: SAGA without compensation
@KafkaListener(topics = "payment-events")
public void onPaymentFailed(PaymentFailedEvent event) {
    Order order = orderRepository.findById(event.getOrderId()).orElseThrow();
    order.setStatus(OrderStatus.FAILED);
    orderRepository.save(order);
    // FORGOT: Inventory was already reserved! Never released!
    // Result: Inventory count permanently decremented for a failed order
}

private void onPaymentFailed(PaymentFailedEvent event) {
    Order order = orderRepository.findById(event.getOrderId()).orElseThrow();
    order.setStatus(OrderStatus.FAILED);
    orderRepository.save(order);
 
    // COMPENSATION: Release inventory that was reserved in Step 2
    inventoryCompensationPublisher.publish(new ReleaseInventoryCommand(
        event.getOrderId(), order.getItems()));
 
    // Log for observability
    log.info("SAGA compensation: releasing inventory for failed order {}", event.getOrderId());
}

// Inventory reservation MUST be atomic and strongly consistent
// Use DynamoDB conditional write: only decrement if quantity > 0
UpdateItemRequest reserve = UpdateItemRequest.builder()
    .tableName("Inventory")
    .key(Map.of("productId", AttributeValue.fromS(productId)))
    .updateExpression("SET quantity = quantity - :qty")
    .conditionExpression("quantity >= :qty")  // Atomic check-and-decrement
    .expressionAttributeValues(Map.of(":qty", AttributeValue.fromN(String.valueOf(quantity))))
    .build();

ExternalPaymentRequest request = ExternalPaymentRequest.builder()
    .customerId(customerId)
    .amount(amount)
    .idempotencyKey(orderId + "-" + attemptNumber)  // or just UUID per payment
    .build();
externalGatewayClient.charge(request);

// NEVER read auth data from replica
@Transactional  // Forces primary read
public boolean isAuthorized(Long userId, String resource) {
    DataSourceContextHolder.setDataSourceType(DataSourceType.WRITE);
    return permissionRepository.hasPermission(userId, resource);
}

// Atomic SQL update instead of read-modify-write
@Transactional
public void applyTransactionBatch(Long accountId, List<Transaction> transactions) {
    BigDecimal totalDelta = transactions.stream()
        .map(t -> t.isDebit() ? t.getAmount().negate() : t.getAmount())
        .reduce(BigDecimal.ZERO, BigDecimal::add);
 
    // Atomic: database calculates new balance, not the application
    int updated = accountRepository.atomicUpdateBalance(accountId, totalDelta);
    if (updated == 0) {
        throw new AccountNotFoundException(accountId);
    }
}
 
// In repository:
@Query("UPDATE accounts SET balance = balance + :delta WHERE id = :id")
@Modifying
int atomicUpdateBalance(@Param("id") Long id, @Param("delta") BigDecimal delta);

# In application.yml for services that are all READ_COMMITTED
spring:
  jpa:
    properties:
      hibernate:
        connection:
          isolation: 2 # READ_COMMITTED

// More efficient: one batch select FOR UPDATE instead of N individual locks
List<Account> accounts = accountRepository.findByIdsForUpdate(accountIds);

-- Find queries holding locks in MySQL
SELECT trx_id, trx_started, trx_requested_lock_id, trx_mysql_thread_id, trx_query
FROM information_schema.INNODB_TRX
WHERE trx_state = 'LOCK WAIT'
ORDER BY trx_started;
 
-- Find which transaction is blocking
SELECT blocking_trx_id, blocking_lock_id, requested_trx_id
FROM information_schema.INNODB_LOCK_WAITS;

// Instead of distributed lock for a job queue, use SKIP LOCKED
// Allows multiple workers to process different items concurrently without blocking
@Query("SELECT j FROM Job j WHERE j.status = 'PENDING' " +
       "ORDER BY j.createdAt LIMIT 10 " +
       "FOR UPDATE SKIP LOCKED")  // Skip items locked by other workers
@Lock(LockModeType.PESSIMISTIC_WRITE)
List<Job> findAndLockPendingJobs(Pageable pageable);

// Include application version or data version in cache key
// Rolling deploy: new version writes to new key, old version reads from old key
// No cache invalidation storms during deployment
String cacheKey = "product:v" + dataVersion + ":" + productId;

// WRONG: Event published before transaction commits
// If transaction rolls back, event is already published!
@Transactional
public void createUser(CreateUserRequest request) {
    userRepository.save(new User(request));
    applicationEventPublisher.publishEvent(new UserCreatedEvent(request.getEmail()));
}
 
// CORRECT: Event published AFTER transaction commits
@Transactional
public void createUser(CreateUserRequest request) {
    User user = userRepository.save(new User(request));
    applicationEventPublisher.publishEvent(new UserCreatedEvent(user));
    // Event not published until commit succeeds
}
 
@Component
public class UserCreatedListener {
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void handleUserCreated(UserCreatedEvent event) {
        emailService.sendWelcomeEmail(event.getEmail());
        // Runs only after DB transaction commits
    }
}

spring:
  jpa:
    open-in-view: false # CRITICAL for production

spring:
  datasource:
    hikari:
      max-lifetime: 1740000 # 29 minutes -- Aurora closes idle connections at 30min
      keepalive-time: 60000 # 1 minute -- prevents AWS security group from closing idle
      connection-timeout: 30000 # 30 seconds -- fail fast if pool exhausted
      leak-detection-threshold: 60000 # Warn if connection held > 60s (detect long transactions)

@Component
@RequiredArgsConstructor
public class ConsistencyMetrics {
 
    private final MeterRegistry meterRegistry;
 
    // Track optimistic lock conflicts
    public void recordOptimisticLockConflict(String entityType) {
        meterRegistry.counter("consistency.optimistic_lock.conflict",
            Tags.of("entity", entityType)).increment();
    }
 
    // Track cache miss rate (high miss rate = potential inconsistency or cold cache)
    public void recordCacheHit(String cacheName, boolean hit) {
        meterRegistry.counter("consistency.cache." + (hit ? "hit" : "miss"),
            Tags.of("cache", cacheName)).increment();
    }
 
    // Track outbox event processing lag
    public void recordOutboxLag(long lagMs) {
        meterRegistry.gauge("consistency.outbox.lag_ms", lagMs);
    }
 
    // Track idempotency key hits (duplicate request detection)
    public void recordIdempotencyHit(String operationType) {
        meterRegistry.counter("consistency.idempotency.hit",
            Tags.of("operation", operationType)).increment();
    }
}

-- Check for replica lag on Aurora reader
SELECT server_id,
       session_id,
       last_update_timestamp,
       TIMESTAMPDIFF(SECOND, last_update_timestamp, NOW()) AS lag_seconds
FROM information_schema.replica_host_status;
 
-- Check for long-running transactions (potential lock holders)
SELECT trx_id,
       trx_started,
       TIMEDIFF(NOW(), trx_started) AS duration,
       trx_state,
       trx_mysql_thread_id,
       SUBSTR(trx_query, 1, 200) AS query
FROM information_schema.INNODB_TRX
WHERE TIMEDIFF(NOW(), trx_started) > '00:00:05'  -- running > 5 seconds
ORDER BY trx_started;
 
-- Check current locks
SELECT r.trx_id waiting_trx_id,
       r.trx_mysql_thread_id waiting_thread,
       r.trx_query waiting_query,
       b.trx_id blocking_trx_id,
       b.trx_mysql_thread_id blocking_thread,
       b.trx_query blocking_query
FROM information_schema.INNODB_LOCK_WAITS w
INNER JOIN information_schema.INNODB_TRX b ON b.trx_id = w.blocking_trx_id
INNER JOIN information_schema.INNODB_TRX r ON r.trx_id = w.requesting_trx_id;
 
-- Check pending outbox events (should be near zero in healthy system)
SELECT status, COUNT(*) AS count, MIN(created_at) AS oldest
FROM outbox_events
GROUP BY status;
 
-- Check for stuck PROCESSING outbox events (publisher crashed mid-process)
SELECT * FROM outbox_events
WHERE status = 'PROCESSING'
  AND created_at < DATE_SUB(NOW(), INTERVAL 5 MINUTE);
-- These should be reset to PENDING and retried