feat: Implement Transactional Outbox pattern

Resolves #2671

Author: tanya-johari

microservices-transactional-outbox/pom.xml

@@ -0,0 +1,100 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>com.iluwatar</groupId>
+        <artifactId>java-design-patterns</artifactId>
+        <version>1.26.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>microservices-transactional-outbox</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>jakarta.persistence</groupId>
+            <artifactId>jakarta.persistence-api</artifactId>
+            <version>3.1.0</version>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.hibernate.orm</groupId>
+            <artifactId>hibernate-core</artifactId>
+            <version>6.4.4.Final</version>
+        </dependency>
+        <dependency>
+            <groupId>com.h2database</groupId>
+            <artifactId>h2</artifactId>
+            <version>2.2.224</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <version>2.17.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
+            <version>5.16.1</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <configuration>
+                            <archive>
+                                <manifest>
+                                    <mainClass>com.iluwatar.transactionaloutbox.App</mainClass>
+                                </manifest>
+                            </archive>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

microservices-transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/App.java

@@ -0,0 +1,57 @@
+/*
+ * 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.Persistence;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class App {
+
+  public static void main(String[] args) throws Exception {
+    var entityManagerFactory = Persistence.createEntityManagerFactory("transactional-outbox-pu");
+    var entityManager = entityManagerFactory.createEntityManager();
+
+    var customerService = new CustomerService(entityManager);
+    var messageBroker = new MessageBroker();
+    var eventPoller = new EventPoller(entityManager, messageBroker);
+
+    eventPoller.start();
+
+    LOGGER.info("Running simulation...");
+    Thread.sleep(1000);
+
+    customerService.createCustomer("john.doe");
+    Thread.sleep(5000);
+
+    customerService.createCustomer("jane.doe");
+    Thread.sleep(5000);
+
+    eventPoller.stop();
+    entityManager.close();
+    entityManagerFactory.close();
+    LOGGER.info("Simulation finished.");
+  }
+}

microservices-transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/Customer.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 jakarta.persistence.Entity;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.Id;
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+@Entity
+public class Customer {
+
+  private final String username;
+  @Setter @Id @GeneratedValue private Integer id;
+
+  public Customer(String username) {
+    this.username = username;
+  }
+}

microservices-transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/CustomerService.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 com.fasterxml.jackson.databind.ObjectMapper;
+import jakarta.persistence.EntityManager;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class CustomerService {
+
+  private final EntityManager entityManager;
+  private final OutboxRepository outboxRepository;
+  private final ObjectMapper objectMapper = new ObjectMapper();
+
+  public CustomerService(EntityManager entityManager) {
+    this.entityManager = entityManager;
+    this.outboxRepository = new OutboxRepository(entityManager);
+  }
+
+  public void createCustomer(String username) throws Exception {
+    entityManager.getTransaction().begin();
+    try {
+      var customer = new Customer(username);
+      entityManager.persist(customer);
+
+      String payload = objectMapper.writeValueAsString(customer);
+      var event = new OutboxEvent("CUSTOMER_CREATED", payload);
+      outboxRepository.save(event);
+
+      entityManager.getTransaction().commit();
+      LOGGER.info("SUCCESS: Customer and OutboxEvent saved transactionally.");
+
+    } catch (Exception e) {
+      entityManager.getTransaction().rollback();
+      LOGGER.error("ERROR: Transaction rolled back.");
+      throw e;
+    }
+  }
+}

microservices-transactional-outbox/src/main/java/com/iluwatar/transactionaloutbox/EventPoller.java

@@ -0,0 +1,122 @@
+/*
+ * 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.EntityManager;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import lombok.Getter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class EventPoller {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(EventPoller.class);
+  private static final int MAX_RETRIES = 3;
+  private static final long RETRY_DELAY_MS = 1000;
+  private static final java.security.SecureRandom RANDOM = new java.security.SecureRandom();
+
+  @Getter
+  private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
+
+  private final EntityManager entityManager;
+  private final OutboxRepository outboxRepository;
+  private final MessageBroker messageBroker;
+  @Getter private long processedEventsCount = 0;
+  @Getter private long failedEventsCount = 0;
+
+  public EventPoller(EntityManager entityManager, MessageBroker messageBroker) {
+    this.entityManager = entityManager;
+    this.outboxRepository = new OutboxRepository(entityManager);
+    this.messageBroker = messageBroker;
+  }
+
+  public void start() {
+    scheduler.scheduleAtFixedRate(this::processOutboxEvents, 0, 2, TimeUnit.SECONDS);
+    LOGGER.info("EventPoller started.");
+  }
+
+  public void stop() {
+    scheduler.shutdown();
+    LOGGER.info(
+        "EventPoller stopped with {} events processed and {} failures.",
+        processedEventsCount,
+        failedEventsCount);
+  }
+
+  void processOutboxEvents() {
+    LOGGER.info("Polling for unprocessed events...");
+    entityManager.getTransaction().begin();
+    try {
+      List<OutboxEvent> events = outboxRepository.findUnprocessedEvents();
+      if (events.isEmpty()) {
+        LOGGER.info("No new events found.");
+      } else {
+        LOGGER.info("Found {} events to process.", events.size());
+        for (var event : events) {
+          processEventWithRetry(event);
+        }
+      }
+      entityManager.getTransaction().commit();
+    } catch (Exception e) {
+      LOGGER.error("Error processing outbox events, rolling back transaction.", e);
+      entityManager.getTransaction().rollback();
+      failedEventsCount++;
+    }
+  }
+
+  private void processEventWithRetry(OutboxEvent event) {
+    int retryCount = 0;
+    while (retryCount < MAX_RETRIES) {
+      try {
+        messageBroker.sendMessage(event);
+        outboxRepository.markAsProcessed(event);
+        processedEventsCount++;
+        LOGGER.info("Successfully processed event.");
+        return;
+      } catch (Exception e) {
+        retryCount++;
+        if (retryCount < MAX_RETRIES) {
+          LOGGER.warn(
+              "Failed to process event (attempt {}/{}). Retrying...", retryCount, MAX_RETRIES);
+          try {
+            long sleepTime =
+                Math.min(RETRY_DELAY_MS * (1L << (retryCount - 1)) + RANDOM.nextLong(100), 10000L);
+            Thread.sleep(sleepTime);
+          } catch (InterruptedException ie) {
+            Thread.currentThread().interrupt();
+            break;
+          }
+        } else {
+          LOGGER.error("Failed to process event after {} attempts", MAX_RETRIES);
+          failedEventsCount++;
+        }
+      }
+    }
+  }
+}