Merge pull request #91 from eugenp/master

update

axon/pom.xml

@@ -53,7 +53,7 @@
     </dependencies>
 
     <properties>
-        <axon.version>4.1.2</axon.version>
+        <axon.version>4.4.7</axon.version>
     </properties>
 
 </project>

axon/src/main/java/com/baeldung/axon/commandmodel/OrderAggregate.java

@@ -1,59 +0,0 @@
-package com.baeldung.axon.commandmodel;
-
-import static org.axonframework.modelling.command.AggregateLifecycle.apply;
-
-import org.axonframework.commandhandling.CommandHandler;
-import org.axonframework.eventsourcing.EventSourcingHandler;
-import org.axonframework.modelling.command.AggregateIdentifier;
-import org.axonframework.spring.stereotype.Aggregate;
-
-import com.baeldung.axon.coreapi.commands.ConfirmOrderCommand;
-import com.baeldung.axon.coreapi.commands.PlaceOrderCommand;
-import com.baeldung.axon.coreapi.commands.ShipOrderCommand;
-import com.baeldung.axon.coreapi.events.OrderConfirmedEvent;
-import com.baeldung.axon.coreapi.events.OrderPlacedEvent;
-import com.baeldung.axon.coreapi.events.OrderShippedEvent;
-import com.baeldung.axon.coreapi.exceptions.UnconfirmedOrderException;
-
-@Aggregate
-public class OrderAggregate {
-
-    @AggregateIdentifier
-    private String orderId;
-    private boolean orderConfirmed;
-
-    @CommandHandler
-    public OrderAggregate(PlaceOrderCommand command) {
-        apply(new OrderPlacedEvent(command.getOrderId(), command.getProduct()));
-    }
-
-    @CommandHandler
-    public void handle(ConfirmOrderCommand command) {
-        apply(new OrderConfirmedEvent(orderId));
-    }
-
-    @CommandHandler
-    public void handle(ShipOrderCommand command) {
-        if (!orderConfirmed) {
-            throw new UnconfirmedOrderException();
-        }
-
-        apply(new OrderShippedEvent(orderId));
-    }
-
-    @EventSourcingHandler
-    public void on(OrderPlacedEvent event) {
-        this.orderId = event.getOrderId();
-        this.orderConfirmed = false;
-    }
-
-    @EventSourcingHandler
-    public void on(OrderConfirmedEvent event) {
-        this.orderConfirmed = true;
-    }
-
-    protected OrderAggregate() {
-        // Required by Axon to build a default Aggregate prior to Event Sourcing
-    }
-
-}

axon/src/main/java/com/baeldung/axon/commandmodel/order/OrderAggregate.java

@@ -0,0 +1,98 @@
+package com.baeldung.axon.commandmodel.order;
+
+import com.baeldung.axon.coreapi.commands.AddProductCommand;
+import com.baeldung.axon.coreapi.commands.ConfirmOrderCommand;
+import com.baeldung.axon.coreapi.commands.CreateOrderCommand;
+import com.baeldung.axon.coreapi.commands.ShipOrderCommand;
+import com.baeldung.axon.coreapi.events.OrderConfirmedEvent;
+import com.baeldung.axon.coreapi.events.OrderCreatedEvent;
+import com.baeldung.axon.coreapi.events.OrderShippedEvent;
+import com.baeldung.axon.coreapi.events.ProductAddedEvent;
+import com.baeldung.axon.coreapi.events.ProductRemovedEvent;
+import com.baeldung.axon.coreapi.exceptions.DuplicateOrderLineException;
+import com.baeldung.axon.coreapi.exceptions.OrderAlreadyConfirmedException;
+import com.baeldung.axon.coreapi.exceptions.UnconfirmedOrderException;
+import org.axonframework.commandhandling.CommandHandler;
+import org.axonframework.eventsourcing.EventSourcingHandler;
+import org.axonframework.modelling.command.AggregateIdentifier;
+import org.axonframework.modelling.command.AggregateMember;
+import org.axonframework.spring.stereotype.Aggregate;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.axonframework.modelling.command.AggregateLifecycle.apply;
+
+@Aggregate
+public class OrderAggregate {
+
+    @AggregateIdentifier
+    private String orderId;
+    private boolean orderConfirmed;
+
+    @AggregateMember
+    private Map<String, OrderLine> orderLines;
+
+    @CommandHandler
+    public OrderAggregate(CreateOrderCommand command) {
+        apply(new OrderCreatedEvent(command.getOrderId()));
+    }
+
+    @CommandHandler
+    public void handle(AddProductCommand command) {
+        if (orderConfirmed) {
+            throw new OrderAlreadyConfirmedException(orderId);
+        }
+
+        String productId = command.getProductId();
+        if (orderLines.containsKey(productId)) {
+            throw new DuplicateOrderLineException(productId);
+        }
+        apply(new ProductAddedEvent(orderId, productId));
+    }
+
+    @CommandHandler
+    public void handle(ConfirmOrderCommand command) {
+        if (orderConfirmed) {
+            return;
+        }
+
+        apply(new OrderConfirmedEvent(orderId));
+    }
+
+    @CommandHandler
+    public void handle(ShipOrderCommand command) {
+        if (!orderConfirmed) {
+            throw new UnconfirmedOrderException();
+        }
+
+        apply(new OrderShippedEvent(orderId));
+    }
+
+    @EventSourcingHandler
+    public void on(OrderCreatedEvent event) {
+        this.orderId = event.getOrderId();
+        this.orderConfirmed = false;
+        this.orderLines = new HashMap<>();
+    }
+
+    @EventSourcingHandler
+    public void on(OrderConfirmedEvent event) {
+        this.orderConfirmed = true;
+    }
+
+    @EventSourcingHandler
+    public void on(ProductAddedEvent event) {
+        String productId = event.getProductId();
+        this.orderLines.put(productId, new OrderLine(productId));
+    }
+
+    @EventSourcingHandler
+    public void on(ProductRemovedEvent event) {
+        this.orderLines.remove(event.getProductId());
+    }
+
+    protected OrderAggregate() {
+        // Required by Axon to build a default Aggregate prior to Event Sourcing
+    }
+}

axon/src/main/java/com/baeldung/axon/commandmodel/order/OrderLine.java

@@ -0,0 +1,83 @@
+package com.baeldung.axon.commandmodel.order;
+
+import com.baeldung.axon.coreapi.commands.DecrementProductCountCommand;
+import com.baeldung.axon.coreapi.commands.IncrementProductCountCommand;
+import com.baeldung.axon.coreapi.events.OrderConfirmedEvent;
+import com.baeldung.axon.coreapi.events.ProductCountDecrementedEvent;
+import com.baeldung.axon.coreapi.events.ProductCountIncrementedEvent;
+import com.baeldung.axon.coreapi.events.ProductRemovedEvent;
+import com.baeldung.axon.coreapi.exceptions.OrderAlreadyConfirmedException;
+import org.axonframework.commandhandling.CommandHandler;
+import org.axonframework.eventsourcing.EventSourcingHandler;
+import org.axonframework.modelling.command.EntityId;
+
+import java.util.Objects;
+
+import static org.axonframework.modelling.command.AggregateLifecycle.apply;
+
+public class OrderLine {
+
+    @EntityId
+    private final String productId;
+    private Integer count;
+    private boolean orderConfirmed;
+
+    public OrderLine(String productId) {
+        this.productId = productId;
+        this.count = 1;
+    }
+
+    @CommandHandler
+    public void handle(IncrementProductCountCommand command) {
+        if (orderConfirmed) {
+            throw new OrderAlreadyConfirmedException(command.getOrderId());
+        }
+
+        apply(new ProductCountIncrementedEvent(command.getOrderId(), productId));
+    }
+
+    @CommandHandler
+    public void handle(DecrementProductCountCommand command) {
+        if (orderConfirmed) {
+            throw new OrderAlreadyConfirmedException(command.getOrderId());
+        }
+
+        if (count <= 1) {
+            apply(new ProductRemovedEvent(command.getOrderId(), productId));
+        } else {
+            apply(new ProductCountDecrementedEvent(command.getOrderId(), productId));
+        }
+    }
+
+    @EventSourcingHandler
+    public void on(ProductCountIncrementedEvent event) {
+        this.count++;
+    }
+
+    @EventSourcingHandler
+    public void on(ProductCountDecrementedEvent event) {
+        this.count--;
+    }
+
+    @EventSourcingHandler
+    public void on(OrderConfirmedEvent event) {
+        this.orderConfirmed = true;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        OrderLine orderLine = (OrderLine) o;
+        return Objects.equals(productId, orderLine.productId) && Objects.equals(count, orderLine.count);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(productId, count);
+    }
+}

axon/src/main/java/com/baeldung/axon/coreapi/commands/AddProductCommand.java

@@ -0,0 +1,50 @@
+package com.baeldung.axon.coreapi.commands;
+
+import org.axonframework.modelling.command.TargetAggregateIdentifier;
+
+import java.util.Objects;
+
+public class AddProductCommand {
+
+    @TargetAggregateIdentifier
+    private final String orderId;
+    private final String productId;
+
+    public AddProductCommand(String orderId, String productId) {
+        this.orderId = orderId;
+        this.productId = productId;
+    }
+
+    public String getOrderId() {
+        return orderId;
+    }
+
+    public String getProductId() {
+        return productId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        AddProductCommand that = (AddProductCommand) o;
+        return Objects.equals(orderId, that.orderId) && Objects.equals(productId, that.productId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(orderId, productId);
+    }
+
+    @Override
+    public String toString() {
+        return "AddProductCommand{" +
+                "orderId='" + orderId + '\'' +
+                ", productId='" + productId + '\'' +
+                '}';
+    }
+}

axon/src/main/java/com/baeldung/axon/coreapi/commands/CreateOrderCommand.java

@@ -0,0 +1,43 @@
+package com.baeldung.axon.coreapi.commands;
+
+import org.axonframework.modelling.command.TargetAggregateIdentifier;
+
+import java.util.Objects;
+
+public class CreateOrderCommand {
+
+    @TargetAggregateIdentifier
+    private final String orderId;
+
+    public CreateOrderCommand(String orderId) {
+        this.orderId = orderId;
+    }
+
+    public String getOrderId() {
+        return orderId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        CreateOrderCommand that = (CreateOrderCommand) o;
+        return Objects.equals(orderId, that.orderId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(orderId);
+    }
+
+    @Override
+    public String toString() {
+        return "CreateOrderCommand{" +
+                "orderId='" + orderId + '\'' +
+                '}';
+    }
+}

axon/src/main/java/com/baeldung/axon/coreapi/commands/DecrementProductCountCommand.java

@@ -0,0 +1,50 @@
+package com.baeldung.axon.coreapi.commands;
+
+import org.axonframework.modelling.command.TargetAggregateIdentifier;
+
+import java.util.Objects;
+
+public class DecrementProductCountCommand {
+
+    @TargetAggregateIdentifier
+    private final String orderId;
+    private final String productId;
+
+    public DecrementProductCountCommand(String orderId, String productId) {
+        this.orderId = orderId;
+        this.productId = productId;
+    }
+
+    public String getOrderId() {
+        return orderId;
+    }
+
+    public String getProductId() {
+        return productId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        DecrementProductCountCommand that = (DecrementProductCountCommand) o;
+        return Objects.equals(orderId, that.orderId) && Objects.equals(productId, that.productId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(orderId, productId);
+    }
+
+    @Override
+    public String toString() {
+        return "DecrementProductCountCommand{" +
+                "orderId='" + orderId + '\'' +
+                ", productId='" + productId + '\'' +
+                '}';
+    }
+}

axon/src/main/java/com/baeldung/axon/coreapi/commands/IncrementProductCountCommand.java

@@ -0,0 +1,50 @@
+package com.baeldung.axon.coreapi.commands;
+
+import org.axonframework.modelling.command.TargetAggregateIdentifier;
+
+import java.util.Objects;
+
+public class IncrementProductCountCommand {
+
+    @TargetAggregateIdentifier
+    private final String orderId;
+    private final String productId;
+
+    public IncrementProductCountCommand(String orderId, String productId) {
+        this.orderId = orderId;
+        this.productId = productId;
+    }
+
+    public String getOrderId() {
+        return orderId;
+    }
+
+    public String getProductId() {
+        return productId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        IncrementProductCountCommand that = (IncrementProductCountCommand) o;
+        return Objects.equals(orderId, that.orderId) && Objects.equals(productId, that.productId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(orderId, productId);
+    }
+
+    @Override
+    public String toString() {
+        return "IncrementProductCountCommand{" +
+                "orderId='" + orderId + '\'' +
+                ", productId='" + productId + '\'' +
+                '}';
+    }
+}

axon/src/main/java/com/baeldung/axon/coreapi/commands/PlaceOrderCommand.java

@@ -1,51 +0,0 @@
-package com.baeldung.axon.coreapi.commands;
-
-import java.util.Objects;
-
-import org.axonframework.modelling.command.TargetAggregateIdentifier;
-
-public class PlaceOrderCommand {
-
-    @TargetAggregateIdentifier
-    private final String orderId;
-    private final String product;
-
-    public PlaceOrderCommand(String orderId, String product) {
-        this.orderId = orderId;
-        this.product = product;
-    }
-
-    public String getOrderId() {
-        return orderId;
-    }
-
-    public String getProduct() {
-        return product;
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(orderId, product);
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-        if (this == obj) {
-            return true;
-        }
-        if (obj == null || getClass() != obj.getClass()) {
-            return false;
-        }
-        final PlaceOrderCommand other = (PlaceOrderCommand) obj;
-        return Objects.equals(this.orderId, other.orderId)
-                && Objects.equals(this.product, other.product);
-    }
-
-    @Override
-    public String toString() {
-        return "PlaceOrderCommand{" +
-                "orderId='" + orderId + '\'' +
-                ", product='" + product + '\'' +
-                '}';
-    }
-}

axon/src/main/java/com/baeldung/axon/coreapi/events/OrderCreatedEvent.java

@@ -0,0 +1,40 @@
+package com.baeldung.axon.coreapi.events;
+
+import java.util.Objects;
+
+public class OrderCreatedEvent {
+
+    private final String orderId;
+
+    public OrderCreatedEvent(String orderId) {
+        this.orderId = orderId;
+    }
+
+    public String getOrderId() {
+        return orderId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        OrderCreatedEvent that = (OrderCreatedEvent) o;
+        return Objects.equals(orderId, that.orderId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(orderId);
+    }
+
+    @Override
+    public String toString() {
+        return "OrderCreatedEvent{" +
+                "orderId='" + orderId + '\'' +
+                '}';
+    }
+}

axon/src/main/java/com/baeldung/axon/coreapi/events/OrderPlacedEvent.java

@@ -1,48 +0,0 @@
-package com.baeldung.axon.coreapi.events;
-
-import java.util.Objects;
-
-public class OrderPlacedEvent {
-
-    private final String orderId;
-    private final String product;
-
-    public OrderPlacedEvent(String orderId, String product) {
-        this.orderId = orderId;
-        this.product = product;
-    }
-
-    public String getOrderId() {
-        return orderId;
-    }
-
-    public String getProduct() {
-        return product;
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(orderId, product);
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-        if (this == obj) {
-            return true;
-        }
-        if (obj == null || getClass() != obj.getClass()) {
-            return false;
-        }
-        final OrderPlacedEvent other = (OrderPlacedEvent) obj;
-        return Objects.equals(this.orderId, other.orderId)
-                && Objects.equals(this.product, other.product);
-    }
-
-    @Override
-    public String toString() {
-        return "OrderPlacedEvent{" +
-                "orderId='" + orderId + '\'' +
-                ", product='" + product + '\'' +
-                '}';
-    }
-}

axon/src/main/java/com/baeldung/axon/coreapi/events/ProductAddedEvent.java

@@ -0,0 +1,47 @@
+package com.baeldung.axon.coreapi.events;
+
+import java.util.Objects;
+
+public class ProductAddedEvent {
+
+    private final String orderId;
+    private final String productId;
+
+    public ProductAddedEvent(String orderId, String productId) {
+        this.orderId = orderId;
+        this.productId = productId;
+    }
+
+    public String getOrderId() {
+        return orderId;
+    }
+
+    public String getProductId() {
+        return productId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ProductAddedEvent that = (ProductAddedEvent) o;
+        return Objects.equals(orderId, that.orderId) && Objects.equals(productId, that.productId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(orderId, productId);
+    }
+
+    @Override
+    public String toString() {
+        return "ProductAddedEvent{" +
+                "orderId='" + orderId + '\'' +
+                ", productId='" + productId + '\'' +
+                '}';
+    }
+}

axon/src/main/java/com/baeldung/axon/coreapi/events/ProductCountDecrementedEvent.java

@@ -0,0 +1,47 @@
+package com.baeldung.axon.coreapi.events;
+
+import java.util.Objects;
+
+public class ProductCountDecrementedEvent {
+
+    private final String orderId;
+    private final String productId;
+
+    public ProductCountDecrementedEvent(String orderId, String productId) {
+        this.orderId = orderId;
+        this.productId = productId;
+    }
+
+    public String getOrderId() {
+        return orderId;
+    }
+
+    public String getProductId() {
+        return productId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ProductCountDecrementedEvent that = (ProductCountDecrementedEvent) o;
+        return Objects.equals(orderId, that.orderId) && Objects.equals(productId, that.productId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(orderId, productId);
+    }
+
+    @Override
+    public String toString() {
+        return "ProductCountDecrementedEvent{" +
+                "orderId='" + orderId + '\'' +
+                ", productId='" + productId + '\'' +
+                '}';
+    }
+}

axon/src/main/java/com/baeldung/axon/coreapi/events/ProductCountIncrementedEvent.java

@@ -0,0 +1,47 @@
+package com.baeldung.axon.coreapi.events;
+
+import java.util.Objects;
+
+public class ProductCountIncrementedEvent {
+
+    private final String orderId;
+    private final String productId;
+
+    public ProductCountIncrementedEvent(String orderId, String productId) {
+        this.orderId = orderId;
+        this.productId = productId;
+    }
+
+    public String getOrderId() {
+        return orderId;
+    }
+
+    public String getProductId() {
+        return productId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ProductCountIncrementedEvent that = (ProductCountIncrementedEvent) o;
+        return Objects.equals(orderId, that.orderId) && Objects.equals(productId, that.productId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(orderId, productId);
+    }
+
+    @Override
+    public String toString() {
+        return "ProductCountIncrementedEvent{" +
+                "orderId='" + orderId + '\'' +
+                ", productId='" + productId + '\'' +
+                '}';
+    }
+}

axon/src/main/java/com/baeldung/axon/coreapi/events/ProductRemovedEvent.java

@@ -0,0 +1,47 @@
+package com.baeldung.axon.coreapi.events;
+
+import java.util.Objects;
+
+public class ProductRemovedEvent {
+
+    private final String orderId;
+    private final String productId;
+
+    public ProductRemovedEvent(String orderId, String productId) {
+        this.orderId = orderId;
+        this.productId = productId;
+    }
+
+    public String getOrderId() {
+        return orderId;
+    }
+
+    public String getProductId() {
+        return productId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ProductRemovedEvent that = (ProductRemovedEvent) o;
+        return Objects.equals(orderId, that.orderId) && Objects.equals(productId, that.productId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(orderId, productId);
+    }
+
+    @Override
+    public String toString() {
+        return "ProductRemovedEvent{" +
+                "orderId='" + orderId + '\'' +
+                ", productId='" + productId + '\'' +
+                '}';
+    }
+}

axon/src/main/java/com/baeldung/axon/coreapi/exceptions/DuplicateOrderLineException.java

@@ -0,0 +1,8 @@
+package com.baeldung.axon.coreapi.exceptions;
+
+public class DuplicateOrderLineException extends IllegalStateException {
+
+    public DuplicateOrderLineException(String productId) {
+        super("Cannot duplicate order line for product identifier [" + productId + "]");
+    }
+}

axon/src/main/java/com/baeldung/axon/coreapi/exceptions/OrderAlreadyConfirmedException.java

@@ -0,0 +1,8 @@
+package com.baeldung.axon.coreapi.exceptions;
+
+public class OrderAlreadyConfirmedException extends IllegalStateException {
+
+    public OrderAlreadyConfirmedException(String orderId) {
+        super("Cannot perform operation because order [" + orderId + "] is already confirmed.");
+    }
+}

axon/src/main/java/com/baeldung/axon/coreapi/queries/Order.java

@@ -0,0 +1,83 @@
+package com.baeldung.axon.coreapi.queries;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+public class Order {
+
+    private final String orderId;
+    private final Map<String, Integer> products;
+    private OrderStatus orderStatus;
+
+    public Order(String orderId) {
+        this.orderId = orderId;
+        this.products = new HashMap<>();
+        orderStatus = OrderStatus.CREATED;
+    }
+
+    public String getOrderId() {
+        return orderId;
+    }
+
+    public Map<String, Integer> getProducts() {
+        return products;
+    }
+
+    public OrderStatus getOrderStatus() {
+        return orderStatus;
+    }
+
+    public void addProduct(String productId) {
+        products.putIfAbsent(productId, 1);
+    }
+
+    public void incrementProductInstance(String productId) {
+        products.computeIfPresent(productId, (id, count) -> ++count);
+    }
+
+    public void decrementProductInstance(String productId) {
+        products.computeIfPresent(productId, (id, count) -> --count);
+    }
+
+
+    public void removeProduct(String productId) {
+        products.remove(productId);
+    }
+
+    public void setOrderConfirmed() {
+        this.orderStatus = OrderStatus.CONFIRMED;
+    }
+
+    public void setOrderShipped() {
+        this.orderStatus = OrderStatus.SHIPPED;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        Order that = (Order) o;
+        return Objects.equals(orderId, that.orderId)
+                && Objects.equals(products, that.products)
+                && orderStatus == that.orderStatus;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(orderId, products, orderStatus);
+    }
+
+    @Override
+    public String toString() {
+        return "Order{" +
+                "orderId='" + orderId + '\'' +
+                ", products=" + products +
+                ", orderStatus=" + orderStatus +
+                '}';
+    }
+}

axon/src/main/java/com/baeldung/axon/coreapi/queries/OrderStatus.java

@@ -2,6 +2,5 @@ package com.baeldung.axon.coreapi.queries;
 
 public enum OrderStatus {
 
-    PLACED, CONFIRMED, SHIPPED
+    CREATED, CONFIRMED, SHIPPED
-
 }

axon/src/main/java/com/baeldung/axon/coreapi/queries/OrderedProduct.java

@@ -1,64 +0,0 @@
-package com.baeldung.axon.coreapi.queries;
-
-import java.util.Objects;
-
-public class OrderedProduct {
-
-    private final String orderId;
-    private final String product;
-    private OrderStatus orderStatus;
-
-    public OrderedProduct(String orderId, String product) {
-        this.orderId = orderId;
-        this.product = product;
-        orderStatus = OrderStatus.PLACED;
-    }
-
-    public String getOrderId() {
-        return orderId;
-    }
-
-    public String getProduct() {
-        return product;
-    }
-
-    public OrderStatus getOrderStatus() {
-        return orderStatus;
-    }
-
-    public void setOrderConfirmed() {
-        this.orderStatus = OrderStatus.CONFIRMED;
-    }
-
-    public void setOrderShipped() {
-        this.orderStatus = OrderStatus.SHIPPED;
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(orderId, product, orderStatus);
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-        if (this == obj) {
-            return true;
-        }
-        if (obj == null || getClass() != obj.getClass()) {
-            return false;
-        }
-        final OrderedProduct other = (OrderedProduct) obj;
-        return Objects.equals(this.orderId, other.orderId)
-                && Objects.equals(this.product, other.product)
-                && Objects.equals(this.orderStatus, other.orderStatus);
-    }
-
-    @Override
-    public String toString() {
-        return "OrderedProduct{" +
-                "orderId='" + orderId + '\'' +
-                ", product='" + product + '\'' +
-                ", orderStatus=" + orderStatus +
-                '}';
-    }
-}

axon/src/main/java/com/baeldung/axon/gui/OrderRestEndpoint.java

@@ -1,20 +1,24 @@
 package com.baeldung.axon.gui;
 
-import java.util.List;
+import com.baeldung.axon.coreapi.commands.AddProductCommand;
-import java.util.UUID;
+import com.baeldung.axon.coreapi.commands.ConfirmOrderCommand;
-
+import com.baeldung.axon.coreapi.commands.CreateOrderCommand;
+import com.baeldung.axon.coreapi.commands.DecrementProductCountCommand;
+import com.baeldung.axon.coreapi.commands.IncrementProductCountCommand;
+import com.baeldung.axon.coreapi.commands.ShipOrderCommand;
+import com.baeldung.axon.coreapi.queries.FindAllOrderedProductsQuery;
+import com.baeldung.axon.coreapi.queries.Order;
 import org.axonframework.commandhandling.gateway.CommandGateway;
 import org.axonframework.messaging.responsetypes.ResponseTypes;
 import org.axonframework.queryhandling.QueryGateway;
 import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RestController;
 
-import com.baeldung.axon.coreapi.commands.ConfirmOrderCommand;
+import java.util.List;
-import com.baeldung.axon.coreapi.commands.PlaceOrderCommand;
+import java.util.UUID;
-import com.baeldung.axon.coreapi.commands.ShipOrderCommand;
+import java.util.concurrent.CompletableFuture;
-import com.baeldung.axon.coreapi.queries.FindAllOrderedProductsQuery;
-import com.baeldung.axon.coreapi.queries.OrderedProduct;
 
 @RestController
 public class OrderRestEndpoint {
@@ -28,25 +32,63 @@ public class OrderRestEndpoint {
     }
 
     @PostMapping("/ship-order")
-    public void shipOrder() {
+    public CompletableFuture<Void> shipOrder() {
         String orderId = UUID.randomUUID().toString();
-        commandGateway.send(new PlaceOrderCommand(orderId, "Deluxe Chair"));
+        return commandGateway.send(new CreateOrderCommand(orderId))
-        commandGateway.send(new ConfirmOrderCommand(orderId));
+                             .thenCompose(result -> commandGateway.send(new AddProductCommand(orderId, "Deluxe Chair")))
-        commandGateway.send(new ShipOrderCommand(orderId));
+                             .thenCompose(result -> commandGateway.send(new ConfirmOrderCommand(orderId)))
+                             .thenCompose(result -> commandGateway.send(new ShipOrderCommand(orderId)));
     }
 
     @PostMapping("/ship-unconfirmed-order")
-    public void shipUnconfirmedOrder() {
+    public CompletableFuture<Void> shipUnconfirmedOrder() {
         String orderId = UUID.randomUUID().toString();
-        commandGateway.send(new PlaceOrderCommand(orderId, "Deluxe Chair"));
+        return commandGateway.send(new CreateOrderCommand(orderId))
-        // This throws an exception, as an Order cannot be shipped if it has not been confirmed yet.
+                             .thenCompose(result -> commandGateway.send(new AddProductCommand(orderId, "Deluxe Chair")))
-        commandGateway.send(new ShipOrderCommand(orderId));
+                             // This throws an exception, as an Order cannot be shipped if it has not been confirmed yet.
+                             .thenCompose(result -> commandGateway.send(new ShipOrderCommand(orderId)));
+    }
+
+    @PostMapping("/order")
+    public CompletableFuture<String> createOrder() {
+        return createOrder(UUID.randomUUID().toString());
+    }
+
+    @PostMapping("/order/{order-id}")
+    public CompletableFuture<String> createOrder(@PathVariable("order-id") String orderId) {
+        return commandGateway.send(new CreateOrderCommand(orderId));
+    }
+
+    @PostMapping("/order/{order-id}/product/{product-id}")
+    public CompletableFuture<Void> addProduct(@PathVariable("order-id") String orderId,
+                                              @PathVariable("product-id") String productId) {
+        return commandGateway.send(new AddProductCommand(orderId, productId));
+    }
+
+    @PostMapping("/order/{order-id}/product/{product-id}/increment")
+    public CompletableFuture<Void> incrementProduct(@PathVariable("order-id") String orderId,
+                                                    @PathVariable("product-id") String productId) {
+        return commandGateway.send(new IncrementProductCountCommand(orderId, productId));
+    }
+
+    @PostMapping("/order/{order-id}/product/{product-id}/decrement")
+    public CompletableFuture<Void> decrementProduct(@PathVariable("order-id") String orderId,
+                                                    @PathVariable("product-id") String productId) {
+        return commandGateway.send(new DecrementProductCountCommand(orderId, productId));
+    }
+
+    @PostMapping("/order/{order-id}/confirm")
+    public CompletableFuture<Void> confirmOrder(@PathVariable("order-id") String orderId) {
+        return commandGateway.send(new ConfirmOrderCommand(orderId));
+    }
+
+    @PostMapping("/order/{order-id}/ship")
+    public CompletableFuture<Void> shipOrder(@PathVariable("order-id") String orderId) {
+        return commandGateway.send(new ShipOrderCommand(orderId));
     }
 
     @GetMapping("/all-orders")
-    public List<OrderedProduct> findAllOrderedProducts() {
+    public CompletableFuture<List<Order>> findAllOrders() {
-        return queryGateway.query(new FindAllOrderedProductsQuery(), ResponseTypes.multipleInstancesOf(OrderedProduct.class))
+        return queryGateway.query(new FindAllOrderedProductsQuery(), ResponseTypes.multipleInstancesOf(Order.class));
-                           .join();
     }
-
 }

axon/src/main/java/com/baeldung/axon/querymodel/OrderedProductsEventHandler.java

@@ -1,52 +0,0 @@
-package com.baeldung.axon.querymodel;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.axonframework.config.ProcessingGroup;
-import org.axonframework.eventhandling.EventHandler;
-import org.axonframework.queryhandling.QueryHandler;
-import org.springframework.stereotype.Service;
-
-import com.baeldung.axon.coreapi.events.OrderConfirmedEvent;
-import com.baeldung.axon.coreapi.events.OrderPlacedEvent;
-import com.baeldung.axon.coreapi.events.OrderShippedEvent;
-import com.baeldung.axon.coreapi.queries.FindAllOrderedProductsQuery;
-import com.baeldung.axon.coreapi.queries.OrderedProduct;
-
-@Service
-@ProcessingGroup("ordered-products")
-public class OrderedProductsEventHandler {
-
-    private final Map<String, OrderedProduct> orderedProducts = new HashMap<>();
-
-    @EventHandler
-    public void on(OrderPlacedEvent event) {
-        String orderId = event.getOrderId();
-        orderedProducts.put(orderId, new OrderedProduct(orderId, event.getProduct()));
-    }
-
-    @EventHandler
-    public void on(OrderConfirmedEvent event) {
-        orderedProducts.computeIfPresent(event.getOrderId(), (orderId, orderedProduct) -> {
-            orderedProduct.setOrderConfirmed();
-            return orderedProduct;
-        });
-    }
-
-    @EventHandler
-    public void on(OrderShippedEvent event) {
-        orderedProducts.computeIfPresent(event.getOrderId(), (orderId, orderedProduct) -> {
-            orderedProduct.setOrderShipped();
-            return orderedProduct;
-        });
-    }
-
-    @QueryHandler
-    public List<OrderedProduct> handle(FindAllOrderedProductsQuery query) {
-        return new ArrayList<>(orderedProducts.values());
-    }
-
-}

axon/src/main/java/com/baeldung/axon/querymodel/OrdersEventHandler.java

@@ -0,0 +1,86 @@
+package com.baeldung.axon.querymodel;
+
+import com.baeldung.axon.coreapi.events.OrderConfirmedEvent;
+import com.baeldung.axon.coreapi.events.OrderCreatedEvent;
+import com.baeldung.axon.coreapi.events.OrderShippedEvent;
+import com.baeldung.axon.coreapi.events.ProductAddedEvent;
+import com.baeldung.axon.coreapi.events.ProductCountDecrementedEvent;
+import com.baeldung.axon.coreapi.events.ProductCountIncrementedEvent;
+import com.baeldung.axon.coreapi.events.ProductRemovedEvent;
+import com.baeldung.axon.coreapi.queries.FindAllOrderedProductsQuery;
+import com.baeldung.axon.coreapi.queries.Order;
+import org.axonframework.config.ProcessingGroup;
+import org.axonframework.eventhandling.EventHandler;
+import org.axonframework.queryhandling.QueryHandler;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Service
+@ProcessingGroup("orders")
+public class OrdersEventHandler {
+
+    private final Map<String, Order> orders = new HashMap<>();
+
+    @EventHandler
+    public void on(OrderCreatedEvent event) {
+        String orderId = event.getOrderId();
+        orders.put(orderId, new Order(orderId));
+    }
+
+    @EventHandler
+    public void on(ProductAddedEvent event) {
+        orders.computeIfPresent(event.getOrderId(), (orderId, order) -> {
+            order.addProduct(event.getProductId());
+            return order;
+        });
+    }
+
+    @EventHandler
+    public void on(ProductCountIncrementedEvent event) {
+        orders.computeIfPresent(event.getOrderId(), (orderId, order) -> {
+            order.incrementProductInstance(event.getProductId());
+            return order;
+        });
+    }
+
+    @EventHandler
+    public void on(ProductCountDecrementedEvent event) {
+        orders.computeIfPresent(event.getOrderId(), (orderId, order) -> {
+            order.decrementProductInstance(event.getProductId());
+            return order;
+        });
+    }
+
+    @EventHandler
+    public void on(ProductRemovedEvent event) {
+        orders.computeIfPresent(event.getOrderId(), (orderId, order) -> {
+            order.removeProduct(event.getProductId());
+            return order;
+        });
+    }
+
+    @EventHandler
+    public void on(OrderConfirmedEvent event) {
+        orders.computeIfPresent(event.getOrderId(), (orderId, order) -> {
+            order.setOrderConfirmed();
+            return order;
+        });
+    }
+
+    @EventHandler
+    public void on(OrderShippedEvent event) {
+        orders.computeIfPresent(event.getOrderId(), (orderId, order) -> {
+            order.setOrderShipped();
+            return order;
+        });
+    }
+
+    @QueryHandler
+    public List<Order> handle(FindAllOrderedProductsQuery query) {
+        return new ArrayList<>(orders.values());
+    }
+}

axon/src/main/resources/order-api.http

@@ -1,11 +1,37 @@
+### Create Order, Add Product, Confirm and Ship Order
+
 POST http://localhost:8080/ship-order
 
-###
+### Create Order, Add Product and Ship Order
 
 POST http://localhost:8080/ship-unconfirmed-order
 
-###
+### Retrieve all existing Orders
 
 GET http://localhost:8080/all-orders
 
+### Create Order with id 666a1661-474d-4046-8b12-8b5896312768
+
+POST http://localhost:8080/order/666a1661-474d-4046-8b12-8b5896312768
+
+### Add Product a6aa01eb-4e38-4dfb-b53b-b5b82961fbf3 to Order 666a1661-474d-4046-8b12-8b5896312768
+
+POST http://localhost:8080/order/666a1661-474d-4046-8b12-8b5896312768/product/a6aa01eb-4e38-4dfb-b53b-b5b82961fbf3
+
+### Increment Product a6aa01eb-4e38-4dfb-b53b-b5b82961fbf3 to Order 666a1661-474d-4046-8b12-8b5896312768
+
+POST http://localhost:8080/order/666a1661-474d-4046-8b12-8b5896312768/product/a6aa01eb-4e38-4dfb-b53b-b5b82961fbf3/increment
+
+### Decrement Product a6aa01eb-4e38-4dfb-b53b-b5b82961fbf3 to Order 666a1661-474d-4046-8b12-8b5896312768
+
+POST http://localhost:8080/order/666a1661-474d-4046-8b12-8b5896312768/product/a6aa01eb-4e38-4dfb-b53b-b5b82961fbf3/decrement
+
+### Confirm Order 666a1661-474d-4046-8b12-8b5896312768
+
+POST http://localhost:8080/order/666a1661-474d-4046-8b12-8b5896312768/confirm
+
+### Ship Order 666a1661-474d-4046-8b12-8b5896312768
+
+POST http://localhost:8080/order/666a1661-474d-4046-8b12-8b5896312768/ship
+
 ###

axon/src/test/java/com/baeldung/axon/commandmodel/OrderAggregateUnitTest.java

@@ -1,62 +1,139 @@
 package com.baeldung.axon.commandmodel;
 
-import java.util.UUID;
+import com.baeldung.axon.commandmodel.order.OrderAggregate;
-
+import com.baeldung.axon.coreapi.commands.AddProductCommand;
+import com.baeldung.axon.coreapi.commands.ConfirmOrderCommand;
+import com.baeldung.axon.coreapi.commands.CreateOrderCommand;
+import com.baeldung.axon.coreapi.commands.DecrementProductCountCommand;
+import com.baeldung.axon.coreapi.commands.IncrementProductCountCommand;
+import com.baeldung.axon.coreapi.commands.ShipOrderCommand;
+import com.baeldung.axon.coreapi.events.OrderConfirmedEvent;
+import com.baeldung.axon.coreapi.events.OrderCreatedEvent;
+import com.baeldung.axon.coreapi.events.OrderShippedEvent;
+import com.baeldung.axon.coreapi.events.ProductAddedEvent;
+import com.baeldung.axon.coreapi.events.ProductCountDecrementedEvent;
+import com.baeldung.axon.coreapi.events.ProductCountIncrementedEvent;
+import com.baeldung.axon.coreapi.events.ProductRemovedEvent;
+import com.baeldung.axon.coreapi.exceptions.DuplicateOrderLineException;
+import com.baeldung.axon.coreapi.exceptions.OrderAlreadyConfirmedException;
 import com.baeldung.axon.coreapi.exceptions.UnconfirmedOrderException;
 import org.axonframework.test.aggregate.AggregateTestFixture;
 import org.axonframework.test.aggregate.FixtureConfiguration;
-import org.junit.*;
+import org.axonframework.test.matchers.Matchers;
+import org.junit.jupiter.api.*;
 
-import com.baeldung.axon.coreapi.commands.ConfirmOrderCommand;
+import java.util.UUID;
-import com.baeldung.axon.coreapi.commands.PlaceOrderCommand;
-import com.baeldung.axon.coreapi.commands.ShipOrderCommand;
-import com.baeldung.axon.coreapi.events.OrderConfirmedEvent;
-import com.baeldung.axon.coreapi.events.OrderPlacedEvent;
-import com.baeldung.axon.coreapi.events.OrderShippedEvent;
 
-public class OrderAggregateUnitTest {
+class OrderAggregateUnitTest {
+
+    private static final String ORDER_ID = UUID.randomUUID().toString();
+    private static final String PRODUCT_ID = UUID.randomUUID().toString();
 
     private FixtureConfiguration<OrderAggregate> fixture;
 
-    @Before
+    @BeforeEach
-    public void setUp() {
+    void setUp() {
         fixture = new AggregateTestFixture<>(OrderAggregate.class);
     }
 
     @Test
-    public void giveNoPriorActivity_whenPlaceOrderCommand_thenShouldPublishOrderPlacedEvent() {
+    void giveNoPriorActivity_whenCreateOrderCommand_thenShouldPublishOrderCreatedEvent() {
-        String orderId = UUID.randomUUID().toString();
-        String product = "Deluxe Chair";
         fixture.givenNoPriorActivity()
-               .when(new PlaceOrderCommand(orderId, product))
+               .when(new CreateOrderCommand(ORDER_ID))
-               .expectEvents(new OrderPlacedEvent(orderId, product));
+               .expectEvents(new OrderCreatedEvent(ORDER_ID));
     }
 
     @Test
-    public void givenOrderPlacedEvent_whenConfirmOrderCommand_thenShouldPublishOrderConfirmedEvent() {
+    void givenOrderCreatedEvent_whenAddProductCommand_thenShouldPublishProductAddedEvent() {
-        String orderId = UUID.randomUUID().toString();
+        fixture.given(new OrderCreatedEvent(ORDER_ID))
-        String product = "Deluxe Chair";
+               .when(new AddProductCommand(ORDER_ID, PRODUCT_ID))
-        fixture.given(new OrderPlacedEvent(orderId, product))
+               .expectEvents(new ProductAddedEvent(ORDER_ID, PRODUCT_ID));
-               .when(new ConfirmOrderCommand(orderId))
-               .expectEvents(new OrderConfirmedEvent(orderId));
     }
 
     @Test
-    public void givenOrderPlacedEvent_whenShipOrderCommand_thenShouldThrowUnconfirmedOrderException() {
+    void givenOrderCreatedEventAndProductAddedEvent_whenAddProductCommandForSameProductId_thenShouldThrowDuplicateOrderLineException() {
-        String orderId = UUID.randomUUID().toString();
+        fixture.given(new OrderCreatedEvent(ORDER_ID), new ProductAddedEvent(ORDER_ID, PRODUCT_ID))
-        String product = "Deluxe Chair";
+               .when(new AddProductCommand(ORDER_ID, PRODUCT_ID))
-        fixture.given(new OrderPlacedEvent(orderId, product))
+               .expectException(DuplicateOrderLineException.class)
-               .when(new ShipOrderCommand(orderId))
+               .expectExceptionMessage(Matchers.predicate(message -> ((String) message).contains(PRODUCT_ID)));
+    }
+
+    @Test
+    void givenOrderCreatedEventAndProductAddedEvent_whenIncrementProductCountCommand_thenShouldPublishProductCountIncrementedEvent() {
+        fixture.given(new OrderCreatedEvent(ORDER_ID), new ProductAddedEvent(ORDER_ID, PRODUCT_ID))
+               .when(new IncrementProductCountCommand(ORDER_ID, PRODUCT_ID))
+               .expectEvents(new ProductCountIncrementedEvent(ORDER_ID, PRODUCT_ID));
+    }
+
+    @Test
+    void givenOrderCreatedEventProductAddedEventAndProductCountIncrementedEvent_whenDecrementProductCountCommand_thenShouldPublishProductCountDecrementedEvent() {
+        fixture.given(new OrderCreatedEvent(ORDER_ID),
+                      new ProductAddedEvent(ORDER_ID, PRODUCT_ID),
+                      new ProductCountIncrementedEvent(ORDER_ID, PRODUCT_ID))
+               .when(new DecrementProductCountCommand(ORDER_ID, PRODUCT_ID))
+               .expectEvents(new ProductCountDecrementedEvent(ORDER_ID, PRODUCT_ID));
+    }
+
+    @Test
+    void givenOrderCreatedEventAndProductAddedEvent_whenDecrementProductCountCommand_thenShouldPublishProductRemovedEvent() {
+        fixture.given(new OrderCreatedEvent(ORDER_ID), new ProductAddedEvent(ORDER_ID, PRODUCT_ID))
+               .when(new DecrementProductCountCommand(ORDER_ID, PRODUCT_ID))
+               .expectEvents(new ProductRemovedEvent(ORDER_ID, PRODUCT_ID));
+    }
+
+    @Test
+    void givenOrderCreatedEvent_whenConfirmOrderCommand_thenShouldPublishOrderConfirmedEvent() {
+        fixture.given(new OrderCreatedEvent(ORDER_ID))
+               .when(new ConfirmOrderCommand(ORDER_ID))
+               .expectEvents(new OrderConfirmedEvent(ORDER_ID));
+    }
+
+    @Test
+    void givenOrderCreatedEventAndOrderConfirmedEvent_whenConfirmOrderCommand_thenExpectNoEvents() {
+        fixture.given(new OrderCreatedEvent(ORDER_ID), new OrderConfirmedEvent(ORDER_ID))
+               .when(new ConfirmOrderCommand(ORDER_ID))
+               .expectNoEvents();
+    }
+
+    @Test
+    void givenOrderCreatedEvent_whenShipOrderCommand_thenShouldThrowUnconfirmedOrderException() {
+        fixture.given(new OrderCreatedEvent(ORDER_ID))
+               .when(new ShipOrderCommand(ORDER_ID))
                .expectException(UnconfirmedOrderException.class);
     }
 
     @Test
-    public void givenOrderPlacedEventAndOrderConfirmedEvent_whenShipOrderCommand_thenShouldPublishOrderShippedEvent() {
+    void givenOrderCreatedEventAndOrderConfirmedEvent_whenShipOrderCommand_thenShouldPublishOrderShippedEvent() {
-        String orderId = UUID.randomUUID().toString();
+        fixture.given(new OrderCreatedEvent(ORDER_ID), new OrderConfirmedEvent(ORDER_ID))
-        String product = "Deluxe Chair";
+               .when(new ShipOrderCommand(ORDER_ID))
-        fixture.given(new OrderPlacedEvent(orderId, product), new OrderConfirmedEvent(orderId))
+               .expectEvents(new OrderShippedEvent(ORDER_ID));
-               .when(new ShipOrderCommand(orderId))
-               .expectEvents(new OrderShippedEvent(orderId));
     }
 
+    @Test
+    void givenOrderCreatedEventProductAndOrderConfirmedEvent_whenAddProductCommand_thenShouldThrowOrderAlreadyConfirmedException() {
+        fixture.given(new OrderCreatedEvent(ORDER_ID), new OrderConfirmedEvent(ORDER_ID))
+               .when(new AddProductCommand(ORDER_ID, PRODUCT_ID))
+               .expectException(OrderAlreadyConfirmedException.class)
+               .expectExceptionMessage(Matchers.predicate(message -> ((String) message).contains(ORDER_ID)));
+    }
+
+    @Test
+    void givenOrderCreatedEventProductAddedEventAndOrderConfirmedEvent_whenIncrementProductCountCommand_thenShouldThrowOrderAlreadyConfirmedException() {
+        fixture.given(new OrderCreatedEvent(ORDER_ID),
+                      new ProductAddedEvent(ORDER_ID, PRODUCT_ID),
+                      new OrderConfirmedEvent(ORDER_ID))
+               .when(new IncrementProductCountCommand(ORDER_ID, PRODUCT_ID))
+               .expectException(OrderAlreadyConfirmedException.class)
+               .expectExceptionMessage(Matchers.predicate(message -> ((String) message).contains(ORDER_ID)));
+    }
+
+    @Test
+    void givenOrderCreatedEventProductAddedEventAndOrderConfirmedEvent_whenDecrementProductCountCommand_thenShouldThrowOrderAlreadyConfirmedException() {
+        fixture.given(new OrderCreatedEvent(ORDER_ID),
+                      new ProductAddedEvent(ORDER_ID, PRODUCT_ID),
+                      new OrderConfirmedEvent(ORDER_ID))
+               .when(new DecrementProductCountCommand(ORDER_ID, PRODUCT_ID))
+               .expectException(OrderAlreadyConfirmedException.class)
+               .expectExceptionMessage(Matchers.predicate(message -> ((String) message).contains(ORDER_ID)));
+    }
 }

core-java-modules/core-java-concurrency-basic-2/src/main/java/com/baeldung/concurrent/interrupt/CustomInterruptedException.java

@@ -0,0 +1,10 @@
+package com.baeldung.concurrent.interrupt;
+
+public class CustomInterruptedException extends Exception {
+   
+    private static final long serialVersionUID = 1L;
+
+    CustomInterruptedException(String message) {
+        super(message);
+    }
+}

core-java-modules/core-java-concurrency-basic-2/src/main/java/com/baeldung/concurrent/interrupt/InterruptExample.java

@@ -0,0 +1,47 @@
+package com.baeldung.concurrent.interrupt;
+
+public class InterruptExample extends Thread {
+    
+    public static void propagateException() throws InterruptedException {
+        Thread.sleep(1000);
+        Thread.currentThread().interrupt();
+        if (Thread.interrupted()) {
+            throw new InterruptedException();
+        }
+    }
+
+    public static Boolean restoreTheState() {
+        InterruptExample thread1 = new InterruptExample();
+        thread1.start();
+        thread1.interrupt();
+        return thread1.isInterrupted();
+    }
+
+    public void run() {
+        try {
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+    
+    public static void throwCustomException() throws Exception {
+
+        Thread.sleep(1000);
+        Thread.currentThread().interrupt();
+        if (Thread.interrupted()) {
+            throw new CustomInterruptedException("This thread was interrupted");
+        }
+    }
+    
+    public static Boolean handleWithCustomException() throws CustomInterruptedException{
+        try {
+            Thread.sleep(1000);
+            Thread.currentThread().interrupt();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new CustomInterruptedException("This thread was interrupted...");
+        }
+        return Thread.currentThread().isInterrupted();
+    }
+}

core-java-modules/core-java-concurrency-basic-2/src/test/java/com/baeldung/concurrent/interrupt/InterruptExampleUnitTest.java

@@ -0,0 +1,33 @@
+package com.baeldung.concurrent.interrupt;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.junit.jupiter.api.Test;
+
+public class InterruptExampleUnitTest {
+
+    @Test
+    public void whenPropagateException_thenThrowsInterruptedException() {
+        assertThrows(InterruptedException.class, () -> InterruptExample.propagateException());
+    }
+
+    @Test
+    public void whenRestoreTheState_thenReturnsTrue() {
+        assertTrue(InterruptExample.restoreTheState());
+    }
+    
+    @Test
+    public void whenThrowCustomException_thenContainsExpectedMessage() {
+        Exception exception = assertThrows(CustomInterruptedException.class, () -> InterruptExample.throwCustomException());
+        String expectedMessage = "This thread was interrupted";
+        String actualMessage = exception.getMessage();
+
+        assertTrue(actualMessage.contains(expectedMessage));
+    }
+   
+    @Test
+    public void whenHandleWithCustomException_thenReturnsTrue() throws CustomInterruptedException{
+        assertTrue(InterruptExample.handleWithCustomException());
+    }
+}

core-java-modules/core-java-jpms/decoupling-pattern1/consumermodule/pom.xml

@@ -17,7 +17,7 @@
     <dependencies>
         <dependency>
             <groupId>com.baeldung.servicemodule</groupId>
-            <artifactId>servicemodule</artifactId>
+            <artifactId>servicemodule1</artifactId>
             <version>${servicemodule.version}</version>
         </dependency>
     </dependencies>

core-java-modules/core-java-jpms/decoupling-pattern1/pom.xml

@@ -6,9 +6,15 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.baeldung.decoupling-pattern1</groupId>
     <artifactId>decoupling-pattern1</artifactId>
-    <version>1.0</version>
+    <version>1.0-SNAPSHOT</version>
     <packaging>pom</packaging>
 
+    <parent>
+        <groupId>com.baeldung.core-java-modules</groupId>
+        <artifactId>core-java-jpms</artifactId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+
     <modules>
         <module>servicemodule</module>
         <module>consumermodule</module>

core-java-modules/core-java-jpms/decoupling-pattern1/servicemodule/pom.xml

@@ -5,7 +5,7 @@
     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>
     <groupId>com.baeldung.servicemodule</groupId>
-    <artifactId>servicemodule</artifactId>
+    <artifactId>servicemodule1</artifactId>
     <packaging>jar</packaging>
 
     <parent>

core-java-modules/core-java-jpms/decoupling-pattern2/consumermodule/pom.xml

@@ -32,6 +32,11 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
+                <version>${compiler.plugin.version}</version>
+                <configuration>
+                    <source>${source.version}</source>
+                    <target>${target.version}</target>
+                </configuration>
             </plugin>
         </plugins>
     </build>

core-java-modules/core-java-jpms/decoupling-pattern2/pom.xml

@@ -9,6 +9,12 @@
     <version>1.0-SNAPSHOT</version>
     <packaging>pom</packaging>
 
+    <parent>
+        <groupId>com.baeldung.core-java-modules</groupId>
+        <artifactId>core-java-jpms</artifactId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+
     <modules>
         <module>servicemodule</module>
         <module>providermodule</module>

core-java-modules/core-java-jpms/decoupling-pattern2/providermodule/pom.xml

@@ -27,6 +27,11 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
+                <version>${compiler.plugin.version}</version>
+                <configuration>
+                    <source>${source.version}</source>
+                    <target>${target.version}</target>
+                </configuration>
             </plugin>
         </plugins>
     </build>

core-java-modules/core-java-jpms/decoupling-pattern2/servicemodule/pom.xml

@@ -5,7 +5,7 @@
     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>
     <groupId>com.baeldung.servicemodule</groupId>
-    <artifactId>servicemodule</artifactId>
+    <artifactId>servicemodule2</artifactId>
     <version>1.0</version>
 
     <parent>
@@ -19,6 +19,11 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
+                <version>${compiler.plugin.version}</version>
+                <configuration>
+                    <source>${source.version}</source>
+                    <target>${target.version}</target>
+                </configuration>
             </plugin>
         </plugins>
     </build>

core-java-modules/core-java-jpms/pom.xml

@@ -20,4 +20,26 @@
         <module>decoupling-pattern2</module>
     </modules>
 
+    <build>
+        <pluginManagement>
+            <plugins>
+                <plugin>
+                    <groupId>org.apache.maven.plugins</groupId>
+                    <artifactId>maven-compiler-plugin</artifactId>
+                    <version>${compiler.plugin.version}</version>
+                    <configuration>
+                        <source>${source.version}</source>
+                        <target>${target.version}</target>
+                    </configuration>
+                </plugin>
+            </plugins>
+        </pluginManagement>
+    </build>
+
+    <properties>
+        <compiler.plugin.version>3.8.0</compiler.plugin.version>
+        <source.version>11</source.version>
+        <target.version>11</target.version>
+    </properties>
+
 </project>

core-java-modules/core-java-lang-4/README.md

@@ -3,3 +3,4 @@
 This module contains articles about core features in the Java language
 
 - [The Java final Keyword – Impact on Performance](https://www.baeldung.com/java-final-performance)
+- [The package-info.java File](https://www.baeldung.com/java-package-info)

core-java-modules/core-java-lang-4/src/main/java/com/baeldung/finalkeyword/package-info.java

@@ -0,0 +1,13 @@
+/**
+* This module is about impact of the final keyword on performance
+ * <p>
+ * This module explores  if there are any performance benefits from
+ * using the final keyword in our code. This module examines the performance
+ * implications of using final on a variable, method, and class level.
+ * </p>
+*
+* @since 1.0
+* @author baeldung
+* @version 1.1
+*/
+package com.baeldung.finalkeyword;

core-java-modules/core-java-lang-oop-types-2/README.md

@@ -0,0 +1,7 @@
+## Core Java Lang OOP - Types
+
+This module contains articles about types in Java
+
+### Relevant Articles: 
+
+-

core-java-modules/core-java-lang-oop-types-2/pom.xml

@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<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">
+    <parent>
+        <artifactId>core-java-modules</artifactId>
+        <groupId>com.baeldung.core-java-modules</groupId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>core-java-lang-oop-types-2</artifactId>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>${commons-lang3.version}</version>
+        </dependency>
+    </dependencies>
+</project>

core-java-modules/core-java-lang-oop-types-2/src/test/java/com/baeldung/conversions/PrimitiveToObjectArrayUnitTest.java

@@ -0,0 +1,81 @@
+package com.baeldung.conversions;
+
+import org.apache.commons.lang3.ArrayUtils;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class PrimitiveToObjectArrayUnitTest {
+
+    @Test
+    void givenUsingIteration_whenConvertingToObjects_thenSuccess() {
+        int[] input = new int[] { 0, 1, 2, 3, 4 };
+        Integer[] expected = new Integer[] { 0, 1, 2, 3, 4 };
+
+        Integer[] output = new Integer[input.length];
+        for (int i = 0; i < input.length; i++) {
+            output[i] = input[i];
+        }
+
+        assertArrayEquals(expected, output);
+    }
+
+    @Test
+    void givenUsingIteration_whenConvertingToPrimitives_thenSuccess() {
+        Integer[] input = new Integer[] { 0, 1, 2, 3, 4 };
+        int[] expected = new int[] { 0, 1, 2, 3, 4 };
+
+        int[] output = new int[input.length];
+        for (int i = 0; i < input.length; i++) {
+            output[i] = input[i];
+        }
+
+        assertArrayEquals(expected, output);
+    }
+
+    @Test
+    void givenUsingStreams_whenConvertingToObjects_thenSuccess() {
+        int[] input = new int[] { 0, 1, 2, 3, 4 };
+        Integer[] expected = new Integer[] { 0, 1, 2, 3, 4 };
+
+        Integer[] output = Arrays.stream(input)
+            .boxed()
+            .toArray(Integer[]::new);
+
+        assertArrayEquals(expected, output);
+    }
+
+    @Test
+    void givenUsingStreams_whenConvertingToPrimitives_thenSuccess() {
+        Integer[] input = new Integer[] { 0, 1, 2, 3, 4 };
+        int[] expected = new int[] { 0, 1, 2, 3, 4 };
+
+        int[] output = Arrays.stream(input)
+            .mapToInt(Integer::intValue)
+            .toArray();
+
+        assertArrayEquals(expected, output);
+    }
+
+    @Test
+    void givenUsingApacheCommons_whenConvertingToObjects_thenSuccess() {
+        int[] input = new int[] { 0, 1, 2, 3, 4 };
+        Integer[] expected = new Integer[] { 0, 1, 2, 3, 4 };
+
+        Integer[] output = ArrayUtils.toObject(input);
+
+        assertArrayEquals(expected, output);
+    }
+
+    @Test
+    void givenUsingApacheCommons_whenConvertingToPrimitives_thenSuccess() {
+        Integer[] input = new Integer[] { 0, 1, 2, 3, 4 };
+        int[] expected = new int[] { 0, 1, 2, 3, 4 };
+
+        int[] output = ArrayUtils.toPrimitive(input);
+
+        assertArrayEquals(expected, output);
+    }
+}

core-java-modules/core-java-nio-2/README.md

@@ -9,4 +9,5 @@ This module contains articles about core Java non-blocking input and output (IO)
 - [Introduction to the Java NIO Selector](https://www.baeldung.com/java-nio-selector)
 - [Using Java MappedByteBuffer](https://www.baeldung.com/java-mapped-byte-buffer)
 - [How to Lock a File in Java](https://www.baeldung.com/java-lock-files)
+- [Java NIO DatagramChannel](https://www.baeldung.com/java-nio-datagramchannel)
 - [[<-- Prev]](/core-java-modules/core-java-nio)

core-java-modules/core-java-os/src/test/java/com/baeldung/java9/process/ProcessUnderstandingUnitTest.java

@@ -101,7 +101,8 @@ class ProcessUnderstandingUnitTest {
                 .replace("/", File.separator));
 
         BufferedReader output = new BufferedReader(new InputStreamReader(process.getInputStream()));
-        int value = Integer.parseInt(output.readLine());
+        String line = output.readLine();
+        int value = Integer.parseInt(line);
 
         assertEquals(3, value);
     }

core-java-modules/core-java-security-2/README.md

@@ -16,4 +16,5 @@ This module contains articles about core Java Security
 - [Java AES Encryption and Decryption](https://www.baeldung.com/java-aes-encryption-decryption)
 - [InvalidAlgorithmParameterException: Wrong IV Length](https://www.baeldung.com/java-invalidalgorithmparameter-exception)
 - [The java.security.egd JVM Option](https://www.baeldung.com/java-security-egd)
+- [RSA in Java](https://www.baeldung.com/java-rsa)
 - More articles: [[<-- prev]](/core-java-modules/core-java-security)

core-java-modules/core-java-streams-3/README.md

@@ -11,4 +11,5 @@ This module contains articles about the Stream API in Java.
 - [Add BigDecimals using the Stream API](https://www.baeldung.com/java-stream-add-bigdecimals)
 - [Should We Close a Java Stream?](https://www.baeldung.com/java-stream-close)
 - [Returning Stream vs. Collection](https://www.baeldung.com/java-return-stream-collection)
+- [Convert a Java Enumeration Into a Stream](https://www.baeldung.com/java-enumeration-to-stream)
 - More articles: [[<-- prev>]](/../core-java-streams-2)

core-java-modules/core-java-streams-3/src/main/java/com/baeldung/streams/conversion/EnumerationSpliterator.java

@@ -0,0 +1,30 @@
+package com.baeldung.streams.conversion;
+
+import java.util.Enumeration;
+import java.util.Spliterators.AbstractSpliterator;
+import java.util.function.Consumer;
+
+public class EnumerationSpliterator<T> extends AbstractSpliterator<T> {
+
+    private final Enumeration<T> enumeration;
+
+    public EnumerationSpliterator(long est, int additionalCharacteristics, Enumeration<T> enumeration) {
+        super(est, additionalCharacteristics);
+        this.enumeration = enumeration;
+    }
+
+    @Override
+    public boolean tryAdvance(Consumer<? super T> action) {
+        if (enumeration.hasMoreElements()) {
+            action.accept(enumeration.nextElement());
+            return true;
+        }
+        return false;
+    }
+
+    @Override
+    public void forEachRemaining(Consumer<? super T> action) {
+        while (enumeration.hasMoreElements())
+            action.accept(enumeration.nextElement());
+    }
+}

core-java-modules/core-java-streams-3/src/main/java/com/baeldung/streams/conversion/EnumerationStreamConversion.java

@@ -0,0 +1,16 @@
+package com.baeldung.streams.conversion;
+
+import java.util.Enumeration;
+import java.util.Spliterator;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
+public class EnumerationStreamConversion {
+
+    public static <T> Stream<T> convert(Enumeration<T> enumeration) {
+        EnumerationSpliterator<T> spliterator = new EnumerationSpliterator<T>(Long.MAX_VALUE, Spliterator.ORDERED, enumeration);
+        Stream<T> stream = StreamSupport.stream(spliterator, false);
+
+        return stream;
+    }
+}

core-java-modules/core-java-streams-3/src/test/java/com/baeldung/streams/conversion/EnumerationStreamConversionUnitTest.java

@@ -0,0 +1,35 @@
+package com.baeldung.streams.conversion;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.collection.IsIterableContainingInOrder.contains;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Vector;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.junit.Assert;
+import org.junit.jupiter.api.Test;
+
+public class EnumerationStreamConversionUnitTest {
+
+    @Test
+    public void givenEnumeration_whenConvertedToStream_thenNotNull() {
+        Vector<Integer> input = new Vector<>(Arrays.asList(1, 2, 3, 4, 5));
+
+        Stream<Integer> resultingStream = EnumerationStreamConversion.convert(input.elements());
+
+        Assert.assertNotNull(resultingStream);
+    }
+
+    @Test
+    public void whenConvertedToList_thenCorrect() {
+        Vector<Integer> input = new Vector<>(Arrays.asList(1, 2, 3, 4, 5));
+
+        Stream<Integer> stream = EnumerationStreamConversion.convert(input.elements());
+        List<Integer> list = stream.filter(e -> e >= 3)
+            .collect(Collectors.toList());
+        assertThat(list, contains(3, 4, 5));
+    }
+}

core-java-modules/core-java-string-operations-3/src/test/java/com/baeldung/splitstringbynewline/SplitStringByNewLineUnitTest.java

@@ -0,0 +1,29 @@
+package com.baeldung.splitstringbynewline;
+
+import org.junit.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class SplitStringByNewLineUnitTest {
+    @Test
+    public void givenString_whenSplitByNewLineUsingSystemLineSeparator_thenReturnsArray() {
+        assertThat("Line1\nLine2\nLine3".split(System.lineSeparator())).containsExactly("Line1", "Line2", "Line3");
+    }
+
+    @Test
+    public void givenString_whenSplitByNewLineUsingRegularExpressionPattern_thenReturnsArray() {
+        assertThat("Line1\nLine2\nLine3".split("\\r?\\n|\\r")).containsExactly("Line1", "Line2", "Line3");
+
+        assertThat("Line1\rLine2\rLine3".split("\\r?\\n|\\r")).containsExactly("Line1", "Line2", "Line3");
+
+        assertThat("Line1\r\nLine2\r\nLine3".split("\\r?\\n|\\r")).containsExactly("Line1", "Line2", "Line3");
+    }
+
+    @Test
+    public void givenString_whenSplitByNewLineUsingJava8Pattern_thenReturnsArray() {
+        assertThat("Line1\nLine2\nLine3".split("\\R")).containsExactly("Line1", "Line2", "Line3");
+
+        assertThat("Line1\rLine2\rLine3".split("\\R")).containsExactly("Line1", "Line2", "Line3");
+
+        assertThat("Line1\r\nLine2\r\nLine3".split("\\R")).containsExactly("Line1", "Line2", "Line3");
+    }
+}

core-java-modules/pom.xml

@@ -91,6 +91,7 @@
         <module>core-java-lang-oop-generics</module>
         <module>core-java-lang-oop-modifiers</module>
         <module>core-java-lang-oop-types</module>
+        <module>core-java-lang-oop-types-2</module>
         <module>core-java-lang-oop-inheritance</module>
         <module>core-java-lang-oop-methods</module>
         <module>core-java-lang-oop-others</module>

kubernetes/k8s-intro/src/main/java/com/baeldung/kubernetes/intro/WatchPods.java

@@ -0,0 +1,72 @@
+package com.baeldung.kubernetes.intro;
+
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.gson.reflect.TypeToken;
+
+import io.kubernetes.client.openapi.ApiClient;
+import io.kubernetes.client.openapi.ApiException;
+import io.kubernetes.client.openapi.apis.CoreV1Api;
+import io.kubernetes.client.openapi.models.V1ObjectMeta;
+import io.kubernetes.client.openapi.models.V1Pod;
+import io.kubernetes.client.util.Config;
+import io.kubernetes.client.util.Watch;
+import io.kubernetes.client.util.Watch.Response;
+import okhttp3.OkHttpClient;
+import okhttp3.logging.HttpLoggingInterceptor;
+
+public class WatchPods {
+
+    private static Logger log = LoggerFactory.getLogger(WatchPods.class);
+
+    public static void main(String[] args) throws Exception {
+
+        ApiClient client = Config.defaultClient();
+
+        // Optional, put helpful during tests: disable client timeout and enable
+        // HTTP wire-level logs
+        HttpLoggingInterceptor interceptor = new HttpLoggingInterceptor(message -> log.info(message));
+        interceptor.setLevel(HttpLoggingInterceptor.Level.BODY);
+        OkHttpClient newClient = client.getHttpClient()
+          .newBuilder()
+          .addInterceptor(interceptor)
+          .readTimeout(0, TimeUnit.SECONDS)
+          .build();
+        
+        client.setHttpClient(newClient);
+        CoreV1Api api = new CoreV1Api(client);
+
+        // Create the watch object that monitors pod creation/deletion/update events
+        while (true) {
+            log.info("[I46] Creating watch...");
+            try (Watch<V1Pod> watch = Watch.createWatch(
+              client,
+              api.listPodForAllNamespacesCall(false, null, null, null, null, "false", null, null, 10, true, null),
+              new TypeToken<Response<V1Pod>>(){}.getType())) {
+
+                log.info("[I52] Receiving events:");
+                for (Response<V1Pod> event : watch) {
+                    V1Pod pod = event.object;
+                    V1ObjectMeta meta = pod.getMetadata();
+                    switch (event.type) {
+                    case "ADDED":
+                    case "MODIFIED":
+                    case "DELETED":
+                        log.info("event.type: {}, namespace={}, name={}",
+                          event.type,
+                          meta.getNamespace(),
+                          meta.getName());
+                        break;
+                    default:
+                        log.warn("[W66] Unknown event type: {}", event.type);
+                    }
+                }
+            } catch (ApiException ex) {
+                log.error("[E70] ApiError", ex);
+            }
+        }
+    }
+}

kubernetes/k8s-intro/src/main/java/com/baeldung/kubernetes/intro/WatchPodsUsingBookmarks.java

@@ -0,0 +1,87 @@
+package com.baeldung.kubernetes.intro;
+
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.gson.reflect.TypeToken;
+
+import io.kubernetes.client.openapi.ApiClient;
+import io.kubernetes.client.openapi.ApiException;
+import io.kubernetes.client.openapi.apis.CoreV1Api;
+import io.kubernetes.client.openapi.models.V1ObjectMeta;
+import io.kubernetes.client.openapi.models.V1Pod;
+import io.kubernetes.client.openapi.models.V1PodList;
+import io.kubernetes.client.util.Config;
+import io.kubernetes.client.util.Watch;
+import io.kubernetes.client.util.Watch.Response;
+import okhttp3.OkHttpClient;
+import okhttp3.logging.HttpLoggingInterceptor;
+
+public class WatchPodsUsingBookmarks {
+
+    private static Logger log = LoggerFactory.getLogger(WatchPodsUsingBookmarks.class);
+
+    public static void main(String[] args) throws Exception {
+
+        ApiClient client = Config.defaultClient();
+
+        // Optional, put helpful during tests: disable client timeout and enable
+        // HTTP wire-level logs
+        HttpLoggingInterceptor interceptor = new HttpLoggingInterceptor(message -> log.info(message));
+        interceptor.setLevel(HttpLoggingInterceptor.Level.BODY);
+        OkHttpClient newClient = client.getHttpClient()
+          .newBuilder()
+          .addInterceptor(interceptor)
+          .readTimeout(0, TimeUnit.SECONDS)
+          .build();
+        
+        client.setHttpClient(newClient);
+        CoreV1Api api = new CoreV1Api(client);
+
+        String resourceVersion = null;
+        while (true) {
+            // Get a fresh list only we need to resync
+            if ( resourceVersion == null ) {
+                log.info("[I48] Creating initial POD list...");
+                V1PodList podList = api.listPodForAllNamespaces(true, null, null, null, null, "false", resourceVersion, null, null, null);
+                resourceVersion = podList.getMetadata().getResourceVersion();
+            }
+
+            while (true) {
+                log.info("[I54] Creating watch: resourceVersion={}", resourceVersion);
+                try (Watch<V1Pod> watch = Watch.createWatch(
+                  client,
+                  api.listPodForAllNamespacesCall(true, null, null, null, null, "false", resourceVersion, null, 10, true, null),
+                  new TypeToken<Response<V1Pod>>(){}.getType())) {
+                    
+                    log.info("[I60] Receiving events:");
+                    for (Response<V1Pod> event : watch) {
+                        V1Pod pod = event.object;
+                        V1ObjectMeta meta = pod.getMetadata();
+                        switch (event.type) {
+                        case "BOOKMARK":
+                            resourceVersion = meta.getResourceVersion();
+                            log.info("[I67] event.type: {}, resourceVersion={}", event.type,resourceVersion);
+                            break;
+                        case "ADDED":
+                        case "MODIFIED":
+                        case "DELETED":
+                            log.info("event.type: {}, namespace={}, name={}",
+                              event.type,
+                              meta.getNamespace(),
+                              meta.getName());
+                            break;
+                        default:
+                            log.warn("[W76] Unknown event type: {}", event.type);
+                        }
+                    }
+                } catch (ApiException ex) {
+                    log.error("[E80] ApiError", ex);
+                    resourceVersion = null;
+                }
+            }
+        }
+    }
+}

kubernetes/k8s-intro/src/main/java/com/baeldung/kubernetes/intro/WatchPodsUsingResourceVersions.java

@@ -0,0 +1,111 @@
+package com.baeldung.kubernetes.intro;
+
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
+
+import io.kubernetes.client.openapi.ApiClient;
+import io.kubernetes.client.openapi.ApiException;
+import io.kubernetes.client.openapi.apis.CoreV1Api;
+import io.kubernetes.client.openapi.models.V1ObjectMeta;
+import io.kubernetes.client.openapi.models.V1Pod;
+import io.kubernetes.client.openapi.models.V1PodList;
+import io.kubernetes.client.util.Config;
+import io.kubernetes.client.util.Watch;
+import io.kubernetes.client.util.Watch.Response;
+import okhttp3.OkHttpClient;
+import okhttp3.logging.HttpLoggingInterceptor;
+
+public class WatchPodsUsingResourceVersions {
+
+    private static Logger log = LoggerFactory.getLogger(WatchPodsUsingResourceVersions.class);
+
+    public static void main(String[] args) throws Exception {
+
+        ApiClient client = Config.defaultClient();
+
+        // Optional, put helpful during tests: disable client timeout and enable
+        // HTTP wire-level logs
+        HttpLoggingInterceptor interceptor = new HttpLoggingInterceptor(message -> log.info(message));
+        interceptor.setLevel(HttpLoggingInterceptor.Level.BODY);
+        OkHttpClient newClient = client.getHttpClient()
+          .newBuilder()
+          .addInterceptor(interceptor)
+          .readTimeout(0, TimeUnit.SECONDS)
+           .build();
+        
+        client.setHttpClient(newClient);
+        CoreV1Api api = new CoreV1Api(client);
+
+        String resourceVersion = null;
+        while (true) {
+            try {
+                if ( resourceVersion == null ) {
+                    V1PodList podList = api.listPodForAllNamespaces(null, null, null, null, null, null, resourceVersion, null, null, null);
+                    resourceVersion = podList.getMetadata().getResourceVersion();
+                }
+                
+                log.info("[I59] Creating watch: resourceVersion={}", resourceVersion);
+                try (Watch<V1Pod> watch = Watch.createWatch(
+                  client,
+                  api.listPodForAllNamespacesCall(null, null, null, null, null, null, resourceVersion, null, 60, true, null),
+                  new TypeToken<Response<V1Pod>>(){}.getType())) {
+                        
+                    log.info("[I65] Receiving events:");
+                    for (Response<V1Pod> event : watch) {
+                        V1Pod pod = event.object;
+                        V1ObjectMeta meta = pod.getMetadata();
+                        switch (event.type) {
+                        case "ADDED":
+                        case "MODIFIED":
+                        case "DELETED":
+                            log.info("event: type={}, namespace={}, name={}",
+                              event.type,
+                              meta.getNamespace(),
+                              meta.getName());
+                            break;
+                        default:
+                            log.warn("[W76] Unknown event type: {}", event.type);
+                        }
+                    }
+                }
+            }
+            catch (ApiException ex) {
+                if ( ex.getCode() == 504 || ex.getCode() == 410 ) {
+                    resourceVersion = extractResourceVersionFromException(ex);
+                }
+                else {
+                    // Reset resource version
+                    resourceVersion = null;
+                }
+            }
+        }
+    }
+
+    private static String extractResourceVersionFromException(ApiException ex) {
+        
+        String body = ex.getResponseBody();
+        if (body == null) {
+            return null;
+        }
+
+        Gson gson = new Gson();
+        Map<?,?> st = gson.fromJson(body, Map.class);
+        Pattern p = Pattern.compile("Timeout: Too large resource version: (\\d+), current: (\\d+)");
+        String msg = (String)st.get("message");
+        Matcher m = p.matcher(msg);
+        if (!m.matches()) {
+            return null;
+        }
+        
+        return m.group(2);
+    }
+
+}

kubernetes/k8s-intro/src/test/java/com/baeldung/kubernetes/intro/WatchPodsLiveTest.java

@@ -0,0 +1,10 @@
+package com.baeldung.kubernetes.intro;
+
+import org.junit.jupiter.api.Test;
+
+class WatchPodsLiveTest {
+    @Test
+    void whenWatchPods_thenSuccess()  throws Exception {
+        WatchPods.main(new String[] {});
+    }
+}

kubernetes/k8s-intro/src/test/java/com/baeldung/kubernetes/intro/WatchPodsUsingBookmarksLiveTest.java

@@ -0,0 +1,10 @@
+package com.baeldung.kubernetes.intro;
+
+import org.junit.jupiter.api.Test;
+
+class WatchPodsUsingBookmarksLiveTest {
+    @Test
+    void whenWatchPods_thenSuccess()  throws Exception {
+        WatchPodsUsingBookmarks.main(new String[] {});
+    }
+}

kubernetes/k8s-intro/src/test/java/com/baeldung/kubernetes/intro/WatchPodsUsingResourceVersionsLiveTest.java

@@ -0,0 +1,10 @@
+package com.baeldung.kubernetes.intro;
+
+import org.junit.jupiter.api.Test;
+
+class WatchPodsUsingResourceVersionsLiveTest {
+    @Test
+    void whenWatchPods_thenSuccess()  throws Exception {
+        WatchPodsUsingResourceVersions.main(new String[] {});
+    }
+}

libraries-data-db/pom.xml

@@ -8,8 +8,9 @@
 
     <parent>
         <groupId>com.baeldung</groupId>
-        <artifactId>parent-modules</artifactId>
+        <artifactId>parent-boot-2</artifactId>
-        <version>1.0.0-SNAPSHOT</version>
+        <version>0.0.1-SNAPSHOT</version>
+        <relativePath>../parent-boot-2</relativePath>
     </parent>
 
     <dependencies>
@@ -76,6 +77,71 @@
             <groupId>io.ebean</groupId>
             <artifactId>ebean</artifactId>
             <version>${ebean.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.fasterxml.jackson.core</groupId>
+                    <artifactId>jackson-core</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <!-- Debezium -->
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-api</artifactId>
+            <version>${debezium.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-embedded</artifactId>
+            <version>${debezium.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-log4j12</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-connector-mysql</artifactId>
+            <version>${debezium.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <version>${testcontainers-version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>mysql</artifactId>
+            <version>${testcontainers-version}</version>
+        </dependency>
+
+        <!-- Spring Core dependencies-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+
+        <!-- Spring Data JPA, crud repository -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-jpa</artifactId>
+        </dependency>
+
+        <!-- MySQL -->
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+        </dependency>
+
+        <!-- Utility dependencies-->
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
         </dependency>
     </dependencies>
 
@@ -213,6 +279,8 @@
         <javax.jdo.version>3.2.0-m7</javax.jdo.version>
         <HikariCP.version>3.4.5</HikariCP.version>
         <ebean.version>11.22.4</ebean.version>
+        <debezium.version>1.4.2.Final</debezium.version>
+        <testcontainers-version>1.15.3</testcontainers-version>
     </properties>
 
-</project>
+</project>

libraries-data-db/src/main/java/com/baeldung/libraries/debezium/DebeziumCDCApplication.java

@@ -0,0 +1,13 @@
+package com.baeldung.libraries.debezium;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class DebeziumCDCApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(DebeziumCDCApplication.class, args);
+    }
+
+}

libraries-data-db/src/main/java/com/baeldung/libraries/debezium/config/DebeziumConnectorConfig.java

@@ -0,0 +1,58 @@
+package com.baeldung.libraries.debezium.config;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.io.File;
+import java.io.IOException;
+
+@Configuration
+public class DebeziumConnectorConfig {
+
+    /**
+     * Database details.
+     */
+    @Value("${customer.datasource.host}")
+    private String customerDbHost;
+
+    @Value("${customer.datasource.database}")
+    private String customerDbName;
+
+    @Value("${customer.datasource.port}")
+    private String customerDbPort;
+
+    @Value("${customer.datasource.username}")
+    private String customerDbUsername;
+
+    @Value("${customer.datasource.password}")
+    private String customerDbPassword;
+
+    /**
+     * Customer Database Connector Configuration
+     */
+    @Bean
+    public io.debezium.config.Configuration customerConnector() throws IOException {
+        File offsetStorageTempFile = File.createTempFile("offsets_", ".dat");
+        File dbHistoryTempFile = File.createTempFile("dbhistory_", ".dat");
+        return io.debezium.config.Configuration.create()
+            .with("name", "customer-mysql-connector")
+            .with("connector.class", "io.debezium.connector.mysql.MySqlConnector")
+            .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
+            .with("offset.storage.file.filename", offsetStorageTempFile.getAbsolutePath())
+            .with("offset.flush.interval.ms", "60000")
+            .with("database.hostname", customerDbHost)
+            .with("database.port", customerDbPort)
+            .with("database.user", customerDbUsername)
+            .with("database.password", customerDbPassword)
+            .with("database.dbname", customerDbName)
+            .with("database.include.list", customerDbName)
+            .with("include.schema.changes", "false")
+            .with("database.allowPublicKeyRetrieval", "true")
+            .with("database.server.id", "10181")
+            .with("database.server.name", "customer-mysql-db-server")
+            .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
+            .with("database.history.file.filename", dbHistoryTempFile.getAbsolutePath())
+            .build();
+    }
+}

libraries-data-db/src/main/java/com/baeldung/libraries/debezium/entity/Customer.java

@@ -0,0 +1,18 @@
+package com.baeldung.libraries.debezium.entity;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import javax.persistence.Entity;
+import javax.persistence.Id;
+
+@Entity
+@Getter
+@Setter
+public class Customer {
+    @Id
+    private Long id;
+
+    private String fullname;
+    private String email;
+}

libraries-data-db/src/main/java/com/baeldung/libraries/debezium/listener/DebeziumListener.java

@@ -0,0 +1,83 @@
+package com.baeldung.libraries.debezium.listener;
+
+import com.baeldung.libraries.debezium.service.CustomerService;
+import io.debezium.config.Configuration;
+import io.debezium.embedded.Connect;
+import io.debezium.engine.DebeziumEngine;
+import io.debezium.engine.RecordChangeEvent;
+import io.debezium.engine.format.ChangeEventFormat;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.kafka.connect.data.Field;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+
+import static io.debezium.data.Envelope.FieldName.*;
+import static io.debezium.data.Envelope.Operation;
+import static java.util.stream.Collectors.toMap;
+
+@Slf4j
+@Component
+public class DebeziumListener {
+
+    private final Executor executor = Executors.newSingleThreadExecutor();
+    private final CustomerService customerService;
+    private final DebeziumEngine<RecordChangeEvent<SourceRecord>> debeziumEngine;
+
+    public DebeziumListener(Configuration customerConnectorConfiguration, CustomerService customerService) {
+
+        this.debeziumEngine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
+            .using(customerConnectorConfiguration.asProperties())
+            .notifying(this::handleChangeEvent)
+            .build();
+
+        this.customerService = customerService;
+    }
+
+    private void handleChangeEvent(RecordChangeEvent<SourceRecord> sourceRecordRecordChangeEvent) {
+        SourceRecord sourceRecord = sourceRecordRecordChangeEvent.record();
+
+        log.info("Key = '" + sourceRecord.key() + "' value = '" + sourceRecord.value() + "'");
+
+        Struct sourceRecordChangeValue= (Struct) sourceRecord.value();
+
+        if (sourceRecordChangeValue != null) {
+            Operation operation = Operation.forCode((String) sourceRecordChangeValue.get(OPERATION));
+
+            if(operation != Operation.READ) {
+                String record = operation == Operation.DELETE ? BEFORE : AFTER; // Handling Update & Insert operations.
+
+                Struct struct = (Struct) sourceRecordChangeValue.get(record);
+                Map<String, Object> payload = struct.schema().fields().stream()
+                    .map(Field::name)
+                    .filter(fieldName -> struct.get(fieldName) != null)
+                    .map(fieldName -> Pair.of(fieldName, struct.get(fieldName)))
+                    .collect(toMap(Pair::getKey, Pair::getValue));
+
+                this.customerService.replicateData(payload, operation);
+                log.info("Updated Data: {} with Operation: {}", payload, operation.name());
+            }
+        }
+    }
+
+    @PostConstruct
+    private void start() {
+        this.executor.execute(debeziumEngine);
+    }
+
+    @PreDestroy
+    private void stop() throws IOException {
+        if (this.debeziumEngine != null) {
+            this.debeziumEngine.close();
+        }
+    }
+
+}

libraries-data-db/src/main/java/com/baeldung/libraries/debezium/repository/CustomerRepository.java

@@ -0,0 +1,9 @@
+package com.baeldung.libraries.debezium.repository;
+
+import com.baeldung.libraries.debezium.entity.Customer;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface CustomerRepository extends JpaRepository<Customer, Long> {
+}

libraries-data-db/src/main/java/com/baeldung/libraries/debezium/service/CustomerService.java

@@ -0,0 +1,30 @@
+package com.baeldung.libraries.debezium.service;
+
+import com.baeldung.libraries.debezium.entity.Customer;
+import com.baeldung.libraries.debezium.repository.CustomerRepository;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.debezium.data.Envelope.Operation;
+import org.springframework.stereotype.Service;
+
+import java.util.Map;
+
+@Service
+public class CustomerService {
+
+    private final CustomerRepository customerRepository;
+
+    public CustomerService(CustomerRepository customerRepository) {
+        this.customerRepository = customerRepository;
+    }
+
+    public void replicateData(Map<String, Object> customerData, Operation operation) {
+        final ObjectMapper mapper = new ObjectMapper();
+        final Customer customer = mapper.convertValue(customerData, Customer.class);
+
+        if (Operation.DELETE.name().equals(operation.name())) {
+            customerRepository.deleteById(customer.getId());
+        } else {
+            customerRepository.save(customer);
+        }
+    }
+}

libraries-data-db/src/main/resources/application.yml

@@ -0,0 +1,34 @@
+## Server properties
+server:
+  port: 8080
+
+## Primary/Target Database Properties
+spring:
+  datasource:
+    url: jdbc:mysql://localhost:3306/customerdb
+    username: root
+    password: root
+  jpa.hibernate.ddl-auto: create-drop
+  jpa.show-sql: true
+
+## Source Database Properties
+customer:
+  datasource:
+    host: localhost
+    port: 3305
+    database: customerdb
+    username: root
+    password: root
+
+ ## Logging properties
+logging:
+  level:
+    root: INFO
+    io:
+      debezium:
+        mysql:
+          BinlogReader: INFO
+    com:
+      baeldung:
+        libraries:
+          debezium: DEBUG

libraries-data-db/src/main/resources/customer.sql

@@ -0,0 +1,9 @@
+drop table if exists customer;
+
+CREATE TABLE customer
+(
+    id integer NOT NULL,
+    fullname character varying(255),
+    email character varying(255),
+    CONSTRAINT customer_pkey PRIMARY KEY (id)
+);

libraries-data-db/src/main/resources/docker-compose.yml

@@ -0,0 +1,25 @@
+version: "3.9"
+services:
+  # Install Source MySQL DB and setup the Customer database
+  mysql-1:
+    container_name: source-database
+    image: mysql
+    command: --default-authentication-plugin=mysql_native_password
+    restart: always
+    ports:
+      - 3305:3306
+    environment:
+      MYSQL_ROOT_PASSWORD: root
+      MYSQL_DATABASE: customerdb
+
+  # Install Target MySQL DB and setup the Customer database
+  mysql-2:
+    container_name: target-database
+    image: mysql
+    command: --default-authentication-plugin=mysql_native_password
+    restart: always
+    ports:
+      - 3306:3306
+    environment:
+      MYSQL_ROOT_PASSWORD: root
+      MYSQL_DATABASE: customerdb

libraries-data-db/src/test/java/com/baeldung/libraries/debezium/DebeziumCDCLiveTest.java

@@ -0,0 +1,61 @@
+package com.baeldung.libraries.debezium;
+
+import com.baeldung.libraries.debezium.repository.CustomerRepository;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = DebeziumCDCApplication.class)
+@ActiveProfiles("test")
+public class DebeziumCDCLiveTest {
+
+    @Autowired
+    private CustomerRepository customerRepository;
+
+    @Autowired
+    @Qualifier("sourceJdbcTemplate")
+    private NamedParameterJdbcTemplate jdbcTemplate;
+
+    @Before
+    public void clearData() {
+        jdbcTemplate.update("delete from customer where id = :id", Collections.singletonMap("id", 1));
+    }
+
+    @DynamicPropertySource
+    static void registerProperties(DynamicPropertyRegistry registry) {
+        registry.add("customer.datasource.port", MySQLTestContainerConfiguration::getPort);
+    }
+
+    @Test
+    public void whenInsertDataToSourceDatabase_thenCdcOk() throws InterruptedException {
+        assertThat(customerRepository.findAll().size()).isZero();
+
+        // insert data to source DB
+        Map<String, Object> map = new HashMap<>();
+        map.put("id", 1);
+        map.put("fullname", "John Doe");
+        map.put("email", "test@test.com");
+
+        jdbcTemplate.update("INSERT INTO customer(id, fullname, email) VALUES (:id, :fullname, :email)", map);
+
+        // verify target DB
+        Thread.sleep(10000);
+        assertThat(customerRepository.findAll().size()).isNotZero();
+    }
+
+}

libraries-data-db/src/test/java/com/baeldung/libraries/debezium/MySQLTestContainerConfiguration.java

@@ -0,0 +1,61 @@
+package com.baeldung.libraries.debezium;
+
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.jdbc.datasource.DriverManagerDataSource;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
+import org.testcontainers.containers.MySQLContainer;
+import org.testcontainers.utility.DockerImageName;
+
+import javax.sql.DataSource;
+
+@Configuration
+public class MySQLTestContainerConfiguration {
+
+    public static final DockerImageName MYSQL_IMAGE = DockerImageName.parse("mysql:8.0");
+
+    private static final MySQLContainer<?> mysqlContainer = new MySQLContainer<>(MYSQL_IMAGE)
+        .withCommand("--default-authentication-plugin=mysql_native_password")
+        .withInitScript("debezium/customer.sql")
+        .withDatabaseName("SOURCE_DB")
+        .withUsername("user")
+        .withPassword("user")
+        .withEnv("MYSQL_ROOT_PASSWORD", "user");
+
+    MySQLTestContainerConfiguration() {
+        mysqlContainer.start();
+    }
+
+    public static int getPort() {
+        return mysqlContainer.getFirstMappedPort();
+    }
+
+    @Bean
+    @Primary
+    public EmbeddedDatabase targetDatasource() {
+        return new EmbeddedDatabaseBuilder()
+                 .setType(EmbeddedDatabaseType.H2)
+                 .setName("TAGRET_DB")
+                 .build();
+    }
+
+    @Bean(name = "SOURCE_DS")
+    public DataSource sourceDataSource() {
+        DriverManagerDataSource dataSource = new DriverManagerDataSource();
+        dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
+        dataSource.setUrl(mysqlContainer.getJdbcUrl());
+        dataSource.setUsername(mysqlContainer.getUsername());
+        dataSource.setPassword(mysqlContainer.getPassword());
+        return dataSource;
+    }
+
+    @Bean(name = "sourceJdbcTemplate")
+    public NamedParameterJdbcTemplate getJdbcTemplate(@Qualifier("SOURCE_DS") DataSource sourceDataSource) {
+        return new NamedParameterJdbcTemplate(sourceDataSource);
+    }
+}

libraries-data-db/src/test/resources/application-test.yml

@@ -0,0 +1,7 @@
+## Source Database Properties
+customer:
+  datasource:
+    host: localhost
+    database: SOURCE_DB
+    username: root
+    password: user

libraries-data-db/src/test/resources/debezium/customer.sql

@@ -0,0 +1,9 @@
+drop table if exists customer;
+
+CREATE TABLE customer
+(
+    id integer NOT NULL,
+    fullname character varying(255),
+    email character varying(255),
+    CONSTRAINT customer_pkey PRIMARY KEY (id)
+);

persistence-modules/hibernate-annotations/README.md

@@ -8,3 +8,4 @@ This module contains articles about Annotations used in Hibernate.
 - [Difference Between @JoinColumn and mappedBy](https://www.baeldung.com/jpa-joincolumn-vs-mappedby)
 - [Hibernate One to Many Annotation Tutorial](https://www.baeldung.com/hibernate-one-to-many)
 - [Hibernate @WhereJoinTable Annotation](https://www.baeldung.com/hibernate-wherejointable)
+- [Usage of the Hibernate @LazyCollection Annotation](https://www.baeldung.com/hibernate-lazycollection)

spring-5-reactive-client/src/main/java/com/baeldung/webclient/timeout/WebClientTimeoutProvider.java

@@ -0,0 +1,89 @@
+package com.baeldung.webclient.timeout;
+
+import io.netty.channel.ChannelOption;
+import io.netty.channel.epoll.EpollChannelOption;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.timeout.ReadTimeoutHandler;
+import io.netty.handler.timeout.WriteTimeoutHandler;
+import lombok.experimental.UtilityClass;
+import org.springframework.http.client.reactive.ReactorClientHttpConnector;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.netty.http.client.HttpClient;
+import reactor.netty.tcp.SslProvider;
+import reactor.netty.transport.ProxyProvider;
+
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+@UtilityClass
+public class WebClientTimeoutProvider {
+
+    public static WebClient defaultWebClient() {
+        HttpClient httpClient = HttpClient.create();
+
+        return buildWebClient(httpClient);
+    }
+
+    public WebClient responseTimeoutClient() {
+        HttpClient httpClient = HttpClient.create()
+          .responseTimeout(Duration.ofSeconds(1));
+
+        return buildWebClient(httpClient);
+    }
+
+    public WebClient connectionTimeoutClient() {
+        HttpClient httpClient = HttpClient.create()
+          .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000);
+
+        return buildWebClient(httpClient);
+    }
+
+    public WebClient connectionTimeoutWithKeepAliveClient() {
+        HttpClient httpClient = HttpClient.create()
+          .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)
+          .option(ChannelOption.SO_KEEPALIVE, true)
+          .option(EpollChannelOption.TCP_KEEPIDLE, 300)
+          .option(EpollChannelOption.TCP_KEEPINTVL, 60)
+          .option(EpollChannelOption.TCP_KEEPCNT, 8);
+
+        return buildWebClient(httpClient);
+    }
+
+    public WebClient readWriteTimeoutClient() {
+        HttpClient httpClient = HttpClient.create()
+          .doOnConnected(conn -> conn
+            .addHandler(new ReadTimeoutHandler(5, TimeUnit.SECONDS))
+            .addHandler(new WriteTimeoutHandler(5)));
+
+        return buildWebClient(httpClient);
+    }
+
+    public WebClient sslTimeoutClient() {
+        HttpClient httpClient = HttpClient.create()
+          .secure(spec -> spec
+            .sslContext(SslContextBuilder.forClient())
+            .defaultConfiguration(SslProvider.DefaultConfigurationType.TCP)
+            .handshakeTimeout(Duration.ofSeconds(30))
+            .closeNotifyFlushTimeout(Duration.ofSeconds(10))
+            .closeNotifyReadTimeout(Duration.ofSeconds(10)));
+
+        return buildWebClient(httpClient);
+    }
+
+    public WebClient proxyTimeoutClient() {
+        HttpClient httpClient = HttpClient.create()
+          .proxy(spec -> spec
+            .type(ProxyProvider.Proxy.HTTP)
+            .host("http://proxy")
+            .port(8080)
+            .connectTimeoutMillis(3000));
+
+        return buildWebClient(httpClient);
+    }
+
+    private WebClient buildWebClient(HttpClient httpClient) {
+        return WebClient.builder()
+          .clientConnector(new ReactorClientHttpConnector(httpClient))
+          .build();
+    }
+}

spring-5-reactive-client/src/test/java/com/baeldung/webclient/timeout/WebClientTimeoutIntegrationTest.java

@@ -0,0 +1,129 @@
+package com.baeldung.webclient.timeout;
+
+import com.github.tomakehurst.wiremock.WireMockServer;
+import io.netty.handler.timeout.ReadTimeoutException;
+import lombok.val;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.reactive.function.client.WebClientRequestException;
+import reactor.core.publisher.Mono;
+import reactor.netty.http.client.HttpClientRequest;
+
+
+import java.time.Duration;
+import java.util.concurrent.TimeoutException;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.configureFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
+import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class WebClientTimeoutIntegrationTest {
+
+    private WireMockServer wireMockServer;
+
+    @Before
+    public void setup() {
+        wireMockServer = new WireMockServer(wireMockConfig().dynamicPort());
+        wireMockServer.start();
+        configureFor("localhost", wireMockServer.port());
+    }
+
+    @After
+    public void tearDown() {
+        wireMockServer.stop();
+    }
+
+    @AfterEach
+    public void tearDownEach() {
+        wireMockServer.resetAll();
+    }
+
+    @Test
+    public void givenResponseTimeoutClientWhenRequestTimeoutThenReadTimeoutException() {
+        val path = "/response-timeout";
+        val delay = Math.toIntExact(Duration.ofSeconds(2).toMillis());
+        stubFor(get(urlEqualTo(path)).willReturn(aResponse().withFixedDelay(delay)
+          .withStatus(HttpStatus.OK.value())));
+
+        val webClient = WebClientTimeoutProvider.responseTimeoutClient();
+
+        val ex = assertThrows(RuntimeException.class, () ->
+          webClient.get()
+            .uri(wireMockServer.baseUrl() + path)
+            .exchangeToMono(Mono::just)
+            .log()
+            .block());
+        assertThat(ex).isInstanceOf(WebClientRequestException.class)
+          .getCause().isInstanceOf(ReadTimeoutException.class);
+    }
+
+    @Test
+    public void givenReadWriteTimeoutClientWhenRequestTimeoutThenReadTimeoutException() {
+        val path = "/read-write-timeout";
+        val delay = Math.toIntExact(Duration.ofSeconds(6).toMillis());
+        stubFor(get(urlEqualTo(path)).willReturn(aResponse().withFixedDelay(delay)
+          .withStatus(HttpStatus.OK.value())));
+
+        val webClient = WebClientTimeoutProvider.readWriteTimeoutClient();
+
+        val ex = assertThrows(RuntimeException.class, () ->
+          webClient.get()
+            .uri(wireMockServer.baseUrl() + path)
+            .exchangeToMono(Mono::just)
+            .log()
+            .block());
+        assertThat(ex).isInstanceOf(WebClientRequestException.class)
+          .getCause().isInstanceOf(ReadTimeoutException.class);
+    }
+
+    @Test
+    public void givenNoTimeoutClientAndReactorTimeoutWhenRequestTimeoutThenTimeoutException() {
+        val path = "/reactor-timeout";
+        val delay = Math.toIntExact(Duration.ofSeconds(5).toMillis());
+        stubFor(get(urlEqualTo(path)).willReturn(aResponse().withFixedDelay(delay)
+          .withStatus(HttpStatus.OK.value())));
+
+        val webClient = WebClientTimeoutProvider.defaultWebClient();
+
+        val ex = assertThrows(RuntimeException.class, () ->
+          webClient.get()
+            .uri(wireMockServer.baseUrl() + path)
+            .exchangeToMono(Mono::just)
+            .timeout(Duration.ofSeconds(1))
+            .log()
+            .block());
+        assertThat(ex).hasMessageContaining("Did not observe any item")
+          .getCause().isInstanceOf(TimeoutException.class);
+    }
+
+    @Test
+    public void givenNoTimeoutClientAndTimeoutHttpRequestWhenRequestTimeoutThenReadTimeoutException() {
+        val path = "/reactor-http-request-timeout";
+        val delay = Math.toIntExact(Duration.ofSeconds(5).toMillis());
+        stubFor(get(urlEqualTo(path)).willReturn(aResponse().withFixedDelay(delay)
+          .withStatus(HttpStatus.OK.value())));
+
+        val webClient = WebClientTimeoutProvider.defaultWebClient();
+
+        val ex = assertThrows(RuntimeException.class, () ->
+          webClient.get()
+            .uri(wireMockServer.baseUrl() + path)
+            .httpRequest(httpRequest -> {
+              HttpClientRequest reactorRequest = httpRequest.getNativeRequest();
+              reactorRequest.responseTimeout(Duration.ofSeconds(1));
+            })
+            .exchangeToMono(Mono::just)
+            .log()
+            .block());
+        assertThat(ex).isInstanceOf(WebClientRequestException.class)
+          .getCause().isInstanceOf(ReadTimeoutException.class);
+    }
+}

spring-aop/pom.xml

@@ -14,6 +14,15 @@
     </parent>
 
     <dependencies>
+        <dependency>
+            <groupId>org.aspectj</groupId>
+            <artifactId>aspectjrt</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.aspectj</groupId>
+            <artifactId>aspectjweaver</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-aop</artifactId>
@@ -23,7 +32,41 @@
             <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>aspectj-maven-plugin</artifactId>
+                <version>${aspectj-plugin.version}</version>
+                <configuration>
+                    <complianceLevel>${java.version}</complianceLevel>
+                    <source>${java.version}</source>
+                    <target>${java.version}</target>
+                    <showWeaveInfo>true</showWeaveInfo>
+                    <verbose>true</verbose>
+                    <Xlint>ignore</Xlint>
+                    <encoding>UTF-8</encoding>
+                </configuration>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>compile</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+    <properties>
+        <aspectj-plugin.version>1.11</aspectj-plugin.version>
+    </properties>
 </project>
 

spring-aop/src/main/java/com/baeldung/aspectj/classmethodadvice/MyTracedService.java

@@ -0,0 +1,19 @@
+package com.baeldung.aspectj.classmethodadvice;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.stereotype.Component;
+
+@Trace
+@Component
+public class MyTracedService {
+    private static final Log LOG = LogFactory.getLog(MyTracedService.class);
+
+    public void performSomeLogic() {
+        LOG.info("Inside performSomeLogic...");
+    }
+
+    public void performSomeAdditionalLogic() {
+        LOG.info("Inside performSomeAdditionalLogic...");
+    }
+}

spring-aop/src/main/java/com/baeldung/aspectj/classmethodadvice/MyTracedServiceConsumer.java

@@ -0,0 +1,12 @@
+package com.baeldung.aspectj.classmethodadvice;
+
+import org.springframework.stereotype.Component;
+
+@Component
+public class MyTracedServiceConsumer {
+
+    public MyTracedServiceConsumer(MyTracedService myTracedService) {
+        myTracedService.performSomeLogic();
+        myTracedService.performSomeAdditionalLogic();
+    }
+}

spring-aop/src/main/java/com/baeldung/aspectj/classmethodadvice/Trace.java

@@ -0,0 +1,12 @@
+package com.baeldung.aspectj.classmethodadvice;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+public @interface Trace {
+
+}

spring-aop/src/main/java/com/baeldung/aspectj/classmethodadvice/TracingAspect.aj

@@ -0,0 +1,24 @@
+package com.baeldung.aspectj.classmethodadvice;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+public aspect TracingAspect {
+    private static final Log LOG = LogFactory.getLog(TracingAspect.class);
+
+    pointcut traceAnnotatedClasses(): within(@Trace *) && execution(* *(..));
+
+    Object around() : traceAnnotatedClasses() {
+        String signature = thisJoinPoint.getSignature().toShortString();
+        LOG.trace("Entering " + signature);
+        try {
+            return proceed();
+        } finally {
+            LOG.trace("Exiting " + signature);
+        }
+    }
+
+    after() throwing (Exception e) : traceAnnotatedClasses() {
+        LOG.trace("Exception thrown from " + thisJoinPoint.getSignature().toShortString(), e);
+    }
+}

spring-aop/src/main/resources/logback.xml

@@ -17,6 +17,8 @@
     
     <logger name="org.springframework.aop.interceptor.PerformanceMonitorInterceptor" level="TRACE" />
 
+    <logger name="com.baeldung.aspectj.classmethodadvice" level="TRACE" />
+
     <root level="TRACE">
         <appender-ref ref="STDOUT" />
     </root>

spring-aop/src/test/java/com/baeldung/aspectj/classmethodadvice/MyTracedServiceConsumerUnitTest.java

@@ -0,0 +1,32 @@
+package com.baeldung.aspectj.classmethodadvice;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.mockito.Spy;
+import org.mockito.junit.MockitoJUnit;
+import org.mockito.junit.MockitoRule;
+
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.verify;
+
+public class MyTracedServiceConsumerUnitTest {
+
+    @Rule
+    public MockitoRule mockitoRule = MockitoJUnit.rule();
+
+    @Spy
+    private MyTracedService myTracedService;
+
+    @Test
+    public void whenCallingConsumer_thenServiceIsCalled() {
+        doNothing().when(myTracedService)
+            .performSomeLogic();
+        doNothing().when(myTracedService)
+            .performSomeAdditionalLogic();
+
+        new MyTracedServiceConsumer(myTracedService);
+
+        verify(myTracedService).performSomeLogic();
+        verify(myTracedService).performSomeAdditionalLogic();
+    }
+}

spring-aop/src/test/java/com/baeldung/aspectj/classmethodadvice/MyTracedServiceUnitTest.java

@@ -0,0 +1,40 @@
+package com.baeldung.aspectj.classmethodadvice;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.springframework.boot.test.system.OutputCaptureRule;
+
+import static org.junit.Assert.assertTrue;
+
+/*
+ * When running this test class, the tests may fail unless you build the code with Maven first. You
+ * must ensure the AspectJ compiler executes to weave in the Aspect's logic. Without the Aspect
+ * weaved into the class under test, the trace logging will not be written to stdout.
+ */
+public class MyTracedServiceUnitTest {
+
+    @Rule
+    public OutputCaptureRule outputCaptureRule = new OutputCaptureRule();
+
+    @Test
+    public void whenPerformingSomeLogic_thenTraceAndInfoOutputIsWritten() {
+        MyTracedService myTracedService = new MyTracedService();
+        myTracedService.performSomeLogic();
+
+        String output = outputCaptureRule.getOut();
+        assertTrue(output.contains("TracingAspect - Entering MyTracedService.performSomeLogic"));
+        assertTrue(output.contains("MyTracedService - Inside performSomeLogic"));
+        assertTrue(output.contains("TracingAspect - Exiting MyTracedService.performSomeLogic"));
+    }
+
+    @Test
+    public void whenPerformingSomeAdditionalLogic_thenTraceAndInfoOutputIsWritten() {
+        MyTracedService myTracedService = new MyTracedService();
+        myTracedService.performSomeAdditionalLogic();
+
+        String output = outputCaptureRule.getOut();
+        assertTrue(output.contains("TracingAspect - Entering MyTracedService.performSomeAdditionalLogic"));
+        assertTrue(output.contains("MyTracedService - Inside performSomeAdditionalLogic"));
+        assertTrue(output.contains("TracingAspect - Exiting MyTracedService.performSomeAdditionalLogic"));
+    }
+}

spring-boot-modules/pom.xml

@@ -34,6 +34,7 @@
         <module>spring-boot-ctx-fluent</module>
         <module>spring-boot-deployment</module>
         <module>spring-boot-di</module>
+        <module>spring-boot-disable-logging</module>
         <module>spring-boot-camel</module>
         <module>spring-boot-ci-cd</module>
         <!-- <module>spring-boot-cli</module> --> <!-- Not a maven project -->
@@ -73,6 +74,7 @@
         <module>spring-boot-xml</module>
         <module>spring-boot-actuator</module>
         <module>spring-boot-data-2</module>
+        <module>spring-boot-react</module>
     </modules>
 
 

spring-boot-modules/spring-boot-disable-logging/README.md

@@ -0,0 +1,8 @@
+## Spring Boot Disable Logging
+
+This module contains articles about disabling logging in Spring Boot
+
+
+### Relevant Articles: 
+
+- [How to Disable Console Logging in Spring Boot](https://www.baeldung.com/spring-boot-disable-console-logging)

spring-boot-modules/spring-boot-disable-logging/disabling-console-jul/pom.xml

@@ -8,7 +8,7 @@
 
     <parent>
         <groupId>com.baeldung.spring-boot-modules</groupId>
-        <artifactId>spring-boot-runtime</artifactId>
+        <artifactId>spring-boot-disable-logging</artifactId>
         <version>1.0.0-SNAPSHOT</version>
         <relativePath>../</relativePath>
     </parent>

spring-boot-modules/spring-boot-disable-logging/disabling-console-log4j2/pom.xml

@@ -8,7 +8,7 @@
 
     <parent>
         <groupId>com.baeldung.spring-boot-modules</groupId>
-        <artifactId>spring-boot-runtime</artifactId>
+        <artifactId>spring-boot-disable-logging</artifactId>
         <version>1.0.0-SNAPSHOT</version>
         <relativePath>../</relativePath>
     </parent>

spring-boot-modules/spring-boot-disable-logging/disabling-console-logback/pom.xml

@@ -8,7 +8,7 @@
 
     <parent>
         <groupId>com.baeldung.spring-boot-modules</groupId>
-        <artifactId>spring-boot-runtime</artifactId>
+        <artifactId>spring-boot-disable-logging</artifactId>
         <version>1.0.0-SNAPSHOT</version>
         <relativePath>../</relativePath>
     </parent>