MACRO NEPAL
Achieving Eventual Consistency in Microservices: Java Implementation Guide
awiskar acharya
November 17, 2025
9:54 am

Introduction

Eventual consistency is a fundamental concept in distributed systems where data replicas eventually become consistent across services, without requiring immediate synchronization. In microservices architectures, maintaining strong consistency is often impractical due to the distributed nature of the system. This guide explores patterns and Java implementations for achieving eventual consistency while maintaining system reliability and performance.

Core Concepts and Patterns

1. Consistency Models

  • Strong Consistency: Immediate consistency (ACID)
  • Eventual Consistency: Guarantees consistency over time
  • Causal Consistency: Preserves causal relationships
  • Read-your-writes Consistency: Ensures you see your own writes

2. Common Patterns

  • Saga Pattern: Distributed transactions with compensation
  • Event Sourcing: Store state changes as events
  • CQRS: Separate read and write models
  • Outbox Pattern: Reliable event publishing
  • Change Data Capture: Capture database changes

Project Setup and Dependencies

1. Maven Dependencies

<dependencies>
<!-- Spring Boot -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- Database -->
<dependency>
<groupId>org.postgresql</groupId>
<artifactId>postgresql</artifactId>
<scope>runtime</scope>
</dependency>
<!-- Messaging -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
<!-- Resilience -->
<dependency>
<groupId>org.springframework.retry</groupId>
<artifactId>spring-retry</artifactId>
</dependency>
<dependency>
<groupId>io.github.resilience4j</groupId>
<artifactId>resilience4j-spring-boot2</artifactId>
</dependency>
<!-- Testing -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>2022.0.4</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>

2. Domain Models

// Order Service Domain
@Entity
@Table(name = "orders")
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class Order {
@Id
private String orderId;
@Enumerated(EnumType.STRING)
private OrderStatus status;
private String customerId;
private BigDecimal totalAmount;
@ElementCollection
@CollectionTable(name = "order_items", joinColumns = @JoinColumn(name = "order_id"))
private List<OrderItem> items;
private LocalDateTime createdAt;
private LocalDateTime updatedAt;
@Version
private Long version;
@PrePersist
protected void onCreate() {
this.createdAt = LocalDateTime.now();
this.updatedAt = LocalDateTime.now();
}
@PreUpdate
protected void onUpdate() {
this.updatedAt = LocalDateTime.now();
}
}
@Embeddable
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class OrderItem {
private String productId;
private String productName;
private Integer quantity;
private BigDecimal unitPrice;
}
public enum OrderStatus {
PENDING, CONFIRMED, PAYMENT_PENDING, PAYMENT_COMPLETED, 
PAYMENT_FAILED, SHIPPED, DELIVERED, CANCELLED
}
// Payment Service Domain
@Entity
@Table(name = "payments")
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class Payment {
@Id
private String paymentId;
private String orderId;
private String customerId;
private BigDecimal amount;
@Enumerated(EnumType.STRING)
private PaymentStatus status;
private String transactionId;
private LocalDateTime paymentDate;
private LocalDateTime createdAt;
}
public enum PaymentStatus {
PENDING, COMPLETED, FAILED, REFUNDED
}
// Inventory Service Domain
@Entity
@Table(name = "inventory")
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class Inventory {
@Id
private String productId;
private String productName;
private Integer availableQuantity;
private Integer reservedQuantity;
@Version
private Long version;
public boolean canFulfill(int quantity) {
return availableQuantity >= quantity;
}
public void reserve(int quantity) {
if (!canFulfill(quantity)) {
throw new InsufficientInventoryException(
"Insufficient inventory for product: " + productId);
}
availableQuantity -= quantity;
reservedQuantity += quantity;
}
public void release(int quantity) {
reservedQuantity -= quantity;
availableQuantity += quantity;
}
public void confirmReservation(int quantity) {
reservedQuantity -= quantity;
}
}
public class InsufficientInventoryException extends RuntimeException {
public InsufficientInventoryException(String message) {
super(message);
}
}

Saga Pattern Implementation

1. Saga Orchestrator

@Component
@Slf4j
public class OrderSagaOrchestrator {
private final OrderService orderService;
private final PaymentService paymentService;
private final InventoryService inventoryService;
private final SagaEventPublisher eventPublisher;
public OrderSagaOrchestrator(OrderService orderService,
PaymentService paymentService,
InventoryService inventoryService,
SagaEventPublisher eventPublisher) {
this.orderService = orderService;
this.paymentService = paymentService;
this.inventoryService = inventoryService;
this.eventPublisher = eventPublisher;
}
@Async
@Transactional
public void createOrderSaga(CreateOrderCommand command) {
SagaContext context = new SagaContext(command.getOrderId());
try {
// Step 1: Create order in PENDING state
log.info("Starting saga for order: {}", command.getOrderId());
Order order = orderService.createPendingOrder(command);
context.setOrder(order);
// Step 2: Reserve inventory
reserveInventory(context, command);
// Step 3: Process payment
processPayment(context, command);
// Step 4: Confirm order
confirmOrder(context);
log.info("Saga completed successfully for order: {}", command.getOrderId());
} catch (Exception e) {
log.error("Saga failed for order: {}", command.getOrderId(), e);
compensate(context, e);
}
}
private void reserveInventory(SagaContext context, CreateOrderCommand command) {
try {
inventoryService.reserveInventory(command.getOrderId(), command.getItems());
context.setInventoryReserved(true);
log.info("Inventory reserved for order: {}", context.getOrderId());
} catch (InsufficientInventoryException e) {
log.warn("Inventory reservation failed for order: {}", context.getOrderId());
throw new SagaException("Inventory reservation failed", e);
}
}
private void processPayment(SagaContext context, CreateOrderCommand command) {
try {
Payment payment = paymentService.processPayment(
command.getOrderId(), 
command.getCustomerId(), 
command.getTotalAmount()
);
context.setPayment(payment);
context.setPaymentProcessed(true);
log.info("Payment processed for order: {}", context.getOrderId());
} catch (PaymentException e) {
log.warn("Payment processing failed for order: {}", context.getOrderId());
throw new SagaException("Payment processing failed", e);
}
}
private void confirmOrder(SagaContext context) {
orderService.confirmOrder(context.getOrderId());
inventoryService.confirmReservation(context.getOrderId());
context.setOrderConfirmed(true);
eventPublisher.publishOrderConfirmed(context.getOrderId());
log.info("Order confirmed: {}", context.getOrderId());
}
private void compensate(SagaContext context, Exception failure) {
log.info("Starting compensation for order: {}", context.getOrderId());
try {
// Compensate in reverse order
if (context.isPaymentProcessed()) {
paymentService.refundPayment(context.getOrderId());
log.info("Payment refunded for order: {}", context.getOrderId());
}
if (context.isInventoryReserved()) {
inventoryService.releaseReservation(context.getOrderId());
log.info("Inventory released for order: {}", context.getOrderId());
}
if (context.getOrder() != null) {
orderService.cancelOrder(context.getOrderId(), failure.getMessage());
log.info("Order cancelled: {}", context.getOrderId());
}
eventPublisher.publishOrderFailed(context.getOrderId(), failure.getMessage());
} catch (Exception compException) {
log.error("Compensation failed for order: {}", context.getOrderId(), compException);
// In real scenario, would trigger alert/monitoring
}
}
@Data
private static class SagaContext {
private final String orderId;
private Order order;
private Payment payment;
private boolean inventoryReserved = false;
private boolean paymentProcessed = false;
private boolean orderConfirmed = false;
public SagaContext(String orderId) {
this.orderId = orderId;
}
}
public static class SagaException extends RuntimeException {
public SagaException(String message) {
super(message);
}
public SagaException(String message, Throwable cause) {
super(message, cause);
}
}
}
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class CreateOrderCommand {
private String orderId;
private String customerId;
private List<OrderItem> items;
private BigDecimal totalAmount;
private String paymentMethod;
}

2. Saga Services

@Service
@Slf4j
@Transactional
public class OrderService {
private final OrderRepository orderRepository;
private final ApplicationEventPublisher eventPublisher;
public OrderService(OrderRepository orderRepository,
ApplicationEventPublisher eventPublisher) {
this.orderRepository = orderRepository;
this.eventPublisher = eventPublisher;
}
public Order createPendingOrder(CreateOrderCommand command) {
Order order = Order.builder()
.orderId(command.getOrderId())
.customerId(command.getCustomerId())
.items(command.getItems())
.totalAmount(command.getTotalAmount())
.status(OrderStatus.PENDING)
.build();
Order saved = orderRepository.save(order);
eventPublisher.publishEvent(new OrderCreatedEvent(
saved.getOrderId(),
saved.getCustomerId(),
saved.getItems(),
saved.getTotalAmount()
));
return saved;
}
public void confirmOrder(String orderId) {
orderRepository.findById(orderId).ifPresent(order -> {
order.setStatus(OrderStatus.CONFIRMED);
orderRepository.save(order);
eventPublisher.publishEvent(new OrderConfirmedEvent(orderId));
});
}
public void cancelOrder(String orderId, String reason) {
orderRepository.findById(orderId).ifPresent(order -> {
order.setStatus(OrderStatus.CANCELLED);
orderRepository.save(order);
eventPublisher.publishEvent(new OrderCancelledEvent(orderId, reason));
});
}
@Transactional(readOnly = true)
public Optional<Order> getOrder(String orderId) {
return orderRepository.findById(orderId);
}
}
@Service
@Slf4j
@Transactional
public class PaymentService {
private final PaymentRepository paymentRepository;
private final PaymentGateway paymentGateway;
public PaymentService(PaymentRepository paymentRepository,
PaymentGateway paymentGateway) {
this.paymentRepository = paymentRepository;
this.paymentGateway = paymentGateway;
}
public Payment processPayment(String orderId, String customerId, BigDecimal amount) {
try {
// Simulate payment processing
PaymentGatewayResponse response = paymentGateway.charge(amount);
Payment payment = Payment.builder()
.paymentId(UUID.randomUUID().toString())
.orderId(orderId)
.customerId(customerId)
.amount(amount)
.status(PaymentStatus.COMPLETED)
.transactionId(response.getTransactionId())
.paymentDate(LocalDateTime.now())
.createdAt(LocalDateTime.now())
.build();
Payment saved = paymentRepository.save(payment);
log.info("Payment processed successfully for order: {}", orderId);
return saved;
} catch (PaymentGatewayException e) {
// Record failed payment attempt
Payment failedPayment = Payment.builder()
.paymentId(UUID.randomUUID().toString())
.orderId(orderId)
.customerId(customerId)
.amount(amount)
.status(PaymentStatus.FAILED)
.createdAt(LocalDateTime.now())
.build();
paymentRepository.save(failedPayment);
throw new PaymentException("Payment processing failed", e);
}
}
public void refundPayment(String orderId) {
paymentRepository.findByOrderId(orderId).ifPresent(payment -> {
if (payment.getStatus() == PaymentStatus.COMPLETED) {
try {
paymentGateway.refund(payment.getTransactionId());
payment.setStatus(PaymentStatus.REFUNDED);
paymentRepository.save(payment);
log.info("Payment refunded for order: {}", orderId);
} catch (PaymentGatewayException e) {
log.error("Refund failed for order: {}", orderId, e);
throw new PaymentException("Refund failed", e);
}
}
});
}
}
@Service
@Slf4j
@Transactional
public class InventoryService {
private final InventoryRepository inventoryRepository;
private final InventoryReservationRepository reservationRepository;
public InventoryService(InventoryRepository inventoryRepository,
InventoryReservationRepository reservationRepository) {
this.inventoryRepository = inventoryRepository;
this.reservationRepository = reservationRepository;
}
public void reserveInventory(String orderId, List<OrderItem> items) {
for (OrderItem item : items) {
Inventory inventory = inventoryRepository.findById(item.getProductId())
.orElseThrow(() -> new InventoryNotFoundException(
"Product not found: " + item.getProductId()));
if (!inventory.canFulfill(item.getQuantity())) {
throw new InsufficientInventoryException(
"Insufficient inventory for product: " + item.getProductId());
}
// Reserve inventory
inventory.reserve(item.getQuantity());
inventoryRepository.save(inventory);
// Record reservation
InventoryReservation reservation = InventoryReservation.builder()
.reservationId(UUID.randomUUID().toString())
.orderId(orderId)
.productId(item.getProductId())
.quantity(item.getQuantity())
.status(ReservationStatus.RESERVED)
.createdAt(LocalDateTime.now())
.build();
reservationRepository.save(reservation);
log.info("Reserved {} units of product {} for order {}", 
item.getQuantity(), item.getProductId(), orderId);
}
}
public void releaseReservation(String orderId) {
List<InventoryReservation> reservations = 
reservationRepository.findByOrderIdAndStatus(orderId, ReservationStatus.RESERVED);
for (InventoryReservation reservation : reservations) {
inventoryRepository.findById(reservation.getProductId()).ifPresent(inventory -> {
inventory.release(reservation.getQuantity());
inventoryRepository.save(inventory);
});
reservation.setStatus(ReservationStatus.RELEASED);
reservationRepository.save(reservation);
log.info("Released reservation for product {} in order {}", 
reservation.getProductId(), orderId);
}
}
public void confirmReservation(String orderId) {
List<InventoryReservation> reservations = 
reservationRepository.findByOrderIdAndStatus(orderId, ReservationStatus.RESERVED);
for (InventoryReservation reservation : reservations) {
inventoryRepository.findById(reservation.getProductId()).ifPresent(inventory -> {
inventory.confirmReservation(reservation.getQuantity());
inventoryRepository.save(inventory);
});
reservation.setStatus(ReservationStatus.CONFIRMED);
reservationRepository.save(reservation);
}
log.info("Confirmed reservations for order: {}", orderId);
}
}

Outbox Pattern Implementation

1. Outbox Entity and Configuration

@Entity
@Table(name = "outbox_events")
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class OutboxEvent {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
@Column(nullable = false)
private String aggregateId;
@Column(nullable = false)
private String aggregateType;
@Column(nullable = false)
private String eventType;
@Column(nullable = false, columnDefinition = "TEXT")
private String payload;
@Enumerated(EnumType.STRING)
private OutboxStatus status;
private LocalDateTime createdAt;
private LocalDateTime processedAt;
@Version
private Long version;
@PrePersist
protected void onCreate() {
this.createdAt = LocalDateTime.now();
this.status = OutboxStatus.PENDING;
}
}
public enum OutboxStatus {
PENDING, PROCESSED, FAILED
}
@Configuration
@EnableScheduling
public class OutboxConfiguration {
@Bean
public OutboxEventProcessor outboxEventProcessor(
OutboxEventRepository repository,
ObjectMapper objectMapper,
ApplicationEventPublisher eventPublisher) {
return new OutboxEventProcessor(repository, objectMapper, eventPublisher);
}
}

2. Outbox Event Processor

@Component
@Slf4j
public class OutboxEventProcessor {
private final OutboxEventRepository repository;
private final ObjectMapper objectMapper;
private final ApplicationEventPublisher eventPublisher;
private static final int BATCH_SIZE = 100;
private static final Duration PROCESSING_TIMEOUT = Duration.ofMinutes(5);
public OutboxEventProcessor(OutboxEventRepository repository,
ObjectMapper objectMapper,
ApplicationEventPublisher eventPublisher) {
this.repository = repository;
this.objectMapper = objectMapper;
this.eventPublisher = eventPublisher;
}
@Scheduled(fixedDelay = 5000) // Process every 5 seconds
@Transactional
public void processOutboxEvents() {
LocalDateTime cutoffTime = LocalDateTime.now().minus(PROCESSING_TIMEOUT);
List<OutboxEvent> pendingEvents = repository.findByStatusAndCreatedAtBefore(
OutboxStatus.PENDING, cutoffTime, PageRequest.of(0, BATCH_SIZE));
for (OutboxEvent event : pendingEvents) {
try {
processEvent(event);
event.setStatus(OutboxStatus.PROCESSED);
event.setProcessedAt(LocalDateTime.now());
repository.save(event);
log.info("Processed outbox event: {} for aggregate: {}", 
event.getEventType(), event.getAggregateId());
} catch (Exception e) {
log.error("Failed to process outbox event: {}", event.getId(), e);
event.setStatus(OutboxStatus.FAILED);
repository.save(event);
}
}
}
private void processEvent(OutboxEvent event) throws Exception {
switch (event.getEventType()) {
case "ORDER_CREATED":
OrderCreatedEvent orderCreated = objectMapper.readValue(
event.getPayload(), OrderCreatedEvent.class);
eventPublisher.publishEvent(orderCreated);
break;
case "ORDER_CONFIRMED":
OrderConfirmedEvent orderConfirmed = objectMapper.readValue(
event.getPayload(), OrderConfirmedEvent.class);
eventPublisher.publishEvent(orderConfirmed);
break;
case "ORDER_CANCELLED":
OrderCancelledEvent orderCancelled = objectMapper.readValue(
event.getPayload(), OrderCancelledEvent.class);
eventPublisher.publishEvent(orderCancelled);
break;
default:
log.warn("Unknown event type: {}", event.getEventType());
}
}
}
@Component
@Transactional
public class OutboxEventService {
private final OutboxEventRepository repository;
private final ObjectMapper objectMapper;
public OutboxEventService(OutboxEventRepository repository,
ObjectMapper objectMapper) {
this.repository = repository;
this.objectMapper = objectMapper;
}
public void saveEvent(String aggregateId, String aggregateType, 
String eventType, Object event) {
try {
String payload = objectMapper.writeValueAsString(event);
OutboxEvent outboxEvent = OutboxEvent.builder()
.aggregateId(aggregateId)
.aggregateType(aggregateType)
.eventType(eventType)
.payload(payload)
.build();
repository.save(outboxEvent);
log.debug("Saved outbox event: {} for aggregate: {}", eventType, aggregateId);
} catch (JsonProcessingException e) {
throw new RuntimeException("Failed to serialize event payload", e);
}
}
}

Event Sourcing Implementation

1. Event Store and Aggregates

@Entity
@Table(name = "event_store")
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class EventStore {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
@Column(nullable = false)
private String aggregateId;
@Column(nullable = false)
private String aggregateType;
@Column(nullable = false)
private String eventType;
@Column(nullable = false, columnDefinition = "TEXT")
private String eventData;
@Column(nullable = false)
private Long version;
@Column(nullable = false)
private LocalDateTime timestamp;
@Column(nullable = false)
private String metadata;
}
@Component
@Slf4j
public class OrderAggregate {
private String orderId;
private OrderStatus status;
private String customerId;
private BigDecimal totalAmount;
private List<OrderItem> items;
private Long version;
private final List<Object> pendingEvents = new ArrayList<>();
public OrderAggregate() {}
public static OrderAggregate create(String orderId, String customerId, 
List<OrderItem> items, BigDecimal totalAmount) {
OrderAggregate aggregate = new OrderAggregate();
aggregate.apply(new OrderCreatedEvent(orderId, customerId, items, totalAmount));
return aggregate;
}
public void confirm() {
if (this.status != OrderStatus.PENDING) {
throw new IllegalStateException("Order cannot be confirmed from state: " + status);
}
apply(new OrderConfirmedEvent(orderId));
}
public void cancel(String reason) {
apply(new OrderCancelledEvent(orderId, reason));
}
private void apply(OrderCreatedEvent event) {
this.orderId = event.getOrderId();
this.customerId = event.getCustomerId();
this.items = event.getItems();
this.totalAmount = event.getTotalAmount();
this.status = OrderStatus.PENDING;
this.version = 0L;
this.pendingEvents.add(event);
}
private void apply(OrderConfirmedEvent event) {
this.status = OrderStatus.CONFIRMED;
this.version++;
this.pendingEvents.add(event);
}
private void apply(OrderCancelledEvent event) {
this.status = OrderStatus.CANCELLED;
this.version++;
this.pendingEvents.add(event);
}
public List<Object> getPendingEvents() {
return new ArrayList<>(pendingEvents);
}
public void clearPendingEvents() {
pendingEvents.clear();
}
// Rehydrate aggregate from events
public static OrderAggregate rehydrate(List<EventStore> events) {
OrderAggregate aggregate = new OrderAggregate();
for (EventStore event : events) {
aggregate.applyEvent(event);
}
return aggregate;
}
private void applyEvent(EventStore event) {
// Deserialize and apply event based on eventType
// Implementation depends on your event deserialization strategy
}
}
@Service
@Slf4j
@Transactional
public class EventSourcingService {
private final EventStoreRepository eventStoreRepository;
private final ObjectMapper objectMapper;
public EventSourcingService(EventStoreRepository eventStoreRepository,
ObjectMapper objectMapper) {
this.eventStoreRepository = eventStoreRepository;
this.objectMapper = objectMapper;
}
public void saveEvents(String aggregateId, String aggregateType, 
List<Object> events, Long currentVersion) {
long expectedVersion = currentVersion != null ? currentVersion : -1;
for (Object event : events) {
expectedVersion++;
try {
String eventType = event.getClass().getSimpleName();
String eventData = objectMapper.writeValueAsString(event);
String metadata = "{}"; // Could include correlation ID, user ID, etc.
EventStore eventStore = EventStore.builder()
.aggregateId(aggregateId)
.aggregateType(aggregateType)
.eventType(eventType)
.eventData(eventData)
.version(expectedVersion)
.timestamp(LocalDateTime.now())
.metadata(metadata)
.build();
eventStoreRepository.save(eventStore);
log.debug("Saved event: {} for aggregate: {} version: {}", 
eventType, aggregateId, expectedVersion);
} catch (JsonProcessingException e) {
throw new RuntimeException("Failed to serialize event", e);
}
}
}
public List<EventStore> getEventsForAggregate(String aggregateId) {
return eventStoreRepository.findByAggregateIdOrderByVersionAsc(aggregateId);
}
public Optional<OrderAggregate> getOrderAggregate(String orderId) {
List<EventStore> events = getEventsForAggregate(orderId);
if (events.isEmpty()) {
return Optional.empty();
}
OrderAggregate aggregate = OrderAggregate.rehydrate(events);
return Optional.of(aggregate);
}
}

Message Broker Integration

1. RabbitMQ Configuration

@Configuration
public class RabbitMQConfig {
public static final String ORDER_EVENTS_EXCHANGE = "order.events.exchange";
public static final String ORDER_CREATED_QUEUE = "order.created.queue";
public static final String ORDER_CONFIRMED_QUEUE = "order.confirmed.queue";
public static final String ORDER_CANCELLED_QUEUE = "order.cancelled.queue";
@Bean
public TopicExchange orderEventsExchange() {
return new TopicExchange(ORDER_EVENTS_EXCHANGE);
}
@Bean
public Queue orderCreatedQueue() {
return new Queue(ORDER_CREATED_QUEUE, true);
}
@Bean
public Queue orderConfirmedQueue() {
return new Queue(ORDER_CONFIRMED_QUEUE, true);
}
@Bean
public Queue orderCancelledQueue() {
return new Queue(ORDER_CANCELLED_QUEUE, true);
}
@Bean
public Binding orderCreatedBinding(Queue orderCreatedQueue, TopicExchange orderEventsExchange) {
return BindingBuilder.bind(orderCreatedQueue)
.to(orderEventsExchange)
.with("order.created");
}
@Bean
public Binding orderConfirmedBinding(Queue orderConfirmedQueue, TopicExchange orderEventsExchange) {
return BindingBuilder.bind(orderConfirmedQueue)
.to(orderEventsExchange)
.with("order.confirmed");
}
@Bean
public Binding orderCancelledBinding(Queue orderCancelledQueue, TopicExchange orderEventsExchange) {
return BindingBuilder.bind(orderCancelledQueue)
.to(orderEventsExchange)
.with("order.cancelled");
}
@Bean
public Jackson2JsonMessageConverter jsonMessageConverter() {
return new Jackson2JsonMessageConverter();
}
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
template.setMessageConverter(jsonMessageConverter());
template.setChannelTransacted(true);
return template;
}
}

2. Event Publishers and Consumers

@Component
@Slf4j
public class OrderEventPublisher {
private final RabbitTemplate rabbitTemplate;
private final OutboxEventService outboxEventService;
public OrderEventPublisher(RabbitTemplate rabbitTemplate,
OutboxEventService outboxEventService) {
this.rabbitTemplate = rabbitTemplate;
this.outboxEventService = outboxEventService;
}
@EventListener
@Transactional
public void handleOrderCreated(OrderCreatedEvent event) {
// Save to outbox first
outboxEventService.saveEvent(
event.getOrderId(), 
"ORDER", 
"ORDER_CREATED", 
event
);
// Then publish to message broker (optional - could be done by outbox processor)
rabbitTemplate.convertAndSend(
RabbitMQConfig.ORDER_EVENTS_EXCHANGE,
"order.created",
event
);
log.info("Published OrderCreatedEvent for order: {}", event.getOrderId());
}
@EventListener
public void handleOrderConfirmed(OrderConfirmedEvent event) {
rabbitTemplate.convertAndSend(
RabbitMQConfig.ORDER_EVENTS_EXCHANGE,
"order.confirmed",
event
);
log.info("Published OrderConfirmedEvent for order: {}", event.getOrderId());
}
@EventListener
public void handleOrderCancelled(OrderCancelledEvent event) {
rabbitTemplate.convertAndSend(
RabbitMQConfig.ORDER_EVENTS_EXCHANGE,
"order.cancelled",
event
);
log.info("Published OrderCancelledEvent for order: {}", event.getOrderId());
}
}
@Component
@Slf4j
public class OrderEventConsumer {
private final InventoryService inventoryService;
private final NotificationService notificationService;
public OrderEventConsumer(InventoryService inventoryService,
NotificationService notificationService) {
this.inventoryService = inventoryService;
this.notificationService = notificationService;
}
@RabbitListener(queues = RabbitMQConfig.ORDER_CREATED_QUEUE)
public void handleOrderCreated(OrderCreatedEvent event) {
log.info("Received OrderCreatedEvent for order: {}", event.getOrderId());
try {
// Update read models, send notifications, etc.
notificationService.sendOrderConfirmation(event.getOrderId(), event.getCustomerId());
} catch (Exception e) {
log.error("Failed to process OrderCreatedEvent for order: {}", event.getOrderId(), e);
throw new AmqpRejectAndDontRequeueException("Processing failed", e);
}
}
@RabbitListener(queues = RabbitMQConfig.ORDER_CONFIRMED_QUEUE)
public void handleOrderConfirmed(OrderConfirmedEvent event) {
log.info("Received OrderConfirmedEvent for order: {}", event.getOrderId());
// Update analytics, send shipping notifications, etc.
notificationService.sendShippingNotification(event.getOrderId());
}
@RabbitListener(queues = RabbitMQConfig.ORDER_CANCELLED_QUEUE)
public void handleOrderCancelled(OrderCancelledEvent event) {
log.info("Received OrderCancelledEvent for order: {}", event.getOrderId());
// Handle cancellation in dependent services
notificationService.sendCancellationNotification(event.getOrderId(), event.getReason());
}
}

Resilience and Monitoring

1. Circuit Breaker and Retry Configuration

@Configuration
public class ResilienceConfig {
@Bean
public Customizer<Resilience4JCircuitBreakerFactory> circuitBreakerFactoryCustomizer() {
return factory -> factory.configureDefault(id -> new Resilience4JConfigBuilder(id)
.circuitBreakerConfig(CircuitBreakerConfig.custom()
.failureRateThreshold(50)
.waitDurationInOpenState(Duration.ofSeconds(30))
.slidingWindowSize(10)
.build())
.timeLimiterConfig(TimeLimiterConfig.custom()
.timeoutDuration(Duration.ofSeconds(5))
.build())
.build());
}
@Bean
public RetryTemplate retryTemplate() {
return RetryTemplate.builder()
.maxAttempts(3)
.exponentialBackoff(1000, 2, 5000)
.retryOn(PaymentException.class)
.build();
}
}
@Service
@Slf4j
public class ResilientPaymentService {
private final PaymentService paymentService;
private final RetryTemplate retryTemplate;
private final CircuitBreaker circuitBreaker;
public ResilientPaymentService(PaymentService paymentService,
RetryTemplate retryTemplate,
CircuitBreakerRegistry circuitBreakerRegistry) {
this.paymentService = paymentService;
this.retryTemplate = retryTemplate;
this.circuitBreaker = circuitBreakerRegistry.circuitBreaker("paymentService");
}
public Payment processPaymentWithRetry(String orderId, String customerId, BigDecimal amount) {
return circuitBreaker.executeSupplier(() -> 
retryTemplate.execute(context -> {
log.info("Attempting payment processing for order: {}, attempt: {}", 
orderId, context.getRetryCount() + 1);
return paymentService.processPayment(orderId, customerId, amount);
})
);
}
}

2. Consistency Monitoring

@Component
@Slf4j
public class ConsistencyMonitor {
private final OrderRepository orderRepository;
private final PaymentRepository paymentRepository;
private final InventoryRepository inventoryRepository;
public ConsistencyMonitor(OrderRepository orderRepository,
PaymentRepository paymentRepository,
InventoryRepository inventoryRepository) {
this.orderRepository = orderRepository;
this.paymentRepository = paymentRepository;
this.inventoryRepository = inventoryRepository;
}
@Scheduled(fixedRate = 60000) // Check every minute
public void checkDataConsistency() {
log.info("Starting data consistency check");
checkOrderPaymentConsistency();
checkInventoryConsistency();
log.info("Completed data consistency check");
}
private void checkOrderPaymentConsistency() {
// Find orders that are confirmed but have no successful payment
List<Order> confirmedOrders = orderRepository.findByStatus(OrderStatus.CONFIRMED);
for (Order order : confirmedOrders) {
Optional<Payment> payment = paymentRepository.findByOrderId(order.getOrderId());
if (payment.isEmpty() || payment.get().getStatus() != PaymentStatus.COMPLETED) {
log.warn("Inconsistency detected: Order {} is confirmed but has no completed payment", 
order.getOrderId());
// Trigger compensation or alert
}
}
}
private void checkInventoryConsistency() {
// Verify that reserved quantities don't exceed available
List<Inventory> allInventory = inventoryRepository.findAll();
for (Inventory inventory : allInventory) {
if (inventory.getReservedQuantity() > inventory.getAvailableQuantity() + inventory.getReservedQuantity()) {
log.error("Critical inventory inconsistency detected for product: {}", 
inventory.getProductId());
// Trigger emergency correction
}
}
}
}

Testing Eventual Consistency

1. Integration Tests

@SpringBootTest
@Testcontainers
@ActiveProfiles("test")
class EventualConsistencyTest {
@Container
static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:15");
@Container
static RabbitMQContainer rabbitMQ = new RabbitMQContainer("rabbitmq:3.12-management");
@Autowired
private OrderSagaOrchestrator sagaOrchestrator;
@Autowired
private OrderRepository orderRepository;
@Autowired
private PaymentRepository paymentRepository;
@Autowired
private InventoryRepository inventoryRepository;
@DynamicPropertySource
static void configureProperties(DynamicPropertyRegistry registry) {
registry.add("spring.datasource.url", postgres::getJdbcUrl);
registry.add("spring.datasource.username", postgres::getUsername);
registry.add("spring.datasource.password", postgres::getPassword);
registry.add("spring.rabbitmq.host", rabbitMQ::getHost);
registry.add("spring.rabbitmq.port", rabbitMQ::getAmqpPort);
}
@Test
void testSuccessfulOrderSaga() throws InterruptedException {
// Given
CreateOrderCommand command = createTestOrderCommand();
// When
sagaOrchestrator.createOrderSaga(command);
// Then - wait for eventual consistency
await().atMost(10, TimeUnit.SECONDS).until(() -> 
orderRepository.findById(command.getOrderId())
.map(Order::getStatus)
.orElse(null) == OrderStatus.CONFIRMED
);
Optional<Order> order = orderRepository.findById(command.getOrderId());
assertTrue(order.isPresent());
assertEquals(OrderStatus.CONFIRMED, order.get().getStatus());
Optional<Payment> payment = paymentRepository.findByOrderId(command.getOrderId());
assertTrue(payment.isPresent());
assertEquals(PaymentStatus.COMPLETED, payment.get().getStatus());
}
@Test
void testOrderSagaWithPaymentFailure() throws InterruptedException {
// Given
CreateOrderCommand command = createTestOrderCommand();
// Mock payment service to throw exception
// When
sagaOrchestrator.createOrderSaga(command);
// Then - verify compensation
await().atMost(10, TimeUnit.SECONDS).until(() -> 
orderRepository.findById(command.getOrderId())
.map(Order::getStatus)
.orElse(null) == OrderStatus.CANCELLED
);
Optional<Order> order = orderRepository.findById(command.getOrderId());
assertTrue(order.isPresent());
assertEquals(OrderStatus.CANCELLED, order.get().getStatus());
// Verify inventory was released
// Verify payment was refunded if it was processed
}
private CreateOrderCommand createTestOrderCommand() {
return CreateOrderCommand.builder()
.orderId(UUID.randomUUID().toString())
.customerId("customer-123")
.items(List.of(
OrderItem.builder()
.productId("product-1")
.productName("Test Product")
.quantity(2)
.unitPrice(new BigDecimal("29.99"))
.build()
))
.totalAmount(new BigDecimal("59.98"))
.paymentMethod("CREDIT_CARD")
.build();
}
}

Best Practices and Monitoring

1. Application Configuration

# application.yml
spring:
datasource:
url: jdbc:postgresql://localhost:5432/microservices
username: postgres
password: password
jpa:
hibernate:
ddl-auto: validate
properties:
hibernate:
dialect: org.hibernate.dialect.PostgreSQLDialect
jdbc.batch_size: 50
order_inserts: true
order_updates: true
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
listener:
simple:
retry:
enabled: true
max-attempts: 3
initial-interval: 1000ms
cloud:
stream:
bindings:
orderCreated-out-0:
destination: order.events.exchange
group: order-service
orderConfirmed-out-0:
destination: order.events.exchange
group: order-service
resilience4j:
circuitbreaker:
instances:
paymentService:
failure-rate-threshold: 50
wait-duration-in-open-state: 30s
sliding-window-size: 10
retry:
instances:
paymentRetry:
max-attempts: 3
wait-duration: 1s
management:
endpoints:
web:
exposure:
include: health,metrics,circuitbreakers
endpoint:
health:
show-details: always

2. Monitoring and Observability

@Component
@Slf4j
public class ConsistencyMetrics {
private final MeterRegistry meterRegistry;
private final Counter successfulSagas;
private final Counter failedSagas;
private final Timer sagaExecutionTimer;
public ConsistencyMetrics(MeterRegistry meterRegistry) {
this.meterRegistry = meterRegistry;
this.successfulSagas = meterRegistry.counter("saga.completed", "outcome", "success");
this.failedSagas = meterRegistry.counter("saga.completed", "outcome", "failure");
this.sagaExecutionTimer = meterRegistry.timer("saga.execution.time");
}
public void recordSagaSuccess(String sagaType, Duration duration) {
successfulSagas.increment();
sagaExecutionTimer.record(duration);
log.info("Saga {} completed successfully in {} ms", sagaType, duration.toMillis());
}
public void recordSagaFailure(String sagaType, Duration duration, String reason) {
failedSagas.increment();
sagaExecutionTimer.record(duration);
log.warn("Saga {} failed after {} ms: {}", sagaType, duration.toMillis(), reason);
}
public void recordEventPublished(String eventType) {
meterRegistry.counter("event.published", "type", eventType).increment();
}
public void recordEventProcessed(String eventType) {
meterRegistry.counter("event.processed", "type", eventType).increment();
}
}

Conclusion

Eventual consistency in microservices requires careful design and implementation. Key takeaways:

Pattern Selection:

  • Saga Pattern: Best for complex business transactions requiring compensation
  • Outbox Pattern: Ensures reliable event publishing
  • Event Sourcing: Maintains complete audit trail and enables temporal queries
  • CQRS: Separates read and write concerns for scalability

Implementation Considerations:

  • Idempotency: Design operations to be safely retryable
  • Compensation: Implement rollback mechanisms for failed operations
  • Monitoring: Track consistency metrics and detect anomalies
  • Testing: Verify behavior under various failure scenarios

Best Practices:

  • Start simple: Begin with orchestrated sagas for complex workflows
  • Embrace asynchronous communication: Use message brokers for loose coupling
  • Implement retry and circuit breakers: Handle transient failures gracefully
  • Monitor data consistency: Regularly check for inconsistencies
  • Design for failure: Assume any service call can fail

Eventual consistency enables microservices to scale independently while maintaining data integrity across service boundaries. By combining these patterns with proper monitoring and testing, you can build robust, scalable systems that gracefully handle the complexities of distributed data management.

Category : JAVA

Leave a Reply

Your email address will not be published. Required fields are marked *


Macro Nepal Helper