diff --git a/.github/workflows/maven-ci.yml b/.github/workflows/maven-ci.yml index 00ad59c7dab2..ebbe1886ba15 100644 --- a/.github/workflows/maven-ci.yml +++ b/.github/workflows/maven-ci.yml @@ -41,6 +41,10 @@ jobs: key: ${{ runner.os }}-sonar restore-keys: ${{ runner.os }}-sonar + # Start Kafka Docker Service for microservices-messaging + - name: Start Kafka Docker Service + run: docker compose -f microservices-messaging/docker-compose.yml up -d --wait + # Some tests need screen access - name: Install xvfb run: sudo apt-get install -y xvfb @@ -52,6 +56,10 @@ jobs: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }} + - name: Stop Kafka Docker Service + if: always() + run: docker compose -f microservices-messaging/docker-compose.yml down + - name: Upload coverage reports to Codecov uses: codecov/codecov-action@v5 with: diff --git a/.github/workflows/maven-pr-builder.yml b/.github/workflows/maven-pr-builder.yml index 149733343781..2a4eff8f0d9c 100644 --- a/.github/workflows/maven-pr-builder.yml +++ b/.github/workflows/maven-pr-builder.yml @@ -33,6 +33,10 @@ jobs: restore-keys: | ${{ runner.os }}-maven- + # Start Kafka Docker Service for microservices-messaging + - name: Start Kafka Docker Service + run: docker compose -f microservices-messaging/docker-compose.yml up -d --wait + # Some tests need screen access - name: Install xvfb run: sudo apt-get install -y xvfb @@ -40,6 +44,10 @@ jobs: - name: Build with Maven run: xvfb-run ./mvnw clean verify + - name: Stop Kafka Docker Service + if: always() + run: docker compose -f microservices-messaging/docker-compose.yml down + - name: Upload coverage reports to Codecov uses: codecov/codecov-action@v5 with: diff --git a/microservices-messaging/README.md b/microservices-messaging/README.md new file mode 100644 index 000000000000..3ca95350e2a8 --- /dev/null +++ b/microservices-messaging/README.md @@ -0,0 +1,247 @@ +--- +title: "Microservices Messaging Pattern in Java: Enabling Asynchronous Communication Between Services" +shortTitle: Microservices Messaging +description: "Learn about the Microservices Messaging pattern, a method for enabling asynchronous communication between services through message brokers to enhance decoupling, scalability, and fault tolerance in distributed systems." +category: Integration +language: en +tag: + - API design + - Asynchronous + - Cloud distributed + - Decoupling + - Enterprise patterns + - Event-driven + - Messaging + - Microservices + - Scalability +--- +## Also known as + +* Asynchronous Messaging +* Event-Driven Communication +* Message-Oriented Middleware (MOM) + +## Intent of Microservices Messaging Design Pattern + +The Microservices Messaging pattern enables asynchronous communication between microservices through message passing, allowing for better decoupling, scalability, and fault tolerance. Services communicate by exchanging messages over messaging channels managed by a message broker. + +## Detailed Explanation of Microservices Messaging Pattern with Real-World Examples + +Real-world example + +> Imagine an e-commerce platform where a customer places an order. The Order Service publishes an "Order Created" message to a message broker. Multiple services listen to this message: the Inventory Service updates stock levels, the Payment Service processes payment, and the Notification Service sends confirmation emails. Each service operates independently, processing messages at its own pace without blocking others. If the Payment Service is temporarily down, the message broker holds the message until it recovers, ensuring no data is lost. + +In plain words + +> The Microservices Messaging pattern allows services to communicate asynchronously through a message broker, enabling them to work independently without waiting for each other. + +Wikipedia says + +> Message-oriented middleware is software or hardware infrastructure supporting sending and receiving messages between distributed systems. MOM allows application modules to be distributed over heterogeneous platforms and reduces the complexity of developing applications that span multiple operating systems and network protocols. + +Flowchart + +![Microservices Messaging flowchart](./etc/microservices-messaging-flowchart.png) + + + +## Programmatic Example of Microservices Messaging Pattern in Java + + +The Microservices Messaging pattern demonstrates how services communicate through a message broker without direct coupling. In this example, we show an order processing system where services exchange messages asynchronously. + +The `Message` class represents the data exchanged between services. + +```java +public class Message { + private final String id; + private final String content; + private final LocalDateTime timestamp; + + public Message(String content) { + this.id = UUID.randomUUID().toString(); + this.content = content; + this.timestamp = LocalDateTime.now(); + } + + // Getters +} +``` + +The `MessageBroker` acts as the intermediary that routes messages between producers and consumers. + +```java +public class MessageBroker { + private final Map subscribers = new ConcurrentHashMap<>(); + + public void subscribe(String topic, Consumer handler) { + subscribers.computeIfAbsent(topic, k -> new ArrayList<>()).add(handler); + } + + public void publish(String topic, Message message) { + List handlers = subscribers.get(topic); + if (handlers != null) { + handlers.forEach(handler -> handler.accept(message)); + } + } +} +``` + +The `OrderService` is a message producer that publishes order messages. + +```java +public class OrderService { + private static final Logger LOGGER = LoggerFactory.getLogger(OrderService.class); + private final MessageBroker broker; + + public OrderService(MessageBroker broker) { + this.broker = broker; + } + + public void createOrder(String orderId) { + Message message = new Message("Order Created: " + orderId); + broker.publish("order-topic", message); + LOGGER.info("Published order message: {}", orderId); + } +} +``` + +The `InventoryService` is a message consumer that processes inventory updates. + +```java +public class InventoryService { + private static final Logger LOGGER = LoggerFactory.getLogger(InventoryService.class); + + public void handleMessage(Message message) { + LOGGER.info("Inventory Service received: {}", message.getContent()); + LOGGER.info("Updating inventory..."); + } +} +``` + +The `PaymentService` handles payment processing messages. + +```java +public class PaymentService { + private static final Logger LOGGER = LoggerFactory.getLogger(PaymentService.class); + + public void handleMessage(Message message) { + LOGGER.info("Payment Service received: {}", message.getContent()); + LOGGER.info("Processing payment..."); + } +} +``` + +The `main` application demonstrates the messaging pattern in action. + +```java +public class App { + private static final Logger LOGGER = LoggerFactory.getLogger(App.class); + + public static void main(String[] args) throws InterruptedException { + final MessageBroker broker = new MessageBroker(); + + final InventoryService inventoryService = new InventoryService(); + final PaymentService paymentService = new PaymentService(); + + broker.subscribe("order-topic", inventoryService::handleMessage); + broker.subscribe("order-topic", paymentService::handleMessage); + + final OrderService orderService = new OrderService(broker); + + orderService.createOrder("ORDER-123"); + + Thread.sleep(1000); + } +} +``` + +Console output: + +``` +Published order message: ORDER-123 +Inventory Service received: Order Created: ORDER-123 +Updating inventory... +Payment Service received: Order Created: ORDER-123 +Processing payment... +``` + +Sequence Diagram + +![Microservices Messaging sequence_diagram](./etc/microservices-messaging-sequence-diagram.png) + +## How to Run the Application + +### Option 1: Automated Script (Recommended) + +Run the helper script from the module directory, which automatically starts Kafka via Docker Compose (if Docker is installed and Kafka is not already running) and launches the application: + +* **Windows (PowerShell)**: + ```powershell + powershell -ExecutionPolicy Bypass -File .\run-app.ps1 + ``` +* **Linux / macOS**: + ```bash + ./run-app.sh + ``` + +### Option 2: Docker Compose + +Start the Kafka container manually via Docker Compose and run the application: + +```bash +# Start Kafka container on port 9092 +docker compose up -d + +# Run the application +../mvnw compile exec:java -Dexec.mainClass="com.iluwatar.messaging.App" + +# Stop Kafka container when finished +docker compose down +``` + +## When to Use the Microservices Messaging Pattern in Java + +* When services need to communicate without blocking each other. +* In systems requiring loose coupling between components. +* For event-driven architectures where multiple services react to events. +* When you need to handle traffic spikes by buffering messages. +* In distributed systems where services may be temporarily unavailable. + +## Real-World Applications of Microservices Messaging Pattern in Java + +* Java applications using Apache Kafka, RabbitMQ, or ActiveMQ for service communication. +* E-commerce platforms for order processing and inventory management. +* Financial services for transaction processing and notifications. +* IoT systems for sensor data processing and event handling. + +## Benefits and Trade-offs of Microservices Messaging Pattern + +* Services are loosely coupled and can be developed and deployed independently. +* Message buffering improves system resilience when services are temporarily unavailable. +* Supports multiple communication patterns like publish/subscribe and request/reply. +* Enhances scalability by allowing parallel message processing. +* Natural support for event-driven architectures. + +Trade-offs: + +* Introduces additional complexity with the message broker infrastructure. +* Requires high availability setup for the message broker. +* Eventual consistency instead of immediate consistency. +* Debugging asynchronous flows is more complex than synchronous calls. +* Need to handle message duplication and ensure idempotent consumers. + +## Related Java Design Patterns + +* [Saga Pattern](https://java-design-patterns.com/patterns/saga/): Uses messaging to coordinate distributed transactions. +* [CQRS Pattern](https://java-design-patterns.com/patterns/cqrs/): Often uses messaging to separate read and write operations. +* [Event Sourcing](https://java-design-patterns.com/patterns/event-sourcing/): Stores state changes as messages. +* [API Gateway](https://java-design-patterns.com/patterns/microservices-api-gateway/): Complements messaging for synchronous requests. + +## References and Credits + +* [Enterprise Integration Patterns: Designing, Building, and Deploying Messaging Solutions](https://amzn.to/3vLKqET) +* [Microservices Patterns: With examples in Java](https://amzn.to/3UyWD5O) +* [Building Event-Driven Microservices: Leveraging Organizational Data at Scale](https://amzn.to/3PihS9R) +* [Pattern: Messaging (microservices.io)](https://microservices.io/patterns/communication-style/messaging.html) +* [Apache Kafka Documentation](https://kafka.apache.org/documentation/) \ No newline at end of file diff --git a/microservices-messaging/docker-compose.yml b/microservices-messaging/docker-compose.yml new file mode 100644 index 000000000000..b77ee549cd51 --- /dev/null +++ b/microservices-messaging/docker-compose.yml @@ -0,0 +1,53 @@ +# +# This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). +# +# The MIT License +# Copyright © 2014-2022 Ilkka Seppälä +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. +# + +version: '3.8' + +services: + kafka: + image: confluentinc/cp-kafka:7.5.0 + container_name: kafka-messaging-demo + ports: + - "9092:9092" + environment: + KAFKA_NODE_ID: 1 + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT' + KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092' + KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 + KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 + KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 + KAFKA_PROCESS_ROLES: 'broker,controller' + KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:29093' + KAFKA_LISTENERS: 'PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,PLAINTEXT_HOST://0.0.0.0:9092' + KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT' + KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER' + KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs' + CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk' + healthcheck: + test: ["CMD-SHELL", "kafka-topics --bootstrap-server localhost:9092 --list"] + interval: 5s + timeout: 10s + retries: 5 diff --git a/microservices-messaging/etc/microservices-messaging-flowchart.png b/microservices-messaging/etc/microservices-messaging-flowchart.png new file mode 100644 index 000000000000..dd26a2e2ce4a Binary files /dev/null and b/microservices-messaging/etc/microservices-messaging-flowchart.png differ diff --git a/microservices-messaging/etc/microservices-messaging-sequence-diagram.png b/microservices-messaging/etc/microservices-messaging-sequence-diagram.png new file mode 100644 index 000000000000..d22be5acbb5d Binary files /dev/null and b/microservices-messaging/etc/microservices-messaging-sequence-diagram.png differ diff --git a/microservices-messaging/pom.xml b/microservices-messaging/pom.xml new file mode 100644 index 000000000000..8fb9d59c4f74 --- /dev/null +++ b/microservices-messaging/pom.xml @@ -0,0 +1,118 @@ + + + + 4.0.0 + + + com.iluwatar + java-design-patterns + 1.26.0-SNAPSHOT + + + microservices-messaging + 1.26.0-SNAPSHOT + + + 3.6.1 + + + + + + org.apache.kafka + kafka-clients + ${kafka.version} + + + + + org.projectlombok + lombok + ${lombok.version} + provided + + + + + org.slf4j + slf4j-api + + + ch.qos.logback + logback-classic + + + + + com.fasterxml.jackson.core + jackson-databind + 2.16.1 + + + + + org.junit.jupiter + junit-jupiter-engine + test + + + org.mockito + mockito-core + test + + + com.fasterxml.jackson.datatype + jackson-datatype-jsr310 + 2.19.2 + compile + + + + + + + org.apache.maven.plugins + maven-assembly-plugin + + + + + + com.iluwatar.messaging.App + + + + + + + + + \ No newline at end of file diff --git a/microservices-messaging/run-app.ps1 b/microservices-messaging/run-app.ps1 new file mode 100644 index 000000000000..6fd4da50dfbe --- /dev/null +++ b/microservices-messaging/run-app.ps1 @@ -0,0 +1,89 @@ +# +# This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). +# +# The MIT License +# Copyright © 2014-2022 Ilkka Seppälä +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. +# + +# PowerShell script to start Kafka container (if Docker is available) and run the Microservices Messaging App + +$ScriptDir = Split-Path -Parent $MyInvocation.MyCommand.Definition +Set-Location $ScriptDir + +Write-Host "======================================================" -ForegroundColor Cyan +Write-Host " Starting Microservices Messaging Pattern Application" -ForegroundColor Cyan +Write-Host "======================================================" -ForegroundColor Cyan + +# Check if Kafka is already running on port 9092 +$portActive = $false +try { + $socket = New-Object System.Net.Sockets.TcpClient("localhost", 9092) + if ($socket.Connected) { + $portActive = $true + $socket.Close() + } +} catch { + $portActive = $false +} + +if ($portActive) { + Write-Host "[INFO] Kafka is already running on port 9092." -ForegroundColor Green +} else { + # Check if Docker is available, or add Docker Desktop to PATH + if (-not (Get-Command docker -ErrorAction SilentlyContinue)) { + $dockerPath = "$env:LOCALAPPDATA\Programs\DockerDesktop\resources\bin" + if (Test-Path $dockerPath) { + $env:PATH += ";$dockerPath" + } + } + $dockerCmd = Get-Command docker -ErrorAction SilentlyContinue + if ($dockerCmd) { + Write-Host "[INFO] Starting Kafka container via Docker Compose..." -ForegroundColor Yellow + docker compose up -d + + Write-Host "[INFO] Waiting for Kafka to become ready on port 9092..." -ForegroundColor Yellow + $retryCount = 0 + while (-not $portActive -and $retryCount -lt 20) { + Start-Sleep -Seconds 2 + try { + $socket = New-Object System.Net.Sockets.TcpClient("localhost", 9092) + if ($socket.Connected) { + $portActive = $true + $socket.Close() + } + } catch { + $retryCount++ + } + } + + if ($portActive) { + Write-Host "[INFO] Kafka container started successfully!" -ForegroundColor Green + } else { + Write-Host "[WARNING] Kafka did not become ready on port 9092 within timeout." -ForegroundColor Red + } + } else { + Write-Host "[WARNING] Docker is not installed or not in PATH." -ForegroundColor Yellow + Write-Host "[INFO] Please ensure Kafka is running locally on localhost:9092 before running the app." -ForegroundColor Yellow + } +} + +Write-Host "[INFO] Compiling and running App.java..." -ForegroundColor Cyan +& "..\mvnw.cmd" compile exec:java "-Dexec.mainClass=com.iluwatar.messaging.App" diff --git a/microservices-messaging/run-app.sh b/microservices-messaging/run-app.sh new file mode 100644 index 000000000000..8c31c185190e --- /dev/null +++ b/microservices-messaging/run-app.sh @@ -0,0 +1,62 @@ +#!/usr/bin/env bash +# +# This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). +# +# The MIT License +# Copyright © 2014-2022 Ilkka Seppälä +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. +# + +# Shell script to start Kafka container (if Docker is available) and run the Microservices Messaging App + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +cd "$SCRIPT_DIR" + +echo "======================================================" +echo " Starting Microservices Messaging Pattern Application" +echo "======================================================" + +if nc -z localhost 9092 2>/dev/null || (echo > /dev/tcp/localhost/9092) 2>/dev/null; then + echo "[INFO] Kafka is already running on port 9092." +else + if command -v docker &> /dev/null; then + echo "[INFO] Starting Kafka container via Docker Compose..." + docker compose up -d + + echo "[INFO] Waiting for Kafka to become ready on port 9092..." + retry=0 + until nc -z localhost 9092 2>/dev/null || (echo > /dev/tcp/localhost/9092) 2>/dev/null || [ $retry -eq 20 ]; do + sleep 2 + retry=$((retry+1)) + done + + if [ $retry -lt 20 ]; then + echo "[INFO] Kafka container started successfully!" + else + echo "[WARNING] Kafka did not become ready on port 9092 within timeout." + fi + else + echo "[WARNING] Docker is not installed or not in PATH." + echo "[INFO] Please ensure Kafka is running locally on localhost:9092." + fi +fi + +echo "[INFO] Compiling and running App.java..." +../mvnw compile exec:java -Dexec.mainClass="com.iluwatar.messaging.App" diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/App.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/App.java new file mode 100644 index 000000000000..66e5f97c7f48 --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/App.java @@ -0,0 +1,144 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * The Microservices Messaging pattern enables asynchronous communication between services through + * Apache Kafka. This example demonstrates how services can communicate without tight coupling. + * + *

In this example: + * + *

+ * + *

Key benefits demonstrated: + * + *

+ * + *

Prerequisites: This example requires a running Kafka instance. Start Kafka locally: + * + * 

+ * # Start Zookeeper
+ * bin/zookeeper-server-start.sh config/zookeeper.properties
+ *
+ * # Start Kafka
+ * bin/kafka-server-start.sh config/server.properties
+ *
+ * # Create topic
+ * bin/kafka-topics.sh --create --topic order-topic --bootstrap-server localhost:9092
+ *
+ */ +public class App { + private static final Logger LOGGER = LoggerFactory.getLogger(App.class); + private static final String BOOTSTRAP_SERVERS = "localhost:9092"; + + /** + * Program entry point. + * + * @param args command line arguments + */ + public static void main(String[] args) throws InterruptedException { + LOGGER.info("Starting Microservices Messaging Pattern with Apache Kafka"); + + // Create Kafka producer + KafkaMessageProducer producer = new KafkaMessageProducer(BOOTSTRAP_SERVERS); + + // Create consumer services + InventoryService inventoryService = new InventoryService(); + PaymentService paymentService = new PaymentService(); + NotificationService notificationService = new NotificationService(); + + // Create Kafka consumers + KafkaMessageConsumer inventoryConsumer = + new KafkaMessageConsumer( + BOOTSTRAP_SERVERS, "inventory-group", "order-topic", inventoryService::handleMessage); + + KafkaMessageConsumer paymentConsumer = + new KafkaMessageConsumer( + BOOTSTRAP_SERVERS, "payment-group", "order-topic", paymentService::handleMessage); + + KafkaMessageConsumer notificationConsumer = + new KafkaMessageConsumer( + BOOTSTRAP_SERVERS, + "notification-group", + "order-topic", + notificationService::handleMessage); + + // Start consumers in separate threads + ExecutorService executor = Executors.newFixedThreadPool(3); + executor.submit(inventoryConsumer); + executor.submit(paymentConsumer); + executor.submit(notificationConsumer); + + // Give consumers time to subscribe + Thread.sleep(2000); + + // Create producer service + OrderService orderService = new OrderService(producer); + + // Demonstrate the messaging pattern + LOGGER.info("\n=== Creating Order ==="); + orderService.createOrder("ORDER-001"); + + Thread.sleep(2000); + + LOGGER.info("\n=== Updating Order ==="); + orderService.updateOrder("ORDER-001"); + + Thread.sleep(2000); + + LOGGER.info("\n=== Cancelling Order ==="); + orderService.cancelOrder("ORDER-001"); + + Thread.sleep(2000); + + // Cleanup + LOGGER.info("\nShutting down..."); + inventoryConsumer.stop(); + paymentConsumer.stop(); + notificationConsumer.stop(); + producer.close(); + + executor.shutdown(); + executor.awaitTermination(5, TimeUnit.SECONDS); + + LOGGER.info("Microservices Messaging Pattern demonstration completed"); + } +} diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/InventoryService.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/InventoryService.java new file mode 100644 index 000000000000..9552e867d1ee --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/InventoryService.java @@ -0,0 +1,86 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * InventoryService is a message consumer that processes inventory-related messages from Kafka. It + * listens to order events and updates inventory accordingly. + * + *

This service runs in its own Kafka consumer group (inventory-group) which allows it to: + * + *

+ */ +public class InventoryService { + private static final Logger LOGGER = LoggerFactory.getLogger(InventoryService.class); + + /** + * Handles incoming messages related to orders from Kafka. + * + * @param message the message to process + */ + public void handleMessage(Message message) { + LOGGER.info( + "Inventory Service received message [{}]: {}", message.getId(), message.getContent()); + + if (message.getContent().contains("Order Created")) { + updateInventory(message); + } else if (message.getContent().contains("Order Cancelled")) { + restoreInventory(message); + } else { + LOGGER.debug("No inventory action needed for: {}", message.getContent()); + } + } + + private void updateInventory(Message message) { + LOGGER.info("Updating inventory for message: {}", message.getId()); + // Simulate inventory update - reserve stock for the order + try { + Thread.sleep(100); + LOGGER.info("Inventory updated successfully for: {}", message.getContent()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Inventory update interrupted", e); + } + } + + private void restoreInventory(Message message) { + LOGGER.info("Restoring inventory for message: {}", message.getId()); + // Simulate inventory restoration - release reserved stock + try { + Thread.sleep(100); + LOGGER.info("Inventory restored successfully for: {}", message.getContent()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Inventory restore interrupted", e); + } + } +} diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/KafkaMessageConsumer.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/KafkaMessageConsumer.java new file mode 100644 index 000000000000..71130e114dd8 --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/KafkaMessageConsumer.java @@ -0,0 +1,125 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import java.time.Duration; +import java.util.Collections; +import java.util.Properties; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Kafka message consumer that subscribes to topics and processes messages. */ +public class KafkaMessageConsumer implements AutoCloseable, Runnable { + private static final Logger LOGGER = LoggerFactory.getLogger(KafkaMessageConsumer.class); + private final Consumer consumer; + private final ObjectMapper objectMapper; + private final String topic; + private final java.util.function.Consumer messageHandler; + private final AtomicBoolean running = new AtomicBoolean(true); + + /** + * Creates a new Kafka message consumer. + * + * @param bootstrapServers Kafka bootstrap servers + * @param groupId consumer group ID + * @param topic topic to subscribe to + * @param messageHandler handler for received messages + */ + public KafkaMessageConsumer( + String bootstrapServers, + String groupId, + String topic, + java.util.function.Consumer messageHandler) { + this(createDefaultConsumer(bootstrapServers, groupId), topic, messageHandler); + } + + KafkaMessageConsumer( + Consumer consumer, + String topic, + java.util.function.Consumer messageHandler) { + this.consumer = consumer; + this.objectMapper = new ObjectMapper(); + this.objectMapper.registerModule(new JavaTimeModule()); + this.topic = topic; + this.messageHandler = messageHandler; + } + + private static Consumer createDefaultConsumer( + String bootstrapServers, String groupId) { + Properties props = new Properties(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"); + return new KafkaConsumer<>(props); + } + + @Override + public void run() { + try { + consumer.subscribe(Collections.singletonList(topic)); + LOGGER.info("Consumer subscribed to topic: {}", topic); + + while (running.get()) { + ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); + records.forEach( + record -> { + try { + Message message = objectMapper.readValue(record.value(), Message.class); + LOGGER.info("Received message from topic '{}': {}", topic, message.getId()); + messageHandler.accept(message); + } catch (Exception e) { + LOGGER.error("Error processing message: {}", e.getMessage(), e); + } + }); + } + } catch (Exception e) { + LOGGER.error("Consumer error: {}", e.getMessage(), e); + } finally { + consumer.close(); + LOGGER.info("Consumer closed for topic: {}", topic); + } + } + + /** Stops the consumer. */ + public void stop() { + running.set(false); + } + + @Override + public void close() { + stop(); + } +} diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/KafkaMessageProducer.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/KafkaMessageProducer.java new file mode 100644 index 000000000000..56154152c008 --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/KafkaMessageProducer.java @@ -0,0 +1,108 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import java.util.Properties; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringSerializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Kafka message producer that publishes messages to Kafka topics. */ +public class KafkaMessageProducer implements AutoCloseable { + private static final Logger LOGGER = LoggerFactory.getLogger(KafkaMessageProducer.class); + private final Producer producer; + private final ObjectMapper objectMapper; + + /** + * Creates a new Kafka message producer. + * + * @param bootstrapServers Kafka bootstrap servers + */ + public KafkaMessageProducer(String bootstrapServers) { + this(createDefaultProducer(bootstrapServers)); + } + + KafkaMessageProducer(Producer producer) { + this.producer = producer; + this.objectMapper = new ObjectMapper(); + this.objectMapper.registerModule(new JavaTimeModule()); + } + + private static Producer createDefaultProducer(String bootstrapServers) { + Properties props = new Properties(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + props.put(ProducerConfig.ACKS_CONFIG, "all"); + props.put(ProducerConfig.RETRIES_CONFIG, 3); + return new KafkaProducer<>(props); + } + + /** + * Publishes a message to a Kafka topic. + * + * @param topic the topic to publish to + * @param message the message to publish + */ + public void publish(String topic, Message message) { + try { + String json = objectMapper.writeValueAsString(message); + ProducerRecord record = new ProducerRecord<>(topic, message.getId(), json); + + producer.send( + record, + (metadata, exception) -> { + if (exception != null) { + LOGGER.error( + "Failed to publish message to topic {}: {}", topic, exception.getMessage()); + } else { + LOGGER.info( + "Published message to topic '{}' [partition={}, offset={}]", + topic, + metadata.partition(), + metadata.offset()); + } + }); + + } catch (Exception e) { + LOGGER.error("Error serializing message: {}", e.getMessage(), e); + } + } + + @Override + public void close() { + if (producer != null) { + producer.flush(); + producer.close(); + LOGGER.info("Kafka producer closed"); + } + } +} diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/Message.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/Message.java new file mode 100644 index 000000000000..5d0e9f96769c --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/Message.java @@ -0,0 +1,75 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import java.time.LocalDateTime; +import java.util.UUID; +import lombok.Getter; + +/** Represents a message exchanged between services. */ +@Getter +public class Message { + private final String id; + private final String content; + private final LocalDateTime timestamp; + + /** + * Creates a new message with the given content. + * + * @param content the message content + */ + public Message(String content) { + this.id = UUID.randomUUID().toString(); + this.content = content; + this.timestamp = LocalDateTime.now(); + } + + /** JSON constructor for deserialization. */ + @JsonCreator + public Message( + @JsonProperty("id") String id, + @JsonProperty("content") String content, + @JsonProperty("timestamp") LocalDateTime timestamp) { + this.id = id; + this.content = content; + this.timestamp = timestamp; + } + + @Override + public String toString() { + return "Message{" + + "id='" + + id + + '\'' + + ", content='" + + content + + '\'' + + ", timestamp=" + + timestamp + + '}'; + } +} diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/NotificationService.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/NotificationService.java new file mode 100644 index 000000000000..461b1a326efb --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/NotificationService.java @@ -0,0 +1,100 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * NotificationService is a message consumer that processes notification-related messages from + * Kafka. It listens to order events and sends notifications to customers. + * + *

This service runs in its own Kafka consumer group (notification-group) which allows it to: + * + *

    + *
  • Process messages independently from other services + *
  • Scale horizontally by adding more instances to the consumer group + *
  • Resume from last committed offset if the service restarts + *
+ */ +public class NotificationService { + private static final Logger LOGGER = LoggerFactory.getLogger(NotificationService.class); + + /** + * Handles incoming messages related to orders from Kafka. + * + * @param message the message to process + */ + public void handleMessage(Message message) { + LOGGER.info( + "Notification Service received message [{}]: {}", message.getId(), message.getContent()); + + if (message.getContent().contains("Order Created")) { + sendOrderConfirmation(message); + } else if (message.getContent().contains("Order Updated")) { + sendOrderUpdate(message); + } else if (message.getContent().contains("Order Cancelled")) { + sendCancellationNotice(message); + } else { + LOGGER.debug("No notification action needed for: {}", message.getContent()); + } + } + + private void sendOrderConfirmation(Message message) { + LOGGER.info("Sending order confirmation for message: {}", message.getId()); + // Simulate sending email/SMS notification to customer + try { + Thread.sleep(50); + LOGGER.info("Order confirmation sent successfully for: {}", message.getContent()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Notification send interrupted", e); + } + } + + private void sendOrderUpdate(Message message) { + LOGGER.info("Sending order update notification for message: {}", message.getId()); + // Simulate sending update notification + try { + Thread.sleep(50); + LOGGER.info("Order update notification sent successfully for: {}", message.getContent()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Notification send interrupted", e); + } + } + + private void sendCancellationNotice(Message message) { + LOGGER.info("Sending cancellation notice for message: {}", message.getId()); + // Simulate sending cancellation notification + try { + Thread.sleep(50); + LOGGER.info("Cancellation notice sent successfully for: {}", message.getContent()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Notification send interrupted", e); + } + } +} diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/OrderService.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/OrderService.java new file mode 100644 index 000000000000..d059e4e3e0fd --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/OrderService.java @@ -0,0 +1,76 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** OrderService is a message producer that publishes order-related messages using Kafka. */ +public class OrderService { + private static final Logger LOGGER = LoggerFactory.getLogger(OrderService.class); + private static final String ORDER_TOPIC = "order-topic"; + + private final KafkaMessageProducer producer; + + public OrderService(KafkaMessageProducer producer) { + this.producer = producer; + } + + /** + * Creates an order and publishes a message to notify other services. + * + * @param orderId the ID of the order to create + */ + public void createOrder(String orderId) { + LOGGER.info("Creating order: {}", orderId); + Message message = new Message("Order Created: " + orderId); + producer.publish(ORDER_TOPIC, message); + LOGGER.info("Order creation message published for: {}", orderId); + } + + /** + * Updates an order and publishes a message to notify other services. + * + * @param orderId the ID of the order to update + */ + public void updateOrder(String orderId) { + LOGGER.info("Updating order: {}", orderId); + Message message = new Message("Order Updated: " + orderId); + producer.publish(ORDER_TOPIC, message); + LOGGER.info("Order update message published for: {}", orderId); + } + + /** + * Cancels an order and publishes a message to notify other services. + * + * @param orderId the ID of the order to cancel + */ + public void cancelOrder(String orderId) { + LOGGER.info("Cancelling order: {}", orderId); + Message message = new Message("Order Cancelled: " + orderId); + producer.publish(ORDER_TOPIC, message); + LOGGER.info("Order cancellation message published for: {}", orderId); + } +} diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/PaymentService.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/PaymentService.java new file mode 100644 index 000000000000..4c90897a6e8f --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/PaymentService.java @@ -0,0 +1,85 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * PaymentService is a message consumer that processes payment-related messages from Kafka. It + * listens to order events and handles payment processing. + * + *

This service runs in its own Kafka consumer group (payment-group) which allows it to: + * + *

    + *
  • Process messages independently from other services + *
  • Scale horizontally by adding more instances to the consumer group + *
  • Resume from last committed offset if the service restarts + *
+ */ +public class PaymentService { + private static final Logger LOGGER = LoggerFactory.getLogger(PaymentService.class); + + /** + * Handles incoming messages related to orders from Kafka. + * + * @param message the message to process + */ + public void handleMessage(Message message) { + LOGGER.info("Payment Service received message [{}]: {}", message.getId(), message.getContent()); + + if (message.getContent().contains("Order Created")) { + processPayment(message); + } else if (message.getContent().contains("Order Cancelled")) { + refundPayment(message); + } else { + LOGGER.debug("No payment action needed for: {}", message.getContent()); + } + } + + private void processPayment(Message message) { + LOGGER.info("Processing payment for message: {}", message.getId()); + // Simulate payment processing - charge the customer + try { + Thread.sleep(150); + LOGGER.info("Payment processed successfully for: {}", message.getContent()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Payment processing interrupted", e); + } + } + + private void refundPayment(Message message) { + LOGGER.info("Refunding payment for message: {}", message.getId()); + // Simulate payment refund - return money to customer + try { + Thread.sleep(150); + LOGGER.info("Payment refunded successfully for: {}", message.getContent()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Payment refund interrupted", e); + } + } +} diff --git a/microservices-messaging/src/main/resources/logback.xml b/microservices-messaging/src/main/resources/logback.xml new file mode 100644 index 000000000000..7f40741c0ff1 --- /dev/null +++ b/microservices-messaging/src/main/resources/logback.xml @@ -0,0 +1,40 @@ + + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + + + + + + + + + \ No newline at end of file diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/AppTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/AppTest.java new file mode 100644 index 000000000000..2bad9fe0b908 --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/AppTest.java @@ -0,0 +1,38 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; + +import org.junit.jupiter.api.Test; + +/** Unit tests for {@link App}. Tests main application entry point. */ +class AppTest { + + @Test + void testAppConstructor() { + assertDoesNotThrow(() -> new App()); + } +} diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/InventoryServiceTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/InventoryServiceTest.java new file mode 100644 index 000000000000..bb0842ce7aef --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/InventoryServiceTest.java @@ -0,0 +1,118 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link InventoryService}. Tests service behavior with various message types + * without Kafka dependencies. + */ +class InventoryServiceTest { + + private InventoryService inventoryService; + + @BeforeEach + void setUp() { + inventoryService = new InventoryService(); + } + + @Test + void testServiceCanBeInstantiated() { + // Arrange & Act & Assert + assertNotNull(inventoryService, "InventoryService should be instantiated"); + } + + @Test + void testHandleOrderCreatedMessage() { + // Arrange + var message = new Message("Order Created: ORDER-001"); + + // Act & Assert + assertDoesNotThrow( + () -> inventoryService.handleMessage(message), + "Should handle order created message without error"); + } + + @Test + void testHandleOrderCancelledMessage() { + // Arrange + var message = new Message("Order Cancelled: ORDER-001"); + + // Act & Assert + assertDoesNotThrow( + () -> inventoryService.handleMessage(message), + "Should handle order cancelled message without error"); + } + + @Test + void testHandleOrderUpdatedMessage() { + // Arrange + var message = new Message("Order Updated: ORDER-001"); + + // Act & Assert + assertDoesNotThrow( + () -> inventoryService.handleMessage(message), + "Should handle order updated message without error"); + } + + @Test + void testHandleUnknownMessage() { + // Arrange + var message = new Message("Unknown Event: ORDER-001"); + + // Act & Assert + assertDoesNotThrow( + () -> inventoryService.handleMessage(message), + "Should handle unknown message without error"); + } + + @Test + void testHandleMultipleMessages() { + // Act & Assert + assertDoesNotThrow( + () -> { + inventoryService.handleMessage(new Message("Order Created: ORDER-001")); + inventoryService.handleMessage(new Message("Order Updated: ORDER-001")); + inventoryService.handleMessage(new Message("Order Cancelled: ORDER-001")); + }, + "Should handle multiple messages without error"); + } + + @Test + void testHandleMessagesWhenInterrupted() { + Thread.currentThread().interrupt(); + inventoryService.handleMessage(new Message("Order Created: ORDER-001")); + org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted()); + + Thread.currentThread().interrupt(); + inventoryService.handleMessage(new Message("Order Cancelled: ORDER-001")); + org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted()); + } +} diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageConsumerTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageConsumerTest.java new file mode 100644 index 000000000000..9143ca759230 --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageConsumerTest.java @@ -0,0 +1,118 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import java.util.Collections; +import java.util.HashMap; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.MockConsumer; +import org.apache.kafka.clients.consumer.OffsetResetStrategy; +import org.apache.kafka.common.TopicPartition; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** Unit tests for {@link KafkaMessageConsumer}. */ +class KafkaMessageConsumerTest { + + private MockConsumer mockConsumer; + private KafkaMessageConsumer kafkaMessageConsumer; + private AtomicBoolean handlerCalled; + + @BeforeEach + void setUp() { + mockConsumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST); + handlerCalled = new AtomicBoolean(false); + kafkaMessageConsumer = + new KafkaMessageConsumer(mockConsumer, "test-topic", msg -> handlerCalled.set(true)); + } + + @Test + void testConsumerCanBeInstantiated() { + assertNotNull(kafkaMessageConsumer, "KafkaMessageConsumer should be instantiated"); + } + + @Test + void testRunProcessesValidMessageAndStops() throws Exception { + ObjectMapper mapper = new ObjectMapper().registerModule(new JavaTimeModule()); + Message msg = new Message("Order Created: 123"); + String jsonStr = mapper.writeValueAsString(msg); + + TopicPartition tp = new TopicPartition("test-topic", 0); + mockConsumer.updateBeginningOffsets( + new HashMap<>() { + { + put(tp, 0L); + } + }); + + mockConsumer.schedulePollTask( + () -> { + mockConsumer.rebalance(Collections.singletonList(tp)); + mockConsumer.addRecord(new ConsumerRecord<>("test-topic", 0, 0L, "key", jsonStr)); + }); + + mockConsumer.schedulePollTask(() -> kafkaMessageConsumer.stop()); + + kafkaMessageConsumer.run(); + + assertTrue(handlerCalled.get(), "Handler should have been invoked"); + assertTrue(mockConsumer.closed(), "Consumer should be closed"); + } + + @Test + void testRunHandlesInvalidJsonMessage() { + TopicPartition tp = new TopicPartition("test-topic", 0); + mockConsumer.updateBeginningOffsets( + new HashMap<>() { + { + put(tp, 0L); + } + }); + + mockConsumer.schedulePollTask( + () -> { + mockConsumer.rebalance(Collections.singletonList(tp)); + mockConsumer.addRecord( + new ConsumerRecord<>("test-topic", 0, 0L, "key", "{invalid json")); + }); + + mockConsumer.schedulePollTask(() -> kafkaMessageConsumer.stop()); + + assertDoesNotThrow(() -> kafkaMessageConsumer.run()); + assertTrue(mockConsumer.closed(), "Consumer should be closed"); + } + + @Test + void testCloseStopsConsumer() { + assertDoesNotThrow(() -> kafkaMessageConsumer.close()); + } +} diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageProducerTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageProducerTest.java new file mode 100644 index 000000000000..f1bb04979b7d --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageProducerTest.java @@ -0,0 +1,82 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.apache.kafka.clients.producer.MockProducer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** Unit tests for {@link KafkaMessageProducer}. */ +class KafkaMessageProducerTest { + + private MockProducer mockProducer; + private KafkaMessageProducer kafkaMessageProducer; + + @BeforeEach + void setUp() { + mockProducer = new MockProducer<>(true, new StringSerializer(), new StringSerializer()); + kafkaMessageProducer = new KafkaMessageProducer(mockProducer); + } + + @Test + void testProducerCanBeInstantiated() { + assertNotNull(kafkaMessageProducer, "KafkaMessageProducer should be instantiated"); + } + + @Test + void testPublishMessageSuccess() { + Message message = new Message("Test Order"); + + assertDoesNotThrow(() -> kafkaMessageProducer.publish("test-topic", message)); + assertEquals(1, mockProducer.history().size()); + assertEquals("test-topic", mockProducer.history().get(0).topic()); + assertEquals(message.getId(), mockProducer.history().get(0).key()); + } + + @Test + void testPublishMessageErrorCallback() { + MockProducer failingProducer = + new MockProducer<>(false, new StringSerializer(), new StringSerializer()); + try (KafkaMessageProducer producerWithError = new KafkaMessageProducer(failingProducer)) { + Message message = new Message("Test Order"); + + assertDoesNotThrow(() -> producerWithError.publish("test-topic", message)); + assertEquals(1, failingProducer.history().size()); + failingProducer.errorNext(new RuntimeException("Kafka publish error")); + } + } + + @Test + void testClose() { + assertDoesNotThrow(() -> kafkaMessageProducer.close()); + assertTrue(mockProducer.closed()); + } +} diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/MessageTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/MessageTest.java new file mode 100644 index 000000000000..d80cc2b806ca --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/MessageTest.java @@ -0,0 +1,165 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import java.time.LocalDateTime; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link Message}. Tests follow FIRST principles: Fast, Isolated, Repeatable, + * Self-validating, Timely. + */ +class MessageTest { + + private ObjectMapper objectMapper; + + @BeforeEach + void setUp() { + objectMapper = new ObjectMapper(); + objectMapper.registerModule(new JavaTimeModule()); + } + + @Test + void testMessageCreation() { + // Arrange & Act + var message = new Message("Test content"); + + // Assert + assertNotNull(message.getId(), "Message ID should not be null"); + assertEquals("Test content", message.getContent(), "Content should match"); + assertNotNull(message.getTimestamp(), "Timestamp should not be null"); + } + + @Test + void testMessageIdIsUnique() { + // Arrange & Act + var message1 = new Message("Content 1"); + var message2 = new Message("Content 2"); + + // Assert + assertNotEquals(message1.getId(), message2.getId(), "Each message should have unique ID"); + } + + @Test + void testMessageTimestamp() { + // Arrange + var beforeCreation = LocalDateTime.now(); + + // Act + var message = new Message("Test"); + var afterCreation = LocalDateTime.now(); + + // Assert + assertTrue( + message.getTimestamp().isAfter(beforeCreation.minusSeconds(1)) + && message.getTimestamp().isBefore(afterCreation.plusSeconds(1)), + "Timestamp should be close to creation time"); + } + + @Test + void testJsonSerialization() throws Exception { + // Arrange + var originalMessage = new Message("Test content"); + + // Act + var json = objectMapper.writeValueAsString(originalMessage); + + // Assert + assertNotNull(json, "JSON should not be null"); + assertTrue(json.contains("Test content"), "JSON should contain content"); + assertTrue(json.contains(originalMessage.getId()), "JSON should contain ID"); + } + + @Test + void testJsonDeserialization() throws Exception { + // Arrange + var originalMessage = new Message("Test content"); + var json = objectMapper.writeValueAsString(originalMessage); + + // Act + var deserializedMessage = objectMapper.readValue(json, Message.class); + + // Assert + assertNotNull(deserializedMessage, "Deserialized message should not be null"); + assertEquals(originalMessage.getId(), deserializedMessage.getId(), "IDs should match"); + assertEquals( + originalMessage.getContent(), deserializedMessage.getContent(), "Content should match"); + } + + @Test + void testToString() { + // Arrange + var message = new Message("Test content"); + + // Act + var result = message.toString(); + + // Assert + assertNotNull(result, "ToString should not return null"); + assertTrue(result.contains("Message{"), "ToString should contain class name"); + assertTrue(result.contains("Test content"), "ToString should contain content"); + assertTrue(result.contains(message.getId()), "ToString should contain ID"); + } + + @Test + void testMessageWithEmptyContent() { + // Arrange & Act + var message = new Message(""); + + // Assert + assertNotNull(message.getId(), "ID should be generated even for empty content"); + assertEquals("", message.getContent(), "Empty content should be preserved"); + } + + @Test + void testMessageWithNullContent() { + // Arrange & Act + var message = new Message(null); + + // Assert + assertNotNull(message.getId(), "ID should be generated even for null content"); + assertEquals(null, message.getContent(), "Null content should be preserved"); + } + + @Test + void testMessageWithSpecialCharacters() { + // Arrange + var specialContent = "Test with special chars: @#$%^&*()"; + + // Act + var message = new Message(specialContent); + + // Assert + assertEquals(specialContent, message.getContent(), "Special characters should be preserved"); + } +} diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/NotificationServiceTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/NotificationServiceTest.java new file mode 100644 index 000000000000..142e59087b5c --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/NotificationServiceTest.java @@ -0,0 +1,134 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link NotificationService}. Tests service behavior with various message types + * without Kafka dependencies. + */ +class NotificationServiceTest { + + private NotificationService notificationService; + + @BeforeEach + void setUp() { + notificationService = new NotificationService(); + } + + @Test + void testServiceCanBeInstantiated() { + // Arrange & Act & Assert + assertNotNull(notificationService, "NotificationService should be instantiated"); + } + + @Test + void testHandleOrderCreatedMessage() { + // Arrange + var message = new Message("Order Created: ORDER-001"); + + // Act & Assert + assertDoesNotThrow( + () -> notificationService.handleMessage(message), + "Should handle order created message without error"); + } + + @Test + void testHandleOrderUpdatedMessage() { + // Arrange + var message = new Message("Order Updated: ORDER-001"); + + // Act & Assert + assertDoesNotThrow( + () -> notificationService.handleMessage(message), + "Should handle order updated message without error"); + } + + @Test + void testHandleOrderCancelledMessage() { + // Arrange + var message = new Message("Order Cancelled: ORDER-001"); + + // Act & Assert + assertDoesNotThrow( + () -> notificationService.handleMessage(message), + "Should handle order cancelled message without error"); + } + + @Test + void testHandleUnknownMessage() { + // Arrange + var message = new Message("Unknown Event: ORDER-001"); + + // Act & Assert + assertDoesNotThrow( + () -> notificationService.handleMessage(message), + "Should handle unknown message without error"); + } + + @Test + void testHandleAllMessageTypes() { + // Act & Assert + assertDoesNotThrow( + () -> { + notificationService.handleMessage(new Message("Order Created: ORDER-001")); + notificationService.handleMessage(new Message("Order Updated: ORDER-001")); + notificationService.handleMessage(new Message("Order Cancelled: ORDER-001")); + }, + "Should handle all message types without error"); + } + + @Test + void testHandleMultipleOrdersSequentially() { + // Act & Assert + assertDoesNotThrow( + () -> { + notificationService.handleMessage(new Message("Order Created: ORDER-001")); + notificationService.handleMessage(new Message("Order Created: ORDER-002")); + notificationService.handleMessage(new Message("Order Created: ORDER-003")); + }, + "Should handle multiple orders sequentially without error"); + } + + @Test + void testHandleMessagesWhenInterrupted() { + Thread.currentThread().interrupt(); + notificationService.handleMessage(new Message("Order Created: ORDER-001")); + org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted()); + + Thread.currentThread().interrupt(); + notificationService.handleMessage(new Message("Order Updated: ORDER-001")); + org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted()); + + Thread.currentThread().interrupt(); + notificationService.handleMessage(new Message("Order Cancelled: ORDER-001")); + org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted()); + } +} diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/OrderServiceTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/OrderServiceTest.java new file mode 100644 index 000000000000..2b3f38584a20 --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/OrderServiceTest.java @@ -0,0 +1,109 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; + +import org.apache.kafka.clients.producer.MockProducer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** Unit tests for {@link OrderService}. Tests follow Arrange-Act-Assert pattern. */ +class OrderServiceTest { + + private MockProducer mockKafkaProducer; + private KafkaMessageProducer messageProducer; + private OrderService orderService; + + @BeforeEach + void setUp() { + mockKafkaProducer = new MockProducer<>(true, new StringSerializer(), new StringSerializer()); + messageProducer = new KafkaMessageProducer(mockKafkaProducer); + orderService = new OrderService(messageProducer); + } + + @Test + void testCreateOrder() { + // Arrange + var orderId = "ORDER-001"; + + // Act + assertDoesNotThrow(() -> orderService.createOrder(orderId)); + + // Assert + assertEquals(1, mockKafkaProducer.history().size()); + } + + @Test + void testUpdateOrder() { + // Arrange + var orderId = "ORDER-002"; + + // Act + assertDoesNotThrow(() -> orderService.updateOrder(orderId)); + + // Assert + assertEquals(1, mockKafkaProducer.history().size()); + } + + @Test + void testCancelOrder() { + // Arrange + var orderId = "ORDER-003"; + + // Act + assertDoesNotThrow(() -> orderService.cancelOrder(orderId)); + + // Assert + assertEquals(1, mockKafkaProducer.history().size()); + } + + @Test + void testMultipleOrderOperations() { + // Arrange + var orderId = "ORDER-004"; + + // Act + orderService.createOrder(orderId); + orderService.updateOrder(orderId); + orderService.cancelOrder(orderId); + + // Assert + assertEquals(3, mockKafkaProducer.history().size()); + } + + @Test + void testCreateOrderWithDifferentIds() { + // Act + orderService.createOrder("ORDER-001"); + orderService.createOrder("ORDER-002"); + orderService.createOrder("ORDER-003"); + + // Assert + assertEquals(3, mockKafkaProducer.history().size()); + } +} diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/PaymentServiceTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/PaymentServiceTest.java new file mode 100644 index 000000000000..ca46dc5b8c98 --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/PaymentServiceTest.java @@ -0,0 +1,117 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link PaymentService}. Tests service behavior with various message types without + * Kafka dependencies. + */ +class PaymentServiceTest { + + private PaymentService paymentService; + + @BeforeEach + void setUp() { + paymentService = new PaymentService(); + } + + @Test + void testServiceCanBeInstantiated() { + // Arrange & Act & Assert + assertNotNull(paymentService, "PaymentService should be instantiated"); + } + + @Test + void testHandleOrderCreatedMessage() { + // Arrange + var message = new Message("Order Created: ORDER-001"); + + // Act & Assert + assertDoesNotThrow( + () -> paymentService.handleMessage(message), + "Should handle order created message without error"); + } + + @Test + void testHandleOrderCancelledMessage() { + // Arrange + var message = new Message("Order Cancelled: ORDER-001"); + + // Act & Assert + assertDoesNotThrow( + () -> paymentService.handleMessage(message), + "Should handle order cancelled message without error"); + } + + @Test + void testHandleOrderUpdatedMessage() { + // Arrange + var message = new Message("Order Updated: ORDER-001"); + + // Act & Assert + assertDoesNotThrow( + () -> paymentService.handleMessage(message), + "Should handle order updated message without error"); + } + + @Test + void testHandleUnknownMessage() { + // Arrange + var message = new Message("Unknown Event: ORDER-001"); + + // Act & Assert + assertDoesNotThrow( + () -> paymentService.handleMessage(message), + "Should handle unknown message without error"); + } + + @Test + void testHandleMultipleMessages() { + // Act & Assert + assertDoesNotThrow( + () -> { + paymentService.handleMessage(new Message("Order Created: ORDER-001")); + paymentService.handleMessage(new Message("Order Cancelled: ORDER-001")); + }, + "Should handle multiple messages without error"); + } + + @Test + void testHandleMessagesWhenInterrupted() { + Thread.currentThread().interrupt(); + paymentService.handleMessage(new Message("Order Created: ORDER-001")); + org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted()); + + Thread.currentThread().interrupt(); + paymentService.handleMessage(new Message("Order Cancelled: ORDER-001")); + org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted()); + } +} diff --git a/pom.xml b/pom.xml index b856945f3b1e..bdfd58066f3f 100644 --- a/pom.xml +++ b/pom.xml @@ -169,6 +169,7 @@ microservices-idempotent-consumer microservices-log-aggregation microservices-self-registration + microservices-messaging model-view-controller model-view-intent model-view-presenter @@ -444,6 +445,7 @@ src/test/resources/** src/main/resources/** checkstyle-suppressions.xml + **/*.ps1