diff --git a/pom.xml b/pom.xml index 4ed64e643219..a71630d289d3 100644 --- a/pom.xml +++ b/pom.xml @@ -244,6 +244,7 @@ tolerant-reader trampoline transaction-script + transactional-outbox twin type-object unit-of-work diff --git a/transactional-outbox/README.md b/transactional-outbox/README.md new file mode 100644 index 000000000000..2a2c8916c206 --- /dev/null +++ b/transactional-outbox/README.md @@ -0,0 +1,240 @@ +--- +title: "Transactional Outbox Pattern in Java: Ensuring Reliable Event Publishing" +shortTitle: Transactional Outbox +description: "Learn how to implement the Transactional Outbox pattern in Java using Spring Boot and H2. Master reliable event publishing and eliminate dual-write inconsistencies in microservices." +category: Architectural +language: en +tag: + - Spring Boot + - Microservices + - Event-Driven + - Messaging + - Persistence +--- + +## Also known as + +* Outbox Pattern +* Application Event Outbox +* Transactional Event Outbox + +## Intent of Transactional Outbox Pattern + +The Transactional Outbox pattern reliably publishes events in microservices architectures without requiring distributed transactions (XA/2PC). By persisting business data and event notifications in the same database transaction, it guarantees that message publishing always stays consistent with database changes. + +## Detailed Explanation of the Pattern with Real-World Examples + +### Real-world analogy + +> Imagine writing an important contract and placing the outgoing notice into a postal outbox tray located right next to your desk in a single action. Even if the mail courier arrives later, the document is securely staged in the outbox tray and cannot be lost. A dedicated mail clerk periodically inspects the outbox tray and delivers the letters to the post office. + +### In plain words + +> Instead of updating the database and publishing a message directly to a message broker in two separate network calls, a service writes both the business entity and an outbox event into the database within a single database transaction. A separate background process periodically reads pending outbox events and publishes them to the message broker. + +### Architecture Flow + +``` ++-------------------------------------------------------------+ +| Service Boundary | +| | +| +--------------------+ +------------------------+ | +| | Order Service | | Outbox Publisher | | +| +--------------------+ +------------------------+ | +| | | (Polls) | +| (Atomic Transaction) v | +| | +------------------------+ | +| +------------------> | Outbox Table (Pending) | | +| | +------------------------+ | +| v | (Publishes) | +| +--------------------+ v | +| | Orders Table | +------------------------+ | +| +--------------------+ | Message Broker | | +| +------------------------+ | ++-------------------------------------------------------------+ +``` + +## Programmatic Example (Spring Boot) + +### Order & Outbox Entities + +The `Order` entity represents business data, while `OutboxEvent` represents the event payload staged for asynchronous publishing. + +```java +@Entity +@Table(name = "orders") +@Data +@Builder +public class Order { + @Id + @GeneratedValue(strategy = GenerationType.IDENTITY) + private Long id; + private String customerName; + private String productName; + private double amount; + @Enumerated(EnumType.STRING) + private OrderStatus status; +} + +@Entity +@Table(name = "outbox_events") +@Data +@Builder +public class OutboxEvent { + @Id + @GeneratedValue(strategy = GenerationType.IDENTITY) + private Long id; + private String aggregateType; + private String aggregateId; + private String eventType; + private String payload; + @Enumerated(EnumType.STRING) + private EventStatus status; + private LocalDateTime createdAt; + private LocalDateTime processedAt; +} +``` + +### Atomic Transactional Write (`OrderService`) + +The service saves the order entity and creates an outbox event in the same transactional context using Spring's `@Transactional`. + +```java +@Service +@RequiredArgsConstructor +public class OrderService { + + private final OrderRepository orderRepository; + private final OutboxRepository outboxRepository; + + @Transactional + public Order createOrder(String customerName, String productName, double amount) { + var order = Order.builder() + .customerName(customerName) + .productName(productName) + .amount(amount) + .status(OrderStatus.CREATED) + .createdAt(LocalDateTime.now()) + .build(); + + var savedOrder = orderRepository.save(order); + + var outboxEvent = OutboxEvent.builder() + .aggregateType("Order") + .aggregateId(String.valueOf(savedOrder.getId())) + .eventType("ORDER_CREATED") + .payload(String.format("{\"orderId\":%d,\"amount\":%.2f}", savedOrder.getId(), amount)) + .status(EventStatus.PENDING) + .createdAt(LocalDateTime.now()) + .build(); + + outboxRepository.save(outboxEvent); + return savedOrder; + } +} +``` + +### Background Polling Publisher (`OutboxPublisher`) + +A scheduled background process polls `PENDING` outbox events, publishes them to the message broker, and marks their status as `PROCESSED`. + +```java +@Component +@RequiredArgsConstructor +public class OutboxPublisher { + + private final OutboxRepository outboxRepository; + private final MessageBroker messageBroker; + + @Scheduled(fixedDelay = 5000) + @Transactional + public void processOutboxEvents() { + List pendingEvents = outboxRepository.findByStatus(EventStatus.PENDING); + for (OutboxEvent event : pendingEvents) { + messageBroker.publish("order-events", event.getPayload()); + event.setStatus(EventStatus.PROCESSED); + event.setProcessedAt(LocalDateTime.now()); + outboxRepository.save(event); + } + } +} +``` + +## Class Diagram + +```mermaid +classDiagram + class Order { + +Long id + +String customerName + +String productName + +double amount + +OrderStatus status + } + class OutboxEvent { + +Long id + +String aggregateType + +String aggregateId + +String eventType + +String payload + +EventStatus status + +LocalDateTime createdAt + +LocalDateTime processedAt + } + class OrderService { + +createOrder(customerName, productName, amount) Order + } + class OutboxPublisher { + +processOutboxEvents() List~OutboxEvent~ + } + class MessageBroker { + <> + +publish(topic, payload) + } + + OrderService ..> Order : creates + OrderService ..> OutboxEvent : creates + OutboxPublisher ..> OutboxEvent : polls & updates + OutboxPublisher --> MessageBroker : dispatches +``` + +## When to Use the Transactional Outbox Pattern + +Use this pattern when: + +* You need to update a database and publish messages to an event broker without data loss or inconsistent dual-writes. +* Distributed transactions (XA 2-phase commit) are not supported, perform poorly, or add unwanted complexity. +* You are building event-driven microservices requiring **at-least-once** event delivery guarantees. + +## Real-World Applications + +* E-commerce checkout systems emitting order creation events for billing and fulfillment services. +* Financial transaction processing services issuing audit log events alongside database updates. +* Microservices using Change Data Capture (CDC) like Debezium for database log mining outbox patterns. + +## Benefits and Trade-offs + +### Benefits + +* **No Dual-Write Inconsistency**: Prevents lost messages or phantom events caused by network/broker outages. +* **At-Least-Once Delivery**: Guarantees event delivery to message consumers. +* **No Distributed Transactions**: Avoids expensive and fragile XA/2PC transactions across services. + +### Trade-Offs + +* **Near Real-time Latency**: Polling intervals add slight delay before events are dispatched. +* **Duplicate Message Handling**: Consumers must implement idempotent processing to handle potential message redeliveries. +* **Outbox Table Cleanup**: Outbox entries must be periodically archived or purged to prevent uncontrolled table growth. + +## Related Java Design Patterns + +* [Polling Publisher](https://java-design-patterns.com/patterns/polling-publisher/) +* [Saga Pattern](https://java-design-patterns.com/patterns/saga/) +* [Idempotent Consumer](https://java-design-patterns.com/patterns/microservices-idempotent-consumer/) +* [Event-Driven Architecture](https://java-design-patterns.com/patterns/event-driven-architecture/) + +## References and Credits + +* [Microservices.io - Pattern: Transactional Outbox](https://microservices.io/patterns/data/transactional-outbox.html) +* [Debezium - Reliable Microservices Data Exchange With the Outbox Pattern](https://debezium.io/blog/2019/02/19/reliable-microservices-data-exchange-with-outbox-pattern/) +* [Designing Data-Intensive Applications - Martin Kleppmann](https://dataintensive.net/) diff --git a/transactional-outbox/pom.xml b/transactional-outbox/pom.xml new file mode 100644 index 000000000000..c8c178d60739 --- /dev/null +++ b/transactional-outbox/pom.xml @@ -0,0 +1,116 @@ + + + + 4.0.0 + + com.iluwatar + java-design-patterns + 1.26.0-SNAPSHOT + + + transactional-outbox + + + + + org.springframework.boot + spring-boot-starter-data-jpa + + + + + org.springframework.boot + spring-boot-starter-web + + + + + org.projectlombok + lombok + ${lombok.version} + provided + + + + + com.h2database + h2 + runtime + + + + + org.springframework.boot + spring-boot-starter-test + test + + + + + org.junit.jupiter + junit-jupiter-engine + test + + + + + org.mockito + mockito-core + test + + + + + + + org.apache.maven.plugins + maven-assembly-plugin + + + package + + single + + + + jar-with-dependencies + + + + com.iluwatar.transactionaloutbox.App + + + + + + + + + diff --git a/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/App.java b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/App.java new file mode 100644 index 000000000000..65f851a7e7e7 --- /dev/null +++ b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/App.java @@ -0,0 +1,83 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +package com.iluwatar.transactionaloutbox; + +import lombok.RequiredArgsConstructor; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.boot.CommandLineRunner; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.scheduling.annotation.EnableScheduling; + +/** + * Transactional Outbox Pattern Demonstration Application. + * + *

The Transactional Outbox pattern ensures reliable event publishing in microservices by + * persisting events into a database outbox table within the same database transaction as the + * business entity update. A background polling process then dispatches pending outbox events to a + * message broker. + */ +@EnableScheduling +@SpringBootApplication +@RequiredArgsConstructor +public class App implements CommandLineRunner { + + private static final Logger LOGGER = LoggerFactory.getLogger(App.class); + + private final OrderService orderService; + private final OutboxPublisher outboxPublisher; + private final MessageConsumer messageConsumer; + + /** + * Main entry point for the Spring Boot Application. + * + * @param args command-line arguments + */ + public static void main(String[] args) { + SpringApplication.run(App.class, args); + } + + @Override + public void run(String... args) { + LOGGER.info("Starting Transactional Outbox Pattern demonstration..."); + + // 1. Create order and outbox record in a single transaction + var order1 = orderService.createOrder("Alice", "Laptop", 1200.00); + var order2 = orderService.createOrder("Bob", "Headphones", 150.00); + + LOGGER.info("Created orders with IDs: [{}], [{}]", order1.getId(), order2.getId()); + + // 2. Poll and publish outbox events to message broker + LOGGER.info("Triggering OutboxPublisher to process pending events..."); + outboxPublisher.processOutboxEvents(); + + // 3. Inspect consumed events + LOGGER.info( + "Total messages consumed by MessageConsumer: {}", + messageConsumer.getConsumedMessages().size()); + } +} diff --git a/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/EventStatus.java b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/EventStatus.java new file mode 100644 index 000000000000..df146f2e2a89 --- /dev/null +++ b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/EventStatus.java @@ -0,0 +1,33 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +package com.iluwatar.transactionaloutbox; + +/** Enum representing the status of an outbox event. */ +public enum EventStatus { + PENDING, + PROCESSED, + FAILED +} diff --git a/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/MessageBroker.java b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/MessageBroker.java new file mode 100644 index 000000000000..d37cc4b13f70 --- /dev/null +++ b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/MessageBroker.java @@ -0,0 +1,38 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +package com.iluwatar.transactionaloutbox; + +/** Interface for message broker publishing. */ +public interface MessageBroker { + + /** + * Publishes an event message to a topic/queue. + * + * @param topic destination topic name + * @param payload event message payload + */ + void publish(String topic, String payload); +} diff --git a/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/MessageConsumer.java b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/MessageConsumer.java new file mode 100644 index 000000000000..fd0243d58aa9 --- /dev/null +++ b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/MessageConsumer.java @@ -0,0 +1,62 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +package com.iluwatar.transactionaloutbox; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; + +/** Message Consumer/Broker implementation demonstrating event receipt. */ +@Service +public class MessageConsumer implements MessageBroker { + + private static final Logger LOGGER = LoggerFactory.getLogger(MessageConsumer.class); + + private final List consumedMessages = Collections.synchronizedList(new ArrayList<>()); + + @Override + public void publish(String topic, String payload) { + LOGGER.info("MessageBroker received message on topic [{}]: {}", topic, payload); + consumedMessages.add(payload); + } + + /** + * Retrieves unmodifiable list of consumed messages. + * + * @return list of payloads received by the broker + */ + public List getConsumedMessages() { + return Collections.unmodifiableList(consumedMessages); + } + + /** Clears accumulated messages. */ + public void clearMessages() { + consumedMessages.clear(); + } +} diff --git a/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/Order.java b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/Order.java new file mode 100644 index 000000000000..b446a9f1f3f6 --- /dev/null +++ b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/Order.java @@ -0,0 +1,64 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +package com.iluwatar.transactionaloutbox; + +import jakarta.persistence.Entity; +import jakarta.persistence.EnumType; +import jakarta.persistence.Enumerated; +import jakarta.persistence.GeneratedValue; +import jakarta.persistence.GenerationType; +import jakarta.persistence.Id; +import jakarta.persistence.Table; +import java.time.LocalDateTime; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** Represents a business Order entity persisted in the database. */ +@Entity +@Table(name = "orders") +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class Order { + + @Id + @GeneratedValue(strategy = GenerationType.IDENTITY) + private Long id; + + private String customerName; + + private String productName; + + private double amount; + + @Enumerated(EnumType.STRING) + private OrderStatus status; + + private LocalDateTime createdAt; +} diff --git a/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OrderRepository.java b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OrderRepository.java new file mode 100644 index 000000000000..4a527743c9bf --- /dev/null +++ b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OrderRepository.java @@ -0,0 +1,33 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +package com.iluwatar.transactionaloutbox; + +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.stereotype.Repository; + +/** Spring Data JPA Repository interface for Order entity. */ +@Repository +public interface OrderRepository extends JpaRepository {} diff --git a/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OrderService.java b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OrderService.java new file mode 100644 index 000000000000..496302eab19c --- /dev/null +++ b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OrderService.java @@ -0,0 +1,89 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +package com.iluwatar.transactionaloutbox; + +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import lombok.RequiredArgsConstructor; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +/** Service responsible for managing orders and writing outbox events atomically. */ +@Service +@RequiredArgsConstructor +public class OrderService { + + private static final Logger LOGGER = LoggerFactory.getLogger(OrderService.class); + + private final OrderRepository orderRepository; + private final OutboxRepository outboxRepository; + + /** + * Creates a new order and inserts an OutboxEvent into the database atomically within a single + * transaction boundary. + * + * @param customerName name of the customer + * @param productName name of the product + * @param amount purchase amount + * @return persisted Order entity + */ + @Transactional + public Order createOrder(String customerName, String productName, double amount) { + var now = LocalDateTime.now(ZoneOffset.UTC); + + var order = + Order.builder() + .customerName(customerName) + .productName(productName) + .amount(amount) + .status(OrderStatus.CREATED) + .createdAt(now) + .build(); + + var savedOrder = orderRepository.save(order); + LOGGER.info("Saved order with ID [{}] in database", savedOrder.getId()); + + var outboxEvent = + OutboxEvent.builder() + .aggregateType("Order") + .aggregateId(String.valueOf(savedOrder.getId())) + .eventType("ORDER_CREATED") + .payload( + String.format( + "{\"orderId\":%d,\"customerName\":\"%s\",\"productName\":\"%s\",\"amount\":%.2f}", + savedOrder.getId(), customerName, productName, amount)) + .status(EventStatus.PENDING) + .createdAt(now) + .build(); + + outboxRepository.save(outboxEvent); + LOGGER.info("Saved OutboxEvent for Order ID [{}] in database", savedOrder.getId()); + + return savedOrder; + } +} diff --git a/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OrderStatus.java b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OrderStatus.java new file mode 100644 index 000000000000..7f0b20d46b16 --- /dev/null +++ b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OrderStatus.java @@ -0,0 +1,34 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +package com.iluwatar.transactionaloutbox; + +/** Enum representing the status of an order. */ +public enum OrderStatus { + CREATED, + PROCESSING, + COMPLETED, + CANCELLED +} diff --git a/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OutboxEvent.java b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OutboxEvent.java new file mode 100644 index 000000000000..7e578ce385d9 --- /dev/null +++ b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OutboxEvent.java @@ -0,0 +1,71 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +package com.iluwatar.transactionaloutbox; + +import jakarta.persistence.Entity; +import jakarta.persistence.EnumType; +import jakarta.persistence.Enumerated; +import jakarta.persistence.GeneratedValue; +import jakarta.persistence.GenerationType; +import jakarta.persistence.Id; +import jakarta.persistence.Table; +import java.time.LocalDateTime; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * Represents an event entry stored in the outbox table within the same transaction as business + * data. + */ +@Entity +@Table(name = "outbox_events") +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class OutboxEvent { + + @Id + @GeneratedValue(strategy = GenerationType.IDENTITY) + private Long id; + + private String aggregateType; + + private String aggregateId; + + private String eventType; + + private String payload; + + @Enumerated(EnumType.STRING) + private EventStatus status; + + private LocalDateTime createdAt; + + private LocalDateTime processedAt; +} diff --git a/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OutboxPublisher.java b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OutboxPublisher.java new file mode 100644 index 000000000000..90196e2de0bf --- /dev/null +++ b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OutboxPublisher.java @@ -0,0 +1,88 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +package com.iluwatar.transactionaloutbox; + +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.util.List; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +/** Background publisher polling PENDING outbox events and dispatching to message broker. */ +@Component +public class OutboxPublisher { + + private static final Logger LOGGER = LoggerFactory.getLogger(OutboxPublisher.class); + + private final OutboxRepository outboxRepository; + private final MessageBroker messageBroker; + + public OutboxPublisher(OutboxRepository outboxRepository, MessageBroker messageBroker) { + this.outboxRepository = outboxRepository; + this.messageBroker = messageBroker; + } + + /** + * Periodically polls pending outbox events from database and dispatches them to the message + * broker. + */ + @Scheduled(fixedDelay = 5000) + public void publishOutboxEvents() { + processOutboxEvents(); + } + + /** + * Process pending outbox events from the database and dispatch them to the message broker. + * + * @return list of processed outbox events + */ + public List processOutboxEvents() { + List pendingEvents = outboxRepository.findByStatus(EventStatus.PENDING); + if (pendingEvents.isEmpty()) { + return pendingEvents; + } + + LOGGER.info("Found [{}] PENDING outbox events to publish", pendingEvents.size()); + + for (OutboxEvent event : pendingEvents) { + try { + messageBroker.publish("order-events", event.getPayload()); + event.setStatus(EventStatus.PROCESSED); + event.setProcessedAt(LocalDateTime.now(ZoneOffset.UTC)); + outboxRepository.save(event); + LOGGER.info("Successfully published outbox event ID [{}]", event.getId()); + } catch (Exception e) { + LOGGER.error("Failed to publish outbox event ID [{}]", event.getId(), e); + event.setStatus(EventStatus.FAILED); + outboxRepository.save(event); + } + } + + return pendingEvents; + } +} diff --git a/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OutboxRepository.java b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OutboxRepository.java new file mode 100644 index 000000000000..dcf409d5f167 --- /dev/null +++ b/transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/OutboxRepository.java @@ -0,0 +1,43 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +package com.iluwatar.transactionaloutbox; + +import java.util.List; +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.stereotype.Repository; + +/** Spring Data JPA Repository interface for OutboxEvent entity. */ +@Repository +public interface OutboxRepository extends JpaRepository { + + /** + * Finds all outbox events matching the given status. + * + * @param status event processing status + * @return list of matching OutboxEvent records + */ + List findByStatus(EventStatus status); +} diff --git a/transactional-outbox/src/test/java/com/iluwatar/transactionaloutbox/AppTest.java b/transactional-outbox/src/test/java/com/iluwatar/transactionaloutbox/AppTest.java new file mode 100644 index 000000000000..35cceb7fa173 --- /dev/null +++ b/transactional-outbox/src/test/java/com/iluwatar/transactionaloutbox/AppTest.java @@ -0,0 +1,40 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +package com.iluwatar.transactionaloutbox; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; + +import org.junit.jupiter.api.Test; +import org.springframework.boot.test.context.SpringBootTest; + +@SpringBootTest +class AppTest { + + @Test + void testAppMainExecutesWithoutErrors() { + assertDoesNotThrow(() -> App.main(new String[] {})); + } +} diff --git a/transactional-outbox/src/test/java/com/iluwatar/transactionaloutbox/OrderServiceTest.java b/transactional-outbox/src/test/java/com/iluwatar/transactionaloutbox/OrderServiceTest.java new file mode 100644 index 000000000000..f7b530ba50bb --- /dev/null +++ b/transactional-outbox/src/test/java/com/iluwatar/transactionaloutbox/OrderServiceTest.java @@ -0,0 +1,72 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +package com.iluwatar.transactionaloutbox; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import java.util.List; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; + +@SpringBootTest +class OrderServiceTest { + + @Autowired private OrderService orderService; + + @Autowired private OrderRepository orderRepository; + + @Autowired private OutboxRepository outboxRepository; + + @BeforeEach + void setUp() { + outboxRepository.deleteAll(); + orderRepository.deleteAll(); + } + + @Test + void testCreateOrderSavesOrderAndOutboxEventAtomically() { + Order order = orderService.createOrder("John Doe", "Smartphone", 799.99); + + assertNotNull(order.getId()); + assertEquals("John Doe", order.getCustomerName()); + assertEquals(OrderStatus.CREATED, order.getStatus()); + + List orders = orderRepository.findAll(); + assertEquals(1, orders.size()); + + List outboxEvents = outboxRepository.findByStatus(EventStatus.PENDING); + assertEquals(1, outboxEvents.size()); + + OutboxEvent event = outboxEvents.get(0); + assertEquals("Order", event.getAggregateType()); + assertEquals(String.valueOf(order.getId()), event.getAggregateId()); + assertEquals("ORDER_CREATED", event.getEventType()); + assertEquals(EventStatus.PENDING, event.getStatus()); + } +} diff --git a/transactional-outbox/src/test/java/com/iluwatar/transactionaloutbox/OutboxPublisherTest.java b/transactional-outbox/src/test/java/com/iluwatar/transactionaloutbox/OutboxPublisherTest.java new file mode 100644 index 000000000000..234dd6032195 --- /dev/null +++ b/transactional-outbox/src/test/java/com/iluwatar/transactionaloutbox/OutboxPublisherTest.java @@ -0,0 +1,80 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +package com.iluwatar.transactionaloutbox; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.List; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; + +@SpringBootTest +class OutboxPublisherTest { + + @Autowired private OrderService orderService; + + @Autowired private OutboxPublisher outboxPublisher; + + @Autowired private OutboxRepository outboxRepository; + + @Autowired private MessageConsumer messageConsumer; + + @BeforeEach + void setUp() { + outboxRepository.deleteAll(); + messageConsumer.clearMessages(); + } + + @Test + void testProcessOutboxEventsWhenNoPendingEvents() { + List processed = outboxPublisher.processOutboxEvents(); + assertTrue(processed.isEmpty()); + } + + @Test + void testProcessOutboxEventsPublishesAndUpdatesStatusToProcessed() { + orderService.createOrder("Jane Smith", "Tablet", 499.00); + + List pendingBefore = outboxRepository.findByStatus(EventStatus.PENDING); + assertEquals(1, pendingBefore.size()); + + List processedEvents = outboxPublisher.processOutboxEvents(); + assertEquals(1, processedEvents.size()); + + List pendingAfter = outboxRepository.findByStatus(EventStatus.PENDING); + assertEquals(0, pendingAfter.size()); + + List processedAfter = outboxRepository.findByStatus(EventStatus.PROCESSED); + assertEquals(1, processedAfter.size()); + assertNotNull(processedAfter.get(0).getProcessedAt()); + + assertEquals(1, messageConsumer.getConsumedMessages().size()); + } +} diff --git a/transactional-outbox/src/test/java/com/iluwatar/transactionaloutbox/OutboxPublisherUnitTest.java b/transactional-outbox/src/test/java/com/iluwatar/transactionaloutbox/OutboxPublisherUnitTest.java new file mode 100644 index 000000000000..4f4cb891459a --- /dev/null +++ b/transactional-outbox/src/test/java/com/iluwatar/transactionaloutbox/OutboxPublisherUnitTest.java @@ -0,0 +1,66 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +package com.iluwatar.transactionaloutbox; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.List; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +@ExtendWith(MockitoExtension.class) +class OutboxPublisherUnitTest { + + @Mock private OutboxRepository outboxRepository; + + @Mock private MessageBroker messageBroker; + + @InjectMocks private OutboxPublisher outboxPublisher; + + @Test + void testProcessOutboxEventsHandlesBrokerException() { + OutboxEvent event = + OutboxEvent.builder().id(1L).payload("test payload").status(EventStatus.PENDING).build(); + + when(outboxRepository.findByStatus(EventStatus.PENDING)).thenReturn(List.of(event)); + doThrow(new RuntimeException("Broker failure")) + .when(messageBroker) + .publish(anyString(), anyString()); + + List result = outboxPublisher.processOutboxEvents(); + + assertEquals(1, result.size()); + assertEquals(EventStatus.FAILED, event.getStatus()); + verify(outboxRepository).save(event); + } +}