Add Google k8n engine files and finish project...

Cqrs pattern

.idea/compiler.xml

@@ -11,8 +11,27 @@
         <module name="order-application-service" />
         <module name="common-domain" />
         <module name="order-core-domain" />
+        <module name="customer-container" />
+        <module name="customer-service-application" />
+        <module name="restaurant-core-domain" />
+        <module name="payment-domain-core" />
+        <module name="saga" />
+        <module name="restaurant-application-service" />
+        <module name="customer-messaging" />
+        <module name="order-application-service" />
+        <module name="kafka-producer" />
+        <module name="customer-data-access" />
         <module name="order-data-access" />
+        <module name="payment-dataaccess" />
+        <module name="common-data-access" />
+        <module name="outbox" />
         <module name="order-container" />
+        <module name="order-app" />
+        <module name="restaurant-dataaccess" />
+        <module name="payment-container" />
+        <module name="customer-application" />
+        <module name="customer-domain-core" />
+        <module name="restaurant-messaging" />
       </profile>
     </annotationProcessing>
     <bytecodeTargetLevel>
@@ -21,9 +40,26 @@
   </component>
   <component name="JavacSettings">
     <option name="ADDITIONAL_OPTIONS_OVERRIDE">
-      <module name="common" options="-parameters" />
+      <module name="clean-hexagonal-architecture-kafka-saga-outbox" options="" />
+      <module name="common" options="" />
+      <module name="common-application" options="-parameters" />
+      <module name="common-data-access" options="-parameters" />
       <module name="common-domain" options="-parameters" />
+      <module name="customer-application" options="-parameters" />
+      <module name="customer-container" options="-parameters" />
+      <module name="customer-data-access" options="-parameters" />
+      <module name="customer-domain" options="" />
+      <module name="customer-domain-core" options="-parameters" />
+      <module name="customer-messaging" options="-parameters" />
+      <module name="customer-service" options="" />
+      <module name="customer-service-application" options="-parameters" />
       <module name="food-ordering-system" options="" />
+      <module name="infrastructure" options="" />
+      <module name="kafka" options="" />
+      <module name="kafka-config" options="-parameters" />
+      <module name="kafka-consumer" options="-parameters" />
+      <module name="kafka-model" options="-parameters" />
+      <module name="kafka-producer" options="-parameters" />
       <module name="order-app" options="-parameters" />
       <module name="order-application-service" options="-parameters" />
       <module name="order-container" options="-parameters" />

.idea/encodings.xml

@@ -4,6 +4,26 @@
     <file url="file://$PROJECT_DIR$/common/common-domain/src/main/java" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/common/src/main/java" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/common/src/main/resources" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/customer-service/customer-application/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/customer-service/customer-container/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/customer-service/customer-data-access/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/customer-service/customer-domain/customer-domain-core/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/customer-service/customer-domain/customer-service-application/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/customer-service/customer-domain/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/customer-service/customer-domain/src/main/resources" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/customer-service/customer-messaging/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/customer-service/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/customer-service/src/main/resources" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/infrastructure/kafka/kafka-config/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/infrastructure/kafka/kafka-consumer/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/infrastructure/kafka/kafka-model/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/infrastructure/kafka/kafka-producer/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/infrastructure/kafka/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/infrastructure/kafka/src/main/resources" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/infrastructure/outbox/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/infrastructure/saga/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/infrastructure/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/infrastructure/src/main/resources" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/order-service/order-app/src/main/java" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/order-service/order-app/src/main/resources" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/order-service/order-container/src/main/java" charset="UTF-8" />

common/common-data-access/src/main/java/com/food/order/system/common/data/access/entity/RestaurantEntity.java

@@ -34,7 +34,7 @@ public class RestaurantEntity {
 
     private BigDecimal productPrice;
 
-    private Boolean productActive;
+    private Boolean productAvailable;
 
     @Override
     public boolean equals(Object o) {

customer-service/customer-application/pom.xml

@@ -0,0 +1,33 @@
+<?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>customer-service</artifactId>
+        <groupId>com.food.order</groupId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>customer-application</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>customer-service-application</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>common-application</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>

customer-service/customer-application/src/main/java/com/food/order/system/customer/rest/application/handler/CustomerGlobalExceptionHandler.java

@@ -0,0 +1,26 @@
+package com.food.order.system.customer.rest.application.handler;
+
+import com.food.order.system.application.handler.ErrorDTO;
+import com.food.order.system.application.handler.GlobalExceptionHandler;
+import com.food.order.system.customer.domain.exception.CustomerDomainException;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.bind.annotation.ControllerAdvice;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+import org.springframework.web.bind.annotation.ResponseBody;
+import org.springframework.web.bind.annotation.ResponseStatus;
+
+@Slf4j
+@ControllerAdvice
+public class CustomerGlobalExceptionHandler extends GlobalExceptionHandler {
+
+    @ResponseBody
+    @ExceptionHandler(value = {CustomerDomainException.class})
+    @ResponseStatus(HttpStatus.BAD_REQUEST)
+    public ErrorDTO handleException(CustomerDomainException exception) {
+        log.error(exception.getMessage(), exception);
+        return ErrorDTO.builder().code(HttpStatus.BAD_REQUEST.getReasonPhrase())
+                .message(exception.getMessage()).build();
+    }
+
+}

customer-service/customer-application/src/main/java/com/food/order/system/customer/rest/application/rest/CustomerController.java

@@ -0,0 +1,31 @@
+package com.food.order.system.customer.rest.application.rest;
+
+import com.food.order.system.customer.service.create.CreateCustomerCommand;
+import com.food.order.system.customer.service.create.CreateCustomerResponse;
+import com.food.order.system.customer.service.ports.input.service.CustomerApplicationService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@Slf4j
+@RequiredArgsConstructor
+@RestController
+@RequestMapping(value = "/customers", produces = "application/vnd.api.v1+json")
+public class CustomerController {
+
+    private final CustomerApplicationService customerApplicationService;
+
+
+    @PostMapping
+    public ResponseEntity<CreateCustomerResponse> createCustomer(@RequestBody CreateCustomerCommand
+                                                                             createCustomerCommand) {
+        log.info("Creating customer with username: {}", createCustomerCommand.username());
+        CreateCustomerResponse response = customerApplicationService.createCustomer(createCustomerCommand);
+        return ResponseEntity.ok(response);
+    }
+
+}

customer-service/customer-container/pom.xml

@@ -0,0 +1,63 @@
+<?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>customer-service</artifactId>
+        <groupId>com.food.order</groupId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>customer-container</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>customer-service-application</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>customer-domain-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>customer-data-access</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>customer-messaging</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>customer-application</artifactId>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+                <configuration>
+                    <image>
+                        <name>${project.groupId}/customer.service:${project.version}</name>
+                    </image>
+                </configuration>
+                <executions>
+                    <execution>
+                        <phase>install</phase>
+                        <goals>
+                            <goal>build-image</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

customer-service/customer-container/src/main/java/com/food/order/system/customer/service/BeanConfiguration.java

@@ -0,0 +1,15 @@
+package com.food.order.system.customer.service;
+
+import com.food.order.system.customer.domain.CustomerDomainService;
+import com.food.order.system.customer.domain.CustomerDomainServiceImpl;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class BeanConfiguration {
+
+    @Bean
+    public CustomerDomainService customerDomainService() {
+        return new CustomerDomainServiceImpl();
+    }
+}

customer-service/customer-container/src/main/java/com/food/order/system/customer/service/CustomerServiceApplication.java

@@ -0,0 +1,19 @@
+package com.food.order.system.customer.service;
+
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.domain.EntityScan;
+import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
+
+@EnableJpaRepositories(basePackages = {"com.food.order.system.customer.dataaccess",
+        "com.food.order.system.common.data.access"})
+@EntityScan(basePackages = {"com.food.order.system.customer.dataaccess",
+        "com.food.order.system.common.data.access"})
+@SpringBootApplication(scanBasePackages = "com.food.order")
+public class CustomerServiceApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(CustomerServiceApplication.class, args);
+    }
+}

customer-service/customer-container/src/main/resources/application.yml

@@ -0,0 +1,41 @@
+server:
+  port: 8184
+
+customer-service:
+  customer-topic-name: customer
+
+
+spring:
+  jpa:
+    open-in-view: false
+    show-sql: false
+    database-platform: org.hibernate.dialect.PostgreSQL9Dialect
+    properties:
+      hibernate:
+        dialect: org.hibernate.dialect.PostgreSQL9Dialect
+  datasource:
+    url: jdbc:postgresql://localhost:5432/postgres?currentSchema=customer&binaryTransfer=true&reWriteBatchedInserts=true
+    username: postgres
+    password: admin
+    driver-class-name: org.postgresql.Driver
+    platform: postgres
+    schema: classpath:init-schema.sql
+    initialization-mode: always
+
+kafka-config:
+  bootstrap-servers: localhost:19092, localhost:29092, localhost:39092
+  schema-registry-url-key: schema.registry.url
+  schema-registry-url: http://localhost:8081
+  num-of-partitions: 3
+  replication-factor: 3
+
+kafka-producer-config:
+  key-serializer-class: org.apache.kafka.common.serialization.StringSerializer
+  value-serializer-class: io.confluent.kafka.serializers.KafkaAvroSerializer
+  compression-type: snappy
+  acks: all
+  batch-size: 16384
+  batch-size-boost-factor: 100
+  linger-ms: 5
+  request-timeout-ms: 60000
+  retry-count: 5

customer-service/customer-data-access/pom.xml

@@ -0,0 +1,33 @@
+<?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>customer-service</artifactId>
+        <groupId>com.food.order</groupId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>customer-data-access</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>customer-service-application</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-jpa</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.postgresql</groupId>
+            <artifactId>postgresql</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>common-data-access</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>

customer-service/customer-data-access/src/main/java/com/food/order/system/customer/dataaccess/adapter/CustomerRepositoryImpl.java

@@ -0,0 +1,27 @@
+package com.food.order.system.customer.dataaccess.adapter;
+
+import com.food.order.system.customer.dataaccess.mapper.CustomerDataAccessMapper;
+import com.food.order.system.customer.dataaccess.repository.CustomerJpaRepository;
+import com.food.order.system.customer.domain.entity.Customer;
+import com.food.order.system.customer.service.ports.output.repository.CustomerRepository;
+import org.springframework.stereotype.Component;
+
+@Component
+public class CustomerRepositoryImpl implements CustomerRepository {
+
+    private final CustomerJpaRepository customerJpaRepository;
+
+    private final CustomerDataAccessMapper customerDataAccessMapper;
+
+    public CustomerRepositoryImpl(CustomerJpaRepository customerJpaRepository,
+                                  CustomerDataAccessMapper customerDataAccessMapper) {
+        this.customerJpaRepository = customerJpaRepository;
+        this.customerDataAccessMapper = customerDataAccessMapper;
+    }
+
+    @Override
+    public Customer createCustomer(Customer customer) {
+        return customerDataAccessMapper.customerEntityToCustomer(
+                customerJpaRepository.save(customerDataAccessMapper.customerToCustomerEntity(customer)));
+    }
+}

customer-service/customer-data-access/src/main/java/com/food/order/system/customer/dataaccess/entity/CustomerEntity.java

@@ -0,0 +1,24 @@
+package com.food.order.system.customer.dataaccess.entity;
+
+import lombok.*;
+
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.Table;
+import java.util.UUID;
+
+@Getter
+@Setter
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@Table(name = "customers")
+@Entity
+public class CustomerEntity {
+
+    @Id
+    private UUID id;
+    private String username;
+    private String firstName;
+    private String lastName;
+}

customer-service/customer-data-access/src/main/java/com/food/order/system/customer/dataaccess/exception/CustomerDataaccessException.java

@@ -0,0 +1,8 @@
+package com.food.order.system.customer.dataaccess.exception;
+
+public class CustomerDataaccessException extends RuntimeException {
+
+    public CustomerDataaccessException(String message) {
+        super(message);
+    }
+}

customer-service/customer-data-access/src/main/java/com/food/order/system/customer/dataaccess/mapper/CustomerDataAccessMapper.java

@@ -0,0 +1,27 @@
+package com.food.order.system.customer.dataaccess.mapper;
+
+import com.food.order.system.customer.dataaccess.entity.CustomerEntity;
+import com.food.order.system.customer.domain.entity.Customer;
+import com.food.order.system.valueobject.CustomerId;
+import org.springframework.stereotype.Component;
+
+@Component
+public class CustomerDataAccessMapper {
+
+    public Customer customerEntityToCustomer(CustomerEntity customerEntity) {
+        return new Customer(new CustomerId(customerEntity.getId()),
+                customerEntity.getUsername(),
+                customerEntity.getFirstName(),
+                customerEntity.getLastName());
+    }
+
+    public CustomerEntity customerToCustomerEntity(Customer customer) {
+        return CustomerEntity.builder()
+                .id(customer.getId().getValue())
+                .username(customer.getUsername())
+                .firstName(customer.getFirstName())
+                .lastName(customer.getLastName())
+                .build();
+    }
+
+}

customer-service/customer-data-access/src/main/java/com/food/order/system/customer/dataaccess/repository/CustomerJpaRepository.java

@@ -0,0 +1,13 @@
+package com.food.order.system.customer.dataaccess.repository;
+
+import com.food.order.system.customer.dataaccess.entity.CustomerEntity;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+import java.util.UUID;
+
+@Repository
+public interface CustomerJpaRepository extends JpaRepository<CustomerEntity, UUID> {
+
+
+}

customer-service/customer-domain/customer-domain-core/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>customer-domain</artifactId>
+        <groupId>com.food.order</groupId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>customer-domain-core</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>common-domain</artifactId>
+        </dependency>
+    </dependencies>
+
+
+</project>

customer-service/customer-domain/customer-domain-core/src/main/java/com/food/order/system/customer/domain/CustomerDomainService.java

@@ -0,0 +1,11 @@
+package com.food.order.system.customer.domain;
+
+
+import com.food.order.system.customer.domain.entity.Customer;
+import com.food.order.system.customer.domain.event.CustomerCreatedEvent;
+
+public interface CustomerDomainService {
+
+    CustomerCreatedEvent validateAndInitiateCustomer(Customer customer);
+
+}

customer-service/customer-domain/customer-domain-core/src/main/java/com/food/order/system/customer/domain/CustomerDomainServiceImpl.java

@@ -0,0 +1,19 @@
+package com.food.order.system.customer.domain;
+
+import com.food.order.system.customer.domain.entity.Customer;
+import com.food.order.system.customer.domain.event.CustomerCreatedEvent;
+import lombok.extern.slf4j.Slf4j;
+
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+
+@Slf4j
+public class CustomerDomainServiceImpl implements CustomerDomainService {
+
+    public CustomerCreatedEvent validateAndInitiateCustomer(Customer customer) {
+        //Any Business logic required to run for a customer creation
+        log.info("Customer with id: {} is initiated", customer.getId().getValue());
+        return new CustomerCreatedEvent(customer, ZonedDateTime.now(ZoneId.of("UTC")));
+    }
+}
+

customer-service/customer-domain/customer-domain-core/src/main/java/com/food/order/system/customer/domain/entity/Customer.java

@@ -0,0 +1,32 @@
+package com.food.order.system.customer.domain.entity;
+
+
+import com.food.order.system.entity.AggregateRoot;
+import com.food.order.system.valueobject.CustomerId;
+
+public class Customer extends AggregateRoot<CustomerId> {
+    private final String username;
+    private final String firstName;
+    private final String lastName;
+
+    public Customer(CustomerId customerId, String username, String firstName, String lastName) {
+        super.setId(customerId);
+        this.username = username;
+        this.firstName = firstName;
+        this.lastName = lastName;
+    }
+
+    public String getUsername() {
+        return username;
+    }
+
+    public String getFirstName() {
+        return firstName;
+    }
+
+    public String getLastName() {
+        return lastName;
+    }
+}
+
+

customer-service/customer-domain/customer-domain-core/src/main/java/com/food/order/system/customer/domain/event/CustomerCreatedEvent.java

@@ -0,0 +1,23 @@
+package com.food.order.system.customer.domain.event;
+
+
+import com.food.order.system.customer.domain.entity.Customer;
+import com.food.order.system.event.DomainEvent;
+
+import java.time.ZonedDateTime;
+
+public class CustomerCreatedEvent implements DomainEvent<Customer> {
+
+    private final Customer customer;
+
+    private final ZonedDateTime createdAt;
+
+    public CustomerCreatedEvent(Customer customer, ZonedDateTime createdAt) {
+        this.customer = customer;
+        this.createdAt = createdAt;
+    }
+
+    public Customer getCustomer() {
+        return customer;
+    }
+}

customer-service/customer-domain/customer-domain-core/src/main/java/com/food/order/system/customer/domain/exception/CustomerDomainException.java

@@ -0,0 +1,11 @@
+package com.food.order.system.customer.domain.exception;
+
+
+import com.food.order.system.exception.DomainException;
+
+public class CustomerDomainException extends DomainException {
+
+    public CustomerDomainException(String message) {
+        super(message);
+    }
+}

customer-service/customer-domain/customer-service-application/pom.xml

@@ -0,0 +1,35 @@
+<?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>customer-domain</artifactId>
+        <groupId>com.food.order</groupId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>customer-service-application</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>customer-domain-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>common-domain</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-tx</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>

customer-service/customer-domain/customer-service-application/src/main/java/com/food/order/system/customer/service/CustomerApplicationServiceImpl.java

@@ -0,0 +1,36 @@
+package com.food.order.system.customer.service;
+
+
+import com.food.order.system.customer.domain.event.CustomerCreatedEvent;
+import com.food.order.system.customer.service.create.CreateCustomerCommand;
+import com.food.order.system.customer.service.create.CreateCustomerResponse;
+import com.food.order.system.customer.service.mapper.CustomerDataMapper;
+import com.food.order.system.customer.service.ports.input.service.CustomerApplicationService;
+import com.food.order.system.customer.service.ports.output.message.publisher.CustomerMessagePublisher;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.validation.annotation.Validated;
+
+@Slf4j
+@Validated
+@Service
+@RequiredArgsConstructor
+class CustomerApplicationServiceImpl implements CustomerApplicationService {
+
+    private final CustomerCreateCommandHandler customerCreateCommandHandler;
+
+    private final CustomerDataMapper customerDataMapper;
+
+    private final CustomerMessagePublisher customerMessagePublisher;
+
+
+    @Override
+    public CreateCustomerResponse createCustomer(CreateCustomerCommand createCustomerCommand) {
+        CustomerCreatedEvent customerCreatedEvent = customerCreateCommandHandler.createCustomer(createCustomerCommand);
+        customerMessagePublisher.publish(customerCreatedEvent);
+        return customerDataMapper
+                .customerToCreateCustomerResponse(customerCreatedEvent.getCustomer(),
+                        "Customer saved successfully!");
+    }
+}

customer-service/customer-domain/customer-service-application/src/main/java/com/food/order/system/customer/service/CustomerCreateCommandHandler.java

@@ -0,0 +1,42 @@
+package com.food.order.system.customer.service;
+
+
+import com.food.order.system.customer.domain.CustomerDomainService;
+import com.food.order.system.customer.domain.entity.Customer;
+import com.food.order.system.customer.domain.event.CustomerCreatedEvent;
+import com.food.order.system.customer.domain.exception.CustomerDomainException;
+import com.food.order.system.customer.service.create.CreateCustomerCommand;
+import com.food.order.system.customer.service.mapper.CustomerDataMapper;
+import com.food.order.system.customer.service.ports.output.repository.CustomerRepository;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+class CustomerCreateCommandHandler {
+
+    private final CustomerDomainService customerDomainService;
+
+    private final CustomerRepository customerRepository;
+
+    private final CustomerDataMapper customerDataMapper;
+
+
+
+    @Transactional
+    public CustomerCreatedEvent createCustomer(CreateCustomerCommand createCustomerCommand) {
+        Customer customer = customerDataMapper.createCustomerCommandToCustomer(createCustomerCommand);
+        CustomerCreatedEvent customerCreatedEvent = customerDomainService.validateAndInitiateCustomer(customer);
+        Customer savedCustomer = customerRepository.createCustomer(customer);
+        if (savedCustomer == null) {
+            log.error("Could not save customer with id: {}", createCustomerCommand.customerId());
+            throw new CustomerDomainException("Could not save customer with id " +
+                    createCustomerCommand.customerId());
+        }
+        log.info("Returning CustomerCreatedEvent for customer id: {}", createCustomerCommand.customerId());
+        return customerCreatedEvent;
+    }
+}

customer-service/customer-domain/customer-service-application/src/main/java/com/food/order/system/customer/service/config/CustomerServiceConfigData.java

@@ -0,0 +1,12 @@
+package com.food.order.system.customer.service.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+@Data
+@Configuration
+@ConfigurationProperties(prefix = "customer-service")
+public class CustomerServiceConfigData {
+    private String customerTopicName;
+}

customer-service/customer-domain/customer-service-application/src/main/java/com/food/order/system/customer/service/create/CreateCustomerCommand.java

@@ -0,0 +1,11 @@
+package com.food.order.system.customer.service.create;
+
+import lombok.Builder;
+
+import javax.validation.constraints.NotNull;
+import java.util.UUID;
+
+@Builder
+public record CreateCustomerCommand(@NotNull UUID customerId, @NotNull String username, @NotNull String firstName,
+                                    @NotNull String lastName) {
+}

customer-service/customer-domain/customer-service-application/src/main/java/com/food/order/system/customer/service/create/CreateCustomerResponse.java

@@ -0,0 +1,10 @@
+package com.food.order.system.customer.service.create;
+
+import lombok.Builder;
+
+import javax.validation.constraints.NotNull;
+import java.util.UUID;
+
+@Builder
+public record CreateCustomerResponse(@NotNull UUID customerId, @NotNull String message) {
+}

customer-service/customer-domain/customer-service-application/src/main/java/com/food/order/system/customer/service/mapper/CustomerDataMapper.java

@@ -0,0 +1,22 @@
+package com.food.order.system.customer.service.mapper;
+
+import com.food.order.system.customer.domain.entity.Customer;
+import com.food.order.system.customer.service.create.CreateCustomerCommand;
+import com.food.order.system.customer.service.create.CreateCustomerResponse;
+import com.food.order.system.valueobject.CustomerId;
+import org.springframework.stereotype.Component;
+
+@Component
+public class CustomerDataMapper {
+
+    public Customer createCustomerCommandToCustomer(CreateCustomerCommand createCustomerCommand) {
+        return new Customer(new CustomerId(createCustomerCommand.customerId()),
+                createCustomerCommand.username(),
+                createCustomerCommand.firstName(),
+                createCustomerCommand.lastName());
+    }
+
+    public CreateCustomerResponse customerToCreateCustomerResponse(Customer customer, String message) {
+        return new CreateCustomerResponse(customer.getId().getValue(), message);
+    }
+}

customer-service/customer-domain/customer-service-application/src/main/java/com/food/order/system/customer/service/ports/input/service/CustomerApplicationService.java

@@ -0,0 +1,13 @@
+package com.food.order.system.customer.service.ports.input.service;
+
+
+import com.food.order.system.customer.service.create.CreateCustomerCommand;
+import com.food.order.system.customer.service.create.CreateCustomerResponse;
+
+import javax.validation.Valid;
+
+public interface CustomerApplicationService {
+
+    CreateCustomerResponse createCustomer(@Valid CreateCustomerCommand createCustomerCommand);
+
+}

customer-service/customer-domain/customer-service-application/src/main/java/com/food/order/system/customer/service/ports/output/message/publisher/CustomerMessagePublisher.java

@@ -0,0 +1,10 @@
+package com.food.order.system.customer.service.ports.output.message.publisher;
+
+
+import com.food.order.system.customer.domain.event.CustomerCreatedEvent;
+
+public interface CustomerMessagePublisher {
+
+    void publish(CustomerCreatedEvent customerCreatedEvent);
+
+}

customer-service/customer-domain/customer-service-application/src/main/java/com/food/order/system/customer/service/ports/output/repository/CustomerRepository.java

@@ -0,0 +1,9 @@
+package com.food.order.system.customer.service.ports.output.repository;
+
+
+import com.food.order.system.customer.domain.entity.Customer;
+
+public interface CustomerRepository {
+
+    Customer createCustomer(Customer customer);
+}

customer-service/customer-domain/pom.xml

@@ -0,0 +1,20 @@
+<?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>customer-service</artifactId>
+        <groupId>com.food.order</groupId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>customer-domain</artifactId>
+    <packaging>pom</packaging>
+    <modules>
+        <module>customer-domain-core</module>
+        <module>customer-service-application</module>
+    </modules>
+
+
+</project>

customer-service/customer-messaging/pom.xml

@@ -0,0 +1,30 @@
+<?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>customer-service</artifactId>
+        <groupId>com.food.order</groupId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>customer-messaging</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>customer-service-application</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>kafka-producer</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>kafka-model</artifactId>
+        </dependency>
+
+    </dependencies>
+
+</project>

customer-service/customer-messaging/src/main/java/com/food/order/system/customer/messaging/mapper/CustomerMessagingDataMapper.java

@@ -0,0 +1,19 @@
+package com.food.order.system.customer.messaging.mapper;
+
+import com.food.order.system.customer.domain.event.CustomerCreatedEvent;
+import com.food.order.system.kafka.order.avro.model.CustomerAvroModel;
+import org.springframework.stereotype.Component;
+
+@Component
+public class CustomerMessagingDataMapper {
+
+    public CustomerAvroModel paymentResponseAvroModelToPaymentResponse(CustomerCreatedEvent
+                                                                               customerCreatedEvent) {
+        return CustomerAvroModel.newBuilder()
+                .setId(customerCreatedEvent.getCustomer().getId().getValue().toString())
+                .setUsername(customerCreatedEvent.getCustomer().getUsername())
+                .setFirstName(customerCreatedEvent.getCustomer().getFirstName())
+                .setLastName(customerCreatedEvent.getCustomer().getLastName())
+                .build();
+    }
+}

customer-service/customer-messaging/src/main/java/com/food/order/system/customer/messaging/publisher/kafka/CustomerCreatedEventKafkaPublisher.java

@@ -0,0 +1,68 @@
+package com.food.order.system.customer.messaging.publisher.kafka;
+
+
+import com.food.order.system.customer.domain.event.CustomerCreatedEvent;
+import com.food.order.system.customer.messaging.mapper.CustomerMessagingDataMapper;
+import com.food.order.system.customer.service.config.CustomerServiceConfigData;
+import com.food.order.system.customer.service.ports.output.message.publisher.CustomerMessagePublisher;
+import com.food.order.system.kafka.order.avro.model.CustomerAvroModel;
+import com.food.order.system.kafka.producer.service.KafkaProducer;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.stereotype.Component;
+import org.springframework.util.concurrent.ListenableFutureCallback;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class CustomerCreatedEventKafkaPublisher implements CustomerMessagePublisher {
+
+    private final CustomerMessagingDataMapper customerMessagingDataMapper;
+
+    private final KafkaProducer<String, CustomerAvroModel> kafkaProducer;
+
+    private final CustomerServiceConfigData customerServiceConfigData;
+
+    @Override
+    public void publish(CustomerCreatedEvent customerCreatedEvent) {
+        log.info("Received CustomerCreatedEvent for customer id: {}",
+                customerCreatedEvent.getCustomer().getId().getValue());
+        try {
+            CustomerAvroModel customerAvroModel = customerMessagingDataMapper
+                    .paymentResponseAvroModelToPaymentResponse(customerCreatedEvent);
+
+            kafkaProducer.send(customerServiceConfigData.getCustomerTopicName(), customerAvroModel.getId(),
+                    customerAvroModel,
+                    getCallback(customerServiceConfigData.getCustomerTopicName(), customerAvroModel));
+
+            log.info("CustomerCreatedEvent sent to kafka for customer id: {}",
+                    customerAvroModel.getId());
+        } catch (Exception e) {
+            log.error("Error while sending CustomerCreatedEvent to kafka for customer id: {}," +
+                    " error: {}", customerCreatedEvent.getCustomer().getId().getValue(), e.getMessage());
+        }
+    }
+
+    private ListenableFutureCallback<SendResult<String, CustomerAvroModel>>
+    getCallback(String topicName, CustomerAvroModel message) {
+        return new ListenableFutureCallback<>() {
+            @Override
+            public void onFailure(Throwable throwable) {
+                log.error("Error while sending message {} to topic {}", message.toString(), topicName, throwable);
+            }
+
+            @Override
+            public void onSuccess(SendResult<String, CustomerAvroModel> result) {
+                RecordMetadata metadata = result.getRecordMetadata();
+                log.info("Received new metadata. Topic: {}; Partition {}; Offset {}; Timestamp {}, at time {}",
+                        metadata.topic(),
+                        metadata.partition(),
+                        metadata.offset(),
+                        metadata.timestamp(),
+                        System.nanoTime());
+            }
+        };
+    }
+}

customer-service/pom.xml

@@ -10,46 +10,15 @@
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>customer-service</artifactId>
+    <packaging>pom</packaging>
+    <modules>
+        <module>customer-container</module>
+        <module>customer-application</module>
+        <module>customer-data-access</module>
+        <module>customer-messaging</module>
+        <module>customer-domain</module>
+    </modules>
 
-    <dependencies>
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter</artifactId>
-        </dependency>
-
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-data-jpa</artifactId>
-        </dependency>
-
-        <dependency>
-            <groupId>org.postgresql</groupId>
-            <artifactId>postgresql</artifactId>
-        </dependency>
-    </dependencies>
-
-
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>org.springframework.boot</groupId>
-                <artifactId>spring-boot-maven-plugin</artifactId>
-                <configuration>
-                    <image>
-                        <name>${project.groupId}/customer.service:${project.version}</name>
-                    </image>
-                </configuration>
-                <executions>
-                    <execution>
-                        <phase>install</phase>
-                        <goals>
-                            <goal>build-image</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
-        </plugins>
-    </build>
 
 
 </project>

customer-service/src/main/java/com/food/order/system/customer/service/CustomerServiceApplication.java

@@ -1,13 +0,0 @@
-package com.food.order.system.customer.service;
-
-
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-
-@SpringBootApplication(scanBasePackages = "com.food.order")
-public class CustomerServiceApplication {
-
-    public static void main(String[] args) {
-        SpringApplication.run(CustomerServiceApplication.class, args);
-    }
-}

customer-service/src/main/resources/application.yml

@@ -1,22 +0,0 @@
-server:
-  port: 8184
-
-spring:
-  jpa:
-    open-in-view: false
-    show-sql: false
-    database-platform: org.hibernate.dialect.PostgreSQL9Dialect
-    properties:
-      hibernate:
-        dialect: org.hibernate.dialect.PostgreSQL9Dialect
-  datasource:
-    url: jdbc:postgresql://localhost:5432/postgres?currentSchema=customer&binaryTransfer=true&reWriteBatchedInserts=true
-    username: postgres
-    password: postgres
-    driver-class-name: org.postgresql.Driver
-  sql:
-    init:
-      mode: always
-      schema-locations: classpath:init-schema.sql
-      data-locations: classpath:init-data.sql
-      platform: postgres

customer-service/src/main/resources/init-data.sql

@@ -1,5 +0,0 @@
-INSERT INTO customer.customers(id, username, first_name, last_name)
-	VALUES ('d215b5f8-0249-4dc5-89a3-51fd148cfb41', 'user_1', 'First', 'User');
-
-INSERT INTO customer.customers(id, username, first_name, last_name)
-	VALUES ('d215b5f8-0249-4dc5-89a3-51fd148cfb42', 'user_2', 'Second', 'User');

infrastructure/docker-compose/init_kafka.yml

@@ -13,12 +13,14 @@ services:
       kafka-topics --bootstrap-server kafka-broker-1:9092 --topic payment-response-value --delete --if-exists
       kafka-topics --bootstrap-server kafka-broker-1:9092 --topic restaurant-approval-request-value --delete --if-exists
       kafka-topics --bootstrap-server kafka-broker-1:9092 --topic restaurant-approval-response-value --delete --if-exists
+      kafka-topics --bootstrap-server kafka-broker-1:9092 --topic customer-value --delete --if-exists
 
       echo -e 'Creating kafka topics'
       kafka-topics --bootstrap-server kafka-broker-1:9092 --create --if-not-exists --topic payment-request-value --replication-factor 3 --partitions 3
       kafka-topics --bootstrap-server kafka-broker-1:9092 --create --if-not-exists --topic payment-response-value --replication-factor 3 --partitions 3
       kafka-topics --bootstrap-server kafka-broker-1:9092 --create --if-not-exists --topic restaurant-approval-request-value --replication-factor 3 --partitions 3
       kafka-topics --bootstrap-server kafka-broker-1:9092 --create --if-not-exists --topic restaurant-approval-response-value --replication-factor 3 --partitions 3
+      kafka-topics --bootstrap-server kafka-broker-1:9092 --create --if-not-exists --topic customer-value --replication-factor 3 --partitions 3
 
 
       echo -e 'Successfully created the following topics:'

infrastructure/docker-compose/kafka_cluster.yml

@@ -13,7 +13,7 @@ services:
       SCHEMA_REGISTRY_HOST_NAME: schema-registry
       SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181'
       SCHEMA_REGISTRY_LISTENERS: http://schema-registry:8081
-      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka-broker-2:9092,LISTENER_LOCAL://localhost:29092
+      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka-broker-1:9092,LISTENER_LOCAL://localhost:19092,PLAINTEXT://kafka-broker-2:9092,LISTENER_LOCAL://localhost:29092,PLAINTEXT://kafka-broker-3:9092,LISTENER_LOCAL://localhost:39092
       SCHEMA_REGISTRY_DEBUG: 'true'
     networks:
       - ${GLOBAL_NETWORK:-kafka}

infrastructure/kafka/kafka-model/src/main/java/com/food/order/system/kafka/order/avro/model/CustomerAvroModel.java

@@ -0,0 +1,512 @@
+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.food.order.system.kafka.order.avro.model;
+
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.SchemaStore;
+import org.apache.avro.specific.SpecificData;
+
+@org.apache.avro.specific.AvroGenerated
+public class CustomerAvroModel extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = 7953499667909084838L;
+
+
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"CustomerAvroModel\",\"namespace\":\"com.food.order.system.kafka.order.avro.model\",\"fields\":[{\"name\":\"id\",\"type\":{\"type\":\"string\",\"logicalType\":\"uuid\"}},{\"name\":\"username\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}},{\"name\":\"firstName\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}},{\"name\":\"lastName\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static final SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<CustomerAvroModel> ENCODER =
+      new BinaryMessageEncoder<CustomerAvroModel>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<CustomerAvroModel> DECODER =
+      new BinaryMessageDecoder<CustomerAvroModel>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageEncoder instance used by this class.
+   * @return the message encoder used by this class
+   */
+  public static BinaryMessageEncoder<CustomerAvroModel> getEncoder() {
+    return ENCODER;
+  }
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   * @return the message decoder used by this class
+   */
+  public static BinaryMessageDecoder<CustomerAvroModel> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   * @return a BinaryMessageDecoder instance for this class backed by the given SchemaStore
+   */
+  public static BinaryMessageDecoder<CustomerAvroModel> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<CustomerAvroModel>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /**
+   * Serializes this CustomerAvroModel to a ByteBuffer.
+   * @return a buffer holding the serialized data for this instance
+   * @throws java.io.IOException if this instance could not be serialized
+   */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /**
+   * Deserializes a CustomerAvroModel from a ByteBuffer.
+   * @param b a byte buffer holding serialized data for an instance of this class
+   * @return a CustomerAvroModel instance decoded from the given buffer
+   * @throws java.io.IOException if the given bytes could not be deserialized into an instance of this class
+   */
+  public static CustomerAvroModel fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  private java.lang.String id;
+  private java.lang.String username;
+  private java.lang.String firstName;
+  private java.lang.String lastName;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public CustomerAvroModel() {}
+
+  /**
+   * All-args constructor.
+   * @param id The new value for id
+   * @param username The new value for username
+   * @param firstName The new value for firstName
+   * @param lastName The new value for lastName
+   */
+  public CustomerAvroModel(java.lang.String id, java.lang.String username, java.lang.String firstName, java.lang.String lastName) {
+    this.id = id;
+    this.username = username;
+    this.firstName = firstName;
+    this.lastName = lastName;
+  }
+
+  public org.apache.avro.specific.SpecificData getSpecificData() { return MODEL$; }
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return id;
+    case 1: return username;
+    case 2: return firstName;
+    case 3: return lastName;
+    default: throw new IndexOutOfBoundsException("Invalid index: " + field$);
+    }
+  }
+
+  private static final org.apache.avro.Conversion<?>[] conversions =
+      new org.apache.avro.Conversion<?>[] {
+      null,
+      null,
+      null,
+      null,
+      null
+  };
+
+  @Override
+  public org.apache.avro.Conversion<?> getConversion(int field) {
+    return conversions[field];
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: id = value$ != null ? value$.toString() : null; break;
+    case 1: username = value$ != null ? value$.toString() : null; break;
+    case 2: firstName = value$ != null ? value$.toString() : null; break;
+    case 3: lastName = value$ != null ? value$.toString() : null; break;
+    default: throw new IndexOutOfBoundsException("Invalid index: " + field$);
+    }
+  }
+
+  /**
+   * Gets the value of the 'id' field.
+   * @return The value of the 'id' field.
+   */
+  public java.lang.String getId() {
+    return id;
+  }
+
+
+  /**
+   * Sets the value of the 'id' field.
+   * @param value the value to set.
+   */
+  public void setId(java.lang.String value) {
+    this.id = value;
+  }
+
+  /**
+   * Gets the value of the 'username' field.
+   * @return The value of the 'username' field.
+   */
+  public java.lang.String getUsername() {
+    return username;
+  }
+
+
+  /**
+   * Sets the value of the 'username' field.
+   * @param value the value to set.
+   */
+  public void setUsername(java.lang.String value) {
+    this.username = value;
+  }
+
+  /**
+   * Gets the value of the 'firstName' field.
+   * @return The value of the 'firstName' field.
+   */
+  public java.lang.String getFirstName() {
+    return firstName;
+  }
+
+
+  /**
+   * Sets the value of the 'firstName' field.
+   * @param value the value to set.
+   */
+  public void setFirstName(java.lang.String value) {
+    this.firstName = value;
+  }
+
+  /**
+   * Gets the value of the 'lastName' field.
+   * @return The value of the 'lastName' field.
+   */
+  public java.lang.String getLastName() {
+    return lastName;
+  }
+
+
+  /**
+   * Sets the value of the 'lastName' field.
+   * @param value the value to set.
+   */
+  public void setLastName(java.lang.String value) {
+    this.lastName = value;
+  }
+
+  /**
+   * Creates a new CustomerAvroModel RecordBuilder.
+   * @return A new CustomerAvroModel RecordBuilder
+   */
+  public static com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder newBuilder() {
+    return new com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder();
+  }
+
+  /**
+   * Creates a new CustomerAvroModel RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new CustomerAvroModel RecordBuilder
+   */
+  public static com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder newBuilder(com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder other) {
+    if (other == null) {
+      return new com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder();
+    } else {
+      return new com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder(other);
+    }
+  }
+
+  /**
+   * Creates a new CustomerAvroModel RecordBuilder by copying an existing CustomerAvroModel instance.
+   * @param other The existing instance to copy.
+   * @return A new CustomerAvroModel RecordBuilder
+   */
+  public static com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder newBuilder(com.food.order.system.kafka.order.avro.model.CustomerAvroModel other) {
+    if (other == null) {
+      return new com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder();
+    } else {
+      return new com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder(other);
+    }
+  }
+
+  /**
+   * RecordBuilder for CustomerAvroModel instances.
+   */
+  @org.apache.avro.specific.AvroGenerated
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<CustomerAvroModel>
+    implements org.apache.avro.data.RecordBuilder<CustomerAvroModel> {
+
+    private java.lang.String id;
+    private java.lang.String username;
+    private java.lang.String firstName;
+    private java.lang.String lastName;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$, MODEL$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.id)) {
+        this.id = data().deepCopy(fields()[0].schema(), other.id);
+        fieldSetFlags()[0] = other.fieldSetFlags()[0];
+      }
+      if (isValidValue(fields()[1], other.username)) {
+        this.username = data().deepCopy(fields()[1].schema(), other.username);
+        fieldSetFlags()[1] = other.fieldSetFlags()[1];
+      }
+      if (isValidValue(fields()[2], other.firstName)) {
+        this.firstName = data().deepCopy(fields()[2].schema(), other.firstName);
+        fieldSetFlags()[2] = other.fieldSetFlags()[2];
+      }
+      if (isValidValue(fields()[3], other.lastName)) {
+        this.lastName = data().deepCopy(fields()[3].schema(), other.lastName);
+        fieldSetFlags()[3] = other.fieldSetFlags()[3];
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing CustomerAvroModel instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.food.order.system.kafka.order.avro.model.CustomerAvroModel other) {
+      super(SCHEMA$, MODEL$);
+      if (isValidValue(fields()[0], other.id)) {
+        this.id = data().deepCopy(fields()[0].schema(), other.id);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.username)) {
+        this.username = data().deepCopy(fields()[1].schema(), other.username);
+        fieldSetFlags()[1] = true;
+      }
+      if (isValidValue(fields()[2], other.firstName)) {
+        this.firstName = data().deepCopy(fields()[2].schema(), other.firstName);
+        fieldSetFlags()[2] = true;
+      }
+      if (isValidValue(fields()[3], other.lastName)) {
+        this.lastName = data().deepCopy(fields()[3].schema(), other.lastName);
+        fieldSetFlags()[3] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'id' field.
+      * @return The value.
+      */
+    public java.lang.String getId() {
+      return id;
+    }
+
+
+    /**
+      * Sets the value of the 'id' field.
+      * @param value The value of 'id'.
+      * @return This builder.
+      */
+    public com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder setId(java.lang.String value) {
+      validate(fields()[0], value);
+      this.id = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'id' field has been set.
+      * @return True if the 'id' field has been set, false otherwise.
+      */
+    public boolean hasId() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'id' field.
+      * @return This builder.
+      */
+    public com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder clearId() {
+      id = null;
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'username' field.
+      * @return The value.
+      */
+    public java.lang.String getUsername() {
+      return username;
+    }
+
+
+    /**
+      * Sets the value of the 'username' field.
+      * @param value The value of 'username'.
+      * @return This builder.
+      */
+    public com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder setUsername(java.lang.String value) {
+      validate(fields()[1], value);
+      this.username = value;
+      fieldSetFlags()[1] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'username' field has been set.
+      * @return True if the 'username' field has been set, false otherwise.
+      */
+    public boolean hasUsername() {
+      return fieldSetFlags()[1];
+    }
+
+
+    /**
+      * Clears the value of the 'username' field.
+      * @return This builder.
+      */
+    public com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder clearUsername() {
+      username = null;
+      fieldSetFlags()[1] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'firstName' field.
+      * @return The value.
+      */
+    public java.lang.String getFirstName() {
+      return firstName;
+    }
+
+
+    /**
+      * Sets the value of the 'firstName' field.
+      * @param value The value of 'firstName'.
+      * @return This builder.
+      */
+    public com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder setFirstName(java.lang.String value) {
+      validate(fields()[2], value);
+      this.firstName = value;
+      fieldSetFlags()[2] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'firstName' field has been set.
+      * @return True if the 'firstName' field has been set, false otherwise.
+      */
+    public boolean hasFirstName() {
+      return fieldSetFlags()[2];
+    }
+
+
+    /**
+      * Clears the value of the 'firstName' field.
+      * @return This builder.
+      */
+    public com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder clearFirstName() {
+      firstName = null;
+      fieldSetFlags()[2] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'lastName' field.
+      * @return The value.
+      */
+    public java.lang.String getLastName() {
+      return lastName;
+    }
+
+
+    /**
+      * Sets the value of the 'lastName' field.
+      * @param value The value of 'lastName'.
+      * @return This builder.
+      */
+    public com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder setLastName(java.lang.String value) {
+      validate(fields()[3], value);
+      this.lastName = value;
+      fieldSetFlags()[3] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'lastName' field has been set.
+      * @return True if the 'lastName' field has been set, false otherwise.
+      */
+    public boolean hasLastName() {
+      return fieldSetFlags()[3];
+    }
+
+
+    /**
+      * Clears the value of the 'lastName' field.
+      * @return This builder.
+      */
+    public com.food.order.system.kafka.order.avro.model.CustomerAvroModel.Builder clearLastName() {
+      lastName = null;
+      fieldSetFlags()[3] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public CustomerAvroModel build() {
+      try {
+        CustomerAvroModel record = new CustomerAvroModel();
+        record.id = fieldSetFlags()[0] ? this.id : (java.lang.String) defaultValue(fields()[0]);
+        record.username = fieldSetFlags()[1] ? this.username : (java.lang.String) defaultValue(fields()[1]);
+        record.firstName = fieldSetFlags()[2] ? this.firstName : (java.lang.String) defaultValue(fields()[2]);
+        record.lastName = fieldSetFlags()[3] ? this.lastName : (java.lang.String) defaultValue(fields()[3]);
+        return record;
+      } catch (org.apache.avro.AvroMissingFieldException e) {
+        throw e;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<CustomerAvroModel>
+    WRITER$ = (org.apache.avro.io.DatumWriter<CustomerAvroModel>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<CustomerAvroModel>
+    READER$ = (org.apache.avro.io.DatumReader<CustomerAvroModel>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}
+
+
+
+
+
+
+
+
+
+

infrastructure/kafka/kafka-model/src/main/resources/avro/customer.avsc

@@ -0,0 +1,32 @@
+{
+  "namespace": "com.food.order.system.kafka.order.avro.model",
+  "type": "record",
+  "name": "CustomerAvroModel",
+  "fields": [
+    {
+      "name": "id",
+      "type": {
+        "type": "string",
+        "logicalType": "uuid"
+      }
+    },
+    {
+      "name": "username",
+      "type": {
+        "type": "string"
+      }
+    },
+    {
+      "name": "firstName",
+      "type": {
+        "type": "string"
+      }
+    },
+    {
+      "name": "lastName",
+      "type": {
+        "type": "string"
+      }
+    }
+  ]
+}

order-service/order-container/src/main/resources/application.yml

@@ -1,5 +1,5 @@
 server:
-  port: 3232
+  port: 8181
 logging:
   level:
     com.food.order.system : DEBUG
@@ -9,6 +9,7 @@ order-service:
   payment-response-topic-name: payment-response-value
   restaurant-approval-request-topic-name: restaurant-approval-request-value
   restaurant-approval-response-topic-name: restaurant-approval-response-value
+  customer-topic-name: customer
   outbox-scheduler-fixed-rate: 10000
   outbox-scheduler-initial-delay: 10000
 
@@ -23,13 +24,11 @@ spring:
   datasource:
     url: jdbc:postgresql://localhost:5432/postgres?currentSchema=order&binaryTransfer=true&reWriteBatchedInserts=true&stringtype=unspecified
     username: postgres
-    password: postgres
-    driverClassName: org.postgresql.Driver
-  sql:
-    init:
-      platform: postgres
-      mode: always
-      schema-locations: classpath:init-schema.sql
+    password: admin
+    driver-class-name: org.postgresql.Driver
+    platform: postgres
+    schema: classpath:init-schema.sql
+    initialization-mode: always
 
 kafka-config:
   bootstrap-servers: localhost:19092, localhost:29092, localhost:39092
@@ -54,6 +53,7 @@ kafka-consumer-config:
   value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
   payment-consumer-group-id: payment-topic-consumer
   restaurant-approval-consumer-group-id: restaurant-approval-topic-consumer
+  customer-group-id: customer-topic-consumer
   auto-offset-reset: earliest
   specific-avro-reader-key: specific.avro.reader
   specific-avro-reader: true

order-service/order-container/src/main/resources/init-schema.sql

@@ -36,9 +36,9 @@ CREATE TABLE "order".order_items
 
 ALTER TABLE "order".order_items
     ADD CONSTRAINT "FK_ORDER_ID" FOREIGN KEY (order_id)
-    REFERENCES "order".orders (id) MATCH SIMPLE
-    ON UPDATE NO ACTION
-    ON DELETE CASCADE
+        REFERENCES "order".orders (id) MATCH SIMPLE
+        ON UPDATE NO ACTION
+        ON DELETE CASCADE
     NOT VALID;
 
 DROP TABLE IF EXISTS "order".order_address CASCADE;
@@ -55,17 +55,16 @@ CREATE TABLE "order".order_address
 
 ALTER TABLE "order".order_address
     ADD CONSTRAINT "FK_ORDER_ID" FOREIGN KEY (order_id)
-    REFERENCES "order".orders (id) MATCH SIMPLE
-    ON UPDATE NO ACTION
-    ON DELETE CASCADE
+        REFERENCES "order".orders (id) MATCH SIMPLE
+        ON UPDATE NO ACTION
+        ON DELETE CASCADE
     NOT VALID;
 
 DROP TYPE IF EXISTS saga_status;
-CREATE TYPE saga_status AS ENUM ('STARTED','FAILED','SUCCEEDED','PROCESSING', 'COMPENSATING','COMPENSATED');
+CREATE TYPE saga_status AS ENUM ('STARTED', 'FAILED', 'SUCCEEDED', 'PROCESSING', 'COMPENSATING', 'COMPENSATED');
 
 DROP TYPE IF EXISTS outbox_status;
-CREATE TYPE outbox_status AS ENUM ('STARTED','COMPLETED','FAILED');
-
+CREATE TYPE outbox_status AS ENUM ('STARTED', 'COMPLETED', 'FAILED');
 
 DROP TABLE IF EXISTS "order".payment_outbox CASCADE;
 
@@ -90,8 +89,7 @@ CREATE INDEX "payment_outbox_saga_status"
 
 CREATE UNIQUE INDEX "payment_outbox_saga_id"
     ON "order".payment_outbox
-        (type, saga_id, saga_status);
-
+    (type, saga_id, saga_status);
 
 DROP TABLE IF EXISTS "order".restaurant_approval_outbox CASCADE;
 
@@ -114,6 +112,17 @@ CREATE INDEX "restaurant_approval_outbox_saga_status"
     ON "order".restaurant_approval_outbox
         (type, outbox_status, saga_status);
 
-CREATE UNIQUE INDEX "restaurant_approval_outbox_saga_id"
-    ON "order".restaurant_approval_outbox
-        (type, saga_id, saga_status);
+--CREATE UNIQUE INDEX "restaurant_approval_outbox_saga_id"
+--    ON "order".restaurant_approval_outbox
+--    (type, saga_id, saga_status);
+
+DROP TABLE IF EXISTS "order".customers CASCADE;
+
+CREATE TABLE "order".customers
+(
+    id uuid NOT NULL,
+    username character varying COLLATE pg_catalog."default" NOT NULL,
+    first_name character varying COLLATE pg_catalog."default" NOT NULL,
+    last_name character varying COLLATE pg_catalog."default" NOT NULL,
+    CONSTRAINT customers_pkey PRIMARY KEY (id)
+);

order-service/order-data-access/src/main/java/com/food/order/system/data/access/customer/adapter/CustomerRepositoryImpl.java

@@ -19,6 +19,12 @@ public class CustomerRepositoryImpl implements CustomerRepository {
     private final CustomerDataAccessMapper customerDataAccessMapper;
 
 
+    @Override
+    public Customer save(Customer customer) {
+        return customerDataAccessMapper.customerEntityToCustomer(
+                customerJpaRepository.save(customerDataAccessMapper.customerToCustomerEntity(customer)));
+    }
+
     @Override
     public Optional<Customer> findCustomer(UUID customerId) {
         return customerJpaRepository.findById(customerId)

order-service/order-data-access/src/main/java/com/food/order/system/data/access/customer/entity/CustomerEntity.java

@@ -10,7 +10,7 @@ import java.util.UUID;
 
 @Entity
 @DynamicUpdate
-@Table(name = "order_customer_m_view",schema = "customer")
+@Table(name = "customers")
 @Getter
 @Setter
 @Builder
@@ -20,4 +20,8 @@ public class CustomerEntity {
 
     @Id
     private UUID id;
+
+    private String username;
+    private String firstName;
+    private String lastName;
 }

order-service/order-data-access/src/main/java/com/food/order/system/data/access/customer/mapper/CustomerDataAccessMapper.java

@@ -11,4 +11,13 @@ public class CustomerDataAccessMapper {
     public Customer customerEntityToCustomer(CustomerEntity customerEntity) {
         return new Customer(new CustomerId(customerEntity.getId()));
     }
+
+    public CustomerEntity customerToCustomerEntity(Customer customer) {
+        return CustomerEntity.builder()
+                .id(customer.getId().getValue())
+                .firstName(customer.getFirstName())
+                .lastName(customer.getLastName())
+                .username(customer.getUsername())
+                .build();
+    }
 }

order-service/order-domain/order-application-service/src/main/java/com/food/order/system/CustomerMessageListenerImpl.java

@@ -0,0 +1,37 @@
+package com.food.order.system;
+
+import com.food.order.system.dto.message.CustomerModel;
+import com.food.order.system.mapper.OrderDataMapper;
+import com.food.order.system.ports.input.message.listener.customer.CustomerMessageListener;
+import com.food.order.system.ports.output.repository.CustomerRepository;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.util.Objects;
+
+@Service
+@Slf4j
+@RequiredArgsConstructor
+public class CustomerMessageListenerImpl implements CustomerMessageListener {
+
+    private final CustomerRepository customerRepository;
+    private final OrderDataMapper orderDataMapper;
+
+
+    @Override
+    public void customerCreated(CustomerModel customerModel) {
+        var customer = customerRepository.save
+                (orderDataMapper.customerModelToCustomer(customerModel));
+
+        if (Objects.isNull(customer)) {
+            log.error("Customer not created");
+        } else {
+            log.info("Customer created");
+        }
+
+
+
+    }
+
+}

order-service/order-domain/order-application-service/src/main/java/com/food/order/system/dto/message/CustomerModel.java

@@ -0,0 +1,15 @@
+package com.food.order.system.dto.message;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+
+@Getter
+@Builder
+@AllArgsConstructor
+public class CustomerModel {
+    private String id;
+    private String username;
+    private String firstName;
+    private String lastName;
+}

order-service/order-domain/order-application-service/src/main/java/com/food/order/system/mapper/OrderDataMapper.java

@@ -1,9 +1,6 @@
 package com.food.order.system.mapper;
 
-import com.food.order.system.domain.entity.Order;
-import com.food.order.system.domain.entity.OrderItem;
-import com.food.order.system.domain.entity.Product;
-import com.food.order.system.domain.entity.Restaurant;
+import com.food.order.system.domain.entity.*;
 import com.food.order.system.domain.event.OrderCancelledEvent;
 import com.food.order.system.domain.event.OrderCreatedEvent;
 import com.food.order.system.domain.event.OrderPaidEvent;
@@ -11,6 +8,7 @@ import com.food.order.system.domain.valueobject.StreetAddress;
 import com.food.order.system.dto.create.CreateOrderCommand;
 import com.food.order.system.dto.create.CreateOrderResponse;
 import com.food.order.system.dto.create.OrderAddress;
+import com.food.order.system.dto.message.CustomerModel;
 import com.food.order.system.dto.track.TrackOrderResponse;
 import com.food.order.system.outbox.model.approval.OrderApprovalEventPayload;
 import com.food.order.system.outbox.model.approval.OrderApprovalProduct;
@@ -24,6 +22,14 @@ import java.util.UUID;
 @Component
 public class OrderDataMapper {
 
+    public Customer customerModelToCustomer(CustomerModel customerModel) {
+        return new Customer(new CustomerId(UUID.fromString(customerModel.getId())),
+                customerModel.getUsername(),
+                customerModel.getFirstName(),
+                customerModel.getLastName());
+    }
+
+
     public OrderPaymentEventPayload orderCancelledEventToOrderPaymentEventPayload(
             OrderCancelledEvent orderCancelledEvent) {
         return OrderPaymentEventPayload.builder()

order-service/order-domain/order-application-service/src/main/java/com/food/order/system/outbox/scheduler/approval/RestaurantApprovalOutboxScheduler.java

@@ -1,18 +1,18 @@
 package com.food.order.system.outbox.scheduler.approval;
 
-import com.food.order.system.domain.exception.OrderDomainException;
 import com.food.order.system.outbox.OutboxScheduler;
 import com.food.order.system.outbox.OutboxStatus;
 import com.food.order.system.outbox.model.approval.OrderApprovalOutboxMessage;
-import com.food.order.system.saga.SagaStatus;
 import com.food.order.system.ports.output.message.publisher.restaurantapproval.RestaurantApprovalRequestMessagePublisher;
+import com.food.order.system.saga.SagaStatus;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
-import java.util.Objects;
+import java.util.List;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 @Slf4j
@@ -28,28 +28,20 @@ public class RestaurantApprovalOutboxScheduler implements OutboxScheduler {
     @Scheduled(fixedDelayString = "${order-service.outbox-scheduler-fixed-rate}",
             initialDelayString = "${order-service.outbox-scheduler-initial-delay}")
     public void processOutboxMessage() {
-
-        log.info("Processing outbox message STARTED !");
-
-        var outboxMessageResponse =
+        Optional<List<OrderApprovalOutboxMessage>> outboxMessagesResponse =
                 approvalOutboxHelper.getApprovalOutboxMessageByOutboxStatusAndSagaStatus(
-                OutboxStatus.STARTED,
-                SagaStatus.STARTED,
-                SagaStatus.COMPENSATING)
-                        .orElseThrow(
-                                () -> new OrderDomainException("No outbox message found for processing"));
+                        OutboxStatus.STARTED,
+                        SagaStatus.PROCESSING);
+        if (outboxMessagesResponse.isPresent() && outboxMessagesResponse.get().size() > 0) {
+            List<OrderApprovalOutboxMessage> outboxMessages = outboxMessagesResponse.get();
+            log.info("Received {} OrderApprovalOutboxMessage with ids: {}, sending to message bus!",
+                    outboxMessages.size(),
+                    outboxMessages.stream().map(outboxMessage ->
+                            outboxMessage.getId().toString()).collect(Collectors.joining(",")));
+            outboxMessages.forEach(outboxMessage ->
+                    restaurantApprovalRequestMessagePublisher.publish(outboxMessage, this::updateOutboxStatus));
+            log.info("{} OrderApprovalOutboxMessage sent to message bus!", outboxMessages.size());
 
-        if (Objects.nonNull(outboxMessageResponse) && outboxMessageResponse.size() > 0) {
-
-            log.info("Received {} OrderPaymentOutboxMessage with ids :  {} , sending message bus !" ,
-                    outboxMessageResponse.size(),
-                    outboxMessageResponse.stream().map(orderPaymentOutboxMessage -> orderPaymentOutboxMessage.getId().toString())
-                            .collect(Collectors.joining(",")));
-            outboxMessageResponse.forEach(orderPaymentOutboxMessage -> {
-                restaurantApprovalRequestMessagePublisher.publish
-                        (orderPaymentOutboxMessage,this::updateOutboxStatus);
-            });
-            log.info("Processing outbox message completed ! ");
         }
 
     }

order-service/order-domain/order-application-service/src/main/java/com/food/order/system/ports/input/message/listener/customer/CustomerMessageListener.java

@@ -0,0 +1,7 @@
+package com.food.order.system.ports.input.message.listener.customer;
+
+import com.food.order.system.dto.message.CustomerModel;
+
+public interface CustomerMessageListener {
+    void customerCreated(CustomerModel customerModel);
+}

order-service/order-domain/order-application-service/src/main/java/com/food/order/system/ports/output/repository/CustomerRepository.java

@@ -6,5 +6,7 @@ import java.util.Optional;
 import java.util.UUID;
 
 public interface CustomerRepository {
+
+    Customer save(Customer customer);
     Optional<Customer> findCustomer(UUID customerId);
 }

order-service/order-domain/order-core-domain/src/main/java/com/food/order/system/domain/entity/Customer.java

@@ -5,10 +5,33 @@ import com.food.order.system.valueobject.CustomerId;
 
 public class Customer extends AggregateRoot<CustomerId> {
 
-    public Customer(){
+    private String username;
+    private String firstName;
+    private String lastName;
 
+
+    public Customer(CustomerId id, String username, String firstName, String lastName) {
+        super.setId(id);
+        this.username = username;
+        this.firstName = firstName;
+        this.lastName = lastName;
     }
+
     public Customer (CustomerId id) {
         super.setId(id);
     }
+
+    public String getUsername() {
+        return username;
+    }
+
+    public String getFirstName() {
+        return firstName;
+    }
+
+    public String getLastName() {
+        return lastName;
+    }
+
+
 }

order-service/order-messaging/src/main/java/com/food/order/system/messaging/listener/kafka/CustomerKafkaListener.java

@@ -0,0 +1,42 @@
+package com.food.order.system.messaging.listener.kafka;
+
+import com.food.order.system.kafka.consumer.KafkaConsumer;
+import com.food.order.system.kafka.order.avro.model.CustomerAvroModel;
+import com.food.order.system.messaging.mapper.OrderMessagingDataMapper;
+import com.food.order.system.ports.input.message.listener.customer.CustomerMessageListener;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class CustomerKafkaListener implements KafkaConsumer<CustomerAvroModel> {
+
+    private final CustomerMessageListener customerMessageListener;
+    private final OrderMessagingDataMapper orderMessagingDataMapper;
+
+    @Override
+    @KafkaListener(id = "${kafka-consumer-config.customer-group-id}",
+            topics = "${order-service.customer-topic-name}")
+    public void receive(List<CustomerAvroModel> messages,
+                        List<String> keys,
+                        List<Integer> partitions,
+                        List<Long> offSets) {
+
+        log.info("{} number of customer create messages received with keys {}, partitions {} and offsets {}",
+                messages.size(),
+                keys.toString(),
+                partitions.toString(),
+                offSets.toString());
+
+        messages.forEach(customerAvroModel ->
+                customerMessageListener.customerCreated(orderMessagingDataMapper
+                        .customerAvroModelToCustomerModel(customerAvroModel)));
+
+
+    }
+}

order-service/order-messaging/src/main/java/com/food/order/system/messaging/listener/kafka/PaymentResponseKafkaListener.java

@@ -1,5 +1,6 @@
 package com.food.order.system.messaging.listener.kafka;
 
+import com.food.order.system.domain.exception.OrderNotFoundException;
 import com.food.order.system.kafka.consumer.KafkaConsumer;
 import com.food.order.system.kafka.order.avro.model.PaymentResponseAvroModel;
 import com.food.order.system.kafka.order.avro.model.PaymentStatus;
@@ -7,6 +8,7 @@ import com.food.order.system.messaging.mapper.OrderMessagingDataMapper;
 import com.food.order.system.ports.input.message.listener.payment.PaymentResponseMessageListener;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.messaging.handler.annotation.Header;
@@ -33,20 +35,31 @@ public class PaymentResponseKafkaListener implements KafkaConsumer<PaymentRespon
                         @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                         @Header(KafkaHeaders.OFFSET) List<Long> offSets) {
 
-        log.info("{} number of payment responses received with keys : {} , partitions : {} , offsets : {}",
-                messages.size(), keys, partitions, offSets);
+        log.info("{} number of payment responses received with keys:{}, partitions:{} and offsets: {}",
+                messages.size(),
+                keys.toString(),
+                partitions.toString(),
+                offSets.toString());
 
-        messages.forEach(message -> {
-            if (PaymentStatus.COMPLETED.equals(message.getPaymentStatus())) {
-                log.info("Processing successful payment response for order id: {}", message.getOrderId());
-                paymentResponseMessageListener.paymentCompleted(orderMessagingDataMapper.
-                        paymentResponseAvroModelToPaymentResponse(message));
-            }
-            else if (PaymentStatus.FAILED.equals(message.getPaymentStatus()) ||
-                     PaymentStatus.CANCELLED.equals(message.getPaymentStatus())) {
-                log.info("Processing failed payment response for order id: {}", message.getOrderId());
-                paymentResponseMessageListener.paymentCancelled(orderMessagingDataMapper.
-                        paymentResponseAvroModelToPaymentResponse(message));
+        messages.forEach(paymentResponseAvroModel -> {
+            try {
+                if (PaymentStatus.COMPLETED == paymentResponseAvroModel.getPaymentStatus()) {
+                    log.info("Processing successful payment for order id: {}", paymentResponseAvroModel.getOrderId());
+                    paymentResponseMessageListener.paymentCompleted(orderMessagingDataMapper
+                            .paymentResponseAvroModelToPaymentResponse(paymentResponseAvroModel));
+                } else if (PaymentStatus.CANCELLED == paymentResponseAvroModel.getPaymentStatus() ||
+                        PaymentStatus.FAILED == paymentResponseAvroModel.getPaymentStatus()) {
+                    log.info("Processing unsuccessful payment for order id: {}", paymentResponseAvroModel.getOrderId());
+                    paymentResponseMessageListener.paymentCancelled(orderMessagingDataMapper
+                            .paymentResponseAvroModelToPaymentResponse(paymentResponseAvroModel));
+                }
+            } catch (OptimisticLockingFailureException e) {
+                //NO-OP for optimistic lock. This means another thread finished the work, do not throw error to prevent reading the data from kafka again!
+                log.error("Caught optimistic locking exception in PaymentResponseKafkaListener for order id: {}",
+                        paymentResponseAvroModel.getOrderId());
+            } catch (OrderNotFoundException e) {
+                //NO-OP for OrderNotFoundException
+                log.error("No order found for order id: {}", paymentResponseAvroModel.getOrderId());
             }
         });
 

order-service/order-messaging/src/main/java/com/food/order/system/messaging/mapper/OrderMessagingDataMapper.java

@@ -1,11 +1,9 @@
 package com.food.order.system.messaging.mapper;
 
-import com.food.order.system.domain.event.OrderCancelledEvent;
-import com.food.order.system.domain.event.OrderCreatedEvent;
-import com.food.order.system.domain.event.OrderPaidEvent;
-import com.food.order.system.kafka.order.avro.model.*;
+import com.food.order.system.dto.message.CustomerModel;
 import com.food.order.system.dto.message.PaymentResponse;
 import com.food.order.system.dto.message.RestaurantApprovalResponse;
+import com.food.order.system.kafka.order.avro.model.*;
 import com.food.order.system.outbox.model.approval.OrderApprovalEventPayload;
 import com.food.order.system.outbox.model.payment.OrderPaymentEventPayload;
 import com.food.order.system.valueobject.OrderApprovalStatus;
@@ -17,54 +15,9 @@ import java.util.UUID;
 @Component
 public class OrderMessagingDataMapper {
 
-    public PaymentRequestAvroModel orderCreatedEventToPaymentRequestAvroModel(OrderCreatedEvent orderCreatedEvent) {
-        var order = orderCreatedEvent.getOrder();
-        return PaymentRequestAvroModel.newBuilder()
-                .setId(UUID.randomUUID().toString())
-                .setSagaId("")
-                .setCustomerId(order.getCustomerId().getValue().toString())
-                .setOrderId(order.getId().getValue().toString())
-                .setPrice(order.getPrice().getAmount())
-                .setCreatedAt(orderCreatedEvent.getCreatedAt().toInstant())
-                .setPaymentOrderStatus(PaymentOrderStatus.PENDING)
-                .build();
-    }
-
-    public PaymentRequestAvroModel orderCancelledEventToPaymentRequestAvroModel(OrderCancelledEvent orderCancelledEvent) {
-        var order = orderCancelledEvent.getOrder();
-        return PaymentRequestAvroModel.newBuilder()
-                .setOrderId(order.getId().getValue().toString())
-                .setSagaId("")
-                .setCustomerId(order.getCustomerId().getValue().toString())
-                .setId(UUID.randomUUID().toString())
-                .setPrice(order.getPrice().getAmount())
-                .setCreatedAt(orderCancelledEvent.getCreatedAt().toInstant())
-                .setPaymentOrderStatus(PaymentOrderStatus.CANCELLED)
-                .build();
-    }
-
-
-    public RestaurantApprovalRequestAvroModel orderPaidEventToRestaurantApprovalRequestAvroModel(OrderPaidEvent event) {
-        var order = event.getOrder();
-        return RestaurantApprovalRequestAvroModel.newBuilder()
-                .setOrderId(order.getId().getValue().toString())
-                .setRestaurantId(order.getRestaurantId().getValue().toString())
-                .setProducts(order.getItems().stream()
-                        .map(item -> Product.newBuilder()
-                                .setId(item.getProduct().getId().getValue().toString())
-                                .setQuantity(item.getQuantity())
-                                .build())
-                        .toList())
-                .setId(UUID.randomUUID().toString())
-                .setSagaId("")
-                .setPrice(order.getPrice().getAmount())
-                .setCreatedAt(event.getCreatedAt().toInstant())
-                .setRestaurantOrderStatus(RestaurantOrderStatus.PAID)
-                .build();
-    }
-
     public PaymentResponse paymentResponseAvroModelToPaymentResponse(PaymentResponseAvroModel message) {
         return PaymentResponse.builder()
+                .id(message.getId())
                 .orderId(message.getOrderId())
                 .sagaId(message.getSagaId())
                 .paymentId(message.getPaymentId())
@@ -123,4 +76,13 @@ public class OrderMessagingDataMapper {
 
 
     }
+    public CustomerModel customerAvroModelToCustomerModel(CustomerAvroModel customerAvroModel) {
+        return CustomerModel.builder()
+                .id(customerAvroModel.getId())
+                .username(customerAvroModel.getUsername())
+                .firstName(customerAvroModel.getFirstName())
+                .lastName(customerAvroModel.getLastName())
+                .build();
+    }
+
 }

payment-service/payment-container/src/main/resources/application.yml

@@ -22,7 +22,7 @@ spring:
   datasource:
     url: jdbc:postgresql://localhost:5432/postgres?currentSchema=payment&binaryTransfer=true&reWriteBatchedInserts=true&stringtype=unspecified
     username: postgres
-    password: postgres
+    password: admin
     driver-class-name: org.postgresql.Driver
     platform: postgres
     schema: classpath:init-schema.sql

payment-service/payment-domain/payment-domain-core/src/main/java/com/food/order/system/payment/service/domain/PaymentDomainServiceImpl.java

@@ -9,7 +9,6 @@ import com.food.order.system.payment.service.domain.event.PaymentEvent;
 import com.food.order.system.payment.service.domain.event.PaymentFailedEvent;
 import com.food.order.system.payment.service.domain.valueobject.CreditHistoryId;
 import com.food.order.system.payment.service.domain.valueobject.TransactionType;
-import com.food.order.system.event.publisher.DomainEventPublisher;
 import com.food.order.system.valueobject.Money;
 import com.food.order.system.valueobject.PaymentStatus;
 import lombok.extern.slf4j.Slf4j;
@@ -79,13 +78,13 @@ public class PaymentDomainServiceImpl implements PaymentDomainService {
         var totalDebitHistory = getTotalHistoryAmount(creditHistory, TransactionType.DEBIT);
 
         if (totalDebitHistory.isGreaterThan(totalCreditHistory)) {
-            failureMessages.add("Customer id " + creditEntry.getCustomerId() + " has insufficient credit");
-            log.error("Customer id {} has insufficient credit", creditEntry.getCustomerId());
+            failureMessages.add("Customer id " + creditEntry.getCustomerId().getValue() + " has insufficient credit");
+            log.error("Customer id {} has insufficient credit", creditEntry.getCustomerId().getValue());
         }
 
         if (!creditEntry.getTotalCreditAmount().equals(totalCreditHistory.subtract(totalDebitHistory))) {
-            failureMessages.add("Customer id " + creditEntry.getCustomerId() + " has total is not equal to credit history");
-            log.error("Customer id {} has total is not equal to credit history", creditEntry.getCustomerId());
+            failureMessages.add("Customer id " + creditEntry.getCustomerId().getValue() + " has total is not equal to credit history");
+            log.error("Customer id {} has total is not equal to credit history", creditEntry.getCustomerId().getValue());
         }
     }
 
@@ -115,8 +114,8 @@ public class PaymentDomainServiceImpl implements PaymentDomainService {
 
     private void validateCreditEntry(Payment payment, CreditEntry creditEntry, List<String> failureMessages) {
         if(payment.getPrice().isGreaterThan(creditEntry.getTotalCreditAmount())){
-            failureMessages.add("Customer id "+ payment.getCustomerId() + " , has insufficient credit amount" +
-                            creditEntry.getTotalCreditAmount() + " to pay for order id " + payment.getOrderId());
+            failureMessages.add("Customer id "+ payment.getCustomerId().getValue() + " , has insufficient credit amount" +
+                            creditEntry.getTotalCreditAmount().getAmount() + " to pay for order id " + payment.getOrderId().getValue());
             log.error("Payment price is greater than credit");
         }
     }

pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>spring-boot-starter-parent</artifactId>
         <groupId>org.springframework.boot</groupId>
-        <version>2.7.1</version>
+        <version>2.6.7</version>
         <relativePath/>
     </parent>
     <modelVersion>4.0.0</modelVersion>
@@ -28,10 +28,12 @@
 
     <properties>
         <maven-compiler-plugin.version>3.9.0</maven-compiler-plugin.version>
-        <mockito.version>4.5.1</mockito.version>
+        <mockito.version>4.3.1</mockito.version>
         <spring-kafka.version>2.8.2</spring-kafka.version>
         <kavka-avro.serializer.version>7.0.1</kavka-avro.serializer.version>
         <avro.version>1.11.0</avro.version>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
     </properties>
 
     <dependencyManagement>
@@ -49,6 +51,40 @@
                 <version>${project.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>com.food.order</groupId>
+                <artifactId>customer-domain-core</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+
+            <dependency>
+                <groupId>com.food.order</groupId>
+                <artifactId>customer-service-application</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+
+
+            <dependency>
+                <groupId>com.food.order</groupId>
+                <artifactId>customer-application</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+
+
+            <dependency>
+                <groupId>com.food.order</groupId>
+                <artifactId>customer-data-access</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+
+
+            <dependency>
+                <groupId>com.food.order</groupId>
+                <artifactId>customer-messaging</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+
+
             <dependency>
                 <groupId>com.food.order</groupId>
                 <artifactId>payment-domain</artifactId>

restaurant-service/restaurant-container/src/main/resources/application.yml

@@ -22,14 +22,12 @@ spring:
   datasource:
     url: jdbc:postgresql://localhost:5432/postgres?currentSchema=restaurant&binaryTransfer=true&reWriteBatchedInserts=true&stringtype=unspecified
     username: postgres
-    password: postgres
+    password: admin
     driver-class-name: org.postgresql.Driver
-  sql:
-    init:
-      platform: postgres
-      mode: always
-      data-locations: classpath:init-data.sql
-      schema-locations: classpath:init-schema.sql
+    platform: postgres
+    schema: classpath:init-schema.sql
+    data: classpath:init-data.sql
+    initialization-mode: always
 
 kafka-config:
   bootstrap-servers: localhost:19092, localhost:29092, localhost:39092

restaurant-service/restaurant-dataaccess/src/main/java/com/food/order/system/data/access/restaurant/mapper/RestaurantDataAccessMapper.java

@@ -36,7 +36,7 @@ public class RestaurantDataAccessMapper {
                                 .productId(new ProductId(entity.getProductId()))
                                 .name(entity.getProductName())
                                 .price(new Money(entity.getProductPrice()))
-                                .available(entity.getProductActive())
+                                .available(entity.getProductAvailable())
                                 .build())
                 .toList();