Consistency testing

.gitignore

@@ -9,6 +9,8 @@ dump.rdb
 *.war
 *.ear
 
+.DS_Store
+
 # virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
 hs_err_pid*
 

.travis.yml

@@ -2,6 +2,9 @@ language: java
 jdk:
 - oraclejdk8
 services:
-- rabbitmq
 - redis
+- docker
 install: mvn clean install -DskipDockerBuild
+before_install:
+- docker pull spotify/kafka
+- docker run -p 2181:2181 -p 9092:9092 spotify/kafka

account/account-web/pom.xml

@@ -41,7 +41,7 @@
         </dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
+            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>

account/account-web/src/main/java/demo/account/action/ActivateAccount.java

@@ -9,10 +9,10 @@ import demo.account.event.AccountEventType;
 import demo.domain.Action;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
 import java.util.Arrays;
-import java.util.function.Function;
 
 import static demo.account.domain.AccountStatus.*;
 
@@ -22,11 +22,11 @@ import static demo.account.domain.AccountStatus.*;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class ActivateAccount extends Action<Account> {
     private final Logger log = Logger.getLogger(this.getClass());
 
-    public Function<Account, Account> getFunction() {
+    public Account apply(Account account) {
-        return (account) -> {
         Assert.isTrue(account.getStatus() != ACCOUNT_ACTIVE, "The account is already active");
         Assert.isTrue(Arrays.asList(ACCOUNT_CONFIRMED, ACCOUNT_SUSPENDED, ACCOUNT_ARCHIVED)
                 .contains(account.getStatus()), "The account cannot be activated");
@@ -48,12 +48,9 @@ public class ActivateAccount extends Action<Account> {
 
             // Rollback the operation
             account.setStatus(status);
-                accountService.update(account);
+            account = accountService.update(account);
-
-                throw ex;
         }
 
         return account;
-        };
     }
 }

account/account-web/src/main/java/demo/account/action/ArchiveAccount.java

@@ -9,10 +9,9 @@ import demo.account.event.AccountEventType;
 import demo.domain.Action;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
-import java.util.function.Function;
-
 import static demo.account.domain.AccountStatus.ACCOUNT_ACTIVE;
 import static demo.account.domain.AccountStatus.ACCOUNT_ARCHIVED;
 
@@ -22,12 +21,12 @@ import static demo.account.domain.AccountStatus.ACCOUNT_ARCHIVED;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class ArchiveAccount extends Action<Account> {
 
     private final Logger log = Logger.getLogger(this.getClass());
 
-    public Function<Account, Account> getFunction() {
+    public Account apply(Account account) {
-        return (account) -> {
         Assert.isTrue(account.getStatus() != ACCOUNT_ARCHIVED, "The account is already archived");
         Assert.isTrue(account.getStatus() == ACCOUNT_ACTIVE, "An inactive account cannot be archived");
 
@@ -48,12 +47,9 @@ public class ArchiveAccount extends Action<Account> {
 
             // Rollback the operation
             account.setStatus(status);
-                accountService.update(account);
+            account = accountService.update(account);
-
-                throw ex;
         }
 
         return account;
-        };
     }
 }

account/account-web/src/main/java/demo/account/action/ConfirmAccount.java

@@ -9,10 +9,9 @@ import demo.account.event.AccountEventType;
 import demo.domain.Action;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
-import java.util.function.Function;
-
 import static demo.account.domain.AccountStatus.ACCOUNT_CONFIRMED;
 import static demo.account.domain.AccountStatus.ACCOUNT_PENDING;
 
@@ -22,12 +21,12 @@ import static demo.account.domain.AccountStatus.ACCOUNT_PENDING;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class ConfirmAccount extends Action<Account> {
 
     private final Logger log = Logger.getLogger(this.getClass());
 
-    public Function<Account, Account> getFunction() {
+    public Account apply(Account account) {
-        return (account) -> {
         Assert.isTrue(account.getStatus() != ACCOUNT_CONFIRMED, "The account has already been confirmed");
         Assert.isTrue(account.getStatus() == ACCOUNT_PENDING, "The account has already been confirmed");
 
@@ -48,12 +47,9 @@ public class ConfirmAccount extends Action<Account> {
 
             // Rollback the operation
             account.setStatus(status);
-                accountService.update(account);
+            account = accountService.update(account);
-
-                throw ex;
         }
 
         return account;
-        };
     }
 }

account/account-web/src/main/java/demo/account/action/GetOrders.java

@@ -5,8 +5,7 @@ import demo.domain.Action;
 import demo.order.domain.OrderModule;
 import demo.order.domain.Orders;
 import org.springframework.stereotype.Service;
-
+import org.springframework.transaction.annotation.Transactional;
-import java.util.function.Function;
 
 /**
  * Query action to get {@link demo.order.domain.Order}s for an an {@link Account}
@@ -14,6 +13,7 @@ import java.util.function.Function;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class GetOrders extends Action<Account> {
 
     private OrderModule orderModule;
@@ -22,11 +22,9 @@ public class GetOrders extends Action<Account> {
         this.orderModule = orderModule;
     }
 
-    public Function<Account, Orders> getFunction() {
+    public Orders apply(Account account) {
-        return (account) -> {
         // Get orders from the order service
         return orderModule.getDefaultService()
                 .findOrdersByAccountId(account.getIdentity());
-        };
     }
 }

account/account-web/src/main/java/demo/account/action/PostOrder.java

@@ -9,6 +9,7 @@ import org.apache.log4j.Logger;
 import org.springframework.hateoas.MediaTypes;
 import org.springframework.hateoas.client.Traverson;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 import org.springframework.web.client.RestClientResponseException;
 
@@ -16,7 +17,6 @@ import java.io.IOException;
 import java.net.URI;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.function.BiFunction;
 
 import static demo.account.domain.AccountStatus.ACCOUNT_ACTIVE;
 
@@ -26,12 +26,12 @@ import static demo.account.domain.AccountStatus.ACCOUNT_ACTIVE;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class PostOrder extends Action<Account> {
 
     private final Logger log = Logger.getLogger(this.getClass());
 
-    public BiFunction<Account, Order, Order> getFunction() {
+    public Order apply(Account account, Order order) {
-        return (account, order) -> {
         Assert.isTrue(account.getStatus() == ACCOUNT_ACTIVE, "Only active accounts can create an order");
         order = order.post();
 
@@ -52,7 +52,6 @@ public class PostOrder extends Action<Account> {
         }
 
         return order;
-        };
     }
 
     private String getHttpStatusMessage(RestClientResponseException ex) {

account/account-web/src/main/java/demo/account/action/SuspendAccount.java

@@ -9,10 +9,9 @@ import demo.account.event.AccountEventType;
 import demo.domain.Action;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
-import java.util.function.Function;
-
 import static demo.account.domain.AccountStatus.ACCOUNT_ACTIVE;
 import static demo.account.domain.AccountStatus.ACCOUNT_SUSPENDED;
 
@@ -22,12 +21,12 @@ import static demo.account.domain.AccountStatus.ACCOUNT_SUSPENDED;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class SuspendAccount extends Action<Account> {
 
     private final Logger log = Logger.getLogger(this.getClass());
 
-    public Function<Account, Account> getFunction() {
+    public Account apply(Account account) {
-        return (account) -> {
         Assert.isTrue(account.getStatus() != ACCOUNT_SUSPENDED, "The account is already suspended");
         Assert.isTrue(account.getStatus() == ACCOUNT_ACTIVE, "An inactive account cannot be suspended");
 
@@ -48,12 +47,9 @@ public class SuspendAccount extends Action<Account> {
 
             // Rollback the operation
             account.setStatus(status);
-                accountService.update(account);
+            account = accountService.update(account);
-
-                throw ex;
         }
 
         return account;
-        };
     }
 }

account/account-web/src/main/java/demo/account/domain/Account.java

@@ -88,42 +88,36 @@ public class Account extends AbstractEntity<AccountEvent, Long> {
     @JsonIgnore
     public Orders getOrders() {
         return getAction(GetOrders.class)
-                .getFunction()
                 .apply(this);
     }
 
     @Command(method = "activate", controller = AccountController.class)
     public Account activate() {
         return getAction(ActivateAccount.class)
-                .getFunction()
                 .apply(this);
     }
 
     @Command(method = "archive", controller = AccountController.class)
     public Account archive() {
         return getAction(ArchiveAccount.class)
-                .getFunction()
                 .apply(this);
     }
 
     @Command(method = "confirm", controller = AccountController.class)
     public Account confirm() {
         return getAction(ConfirmAccount.class)
-                .getFunction()
                 .apply(this);
     }
 
     @Command(method = "suspend", controller = AccountController.class)
     public Account suspend() {
         return getAction(SuspendAccount.class)
-                .getFunction()
                 .apply(this);
     }
 
     @Command(method = "postOrder", controller = AccountController.class)
     public Order postOrder(Order order) {
         return getAction(PostOrder.class)
-                .getFunction()
                 .apply(this, order);
     }
 

account/account-web/src/main/resources/application.yml

@@ -23,11 +23,12 @@ spring:
 ---
 spring:
   profiles: docker
-  rabbitmq:
-    host: ${DOCKER_IP:192.168.99.100}
-    port: 5672
   cloud:
     stream:
+      kafka:
+        binder:
+          brokers: ${DOCKER_IP:192.168.99.100}
+          zk-nodes: ${DOCKER_IP:192.168.99.100}
       bindings:
         output:
           destination: account

account/account-worker/pom.xml

@@ -37,7 +37,7 @@
         </dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
+            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>

account/account-worker/src/main/resources/application.yml

@@ -28,11 +28,12 @@ eureka:
 ---
 spring:
   profiles: docker
-  rabbitmq:
-    host: ${DOCKER_IP:192.168.99.100}
-    port: 5672
   cloud:
     stream:
+      kafka:
+        binder:
+          brokers: ${DOCKER_IP:192.168.99.100}
+          zk-nodes: ${DOCKER_IP:192.168.99.100}
       bindings:
         input:
           destination: account

docker-compose.yml

@@ -10,10 +10,13 @@ redis:
   container_name: redis
   image: redis:latest
   net: host
-rabbit:
+kafka:
-  container_name: rabbit
+  container_name: kafka
-  image: rabbitmq:3-management
+  image: spotify/kafka:latest
   net: host
+  ports:
+     - 2181:2181
+     - 9092:9092
 account-web:
   image: account-web
   environment:

order/order-web/pom.xml

@@ -41,7 +41,7 @@
         </dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
+            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>

order/order-web/src/main/java/demo/order/action/AddReservation.java

@@ -12,21 +12,20 @@ import org.springframework.hateoas.Link;
 import org.springframework.hateoas.TemplateVariable;
 import org.springframework.hateoas.UriTemplate;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
-import java.util.function.BiFunction;
-
 /**
  * Connects an {@link Order} to an Account.
  *
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class AddReservation extends Action<Order> {
     private final Logger log = Logger.getLogger(this.getClass());
 
-    public BiFunction<Order, Long, Order> getFunction() {
+    public Order apply(Order order, Long reservationId) {
-        return (order, reservationId) -> {
         Assert.isTrue(order
                 .getStatus() == OrderStatus.RESERVATION_PENDING, "Order must be in a pending reservation state");
         Assert.isTrue(!order.getReservationIds().contains(reservationId), "Reservation already added to order");
@@ -48,13 +47,11 @@ public class AddReservation extends Action<Order> {
             log.error("Could not add reservation to order", ex);
             order.getReservationIds().remove(reservationId);
             order.setStatus(OrderStatus.RESERVATION_FAILED);
-                orderService.update(order);
+            order = orderService.update(order);
             order.sendAsyncEvent(new OrderEvent(OrderEventType.RESERVATION_FAILED, order), reservationLink);
-                throw ex;
         }
 
         return order;
-        };
     }
 
 }

order/order-web/src/main/java/demo/order/action/CompleteOrder.java

@@ -9,10 +9,10 @@ import demo.order.event.OrderEvent;
 import demo.order.event.OrderEventType;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
 import java.util.Arrays;
-import java.util.function.Function;
 
 /**
  * Completes the {@link Order} and applies a final status
@@ -20,12 +20,12 @@ import java.util.function.Function;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class CompleteOrder extends Action<Order> {
 
     private final Logger log = Logger.getLogger(CompleteOrder.class);
 
-    public Function<Order, Order> getFunction() {
+    public Order apply(Order order) {
-        return (order) -> {
         Assert.isTrue(Arrays.asList(OrderStatus.PAYMENT_FAILED,
                 OrderStatus.PAYMENT_SUCCEEDED,
                 OrderStatus.RESERVATION_FAILED).contains(order.getStatus()), "Order must be in a terminal state");
@@ -49,10 +49,8 @@ public class CompleteOrder extends Action<Order> {
             // Rollback status change
             order.setStatus(status);
             order = orderService.update(order);
-                throw ex;
         }
 
         return order;
-        };
     }
 }

order/order-web/src/main/java/demo/order/action/CompleteReservation.java

@@ -11,10 +11,10 @@ import demo.reservation.domain.Reservation;
 import demo.reservation.domain.ReservationStatus;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
 import java.util.List;
-import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import static demo.order.domain.OrderStatus.RESERVATION_FAILED;
@@ -26,12 +26,12 @@ import static demo.order.domain.OrderStatus.RESERVATION_SUCCEEDED;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class CompleteReservation extends Action<Order> {
 
     private final Logger log = Logger.getLogger(CompleteReservation.class);
 
-    public Function<Order, Order> getFunction() {
+    public Order apply(Order order) {
-        return (order) -> {
         if (order.getStatus() != RESERVATION_SUCCEEDED && order.getStatus() != RESERVATION_FAILED) {
             Assert.isTrue(order.getStatus() == OrderStatus.RESERVATION_PENDING,
                     "The order must be in a reservation pending state");
@@ -72,10 +72,8 @@ public class CompleteReservation extends Action<Order> {
             // Rollback status change
             order.setStatus(status);
             order = orderService.update(order);
-                throw ex;
         }
 
         return order;
-        };
     }
 }

order/order-web/src/main/java/demo/order/action/ConnectAccount.java

@@ -9,21 +9,20 @@ import demo.order.event.OrderEvent;
 import demo.order.event.OrderEventType;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
-import java.util.function.BiFunction;
-
 /**
  * Connects an {@link Order} to an Account.
  *
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class ConnectAccount extends Action<Order> {
     private final Logger log = Logger.getLogger(this.getClass());
 
-    public BiFunction<Order, Long, Order> getFunction() {
+    public Order apply(Order order, Long accountId) {
-        return (order, accountId) -> {
         Assert.isTrue(order.getStatus() == OrderStatus.ORDER_CREATED, "Order must be in a created state");
 
         OrderService orderService = order.getModule(OrderModule.class).getDefaultService();
@@ -40,12 +39,10 @@ public class ConnectAccount extends Action<Order> {
             log.error("Could not connect order to account", ex);
             order.setAccountId(null);
             order.setStatus(OrderStatus.ORDER_CREATED);
-                orderService.update(order);
+            order = orderService.update(order);
-                throw ex;
         }
 
         return order;
-        };
     }
 
 }

order/order-web/src/main/java/demo/order/action/ConnectPayment.java

@@ -10,21 +10,20 @@ import demo.order.event.OrderEventType;
 import demo.payment.domain.Payment;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
-import java.util.function.BiFunction;
-
 /**
  * Connects a {@link Payment} to an {@link Order}.
  *
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class ConnectPayment extends Action<Order> {
     private final Logger log = Logger.getLogger(this.getClass());
 
-    public BiFunction<Order, Long, Order> getFunction() {
+    public Order apply(Order order, Long paymentId) {
-        return (order, paymentId) -> {
         Assert.isTrue(order
                 .getStatus() == OrderStatus.PAYMENT_CREATED, "Order must be in a payment created state");
 
@@ -42,11 +41,9 @@ public class ConnectPayment extends Action<Order> {
             log.error("Could not connect payment to order", ex);
             order.setPaymentId(null);
             order.setStatus(OrderStatus.ORDER_CREATED);
-                orderService.update(order);
+            order = orderService.update(order);
-                throw ex;
         }
 
         return order;
-        };
     }
 }

order/order-web/src/main/java/demo/order/action/CreatePayment.java

@@ -12,10 +12,10 @@ import demo.payment.domain.PaymentMethod;
 import demo.payment.domain.PaymentService;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
 import java.util.Arrays;
-import java.util.function.Function;
 
 /**
  * Creates a {@link Payment} for an {@link Order}.
@@ -23,6 +23,7 @@ import java.util.function.Function;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class CreatePayment extends Action<Order> {
 
     private final Logger log = Logger.getLogger(this.getClass());
@@ -32,8 +33,7 @@ public class CreatePayment extends Action<Order> {
         this.paymentService = paymentService;
     }
 
-    public Function<Order, Order> getFunction() {
+    public Order apply(Order order) {
-        return order -> {
         Assert.isTrue(order.getPaymentId() == null, "Payment has already been created");
         Assert.isTrue(!Arrays.asList(OrderStatus.PAYMENT_CREATED,
                 OrderStatus.PAYMENT_CONNECTED,
@@ -73,12 +73,9 @@ public class CreatePayment extends Action<Order> {
 
             order.setPaymentId(null);
             order.setStatus(OrderStatus.ACCOUNT_CONNECTED);
-                orderService.update(order);
+            order = orderService.update(order);
-
-                throw ex;
         }
 
         return order;
-        };
     }
 }

order/order-web/src/main/java/demo/order/action/DeleteOrder.java

@@ -7,8 +7,7 @@ import demo.payment.domain.Payment;
 import demo.payment.domain.PaymentService;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
-
+import org.springframework.transaction.annotation.Transactional;
-import java.util.function.Consumer;
 
 /**
  * Processes a {@link Payment} for an {@link Order}.
@@ -16,6 +15,7 @@ import java.util.function.Consumer;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class DeleteOrder extends Action<Order> {
 
     private final Logger log = Logger.getLogger(this.getClass());
@@ -25,8 +25,7 @@ public class DeleteOrder extends Action<Order> {
         this.paymentService = paymentService;
     }
 
-    public Consumer<Order> getConsumer() {
+    public void apply(Order order) {
-        return (order) -> {
         // Delete payment
         if (order.getPaymentId() != null)
             paymentService.delete(order.getPaymentId());
@@ -35,6 +34,5 @@ public class DeleteOrder extends Action<Order> {
         order.getModule(OrderModule.class)
                 .getDefaultService()
                 .delete(order.getIdentity());
-        };
     }
 }

order/order-web/src/main/java/demo/order/action/GetReservations.java

@@ -5,8 +5,7 @@ import demo.order.domain.Order;
 import demo.reservation.domain.ReservationModule;
 import demo.reservation.domain.Reservations;
 import org.springframework.stereotype.Service;
-
+import org.springframework.transaction.annotation.Transactional;
-import java.util.function.Function;
 
 /**
  * Query action to get {@link demo.order.domain.Order}s for an an {@link Order}
@@ -14,6 +13,7 @@ import java.util.function.Function;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class GetReservations extends Action<Order> {
 
     private final ReservationModule reservationModule;
@@ -22,11 +22,9 @@ public class GetReservations extends Action<Order> {
         this.reservationModule = reservationModule;
     }
 
-    public Function<Order, Reservations> getFunction() {
+    public Reservations apply(Order order) {
-        return (order) -> {
         // Get orders from the order service
         return reservationModule.getDefaultService()
                 .findReservationsByOrderId(order.getIdentity());
-        };
     }
 }

order/order-web/src/main/java/demo/order/action/ProcessPayment.java

@@ -12,11 +12,11 @@ import org.apache.log4j.Logger;
 import org.springframework.hateoas.MediaTypes;
 import org.springframework.hateoas.client.Traverson;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
 import java.net.URI;
 import java.util.Arrays;
-import java.util.function.Function;
 
 /**
  * Processes a {@link Payment} for an {@link Order}.
@@ -24,12 +24,12 @@ import java.util.function.Function;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class ProcessPayment extends Action<Order> {
 
     private final Logger log = Logger.getLogger(this.getClass());
 
-    public Function<Order, Order> getFunction() {
+    public Order apply(Order order) {
-        return order -> {
         Assert.isTrue(!Arrays
                 .asList(OrderStatus.PAYMENT_SUCCEEDED, OrderStatus.PAYMENT_PENDING, OrderStatus.PAYMENT_FAILED)
                 .contains(order.getStatus()), "Payment has already been processed");
@@ -46,10 +46,13 @@ public class ProcessPayment extends Action<Order> {
         order.setStatus(OrderStatus.PAYMENT_PENDING);
         order = orderService.update(order);
 
+        boolean paymentSuccess = false;
+
         try {
             // Create traverson for the new order
             Traverson traverson = new Traverson(URI.create(payment.getLink("self").getHref()), MediaTypes.HAL_JSON);
             payment = traverson.follow("commands", "processPayment").toObject(Payment.class);
+            paymentSuccess = true;
         } catch (Exception ex) {
             log.error("The order's payment could not be processed", ex);
 
@@ -59,16 +62,17 @@ public class ProcessPayment extends Action<Order> {
             // Trigger payment failed event
             order.sendAsyncEvent(event);
 
-                throw ex;
+            paymentSuccess = false;
         } finally {
+            if(paymentSuccess) {
                 OrderEvent event = new OrderEvent(OrderEventType.PAYMENT_SUCCEEDED, order);
                 event.add(payment.getLink("self").withRel("payment"));
 
                 // Trigger payment succeeded event
                 order.sendAsyncEvent(event);
             }
+        }
 
         return order;
-        };
     }
 }

order/order-web/src/main/java/demo/order/action/ReserveInventory.java

@@ -12,10 +12,10 @@ import demo.warehouse.domain.WarehouseService;
 import demo.warehouse.exception.WarehouseNotFoundException;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
 import java.util.Arrays;
-import java.util.function.Function;
 
 /**
  * Reserves inventory for an {@link Order}.
@@ -23,6 +23,7 @@ import java.util.function.Function;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class ReserveInventory extends Action<Order> {
 
     private final Logger log = Logger.getLogger(ReserveInventory.class);
@@ -32,8 +33,7 @@ public class ReserveInventory extends Action<Order> {
         this.warehouseService = warehouseService;
     }
 
-    public Function<Order, Order> getFunction() {
+    public Order apply(Order order) {
-        return (order) -> {
         Assert.isTrue(!Arrays
                 .asList(OrderStatus.PAYMENT_SUCCEEDED, OrderStatus.PAYMENT_PENDING,
                         OrderStatus.PAYMENT_FAILED, OrderStatus.INVENTORY_RESERVED,
@@ -78,17 +78,16 @@ public class ReserveInventory extends Action<Order> {
 
             // Trigger reservation failed
             order.sendAsyncEvent(event);
-
-                throw ex;
         } finally {
+            if(order.getStatus() != OrderStatus.ACCOUNT_CONNECTED) {
                 OrderEvent event = new OrderEvent(OrderEventType.RESERVATION_PENDING, order);
                 event.add(warehouse.getLink("self").withRel("warehouse"));
 
                 // Trigger reservation pending event
                 order.sendAsyncEvent(event);
             }
+        }
 
         return order;
-        };
     }
 }

order/order-web/src/main/java/demo/order/action/UpdateOrderStatus.java

@@ -6,8 +6,7 @@ import demo.order.domain.OrderService;
 import demo.order.domain.OrderStatus;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
-
+import org.springframework.transaction.annotation.Transactional;
-import java.util.function.BiFunction;
 
 /**
  * Updates the status of a {@link Order} entity.
@@ -15,6 +14,7 @@ import java.util.function.BiFunction;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class UpdateOrderStatus extends Action<Order> {
     private final Logger log = Logger.getLogger(this.getClass());
 
@@ -24,8 +24,7 @@ public class UpdateOrderStatus extends Action<Order> {
         this.orderService = orderService;
     }
 
-    public BiFunction<Order, OrderStatus, Order> getFunction() {
+    public Order apply(Order order, OrderStatus orderStatus) {
-        return (order, orderStatus) -> {
 
         // Save rollback status
         OrderStatus rollbackStatus = order.getStatus();
@@ -38,10 +37,8 @@ public class UpdateOrderStatus extends Action<Order> {
             log.error("Could not update the status", ex);
             order.setStatus(rollbackStatus);
             order = orderService.update(order);
-                throw ex;
         }
 
         return order;
-        };
     }
 }

order/order-web/src/main/java/demo/order/domain/Order.java

@@ -117,77 +117,66 @@ public class Order extends AbstractEntity<OrderEvent, Long> {
     @JsonIgnore
     public Reservations getReservations() {
         return getAction(GetReservations.class)
-                .getFunction()
                 .apply(this);
     }
 
     @Command(method = "connectAccount", controller = OrderController.class)
     public Order connectAccount(Long accountId) {
         return getAction(ConnectAccount.class)
-                .getFunction()
                 .apply(this, accountId);
     }
 
     @Command(method = "connectPayment", controller = OrderController.class)
     public Order connectPayment(Long paymentId) {
         return getAction(ConnectPayment.class)
-                .getFunction()
                 .apply(this, paymentId);
     }
 
     @Command(method = "createPayment", controller = OrderController.class)
     public Order createPayment() {
         return getAction(CreatePayment.class)
-                .getFunction()
                 .apply(this);
     }
 
     @Command(method = "processPayment", controller = OrderController.class)
     public Order processPayment() {
         return getAction(ProcessPayment.class)
-                .getFunction()
                 .apply(this);
     }
 
     @Command(method = "reserveInventory", controller = OrderController.class)
     public Order reserveInventory() {
         return getAction(ReserveInventory.class)
-                .getFunction()
                 .apply(this);
     }
 
     @Command(method = "addReservation", controller = OrderController.class)
     public Order addReservation(Long reservationId) {
         return getAction(AddReservation.class)
-                .getFunction()
                 .apply(this, reservationId);
     }
 
     @Command(method = "completeReservation", controller = OrderController.class)
     public Order completeReservation() {
         return getAction(CompleteReservation.class)
-                .getFunction()
                 .apply(this);
     }
 
     @Command(method = "completeOrder", controller = OrderController.class)
     public Order completeOrder() {
         return getAction(CompleteOrder.class)
-                .getFunction()
                 .apply(this);
     }
 
     @Command(method = "updateOrderStatus", controller = OrderController.class)
     public Order updateOrderStatus(OrderStatus orderStatus) {
         return getAction(UpdateOrderStatus.class)
-                .getFunction()
                 .apply(this, orderStatus);
     }
 
     public boolean delete() {
         getAction(DeleteOrder.class)
-                .getConsumer()
+                .apply(this);
-                .accept(this);
 
         return true;
     }

order/order-web/src/main/resources/application.yml

@@ -23,11 +23,12 @@ spring:
 ---
 spring:
   profiles: docker
-  rabbitmq:
-    host: ${DOCKER_IP:192.168.99.100}
-    port: 5672
   cloud:
     stream:
+      kafka:
+        binder:
+          brokers: ${DOCKER_IP:192.168.99.100}
+          zk-nodes: ${DOCKER_IP:192.168.99.100}
       bindings:
         output:
           destination: order

order/order-worker/pom.xml

@@ -37,7 +37,7 @@
         </dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
+            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>

order/order-worker/src/main/java/demo/config/StateMachineConfig.java

@@ -375,14 +375,19 @@ public class StateMachineConfig extends EnumStateMachineConfigurerAdapter<OrderS
                             MediaTypes.HAL_JSON
                     );
 
+                    Order order = traverson.follow("self", "commands", "completeOrder")
+                            .toEntity(Order.class)
+                            .getBody();
+
                     // Release the reservations
                     Reservations reservations = traverson.follow("self", "reservations")
                             .toObject(Reservations.class);
 
                     reservations.getContent().stream()
-                            .filter(r -> r.getStatus() == ReservationStatus.RESERVATION_SUCCEEDED)
+                            .filter(r -> r.getStatus() != ReservationStatus.RESERVATION_FAILED)
                             .parallel()
                             .forEach(r -> {
+                                try {
                                     Traverson res = new Traverson(
                                             URI.create(r.getLink("self").getHref()),
                                             MediaTypes.HAL_JSON
@@ -390,11 +395,13 @@ public class StateMachineConfig extends EnumStateMachineConfigurerAdapter<OrderS
 
                                     res.follow("self", "commands", "releaseInventory")
                                             .toObject(Reservation.class);
+                                } catch (Exception ex) {
+                                    log.error("Could not release inventory for reservation", ex);
+                                }
                             });
 
-                    return traverson.follow("self", "commands", "completeOrder")
+                    return order;
-                            .toEntity(Order.class)
+
-                            .getBody();
                 }));
     }
 

order/order-worker/src/main/resources/application.yml

@@ -22,11 +22,12 @@ spring:
 ---
 spring:
   profiles: docker
-  rabbitmq:
-    host: ${DOCKER_IP:192.168.99.100}
-    port: 5672
   cloud:
     stream:
+      kafka:
+        binder:
+          brokers: ${DOCKER_IP:192.168.99.100}
+          zk-nodes: ${DOCKER_IP:192.168.99.100}
       bindings:
         input:
           contentType: 'application/json'

payment/payment-web/pom.xml

@@ -41,7 +41,7 @@
         </dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
+            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>

payment/payment-web/src/main/java/demo/payment/action/ConnectOrder.java

@@ -9,16 +9,15 @@ import demo.payment.event.PaymentEvent;
 import demo.payment.event.PaymentEventType;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
-import java.util.function.BiFunction;
-
 @Service
+@Transactional
 public class ConnectOrder extends Action<Payment> {
     private final Logger log = Logger.getLogger(this.getClass());
 
-    public BiFunction<Payment, Long, Payment> getFunction() {
+    public Payment apply(Payment payment, Long orderId) {
-        return (payment, orderId) -> {
         Assert.isTrue(payment
                 .getStatus() == PaymentStatus.PAYMENT_CREATED, "Payment has already been connected to an order");
 
@@ -39,12 +38,9 @@ public class ConnectOrder extends Action<Payment> {
             // Rollback operation
             payment.setStatus(PaymentStatus.PAYMENT_CREATED);
             payment.setOrderId(null);
-                paymentService.update(payment);
+            payment = paymentService.update(payment);
-
-                throw ex;
         }
 
         return payment;
-        };
     }
 }

payment/payment-web/src/main/java/demo/payment/action/ProcessPayment.java

@@ -8,15 +8,16 @@ import demo.payment.event.PaymentEvent;
 import demo.payment.event.PaymentEventType;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
 import java.util.Arrays;
-import java.util.function.Function;
 
 import static demo.payment.domain.PaymentStatus.PAYMENT_FAILED;
 import static demo.payment.domain.PaymentStatus.PAYMENT_SUCCEEDED;
 
 @Service
+@Transactional
 public class ProcessPayment extends Action<Payment> {
     private final Logger log = Logger.getLogger(this.getClass());
     private final PaymentService paymentService;
@@ -25,8 +26,7 @@ public class ProcessPayment extends Action<Payment> {
         this.paymentService = paymentService;
     }
 
-    public Function<Payment, Payment> getFunction() {
+    public Payment apply(Payment payment) {
-        return payment -> {
         // Validations
         Assert.isTrue(!Arrays.asList(PAYMENT_SUCCEEDED,
                 PaymentStatus.PAYMENT_PENDING,
@@ -42,15 +42,15 @@ public class ProcessPayment extends Action<Payment> {
             payment.sendAsyncEvent(new PaymentEvent(PaymentEventType.PAYMENT_PROCESSED, payment));
         } catch (Exception ex) {
             log.error("Payment could not be processed", ex);
-                finalizePayment(payment, PAYMENT_FAILED);
+            payment = finalizePayment(payment, PAYMENT_FAILED);
-                throw ex;
         } finally {
+            if(payment.getStatus() != PAYMENT_FAILED) {
                 // Handle the result asynchronously
-                finalizePayment(payment, PAYMENT_SUCCEEDED);
+                payment = finalizePayment(payment, PAYMENT_SUCCEEDED);
+            }
         }
 
         return payment;
-        };
     }
 
     private Payment finalizePayment(Payment payment, PaymentStatus paymentStatus) {

payment/payment-web/src/main/java/demo/payment/domain/Payment.java

@@ -91,14 +91,12 @@ public class Payment extends AbstractEntity<PaymentEvent, Long> {
     @Command(method = "connectOrder", controller = PaymentController.class)
     public Payment connectOrder(Long orderId) {
         return getAction(ConnectOrder.class)
-                .getFunction()
                 .apply(this, orderId);
     }
 
     @Command(method = "processPayment", controller = PaymentController.class)
     public Payment processPayment() {
         return getAction(ProcessPayment.class)
-                .getFunction()
                 .apply(this);
     }
 

payment/payment-web/src/main/resources/application.yml

@@ -23,11 +23,12 @@ spring:
 ---
 spring:
   profiles: docker
-  rabbitmq:
-    host: ${DOCKER_IP:192.168.99.100}
-    port: 5672
   cloud:
     stream:
+      kafka:
+        binder:
+          brokers: ${DOCKER_IP:192.168.99.100}
+          zk-nodes: ${DOCKER_IP:192.168.99.100}
       bindings:
         output:
           contentType: 'application/json'

payment/payment-worker/pom.xml

@@ -37,7 +37,7 @@
         </dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
+            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>

payment/payment-worker/src/main/resources/application.yml

@@ -22,11 +22,12 @@ spring:
 ---
 spring:
   profiles: docker
-  rabbitmq:
-    host: ${DOCKER_IP:192.168.99.100}
-    port: 5672
   cloud:
     stream:
+      kafka:
+        binder:
+          brokers: ${DOCKER_IP:192.168.99.100}
+          zk-nodes: ${DOCKER_IP:192.168.99.100}
       bindings:
         input:
           contentType: 'application/json'

warehouse/warehouse-web/pom.xml

@@ -51,7 +51,7 @@
         </dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
+            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>

warehouse/warehouse-web/src/main/java/demo/inventory/action/ReserveInventory.java

@@ -12,8 +12,6 @@ import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
 import org.springframework.util.Assert;
 
-import java.util.function.BiFunction;
-
 import static demo.inventory.domain.InventoryStatus.RESERVATION_CONNECTED;
 
 /**
@@ -33,8 +31,7 @@ public class ReserveInventory extends Action<Inventory> {
         this.inventoryService = inventoryService;
     }
 
-    public BiFunction<Inventory, Long, Inventory> getFunction() {
+    public Inventory apply(Inventory inventory, Long reservationId) {
-        return (inventory, reservationId) -> {
         Assert.isTrue(inventory.getStatus() == InventoryStatus.RESERVATION_CONNECTED,
                 "Inventory must be in a reservation connected state");
         Assert.isTrue(inventory.getReservation() == null,
@@ -50,8 +47,7 @@ public class ReserveInventory extends Action<Inventory> {
             log.error("Could not connect reservation to inventory", ex);
             inventory.setReservation(null);
             inventory.setStatus(InventoryStatus.RESERVATION_PENDING);
-                inventoryService.update(inventory);
+            inventory = inventoryService.update(inventory);
-                throw ex;
         } finally {
             if (inventory.getStatus() == RESERVATION_CONNECTED && inventory.getReservation() != null) {
                 inventory.setStatus(InventoryStatus.INVENTORY_RESERVED);
@@ -61,6 +57,5 @@ public class ReserveInventory extends Action<Inventory> {
         }
 
         return inventory;
-        };
     }
 }

warehouse/warehouse-web/src/main/java/demo/inventory/action/UpdateInventoryStatus.java

@@ -7,8 +7,7 @@ import demo.inventory.domain.InventoryStatus;
 import demo.reservation.domain.ReservationService;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
-
+import org.springframework.transaction.annotation.Transactional;
-import java.util.function.BiFunction;
 
 /**
  * Updates the status of a {@link Inventory} entity.
@@ -16,6 +15,7 @@ import java.util.function.BiFunction;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class UpdateInventoryStatus extends Action<Inventory> {
     private final Logger log = Logger.getLogger(this.getClass());
 
@@ -27,9 +27,7 @@ public class UpdateInventoryStatus extends Action<Inventory> {
         this.inventoryService = inventoryService;
     }
 
-    public BiFunction<Inventory, InventoryStatus, Inventory> getFunction() {
+    public Inventory apply(Inventory inventory, InventoryStatus inventoryStatus) {
-        return (inventory, inventoryStatus) -> {
-
         // Save rollback status
         InventoryStatus rollbackStatus = inventory.getStatus();
 
@@ -41,10 +39,8 @@ public class UpdateInventoryStatus extends Action<Inventory> {
             log.error("Could not update the status", ex);
             inventory.setStatus(rollbackStatus);
             inventory = inventoryService.update(inventory);
-                throw ex;
         }
 
         return inventory;
-        };
     }
 }

warehouse/warehouse-web/src/main/java/demo/inventory/domain/Inventory.java

@@ -87,14 +87,12 @@ public class Inventory extends AbstractEntity<InventoryEvent, Long> {
     @Command(method = "reserve", controller = InventoryController.class)
     public Inventory reserve(Long reservationId) {
         return getAction(ReserveInventory.class)
-                .getFunction()
                 .apply(this, reservationId);
     }
 
     @Command(method = "updateInventoryStatus", controller = InventoryController.class)
     public Inventory updateStatus(InventoryStatus status) {
         return getAction(UpdateInventoryStatus.class)
-                .getFunction()
                 .apply(this, status);
     }
 

warehouse/warehouse-web/src/main/java/demo/inventory/domain/InventoryService.java

@@ -105,7 +105,7 @@ public class InventoryService extends Service<Inventory, Long> {
                 Boolean lock = false;
 
                 try {
-                    lock = inventoryLock.tryLock(30, 5000, TimeUnit.MILLISECONDS);
+                    lock = inventoryLock.tryLock(1, 5000, TimeUnit.MILLISECONDS);
                 } catch (InterruptedException e) {
                     log.error("Interrupted while acquiring lock on inventory", e);
                 }

warehouse/warehouse-web/src/main/java/demo/reservation/action/ConnectInventory.java

@@ -21,7 +21,6 @@ import org.springframework.util.Assert;
 
 import java.util.List;
 import java.util.Random;
-import java.util.function.Function;
 
 import static demo.reservation.event.ReservationEventType.*;
 
@@ -41,8 +40,7 @@ public class ConnectInventory extends Action<Reservation> {
         this.inventoryService = inventoryService;
     }
 
-    public Function<Reservation, Reservation> getFunction() {
+    public Reservation apply(Reservation reservation) {
-        return (reservation) -> {
         Assert.isTrue(reservation.getStatus() == ReservationStatus.ORDER_CONNECTED,
                 "Reservation must be in an order connected state");
 
@@ -69,11 +67,16 @@ public class ConnectInventory extends Action<Reservation> {
                         .concat(reservation.getId().toString()));
             }
 
+            inventory.setReservation(reservation);
+            inventory.setStatus(InventoryStatus.RESERVATION_CONNECTED);
+            inventory = inventoryService.update(inventory);
+
             // Set inventory on reservation and mark successful
             reservation.setInventory(inventory);
             reservation.setStatus(ReservationStatus.RESERVATION_SUCCEEDED);
             reservation = reservationService.update(reservation);
 
+
             // Trigger the inventory connected event
             reservation.sendAsyncEvent(new ReservationEvent(INVENTORY_CONNECTED, reservation),
                     reservation.getInventory().getId().withRel("inventory"));
@@ -84,24 +87,26 @@ public class ConnectInventory extends Action<Reservation> {
                 if (inventory != null) {
                     inventory.setReservation(null);
                     inventory.setStatus(InventoryStatus.RESERVATION_PENDING);
-                        inventory = inventoryService.update(inventory);
+                    inventoryService.update(inventory);
                 }
 
                 reservation.setInventory(null);
                 reservation.setStatus(ReservationStatus.ORDER_CONNECTED);
                 reservation = reservationService.update(reservation);
             }
+
             throw ex;
         } finally {
             if (reservation.getStatus() == ReservationStatus.RESERVATION_SUCCEEDED) {
                 Link inventoryLink = reservation.getInventory().getId().withRel("inventory");
                 Link orderLink = getRemoteLink("order-web", "/v1/orders/{id}", reservation.getOrderId(), "order");
-                    reservation.sendAsyncEvent(new ReservationEvent(RESERVATION_SUCCEEDED, reservation), inventoryLink, orderLink);
+                reservation
+                        .sendAsyncEvent(new ReservationEvent(RESERVATION_SUCCEEDED, reservation), inventoryLink,
+                                orderLink);
             }
         }
 
         return reservation;
-        };
     }
 
     private Link getRemoteLink(String service, String relative, Object identifier, String rel) {

warehouse/warehouse-web/src/main/java/demo/reservation/action/ConnectOrder.java

@@ -9,22 +9,22 @@ import demo.reservation.event.ReservationEvent;
 import demo.reservation.event.ReservationEventType;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
-import java.util.function.BiFunction;
-
 /**
  * Connects an {@link Reservation} to an Order.
  *
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class ConnectOrder extends Action<Reservation> {
     private final Logger log = Logger.getLogger(this.getClass());
 
-    public BiFunction<Reservation, Long, Reservation> getFunction() {
+    public Reservation apply(Reservation reservation, Long orderId) {
-        return (reservation, orderId) -> {
+        Assert.isTrue(reservation
-            Assert.isTrue(reservation.getStatus() == ReservationStatus.RESERVATION_CREATED, "Reservation must be in a created state");
+                .getStatus() == ReservationStatus.RESERVATION_CREATED, "Reservation must be in a created state");
 
         ReservationService reservationService = reservation.getModule(ReservationModule.class).getDefaultService();
 
@@ -40,11 +40,9 @@ public class ConnectOrder extends Action<Reservation> {
             log.error("Could not connect reservation to order", ex);
             reservation.setOrderId(null);
             reservation.setStatus(ReservationStatus.RESERVATION_CREATED);
-                reservationService.update(reservation);
+            reservation = reservationService.update(reservation);
-                throw ex;
         }
 
         return reservation;
-        };
     }
 }

warehouse/warehouse-web/src/main/java/demo/reservation/action/ReleaseInventory.java

@@ -12,10 +12,9 @@ import demo.reservation.domain.ReservationStatus;
 import demo.reservation.event.ReservationEvent;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
-import java.util.function.Function;
-
 import static demo.inventory.event.InventoryEventType.INVENTORY_RELEASED;
 import static demo.reservation.event.ReservationEventType.RESERVATION_FAILED;
 
@@ -25,6 +24,7 @@ import static demo.reservation.event.ReservationEventType.RESERVATION_FAILED;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class ReleaseInventory extends Action<Reservation> {
     private final Logger log = Logger.getLogger(this.getClass());
     private final InventoryService inventoryService;
@@ -33,11 +33,9 @@ public class ReleaseInventory extends Action<Reservation> {
         this.inventoryService = inventoryService;
     }
 
-    public Function<Reservation, Reservation> getFunction() {
+    public Reservation apply(Reservation reservation) {
-        return (reservation) -> {
+        Assert.isTrue(reservation.getStatus() != ReservationStatus.RESERVATION_FAILED,
-            Assert.isTrue(reservation.getStatus() == ReservationStatus.RESERVATION_SUCCEEDED,
+                "Reservation is already in a failed state");
-                    "Reservation must be in a succeeded state");
-            Assert.notNull(reservation.getInventory(), "The reservation has no connected inventory");
 
         ReservationService reservationService = reservation.getModule(ReservationModule.class).getDefaultService();
 
@@ -59,8 +57,8 @@ public class ReleaseInventory extends Action<Reservation> {
                 reservation.setStatus(ReservationStatus.RESERVATION_SUCCEEDED);
                 reservation = reservationService.update(reservation);
             }
-                throw ex;
         } finally {
+            if (inventory != null && reservation.getStatus() != ReservationStatus.RESERVATION_SUCCEEDED) {
                 // Release the inventory
                 inventory.setReservation(null);
                 inventory.setStatus(InventoryStatus.RESERVATION_PENDING);
@@ -69,8 +67,8 @@ public class ReleaseInventory extends Action<Reservation> {
                 // Trigger the inventory released event
                 inventory.sendAsyncEvent(new InventoryEvent(INVENTORY_RELEASED, inventory));
             }
+        }
 
         return reservation;
-        };
     }
 }

warehouse/warehouse-web/src/main/java/demo/reservation/domain/Reservation.java

@@ -111,21 +111,18 @@ public class Reservation extends AbstractEntity<ReservationEvent, Long> {
     @Command(method = "connectInventory", controller = ReservationController.class)
     public Reservation connectInventory() {
         return getAction(ConnectInventory.class)
-                .getFunction()
                 .apply(this);
     }
 
     @Command(method = "releaseInventory", controller = ReservationController.class)
     public Reservation releaseInventory() {
         return getAction(ReleaseInventory.class)
-                .getFunction()
                 .apply(this);
     }
 
     @Command(method = "connectOrder", controller = ReservationController.class)
     public Reservation connectOrder(Long orderId) {
         return getAction(ConnectOrder.class)
-                .getFunction()
                 .apply(this, orderId);
     }
 

warehouse/warehouse-web/src/main/java/demo/warehouse/action/ReserveOrder.java

@@ -9,9 +9,9 @@ import demo.reservation.event.ReservationEventType;
 import demo.warehouse.domain.Warehouse;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 
 import java.util.List;
-import java.util.function.BiConsumer;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -21,6 +21,7 @@ import java.util.stream.IntStream;
  * @author Kenny Bastani
  */
 @Service
+@Transactional
 public class ReserveOrder extends Action<Warehouse> {
 
     private final Logger log = Logger.getLogger(ReserveOrder.class);
@@ -30,9 +31,7 @@ public class ReserveOrder extends Action<Warehouse> {
         this.reservationService = reservationService;
     }
 
-    public BiConsumer<Warehouse, Order> getConsumer() {
+    public void apply(Warehouse warehouse, Order order) {
-        return (warehouse, order) -> {
-
         // Create reservations for each order item
         List<Reservation> reservations = order.getLineItems().stream()
                 .map(item -> IntStream.rangeClosed(1, item.getQuantity())
@@ -49,6 +48,5 @@ public class ReserveOrder extends Action<Warehouse> {
             event.add(order.getLink("self").withRel("order"));
             r.sendAsyncEvent(event);
         });
-        };
     }
 }

warehouse/warehouse-web/src/main/java/demo/warehouse/domain/Warehouse.java

@@ -77,8 +77,7 @@ public class Warehouse extends AbstractEntity<WarehouseEvent, Long> {
     @Command(method = "reserveOrder", controller = WarehouseController.class)
     public Warehouse reserveOrder(Order order) {
         getAction(ReserveOrder.class)
-                .getConsumer()
+                .apply(this, order);
-                .accept(this, order);
         return this;
     }
 

warehouse/warehouse-worker/pom.xml

@@ -37,7 +37,7 @@
         </dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
+            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>

warehouse/warehouse-worker/src/main/java/demo/reservation/config/ReservationStateMachineConfig.java

@@ -26,6 +26,8 @@ import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.Map;
 
+import static demo.order.domain.OrderStatus.ORDER_FAILED;
+import static demo.order.domain.OrderStatus.RESERVATION_FAILED;
 import static demo.order.domain.OrderStatus.RESERVATION_PENDING;
 
 /**
@@ -243,12 +245,24 @@ public class ReservationStateMachineConfig extends EnumStateMachineConfigurerAda
                             MediaTypes.HAL_JSON
                     );
 
+                    Order order = traverson.follow("self", "order").toObject(Order.class);
+                    Reservation reservation = null;
+
+                    // Check order status and release inventory if it has failed
+                    if (order.getStatus() == RESERVATION_FAILED || order.getStatus() == ORDER_FAILED) {
+                        reservation = traverson.follow("self", "commands", "releaseInventory")
+                                .toObject(Reservation.class);
+                    } else if (order.getStatus() == RESERVATION_PENDING) {
                         traverson.follow("self", "order", "commands", "completeReservation")
                                 .toObject(Order.class);
+                    }
 
-                    return traverson.follow("self")
+                    if (reservation == null)
+                        reservation = traverson.follow("self")
                                 .toEntity(Reservation.class)
                                 .getBody();
+
+                    return reservation;
                 }));
     }
 

warehouse/warehouse-worker/src/main/resources/application.yml

@@ -33,11 +33,12 @@ spring:
 ---
 spring:
   profiles: docker
-  rabbitmq:
-    host: ${DOCKER_IP:192.168.99.100}
-    port: 5672
   cloud:
     stream:
+      kafka:
+        binder:
+          brokers: ${DOCKER_IP:192.168.99.100}
+          zk-nodes: ${DOCKER_IP:192.168.99.100}
       bindings:
         warehouse:
           contentType: 'application/json'