Initila commit

DeviceManagement/.gitignore

+HELP.md
+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### VS Code ###
+.vscode/

DeviceManagement/.mvn/wrapper/maven-wrapper.properties

+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+wrapperVersion=3.3.1
+distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.6/apache-maven-3.9.6-bin.zip

DeviceManagement/mvnw.cmd

+<# : batch portion
+@REM ----------------------------------------------------------------------------
+@REM Licensed to the Apache Software Foundation (ASF) under one
+@REM or more contributor license agreements.  See the NOTICE file
+@REM distributed with this work for additional information
+@REM regarding copyright ownership.  The ASF licenses this file
+@REM to you under the Apache License, Version 2.0 (the
+@REM "License"); you may not use this file except in compliance
+@REM with the License.  You may obtain a copy of the License at
+@REM
+@REM    https://www.apache.org/licenses/LICENSE-2.0
+@REM
+@REM Unless required by applicable law or agreed to in writing,
+@REM software distributed under the License is distributed on an
+@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+@REM KIND, either express or implied.  See the License for the
+@REM specific language governing permissions and limitations
+@REM under the License.
+@REM ----------------------------------------------------------------------------
+
+@REM ----------------------------------------------------------------------------
+@REM Apache Maven Wrapper startup batch script, version 3.3.1
+@REM
+@REM Optional ENV vars
+@REM   MVNW_REPOURL - repo url base for downloading maven distribution
+@REM   MVNW_USERNAME/MVNW_PASSWORD - user and password for downloading maven
+@REM   MVNW_VERBOSE - true: enable verbose log; others: silence the output
+@REM ----------------------------------------------------------------------------
+
+@IF "%__MVNW_ARG0_NAME__%"=="" (SET __MVNW_ARG0_NAME__=%~nx0)
+@SET __MVNW_CMD__=
+@SET __MVNW_ERROR__=
+@SET __MVNW_PSMODULEP_SAVE=%PSModulePath%
+@SET PSModulePath=
+@FOR /F "usebackq tokens=1* delims==" %%A IN (`powershell -noprofile "& {$scriptDir='%~dp0'; $script='%__MVNW_ARG0_NAME__%'; icm -ScriptBlock ([Scriptblock]::Create((Get-Content -Raw '%~f0'))) -NoNewScope}"`) DO @(
+  IF "%%A"=="MVN_CMD" (set __MVNW_CMD__=%%B) ELSE IF "%%B"=="" (echo %%A) ELSE (echo %%A=%%B)
+)
+@SET PSModulePath=%__MVNW_PSMODULEP_SAVE%
+@SET __MVNW_PSMODULEP_SAVE=
+@SET __MVNW_ARG0_NAME__=
+@SET MVNW_USERNAME=
+@SET MVNW_PASSWORD=
+@IF NOT "%__MVNW_CMD__%"=="" (%__MVNW_CMD__% %*)
+@echo Cannot start maven from wrapper >&2 && exit /b 1
+@GOTO :EOF
+: end batch / begin powershell #>
+
+$ErrorActionPreference = "Stop"
+if ($env:MVNW_VERBOSE -eq "true") {
+  $VerbosePreference = "Continue"
+}
+
+# calculate distributionUrl, requires .mvn/wrapper/maven-wrapper.properties
+$distributionUrl = (Get-Content -Raw "$scriptDir/.mvn/wrapper/maven-wrapper.properties" | ConvertFrom-StringData).distributionUrl
+if (!$distributionUrl) {
+  Write-Error "cannot read distributionUrl property in $scriptDir/.mvn/wrapper/maven-wrapper.properties"
+}
+
+switch -wildcard -casesensitive ( $($distributionUrl -replace '^.*/','') ) {
+  "maven-mvnd-*" {
+    $USE_MVND = $true
+    $distributionUrl = $distributionUrl -replace '-bin\.[^.]*$',"-windows-amd64.zip"
+    $MVN_CMD = "mvnd.cmd"
+    break
+  }
+  default {
+    $USE_MVND = $false
+    $MVN_CMD = $script -replace '^mvnw','mvn'
+    break
+  }
+}
+
+# apply MVNW_REPOURL and calculate MAVEN_HOME
+# maven home pattern: ~/.m2/wrapper/dists/{apache-maven-<version>,maven-mvnd-<version>-<platform>}/<hash>
+if ($env:MVNW_REPOURL) {
+  $MVNW_REPO_PATTERN = if ($USE_MVND) { "/org/apache/maven/" } else { "/maven/mvnd/" }
+  $distributionUrl = "$env:MVNW_REPOURL$MVNW_REPO_PATTERN$($distributionUrl -replace '^.*'+$MVNW_REPO_PATTERN,'')"
+}
+$distributionUrlName = $distributionUrl -replace '^.*/',''
+$distributionUrlNameMain = $distributionUrlName -replace '\.[^.]*$','' -replace '-bin$',''
+$MAVEN_HOME_PARENT = "$HOME/.m2/wrapper/dists/$distributionUrlNameMain"
+$MAVEN_HOME_NAME = ([System.Security.Cryptography.MD5]::Create().ComputeHash([byte[]][char[]]$distributionUrl) | ForEach-Object {$_.ToString("x2")}) -join ''
+$MAVEN_HOME = "$MAVEN_HOME_PARENT/$MAVEN_HOME_NAME"
+
+if (Test-Path -Path "$MAVEN_HOME" -PathType Container) {
+  Write-Verbose "found existing MAVEN_HOME at $MAVEN_HOME"
+  Write-Output "MVN_CMD=$MAVEN_HOME/bin/$MVN_CMD"
+  exit $?
+}
+
+if (! $distributionUrlNameMain -or ($distributionUrlName -eq $distributionUrlNameMain)) {
+  Write-Error "distributionUrl is not valid, must end with *-bin.zip, but found $distributionUrl"
+}
+
+# prepare tmp dir
+$TMP_DOWNLOAD_DIR_HOLDER = New-TemporaryFile
+$TMP_DOWNLOAD_DIR = New-Item -Itemtype Directory -Path "$TMP_DOWNLOAD_DIR_HOLDER.dir"
+$TMP_DOWNLOAD_DIR_HOLDER.Delete() | Out-Null
+trap {
+  if ($TMP_DOWNLOAD_DIR.Exists) {
+    try { Remove-Item $TMP_DOWNLOAD_DIR -Recurse -Force | Out-Null }
+    catch { Write-Warning "Cannot remove $TMP_DOWNLOAD_DIR" }
+  }
+}
+
+New-Item -Itemtype Directory -Path "$MAVEN_HOME_PARENT" -Force | Out-Null
+
+# Download and Install Apache Maven
+Write-Verbose "Couldn't find MAVEN_HOME, downloading and installing it ..."
+Write-Verbose "Downloading from: $distributionUrl"
+Write-Verbose "Downloading to: $TMP_DOWNLOAD_DIR/$distributionUrlName"
+
+$webclient = New-Object System.Net.WebClient
+if ($env:MVNW_USERNAME -and $env:MVNW_PASSWORD) {
+  $webclient.Credentials = New-Object System.Net.NetworkCredential($env:MVNW_USERNAME, $env:MVNW_PASSWORD)
+}
+[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12
+$webclient.DownloadFile($distributionUrl, "$TMP_DOWNLOAD_DIR/$distributionUrlName") | Out-Null
+
+# If specified, validate the SHA-256 sum of the Maven distribution zip file
+$distributionSha256Sum = (Get-Content -Raw "$scriptDir/.mvn/wrapper/maven-wrapper.properties" | ConvertFrom-StringData).distributionSha256Sum
+if ($distributionSha256Sum) {
+  if ($USE_MVND) {
+    Write-Error "Checksum validation is not supported for maven-mvnd. `nPlease disable validation by removing 'distributionSha256Sum' from your maven-wrapper.properties."
+  }
+  Import-Module $PSHOME\Modules\Microsoft.PowerShell.Utility -Function Get-FileHash
+  if ((Get-FileHash "$TMP_DOWNLOAD_DIR/$distributionUrlName" -Algorithm SHA256).Hash.ToLower() -ne $distributionSha256Sum) {
+    Write-Error "Error: Failed to validate Maven distribution SHA-256, your Maven distribution might be compromised. If you updated your Maven version, you need to update the specified distributionSha256Sum property."
+  }
+}
+
+# unzip and move
+Expand-Archive "$TMP_DOWNLOAD_DIR/$distributionUrlName" -DestinationPath "$TMP_DOWNLOAD_DIR" | Out-Null
+Rename-Item -Path "$TMP_DOWNLOAD_DIR/$distributionUrlNameMain" -NewName $MAVEN_HOME_NAME | Out-Null
+try {
+  Move-Item -Path "$TMP_DOWNLOAD_DIR/$MAVEN_HOME_NAME" -Destination $MAVEN_HOME_PARENT | Out-Null
+} catch {
+  if (! (Test-Path -Path "$MAVEN_HOME" -PathType Container)) {
+    Write-Error "fail to move MAVEN_HOME"
+  }
+} finally {
+  try { Remove-Item $TMP_DOWNLOAD_DIR -Recurse -Force | Out-Null }
+  catch { Write-Warning "Cannot remove $TMP_DOWNLOAD_DIR" }
+}
+
+Write-Output "MVN_CMD=$MAVEN_HOME/bin/$MVN_CMD"

DeviceManagement/pom.xml

+<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>org.springframework.boot</groupId>
+		<artifactId>spring-boot-starter-parent</artifactId>
+		<version>3.2.6</version>
+		<relativePath/> <!-- lookup parent from repository -->
+	</parent>
+	<groupId>com.springpoc</groupId>
+	<artifactId>DeviceManagement</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<name>DeviceManagement</name>
+	<description>Device Management project for Spring Boot</description>
+	<properties>
+		<java.version>17</java.version>
+	</properties>
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-jpa</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-mongodb</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-validation</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-devtools</artifactId>
+			<scope>runtime</scope>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.projectlombok</groupId>
+			<artifactId>lombok</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+        <dependency>
+            <groupId>com.springpoc.userservice</groupId>
+            <artifactId>UserService</artifactId>
+            <version>0.0.1-SNAPSHOT</version>
+            <scope>compile</scope>
+        </dependency>
+		<dependency>
+			<groupId>com.springpoc.userservice</groupId>
+			<artifactId>UserService</artifactId>
+			<version>0.0.1-SNAPSHOT</version>
+		</dependency>
+		<dependency>
+			<groupId>com.springpoc.userservice</groupId>
+			<artifactId>UserService</artifactId>
+			<version>0.0.1-SNAPSHOT</version>
+			<scope>compile</scope>
+		</dependency>
+		<!-- Use the latest version -->
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+				<configuration>
+					<excludes>
+						<exclude>
+							<groupId>org.projectlombok</groupId>
+							<artifactId>lombok</artifactId>
+						</exclude>
+					</excludes>
+				</configuration>
+			</plugin>
+		</plugins>
+	</build>
+
+</project>

DeviceManagement/src/main/java/com/springpoc/devicemanagement/DeviceManagementApplication.java

+package com.springpoc.devicemanagement;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.ComponentScan;
+
+@SpringBootApplication
+public class DeviceManagementApplication {
+
+	public static void main(String[] args) {
+		SpringApplication.run(DeviceManagementApplication.class, args);
+
+	}
+
+}

DeviceManagement/src/main/java/com/springpoc/devicemanagement/config/AppConfig.java

+package com.springpoc.devicemanagement.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.client.RestTemplate;
+
+@Configuration
+public class AppConfig {
+
+    @Bean
+    public RestTemplate restTemplate() {
+        return new RestTemplate();
+    }
+}
\ No newline at end of file

DeviceManagement/src/main/java/com/springpoc/devicemanagement/config/WebClientConfig.java

+package com.springpoc.devicemanagement.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.reactive.function.client.WebClient;
+
+@Configuration
+public class WebClientConfig {
+    @Bean
+    public WebClient.Builder webClientBuilder() {
+        return WebClient.builder();
+    }
+}
+

DeviceManagement/src/main/java/com/springpoc/devicemanagement/controller/DeviceController.java

+package com.springpoc.devicemanagement.controller;
+
+import com.springpoc.devicemanagement.dto.DeviceDTO;
+import com.springpoc.devicemanagement.dto.MerchantDeviceAssReq;
+import com.springpoc.devicemanagement.model.Device;
+import com.springpoc.devicemanagement.model.MerchantDeviceAss;
+import com.springpoc.devicemanagement.service.DeviceService;
+import org.apache.kafka.common.protocol.types.Field;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.*;
+import reactor.core.publisher.Mono;
+
+@RestController
+@RequestMapping("/device")
+public class DeviceController {
+
+    @Autowired
+    DeviceService deviceService;
+
+    @PostMapping("/createDevice")
+    public Mono<ResponseEntity<DeviceDTO>> createDevice(@RequestBody Mono<DeviceDTO> deviceDTO) {
+        return deviceService.saveDeviceDetails(deviceDTO)
+                .map(ResponseEntity::ok);
+    }
+
+    @DeleteMapping("/deleteDevice")
+    public Mono<String> deleteDevice() {
+        return deviceService.deleteDeviceDetails();
+    }
+
+    @PostMapping("/merchantDeviceAssociation")
+    public  Mono<String> merchantDeviceAssociation(@RequestBody MerchantDeviceAssReq merchantDeviceAssReq) {
+        return deviceService.createAssociation(merchantDeviceAssReq);
+    }
+}

DeviceManagement/src/main/java/com/springpoc/devicemanagement/dto/DeviceDTO.java

+package com.springpoc.devicemanagement.dto;
+
+import jakarta.persistence.Column;
+import jakarta.validation.constraints.NotBlank;
+import lombok.*;
+
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+@Setter
+@Getter
+@AllArgsConstructor
+@NoArgsConstructor
+@Builder
+@Data
+public class DeviceDTO {
+
+    @NotBlank(message = "Device type is required")
+    private String deviceType;
+    private long serialNo;
+    private String version;
+    private String description;
+    @NotBlank(message = "created By User cannot be blank")
+    private String createdBy;
+    private LocalDateTime createdTime;
+    private String updatedBy;
+    private LocalDateTime updatedTime;
+}

DeviceManagement/src/main/java/com/springpoc/devicemanagement/dto/MerchantDeviceAssReq.java

+package com.springpoc.devicemanagement.dto;
+
+import lombok.*;
+import org.bson.types.ObjectId;
+
+@Setter
+@Getter
+@AllArgsConstructor
+@NoArgsConstructor
+@Builder
+@Data
+public class MerchantDeviceAssReq {
+
+    private ObjectId merchant_id;
+    private ObjectId device_id;
+}

DeviceManagement/src/main/java/com/springpoc/devicemanagement/exception/CustomException.java

+package com.springpoc.devicemanagement.exception;
+
+public class CustomException extends RuntimeException{
+    public CustomException(String message) {
+        super(message);
+    }
+}

DeviceManagement/src/main/java/com/springpoc/devicemanagement/exception/GlobalExceptionHandler.java

+package com.springpoc.devicemanagement.exception;
+
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.ControllerAdvice;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+import reactor.core.publisher.Mono;
+
+@ControllerAdvice
+public class GlobalExceptionHandler {
+
+    @ExceptionHandler(RuntimeException.class)
+    public Mono<ResponseEntity<String>> handleRuntimeException(RuntimeException e) {
+        return Mono.just(new ResponseEntity<>("A runtime exception occurred: " + e.getMessage(), HttpStatus.INTERNAL_SERVER_ERROR));
+    }
+}

DeviceManagement/src/main/java/com/springpoc/devicemanagement/model/Device.java

+package com.springpoc.devicemanagement.model;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.Id;
+import jakarta.validation.constraints.NotBlank;
+import lombok.*;
+import org.bson.types.ObjectId;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+@Setter
+@Getter
+@AllArgsConstructor
+@NoArgsConstructor
+@Builder
+@Data
+@ToString
+@Document(collection = "device-details")
+public class Device {
+
+    @Id
+    private ObjectId id;
+    @Column
+    @NotBlank(message = "Device type is required")
+    private String deviceType;
+
+    @Column
+    private long serialNo;
+
+    @Column
+    private String version;
+
+    @Column
+    private String description;
+
+    @Column
+    @NotBlank(message = "created By User cannot be blank")
+    private String createdBy;
+    @Column
+    private LocalDateTime createdTime;
+    @Column
+    private String updatedBy;
+    @Column
+    private LocalDateTime updatedTime;
+
+
+}

DeviceManagement/src/main/java/com/springpoc/devicemanagement/model/MerchantDeviceAss.java

+package com.springpoc.devicemanagement.model;
+
+import com.springpoc.userservice.model.User;
+import jakarta.persistence.Id;
+import lombok.*;
+import org.bson.types.ObjectId;
+import org.springframework.data.mongodb.core.mapping.DBRef;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Setter
+@Getter
+@AllArgsConstructor
+@NoArgsConstructor
+@Builder
+@Data
+@ToString
+@Document(collection = "merchant-device-ass")
+public class MerchantDeviceAss {
+    @Id
+    private ObjectId device_id;
+    private ObjectId merchant_id;
+
+    private String status;
+
+}

DeviceManagement/src/main/java/com/springpoc/devicemanagement/repo/DeviceRepository.java

+package com.springpoc.devicemanagement.repo;
+
+import com.springpoc.devicemanagement.model.Device;
+import org.bson.types.ObjectId;
+import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
+import org.springframework.stereotype.Repository;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Mono;
+
+import java.util.UUID;
+
+@Repository
+public interface DeviceRepository extends ReactiveMongoRepository<Device, ObjectId> {
+    Mono<String> deleteByDeviceType(String deviceType);
+
+   Mono<String> findByDeviceType(String deviceType);
+
+
+    Device findByCreatedBy(String userType);
+}

DeviceManagement/src/main/java/com/springpoc/devicemanagement/repo/MerchantDeviceAssociationRepository.java

+package com.springpoc.devicemanagement.repo;
+
+import com.springpoc.devicemanagement.model.Device;
+import com.springpoc.devicemanagement.model.MerchantDeviceAss;
+
+
+import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
+import org.springframework.stereotype.Repository;
+import reactor.core.publisher.Mono;
+
+@Repository
+public interface MerchantDeviceAssociationRepository extends ReactiveMongoRepository<MerchantDeviceAss, String>{
+
+
+
+
+
+}

DeviceManagement/src/main/java/com/springpoc/devicemanagement/repo/UserRepository.java

+package com.springpoc.devicemanagement.repo;
+
+import com.springpoc.userservice.model.User;
+import org.bson.types.ObjectId;
+import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface UserRepository extends ReactiveMongoRepository<User, ObjectId> {
+    User findByUserType(String userType);
+}

DeviceManagement/src/main/java/com/springpoc/devicemanagement/service/DeviceService.java

+package com.springpoc.devicemanagement.service;
+
+import com.springpoc.devicemanagement.dto.DeviceDTO;
+import com.springpoc.devicemanagement.dto.MerchantDeviceAssReq;
+import reactor.core.publisher.Mono;
+
+public interface DeviceService {
+
+    Mono<DeviceDTO> saveDeviceDetails(Mono<DeviceDTO> deviceDTO);
+
+    Mono<String> deleteDeviceDetails();
+
+
+    Mono<String> createAssociation(MerchantDeviceAssReq merchantDeviceAssReq);
+}

DeviceManagement/src/main/java/com/springpoc/devicemanagement/service/impl/DeviceServiceImpl.java

+package com.springpoc.devicemanagement.service.impl;
+
+
+import com.springpoc.devicemanagement.dto.DeviceDTO;
+import com.springpoc.devicemanagement.dto.MerchantDeviceAssReq;
+import com.springpoc.devicemanagement.model.MerchantDeviceAss;
+import com.springpoc.devicemanagement.repo.DeviceRepository;
+import com.springpoc.devicemanagement.repo.MerchantDeviceAssociationRepository;
+import com.springpoc.devicemanagement.repo.UserRepository;
+import com.springpoc.devicemanagement.service.DeviceService;
+import com.springpoc.devicemanagement.utils.DeviceMapper;
+import com.springpoc.userservice.exception.CustomException;
+
+import org.bson.types.ObjectId;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Mono;
+
+@Service
+public class DeviceServiceImpl implements DeviceService {
+    private final UserRepository userRepository;
+    @Autowired
+    DeviceRepository deviceRepository;
+
+
+    @Autowired
+    MerchantDeviceAssociationRepository merchantDeviceAssociationRepository;
+
+
+    public DeviceServiceImpl(UserRepository userRepository) {
+        this.userRepository = userRepository;
+    }
+
+    @Override
+    public Mono<DeviceDTO> saveDeviceDetails(Mono<DeviceDTO> deviceDTOMono) {
+        return DeviceMapper.mapToDevice(deviceDTOMono)
+                .flatMap(device -> deviceRepository.save(device))
+                .map(devices -> DeviceMapper.mapToDeviceDTO(devices))
+                .switchIfEmpty(Mono.error(new RuntimeException("Invalid details.")));
+    }
+
+    @Override
+    public Mono<String> deleteDeviceDetails() {
+        return deviceRepository.deleteAll()
+                .thenReturn("Device details deleted successfully.");
+    }
+
+    @Override
+    public Mono<String> createAssociation(MerchantDeviceAssReq merchantdeviceassociation) {
+        ObjectId deviceID = merchantdeviceassociation.getDevice_id();
+        ObjectId merchant_id = merchantdeviceassociation.getMerchant_id();
+        if (merchant_id == null && deviceID == null) {
+            return Mono.error(new CustomException("Merchant Id or device Id is required."));
+        } else {
+            return userRepository.findById(merchant_id)
+                    .flatMap(user -> {
+                        return deviceRepository.findById(deviceID)
+                                .flatMap(device -> {
+                                    // Check if the OTP matches the predefined OTP
+                                    if (!user.getId().equals(merchant_id) && !device.getId().equals(deviceID)) {
+                                        return Mono.error(new CustomException("Merchant id or Device  is not valid"));
+                                    } else if (!user.getUserType().equalsIgnoreCase("Merchant")) {
+                                        return Mono.error(new CustomException("User should merchant user."));
+                                    } else {
+                                        MerchantDeviceAss merchantDeviceAss =
+                                                MerchantDeviceAss.builder()
+                                                        .device_id(deviceID)
+                                                        .merchant_id(merchant_id)
+                                                        .status("Assigned")
+                                                        .build();
+                                        return merchantDeviceAssociationRepository.save(merchantDeviceAss)
+                                                .thenReturn("Device and Merchant Association done successfully.");
+                                    }
+                                }).switchIfEmpty(Mono.error(new CustomException("Device Id is not Valid.")));
+                    })
+                    .switchIfEmpty(Mono.error(new CustomException("Merchant Id is not Valid.")));
+        }
+    }
+
+}
\ No newline at end of file

DeviceManagement/src/main/java/com/springpoc/devicemanagement/utils/ConstantsUtils.java

+package com.springpoc.devicemanagement.utils;
+
+public class ConstantsUtils {
+    public static final String userNotExists = "User does not exists.";
+    public static final String userExists = "User is already  exists.";
+    public static final String userCreationFail = "User creation failed.";
+
+    public static final String  notAuthrozedToCreate ="User is not authorized to create.";
+
+    public static final String saveUrl = "http://localhost:9092/login/createUserEntryLogin";
+    public static final String deleteUrl = "http://localhost:9092/login/delete/{email}";
+    public static final String saveAllUrl = "http://localhost:9092/login/createAllUsersEntryLogin";
+
+    public static final String failToNotify = "Failed to notify other service";
+    public static final String userDeletedSuccessfully =    "User deleted successfully";
+
+
+}

DeviceManagement/src/main/java/com/springpoc/devicemanagement/utils/DeviceMapper.java

+package com.springpoc.devicemanagement.utils;
+
+import com.springpoc.devicemanagement.dto.DeviceDTO;
+import com.springpoc.devicemanagement.model.Device;
+import reactor.core.publisher.Mono;
+
+
+public class DeviceMapper {
+    public static Mono<Device> mapToDevice(Mono<DeviceDTO> deviceDTOMono) {
+        return deviceDTOMono.map(deviceDTO ->
+                Device.builder()
+                        .deviceType(deviceDTO.getDeviceType())
+                        .serialNo(deviceDTO.getSerialNo())
+                        .version(deviceDTO.getVersion())
+                        .description(deviceDTO.getDescription())
+                        .createdBy(deviceDTO.getCreatedBy())
+                        .createdTime(deviceDTO.getCreatedTime())
+                        .updatedBy(deviceDTO.getUpdatedBy())
+                        .updatedTime(deviceDTO.getUpdatedTime())
+                        .build()
+        );
+    }
+
+    public static DeviceDTO mapToDeviceDTO(Device devices) {
+        return DeviceDTO.builder()
+                .deviceType(devices.getDeviceType())
+                .serialNo(devices.getSerialNo())
+                .version(devices.getVersion())
+                .description(devices.getDescription())
+                .createdBy(devices.getCreatedBy())
+                .createdTime(devices.getCreatedTime())
+                .updatedBy(devices.getUpdatedBy())
+                .updatedTime(devices.getUpdatedTime())
+                .build();
+
+    }
+}

DeviceManagement/src/main/resources/application.properties

+spring.application.name=DeviceManagement
+spring.data.mongodb.host=localhost
+spring.data.mongodb.port=27017
+spring.data.mongodb.database=epepocdb
+spring.data.mongodb.url = mongodb://localhost:27017
+spring.autoconfigure.exclude=org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration
+myapp.entry-with-special-characters=@~!%&=""
+spring.mvc.static-path-pattern=/static/**
+spring.kafka.consumer.group-id=device-group
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
+spring.kafka.consumer.auto-offset-reset=earliest
+
+# Kafka topic name
+spring.kafka.device.topic=create-device-topic

DeviceManagement/src/test/java/com/springpoc/devicemanagement/DeviceManagementApplicationTests.java

+package com.springpoc.devicemanagement;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+
+@SpringBootTest
+class DeviceManagementApplicationTests {
+
+	@Test
+	void contextLoads() {
+	}
+
+}

LoginService/.gitignore

+HELP.md
+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### VS Code ###
+.vscode/

LoginService/.mvn/wrapper/maven-wrapper.properties

+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+wrapperVersion=3.3.1
+distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.6/apache-maven-3.9.6-bin.zip

LoginService/mvnw.cmd

+<# : batch portion
+@REM ----------------------------------------------------------------------------
+@REM Licensed to the Apache Software Foundation (ASF) under one
+@REM or more contributor license agreements.  See the NOTICE file
+@REM distributed with this work for additional information
+@REM regarding copyright ownership.  The ASF licenses this file
+@REM to you under the Apache License, Version 2.0 (the
+@REM "License"); you may not use this file except in compliance
+@REM with the License.  You may obtain a copy of the License at
+@REM
+@REM    https://www.apache.org/licenses/LICENSE-2.0
+@REM
+@REM Unless required by applicable law or agreed to in writing,
+@REM software distributed under the License is distributed on an
+@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+@REM KIND, either express or implied.  See the License for the
+@REM specific language governing permissions and limitations
+@REM under the License.
+@REM ----------------------------------------------------------------------------
+
+@REM ----------------------------------------------------------------------------
+@REM Apache Maven Wrapper startup batch script, version 3.3.1
+@REM
+@REM Optional ENV vars
+@REM   MVNW_REPOURL - repo url base for downloading maven distribution
+@REM   MVNW_USERNAME/MVNW_PASSWORD - user and password for downloading maven
+@REM   MVNW_VERBOSE - true: enable verbose log; others: silence the output
+@REM ----------------------------------------------------------------------------
+
+@IF "%__MVNW_ARG0_NAME__%"=="" (SET __MVNW_ARG0_NAME__=%~nx0)
+@SET __MVNW_CMD__=
+@SET __MVNW_ERROR__=
+@SET __MVNW_PSMODULEP_SAVE=%PSModulePath%
+@SET PSModulePath=
+@FOR /F "usebackq tokens=1* delims==" %%A IN (`powershell -noprofile "& {$scriptDir='%~dp0'; $script='%__MVNW_ARG0_NAME__%'; icm -ScriptBlock ([Scriptblock]::Create((Get-Content -Raw '%~f0'))) -NoNewScope}"`) DO @(
+  IF "%%A"=="MVN_CMD" (set __MVNW_CMD__=%%B) ELSE IF "%%B"=="" (echo %%A) ELSE (echo %%A=%%B)
+)
+@SET PSModulePath=%__MVNW_PSMODULEP_SAVE%
+@SET __MVNW_PSMODULEP_SAVE=
+@SET __MVNW_ARG0_NAME__=
+@SET MVNW_USERNAME=
+@SET MVNW_PASSWORD=
+@IF NOT "%__MVNW_CMD__%"=="" (%__MVNW_CMD__% %*)
+@echo Cannot start maven from wrapper >&2 && exit /b 1
+@GOTO :EOF
+: end batch / begin powershell #>
+
+$ErrorActionPreference = "Stop"
+if ($env:MVNW_VERBOSE -eq "true") {
+  $VerbosePreference = "Continue"
+}
+
+# calculate distributionUrl, requires .mvn/wrapper/maven-wrapper.properties
+$distributionUrl = (Get-Content -Raw "$scriptDir/.mvn/wrapper/maven-wrapper.properties" | ConvertFrom-StringData).distributionUrl
+if (!$distributionUrl) {
+  Write-Error "cannot read distributionUrl property in $scriptDir/.mvn/wrapper/maven-wrapper.properties"
+}
+
+switch -wildcard -casesensitive ( $($distributionUrl -replace '^.*/','') ) {
+  "maven-mvnd-*" {
+    $USE_MVND = $true
+    $distributionUrl = $distributionUrl -replace '-bin\.[^.]*$',"-windows-amd64.zip"
+    $MVN_CMD = "mvnd.cmd"
+    break
+  }
+  default {
+    $USE_MVND = $false
+    $MVN_CMD = $script -replace '^mvnw','mvn'
+    break
+  }
+}
+
+# apply MVNW_REPOURL and calculate MAVEN_HOME
+# maven home pattern: ~/.m2/wrapper/dists/{apache-maven-<version>,maven-mvnd-<version>-<platform>}/<hash>
+if ($env:MVNW_REPOURL) {
+  $MVNW_REPO_PATTERN = if ($USE_MVND) { "/org/apache/maven/" } else { "/maven/mvnd/" }
+  $distributionUrl = "$env:MVNW_REPOURL$MVNW_REPO_PATTERN$($distributionUrl -replace '^.*'+$MVNW_REPO_PATTERN,'')"
+}
+$distributionUrlName = $distributionUrl -replace '^.*/',''
+$distributionUrlNameMain = $distributionUrlName -replace '\.[^.]*$','' -replace '-bin$',''
+$MAVEN_HOME_PARENT = "$HOME/.m2/wrapper/dists/$distributionUrlNameMain"
+$MAVEN_HOME_NAME = ([System.Security.Cryptography.MD5]::Create().ComputeHash([byte[]][char[]]$distributionUrl) | ForEach-Object {$_.ToString("x2")}) -join ''
+$MAVEN_HOME = "$MAVEN_HOME_PARENT/$MAVEN_HOME_NAME"
+
+if (Test-Path -Path "$MAVEN_HOME" -PathType Container) {
+  Write-Verbose "found existing MAVEN_HOME at $MAVEN_HOME"
+  Write-Output "MVN_CMD=$MAVEN_HOME/bin/$MVN_CMD"
+  exit $?
+}
+
+if (! $distributionUrlNameMain -or ($distributionUrlName -eq $distributionUrlNameMain)) {
+  Write-Error "distributionUrl is not valid, must end with *-bin.zip, but found $distributionUrl"
+}
+
+# prepare tmp dir
+$TMP_DOWNLOAD_DIR_HOLDER = New-TemporaryFile
+$TMP_DOWNLOAD_DIR = New-Item -Itemtype Directory -Path "$TMP_DOWNLOAD_DIR_HOLDER.dir"
+$TMP_DOWNLOAD_DIR_HOLDER.Delete() | Out-Null
+trap {
+  if ($TMP_DOWNLOAD_DIR.Exists) {
+    try { Remove-Item $TMP_DOWNLOAD_DIR -Recurse -Force | Out-Null }
+    catch { Write-Warning "Cannot remove $TMP_DOWNLOAD_DIR" }
+  }
+}
+
+New-Item -Itemtype Directory -Path "$MAVEN_HOME_PARENT" -Force | Out-Null
+
+# Download and Install Apache Maven
+Write-Verbose "Couldn't find MAVEN_HOME, downloading and installing it ..."
+Write-Verbose "Downloading from: $distributionUrl"
+Write-Verbose "Downloading to: $TMP_DOWNLOAD_DIR/$distributionUrlName"
+
+$webclient = New-Object System.Net.WebClient
+if ($env:MVNW_USERNAME -and $env:MVNW_PASSWORD) {
+  $webclient.Credentials = New-Object System.Net.NetworkCredential($env:MVNW_USERNAME, $env:MVNW_PASSWORD)
+}
+[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12
+$webclient.DownloadFile($distributionUrl, "$TMP_DOWNLOAD_DIR/$distributionUrlName") | Out-Null
+
+# If specified, validate the SHA-256 sum of the Maven distribution zip file
+$distributionSha256Sum = (Get-Content -Raw "$scriptDir/.mvn/wrapper/maven-wrapper.properties" | ConvertFrom-StringData).distributionSha256Sum
+if ($distributionSha256Sum) {
+  if ($USE_MVND) {
+    Write-Error "Checksum validation is not supported for maven-mvnd. `nPlease disable validation by removing 'distributionSha256Sum' from your maven-wrapper.properties."
+  }
+  Import-Module $PSHOME\Modules\Microsoft.PowerShell.Utility -Function Get-FileHash
+  if ((Get-FileHash "$TMP_DOWNLOAD_DIR/$distributionUrlName" -Algorithm SHA256).Hash.ToLower() -ne $distributionSha256Sum) {
+    Write-Error "Error: Failed to validate Maven distribution SHA-256, your Maven distribution might be compromised. If you updated your Maven version, you need to update the specified distributionSha256Sum property."
+  }
+}
+
+# unzip and move
+Expand-Archive "$TMP_DOWNLOAD_DIR/$distributionUrlName" -DestinationPath "$TMP_DOWNLOAD_DIR" | Out-Null
+Rename-Item -Path "$TMP_DOWNLOAD_DIR/$distributionUrlNameMain" -NewName $MAVEN_HOME_NAME | Out-Null
+try {
+  Move-Item -Path "$TMP_DOWNLOAD_DIR/$MAVEN_HOME_NAME" -Destination $MAVEN_HOME_PARENT | Out-Null
+} catch {
+  if (! (Test-Path -Path "$MAVEN_HOME" -PathType Container)) {
+    Write-Error "fail to move MAVEN_HOME"
+  }
+} finally {
+  try { Remove-Item $TMP_DOWNLOAD_DIR -Recurse -Force | Out-Null }
+  catch { Write-Warning "Cannot remove $TMP_DOWNLOAD_DIR" }
+}
+
+Write-Output "MVN_CMD=$MAVEN_HOME/bin/$MVN_CMD"

LoginService/pom.xml

+<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>org.springframework.boot</groupId>
+		<artifactId>spring-boot-starter-parent</artifactId>
+		<version>3.2.6</version>
+		<relativePath/> <!-- lookup parent from repository -->
+	</parent>
+	<groupId>com.springpo.loginservice</groupId>
+	<artifactId>LoginService</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<name>LoginService</name>
+	<description>LoginService project for Spring Boot</description>
+	<properties>
+		<java.version>17</java.version>
+	</properties>
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-jpa</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-mongodb</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-validation</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-devtools</artifactId>
+			<scope>runtime</scope>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.projectlombok</groupId>
+			<artifactId>lombok</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>com.springpoc.userservice</groupId>
+			<artifactId>UserService</artifactId>
+			<version>0.0.1-SNAPSHOT</version>
+			<scope>compile</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<version>3.1.0</version>
+		</dependency>
+		<!-- https://mvnrepository.com/artifact/io.projectreactor.kafka/reactor-kafka -->
+		<dependency>
+			<groupId>io.projectreactor.kafka</groupId>
+			<artifactId>reactor-kafka</artifactId>
+			<version>1.3.23</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+				<configuration>
+					<excludes>
+						<exclude>
+							<groupId>org.projectlombok</groupId>
+							<artifactId>lombok</artifactId>
+						</exclude>
+					</excludes>
+				</configuration>
+			</plugin>
+		</plugins>
+	</build>
+
+</project>

LoginService/src/main/java/com/springpo/loginservice/LoginServiceApplication.java

+package com.springpo.loginservice;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class LoginServiceApplication {
+    public static void main(String[] args) {
+        SpringApplication.run(LoginServiceApplication.class, args);
+    }
+}

LoginService/src/main/java/com/springpo/loginservice/config/AppConfig.java

+package com.springpo.loginservice.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.client.RestTemplate;
+
+@Configuration
+public class AppConfig {
+
+    @Bean
+    public RestTemplate restTemplate() {
+        return new RestTemplate();
+    }
+}
\ No newline at end of file

LoginService/src/main/java/com/springpo/loginservice/config/KafkaConsumerConfiguration.java

+package com.springpo.loginservice.config;
+
+import com.springpoc.userservice.model.User;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.*;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@EnableKafka
+@Configuration
+public class KafkaConsumerConfiguration {
+   @Bean
+    public ConsumerFactory<String, String> consumerFactory() {
+        Map<String, Object> config = new HashMap<>();
+
+        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
+        config.put(ConsumerConfig.GROUP_ID_CONFIG, "group_id");
+        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+
+        return new DefaultKafkaConsumerFactory<>(config);
+    }
+
+
+    @Bean
+    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
+        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory();
+        factory.setConsumerFactory(consumerFactory());
+        return factory;
+    }
+
+
+
+    @Bean
+    public ConsumerFactory<String, User> userConsumerFactory() {
+        Map<String, Object> config = new HashMap<>();
+
+        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
+        config.put(ConsumerConfig.GROUP_ID_CONFIG, "group_json");
+        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
+        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+        return new DefaultKafkaConsumerFactory<>(config, new StringDeserializer(),
+                new JsonDeserializer<>(User.class));
+    }
+
+    @Bean
+    public ConcurrentKafkaListenerContainerFactory<String, User> userKafkaListenerFactory() {
+        ConcurrentKafkaListenerContainerFactory<String, User> factory = new ConcurrentKafkaListenerContainerFactory<>();
+        factory.setConsumerFactory(userConsumerFactory());
+        //factory.setMissingTopicsFatal(false);
+        return factory;
+    }
+}

LoginService/src/main/java/com/springpo/loginservice/config/KafkaProducerConfiguration.java

+package com.springpo.loginservice.config;
+
+import com.springpo.loginservice.model.LoginUser;
+import com.springpoc.userservice.model.User;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaProducerConfiguration {
+    @Bean
+    public ProducerFactory producerFactory(){
+        Map<String,Object> config = new HashMap<>();
+
+        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
+        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.springframework.kafka.support.serializer.JsonSerializer.class);
+
+        return new DefaultKafkaProducerFactory<>(config);
+    }
+
+    @Bean
+    public KafkaTemplate<String, LoginUser> kafkaTemplate(){
+        return new KafkaTemplate<>(producerFactory());
+    }
+/*    @Bean
+    public KafkaTemplate<String, String > kafkaTemplate1() {
+        return new KafkaTemplate<>(producerFactory());
+    }*/
+}

LoginService/src/main/java/com/springpo/loginservice/config/ReactiveKafkaConsumerConfig.java

+/*
+package com.springpo.loginservice.config;
+
+import com.springpoc.userservice.model.User;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+import reactor.kafka.receiver.ReceiverOptions;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+
+@Configuration
+public class ReactiveKafkaConsumerConfig {
+
+    @Value("${spring.kafka.bootstrap-servers}")
+    private String bootstrapServers;
+
+   @Value("${spring.kafka.consumer.group-id}")
+    private String groupId;
+    @Value("${spring.kafka.consumer.topic}")
+    private String kafkaTopic;
+
+
+    @Bean
+    public Map<String, Object> consumerConfigs() {
+        Map<String, Object> props = new HashMap<>();
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+       // props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
+        return props;
+    }
+
+    @Bean
+    public ReactiveKafkaConsumerTemplate<String, User> reactiveKafkaConsumerTemplate() {
+        ReceiverOptions<String, User> receiverOptions = ReceiverOptions.<String, User>create(consumerConfigs())
+                .subscription(Collections.singleton(kafkaTopic))
+                .withValueDeserializer(new JsonDeserializer<>(User.class));
+
+        return new ReactiveKafkaConsumerTemplate<>(receiverOptions);
+    }
+
+}
+*/

LoginService/src/main/java/com/springpo/loginservice/config/WebClientConfig.java

+package com.springpo.loginservice.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.reactive.function.client.WebClient;
+
+@Configuration
+public class WebClientConfig {
+    @Bean
+    public WebClient.Builder webClientBuilder() {
+        return WebClient.builder();
+    }
+}
+

LoginService/src/main/java/com/springpo/loginservice/controller/LoginController.java

+package com.springpo.loginservice.controller;
+
+
+import com.springpo.loginservice.dto.LoginOutUserRequest;
+import com.springpo.loginservice.dto.LoginUserRequest;
+import com.springpo.loginservice.model.LoginUser;
+import com.springpo.loginservice.service.LoginService;
+
+import com.springpoc.userservice.model.User;
+import jakarta.validation.Valid;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.*;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.io.UnsupportedEncodingException;
+
+@RestController
+@RequestMapping("/login")
+public class LoginController {
+
+
+    @Autowired
+    LoginService loginService;
+
+    @PostMapping("/validateOtpAndGeneratePass")
+    public Mono<String> validateOtpAndGeneratePass(@RequestBody LoginUserRequest loginUserRequest) {
+        return loginService.validateOtpAndGeneratePass(loginUserRequest);
+    }
+
+  /*  @PostMapping("/createUserEntryLogin")
+    public Void createUserEntryLogin(@Validated @RequestBody User user) {
+         loginService.fetchUserAndSave(user);
+        return null;
+    }*/
+
+    @PostMapping("/createAllUsersEntryLogin")
+    public Flux<ResponseEntity<LoginUser>> createUserEntryLogin(@Validated @RequestBody Flux<User> users) {
+        return loginService.fetchAllUsersAndSave(users)
+                .map(ResponseEntity::ok)
+                .onErrorResume(e -> Flux.just(ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
+                        .body(null))); // or handle error with a proper error response
+
+    }
+
+    @PostMapping("/authenticatePassword")
+    public Mono<ResponseEntity<User>> authenticatePassword(@RequestBody LoginUserRequest loginUserRequest) {
+        return loginService.authenticatePassword(loginUserRequest);
+    }
+
+    @DeleteMapping(path = "/delete/{email}", produces = MediaType.APPLICATION_JSON_VALUE)
+    public Mono<String> deleteByEmailId(@PathVariable String email) {
+        return loginService.deleteByEmailId(email);
+    }
+
+    @PostMapping(path = "/logoutUser", produces = MediaType.APPLICATION_JSON_VALUE)
+    public Mono<String> logoutUser(@RequestBody LoginOutUserRequest LoginOutUserRequest) {
+
+        return loginService.logoutUser(LoginOutUserRequest);
+    }
+}

LoginService/src/main/java/com/springpo/loginservice/dto/LoginDTO.java

+package com.springpo.loginservice.dto;
+
+import com.springpoc.userservice.model.UserAddress;
+import jakarta.persistence.Column;
+import jakarta.persistence.Id;
+import jakarta.validation.constraints.Email;
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.Pattern;
+import lombok.*;
+
+import java.time.LocalDateTime;
+
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+@Setter
+@Getter
+@Data
+public class LoginDTO {
+
+    private String otp;
+    @NotBlank(message = "User password is require")
+    private String password;
+    private String loginFlag;
+    private String logoutFlag;
+    @NotBlank(message = "User first name cannot be blank")
+    private String firstName;
+    @NotBlank(message = "User last name cannot be blank")
+    private String lastName;
+    @NotBlank(message = "Phone number is required")
+    @Pattern(regexp="^\\+(?:[0-9] ?){6,14}[0-9]$", message="Invalid phone number format")
+    private String phNo;
+    @Email(message = "Invalid email format")
+    private String email;
+    @NotBlank(message = "User userType cannot be blank")
+    private String userType;
+    @NotBlank(message = "User status cannot be blank")
+    private String userStatus;
+    @NotBlank(message = "created By User cannot be blank")
+    private  String  createdBy;
+    private String updatedBy;
+
+}

LoginService/src/main/java/com/springpo/loginservice/dto/LoginOutUserRequest.java

+package com.springpo.loginservice.dto;
+
+import jakarta.validation.constraints.NotBlank;
+import lombok.Getter;
+import lombok.Setter;
+import org.bson.types.ObjectId;
+
+@Getter
+@Setter
+public class LoginOutUserRequest {
+
+    private ObjectId id;
+    @NotBlank(message = "User email is required.")
+    private String email;
+    @NotBlank(message = "User password is required.")
+    private String password;
+}

LoginService/src/main/java/com/springpo/loginservice/dto/LoginUserRequest.java

+package com.springpo.loginservice.dto;
+
+import jakarta.validation.constraints.NotBlank;
+import lombok.Getter;
+import lombok.Setter;
+import org.bson.types.ObjectId;
+
+@Getter
+@Setter
+public class LoginUserRequest {
+
+
+    private String email;
+
+    private ObjectId id;
+    @NotBlank(message = "User password is require")
+    private String password;
+
+    private String otp;
+
+}

LoginService/src/main/java/com/springpo/loginservice/exception/CustomException.java

+package com.springpo.loginservice.exception;
+
+public class CustomException extends RuntimeException{
+    public CustomException(String message) {
+        super(message);
+    }
+}

LoginService/src/main/java/com/springpo/loginservice/exception/GlobalExceptionHandler.java

+package com.springpo.loginservice.exception;
+
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.ControllerAdvice;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+import reactor.core.publisher.Mono;
+
+@ControllerAdvice
+public class GlobalExceptionHandler {
+
+    @ExceptionHandler(RuntimeException.class)
+    public Mono<ResponseEntity<String>> handleRuntimeException(RuntimeException e) {
+        return Mono.just(new ResponseEntity<>("A runtime exception occurred: " + e.getMessage(), HttpStatus.INTERNAL_SERVER_ERROR));
+    }
+}

LoginService/src/main/java/com/springpo/loginservice/model/LoginUser.java

+package com.springpo.loginservice.model;
+
+import com.springpoc.userservice.model.UserAddress;
+import jakarta.persistence.*;
+import jakarta.validation.constraints.NotBlank;
+import lombok.*;
+import org.bson.types.ObjectId;
+import org.springframework.data.mongodb.core.mapping.Document;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+@Setter
+@Getter
+@Data
+@Document(value = "login-details")
+public class LoginUser {
+
+    @Id
+   private ObjectId id;
+  /*  private UUID loginId;*/
+    @Column
+    @NotBlank(message = "User otp is require")
+    private String otp;
+    @Column
+    @NotBlank(message = "User password is require")
+    private String password;
+    @Column
+    private Boolean loginFlag;
+    @Column
+    private Boolean logoutFlag;
+    @Column
+    private String phNo;
+    @Column
+    private String email;
+    @Column
+    private String userType;
+    @Column
+    private String userStatus;
+    @Column
+    private String createdBy;
+    @Column
+    private String updatedBy;
+
+
+}

LoginService/src/main/java/com/springpo/loginservice/repo/LoginRepository.java

+package com.springpo.loginservice.repo;
+
+import com.springpo.loginservice.model.LoginUser;
+import com.springpoc.userservice.model.User;
+import org.apache.el.stream.Stream;
+import org.bson.types.ObjectId;
+import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
+import org.springframework.stereotype.Repository;
+import reactor.core.publisher.Mono;
+
+import java.util.List;
+import java.util.UUID;
+
+@Repository
+public interface LoginRepository extends ReactiveMongoRepository<LoginUser, ObjectId> {
+    Mono<LoginUser> findByEmail(String email);
+
+    Mono<String> deleteByEmail(String email);
+}

LoginService/src/main/java/com/springpo/loginservice/service/Impl/LoginServiceImpl.java

+package com.springpo.loginservice.service.Impl;
+
+import com.springpo.loginservice.dto.LoginOutUserRequest;
+import com.springpo.loginservice.dto.LoginUserRequest;
+import com.springpo.loginservice.utils.LoginMapper;
+import com.springpo.loginservice.model.LoginUser;
+import com.springpo.loginservice.repo.LoginRepository;
+import com.springpo.loginservice.service.LoginService;
+import com.springpo.loginservice.utils.ConstantsUtils;
+import com.springpoc.userservice.exception.CustomException;
+import com.springpoc.userservice.model.User;
+import com.springpoc.userservice.utils.UserStatus;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.common.protocol.types.Field;
+import org.bson.types.ObjectId;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.http.ResponseEntity;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.listener.ListenerExecutionFailedException;
+import org.springframework.stereotype.Service;
+import org.springframework.web.client.RestTemplate;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+
+@Service
+@Slf4j
+public class LoginServiceImpl implements LoginService {
+
+    @Autowired
+    LoginRepository loginRepository;
+    @Autowired
+    RestTemplate restTemplate;
+
+    private static final Logger logger = LoggerFactory.getLogger(LoginServiceImpl.class);
+
+    //   private final ReactiveKafkaConsumerTemplate<String, User> kafkaConsumerTemplate;
+/*
+    @Autowired
+    private KafkaTemplate<String, User> kafkaTemplate;
+    private KafkaTemplate<String, String> kafkaTemplate1;
+*/
+
+
+    @Value("${spring.kafka.consumer.topic}")
+    private String kafkaTopic;
+
+    @Value("${spring.kafka.consumer.group-id}")
+    private String groupId;
+
+    private String generatedPassword;
+
+    @Autowired
+    WebClient.Builder webClientBuilder;
+
+/* public LoginServiceImpl(ReactiveKafkaConsumerTemplate<String, User> reactiveKafkaConsumerTemplate) {
+        this.reactiveKafkaConsumerTemplate = reactiveKafkaConsumerTemplate;
+    }*/
+
+    @Override
+    // @Transactional
+    public void fetchUserAndSave(User user) {
+        //  generatedPassword = LoginMapper.createPassword(user);
+        LoginUser loginUsers = LoginMapper.mapUserToLoginUsers(user);
+
+        loginRepository.save(loginUsers)
+                .doOnSuccess(savedLoginUser -> {
+                    String generatedPassword = LoginMapper.createPassword(user);
+                    // LoginMapper.savePasswordInTopic(generatedPassword);
+                    System.out.println("Saved loginUsers in MongoDB: " + savedLoginUser);
+                })
+                .thenReturn("User entry created in login table.")
+                .subscribe();
+
+    }
+
+    @KafkaListener(topics = "${spring.kafka.consumer.topic}", groupId = "${spring.kafka.consumer.group-id}", containerFactory = "userKafkaListenerFactory")
+    private void consumeUser(User user) {
+        log.info("inside consumer" + user);
+        try {
+
+            fetchUserAndSave(user);
+            // Some code that may throw the exception
+        } catch (ListenerExecutionFailedException e) {
+            Throwable cause = e.getCause();
+            System.out.println("Root cause: " + cause.getMessage());
+        }
+
+    }
+
+
+    @Override
+    public Flux<LoginUser> fetchAllUsersAndSave(Flux<User> users) {
+        return users.flatMap(user -> {
+            LoginUser loginUser = LoginMapper.mapUserToLoginUsers(user);
+            return loginRepository.save(loginUser);
+        });
+    }
+
+    public Mono<String> validateOtpAndGeneratePass(LoginUserRequest loginUserRequest) {
+        String otp = loginUserRequest.getOtp();
+        String email = loginUserRequest.getEmail();
+        ObjectId Id = loginUserRequest.getId();
+        log.info("otp :" + loginUserRequest.getOtp());
+        log.info("EMAIL : " + Id);
+        return webClientBuilder.build()
+                .get()
+                .uri(ConstantsUtils.userUrl, email)
+                .retrieve()
+                .bodyToMono(User.class)
+                .flatMap(user -> {
+                    // Check if there's a login entity for the given email
+                    return loginRepository.findByEmail(email)
+                            .flatMap(loginUser -> {
+                                // Check OTP and update password if OTP is valid
+                                if (loginUser.getOtp() != null && loginUser.getOtp().equalsIgnoreCase(ConstantsUtils.otp)) {
+                                    String generatedPassword = LoginMapper.createPassword(user); // Generate secure password
+                                    loginUser.setPassword(generatedPassword); // Set generated password
+                                    loginUser.setOtp(null);
+                                    loginUser.setLoginFlag(true);
+                                    loginUser.setUserStatus(UserStatus.ACTIVE.getUserStatus());
+                                    loginUser.setLogoutFlag(false);
+                                    return loginRepository.save(loginUser)
+                                            .thenReturn(ConstantsUtils.GeneratePassWord); // Save and return password generation success message
+                                } else {
+                                    return Mono.error(new CustomException(ConstantsUtils.InValidOtp)); // Throw error for invalid OTP
+                                }
+                            })
+                            .switchIfEmpty(Mono.error(new CustomException("User not found"))); // Throw error if no login entity found
+                }).doOnError(t -> log.error(ConstantsUtils.InValidOtp))
+                .switchIfEmpty(Mono.error(new CustomException("User not exists")));
+
+    }
+
+    public Mono<ResponseEntity<User>> authenticatePassword(LoginUserRequest loginUserRequest) {
+        String otp = loginUserRequest.getOtp();
+        String email = loginUserRequest.getEmail();
+
+        ObjectId Id = loginUserRequest.getId();
+        String password = loginUserRequest.getPassword();
+        return webClientBuilder.build()
+                .get()
+                .uri(ConstantsUtils.userUrl, email)
+                .retrieve()
+                .bodyToMono(User.class)
+                .flatMap(user -> {
+                    // Fetch the user by email from the repository
+                    return loginRepository.findByEmail(email)
+                            .flatMap(loginUser -> {
+                                // Check if the OTP matches the predefined OTP
+                                if (!loginUser.getPassword().isEmpty() && loginUser.getPassword().equalsIgnoreCase(password)) {
+                                    // Set flags and update password here
+                                    loginUser.setId(Id);
+                                    loginUser.setLoginFlag(Boolean.TRUE);
+                                    loginUser.setLogoutFlag(Boolean.FALSE);
+                                    loginUser.setUserStatus(UserStatus.ACTIVE.getUserStatus());
+                                    // Generate a secure password and update other flags if needed
+                                    // For now, I'm assuming you have the logic to update password securely
+                                    // Save the updated user in the repository
+                                    return loginRepository.save(loginUser)
+                                            .thenReturn(ResponseEntity.ok(user));
+                                } else {
+                                    // Password doesn't match, return an error
+                                    return Mono.error(new CustomException(ConstantsUtils.invalidCredentials));
+                                }
+                            })
+                            .switchIfEmpty(Mono.error(new CustomException(ConstantsUtils.unableCreateLogin)));
+                });
+    }
+
+
+    @Override
+    public Mono<String> deleteByEmailId(String email) {
+        return loginRepository.deleteByEmail(email);
+    }
+
+    @Override
+    public Mono<String> logoutUser(LoginOutUserRequest loginOutUserRequest) {
+        ObjectId Id = loginOutUserRequest.getId();
+        String email = loginOutUserRequest.getEmail();
+        String password = loginOutUserRequest.getPassword();
+        return loginRepository.findByEmail(email)
+                .flatMap(loginUser -> {
+                    // Check if the OTP matches the predefined OTP
+                    if (!loginUser.getPassword().isEmpty() && loginUser.getPassword().equalsIgnoreCase(password)
+                            && loginUser.getLoginFlag() == true ) {
+                        // Set flags and update password here
+                        loginUser.setId(Id);
+                        loginUser.setLoginFlag(Boolean.FALSE);
+                        loginUser.setLogoutFlag(Boolean.TRUE);
+                        loginUser.setUserStatus(UserStatus.INACTIVE.getUserStatus());
+                        // Generate a secure password and update other flags if needed
+                        // For now, I'm assuming you have the logic to update password securely
+                        // Save the updated user in the repository
+                        return   loginRepository.save(loginUser)
+                                 .thenReturn("Logout done successfully.");
+                    } else {
+                        // Password doesn't match, return an error
+                        return Mono.error(new CustomException(ConstantsUtils.invalidCredentials));
+                    }
+                });
+
+    }
+}
+
+
+
+

LoginService/src/main/java/com/springpo/loginservice/service/LoginService.java

+package com.springpo.loginservice.service;
+
+
+
+import com.springpo.loginservice.dto.LoginOutUserRequest;
+import com.springpo.loginservice.dto.LoginUserRequest;
+import com.springpo.loginservice.model.LoginUser;
+import com.springpoc.userservice.model.User;
+import org.springframework.http.ResponseEntity;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.io.UnsupportedEncodingException;
+
+public interface LoginService {
+
+    Mono<String> validateOtpAndGeneratePass(LoginUserRequest loginUserRequest) ;
+
+    void fetchUserAndSave(User user);
+
+    Flux<LoginUser> fetchAllUsersAndSave(Flux<User> users);
+
+    Mono<ResponseEntity<User>> authenticatePassword(LoginUserRequest loginUserRequest);
+
+    Mono<String> deleteByEmailId(String email);
+
+    Mono<String> logoutUser(LoginOutUserRequest loginOutUserRequest);
+}

LoginService/src/main/java/com/springpo/loginservice/utils/ConstantsUtils.java

+package com.springpo.loginservice.utils;
+
+public class ConstantsUtils {
+    public static final String unableCreateLogin = "Unable to create a login.";
+    public static final String FailedFetchDetails = "Failed to fetch user details from user service.";
+    public static final String userCreationFail = "User creation failed.";
+
+    public static final String  notAuthrozedToCreate ="User is not authorized to create.";
+
+    public static final String userUrl  = "http://localhost:9091/user/searchUser/{email}";
+
+    public static final String invalidCredentials  ="Please Enter Valid credentials.";
+    public static final String  otp = "12389";
+
+    public static final  String  GeneratePassWord = "Password generated successfully.";
+
+    public static final  String InValidOtp = "OTP is invalid or otp is already validated.";
+
+    public static final  String InValidPassword = "Please Enter valid password";
+    public static final  String InvalidEmail = "Please Enter valid Email";
+
+}

LoginService/src/main/java/com/springpo/loginservice/utils/LoginMapper.java

+package com.springpo.loginservice.utils;
+
+
+import com.springpo.loginservice.dto.LoginDTO;
+import com.springpo.loginservice.model.LoginUser;
+import com.springpoc.userservice.dto.UserDTO;
+import com.springpoc.userservice.model.User;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+import reactor.core.publisher.Mono;
+
+import java.util.concurrent.CompletableFuture;
+
+public class LoginMapper {
+    @Value("${spring.kafka.login.topic}")
+    private static String kafkaTopic;
+
+    @Value("${spring.kafka.password.topic}")
+    private static String kafkaPasswordTopic;
+
+    @Autowired
+    private static KafkaTemplate<String, LoginUser> kafkaTemplate;
+   /* @Autowired
+    private static KafkaTemplate<String, String> kafkaTemplate1;*/
+
+    public static LoginUser mapUserToLoginUsers(User user){
+       // String generatedPassword = LoginMapper.createPassword(user);
+       return LoginUser.builder()
+                .id(user.getId())
+                .loginFlag(Boolean.FALSE)
+                .logoutFlag(Boolean.FALSE)
+                .otp(ConstantsUtils.otp)
+                .email(user.getEmail())
+                .phNo(user.getPhNo())
+                .userStatus(user.getUserStatus())
+                .userType(user.getUserType())
+                .password(null)
+                .createdBy(user.getCreatedBy())
+                .updatedBy(user.getUpdatedBy())
+                .build();
+    }
+
+    public static Mono<LoginUser> saveUserLoginTopic(LoginUser loginUser) {
+        CompletableFuture<SendResult<String, LoginUser>> completableFuture = kafkaTemplate.send(kafkaTopic, loginUser);
+        return Mono.fromFuture(completableFuture)
+                .map(sendResult -> sendResult.getProducerRecord().value());
+    }
+
+ /*   public static Mono<String> savePasswordInTopic(String password) {
+        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate1.send(kafkaPasswordTopic, password);
+        return Mono.fromFuture(completableFuture)
+                .map(sendResult -> sendResult.getProducerRecord().value());
+    }*/
+    public static String createPassword(User user) {
+        String firstNamePart = user.getFirstName().substring(0, 2);
+        String phoneNumberPart = user.getPhNo().substring(3, 7);
+        String lastNamePart = user.getLastName().substring(user.getLastName().length() - 2);
+        return firstNamePart + phoneNumberPart + lastNamePart;
+    }
+
+}

LoginService/src/main/java/com/springpo/loginservice/utils/Roles.java

+package com.springpo.loginservice.utils;
+
+
+public enum Roles {
+    ADMIN("admin"),
+    RCADMIN("rcadmin"),
+    MERCHANT("merchant");
+
+    private final String role;
+
+    Roles(String role) {
+        this.role = role;
+    }
+
+    public String getRole() {
+        return role;
+    }
+}

LoginService/src/main/resources/application.properties

+spring.application.name=LoginService
+server.port =9093
+#data source configuration
+
+spring.data.mongodb.host=localhost
+spring.data.mongodb.port=27017
+spring.data.mongodb.database=epepocdb
+spring.data.mongodb.url = mongodb://localhost:27017
+spring.autoconfigure.exclude=org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration
+myapp.entry-with-special-characters=@~!%&=""
+
+
+# Kafka broker address
+spring.kafka.bootstrap-servers=localhost:9092
+
+# Consumer specific properties
+spring.kafka.consumer.group-id=user-group
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
+spring.kafka.consumer.auto-offset-reset=earliest
+
+# Kafka topic name
+spring.kafka.consumer.topic=create-users-topic
+spring.kafka.login.topic=create-login-topic
+spring.kafka.password.topic=password-topic
\ No newline at end of file

LoginService/src/test/java/com/springpo/loginservice/LoginServiceApplicationTests.java

+package com.springpo.loginservice;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+
+@SpringBootTest
+class LoginServiceApplicationTests {
+
+	@Test
+	void contextLoads() {
+	}
+
+}

UserService/.gitignore

+HELP.md
+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### VS Code ###
+.vscode/

UserService/.mvn/wrapper/maven-wrapper.properties

+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+wrapperVersion=3.3.1
+distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.6/apache-maven-3.9.6-bin.zip

UserService/mvnw.cmd

+<# : batch portion
+@REM ----------------------------------------------------------------------------
+@REM Licensed to the Apache Software Foundation (ASF) under one
+@REM or more contributor license agreements.  See the NOTICE file
+@REM distributed with this work for additional information
+@REM regarding copyright ownership.  The ASF licenses this file
+@REM to you under the Apache License, Version 2.0 (the
+@REM "License"); you may not use this file except in compliance
+@REM with the License.  You may obtain a copy of the License at
+@REM
+@REM    https://www.apache.org/licenses/LICENSE-2.0
+@REM
+@REM Unless required by applicable law or agreed to in writing,
+@REM software distributed under the License is distributed on an
+@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+@REM KIND, either express or implied.  See the License for the
+@REM specific language governing permissions and limitations
+@REM under the License.
+@REM ----------------------------------------------------------------------------
+
+@REM ----------------------------------------------------------------------------
+@REM Apache Maven Wrapper startup batch script, version 3.3.1
+@REM
+@REM Optional ENV vars
+@REM   MVNW_REPOURL - repo url base for downloading maven distribution
+@REM   MVNW_USERNAME/MVNW_PASSWORD - user and password for downloading maven
+@REM   MVNW_VERBOSE - true: enable verbose log; others: silence the output
+@REM ----------------------------------------------------------------------------
+
+@IF "%__MVNW_ARG0_NAME__%"=="" (SET __MVNW_ARG0_NAME__=%~nx0)
+@SET __MVNW_CMD__=
+@SET __MVNW_ERROR__=
+@SET __MVNW_PSMODULEP_SAVE=%PSModulePath%
+@SET PSModulePath=
+@FOR /F "usebackq tokens=1* delims==" %%A IN (`powershell -noprofile "& {$scriptDir='%~dp0'; $script='%__MVNW_ARG0_NAME__%'; icm -ScriptBlock ([Scriptblock]::Create((Get-Content -Raw '%~f0'))) -NoNewScope}"`) DO @(
+  IF "%%A"=="MVN_CMD" (set __MVNW_CMD__=%%B) ELSE IF "%%B"=="" (echo %%A) ELSE (echo %%A=%%B)
+)
+@SET PSModulePath=%__MVNW_PSMODULEP_SAVE%
+@SET __MVNW_PSMODULEP_SAVE=
+@SET __MVNW_ARG0_NAME__=
+@SET MVNW_USERNAME=
+@SET MVNW_PASSWORD=
+@IF NOT "%__MVNW_CMD__%"=="" (%__MVNW_CMD__% %*)
+@echo Cannot start maven from wrapper >&2 && exit /b 1
+@GOTO :EOF
+: end batch / begin powershell #>
+
+$ErrorActionPreference = "Stop"
+if ($env:MVNW_VERBOSE -eq "true") {
+  $VerbosePreference = "Continue"
+}
+
+# calculate distributionUrl, requires .mvn/wrapper/maven-wrapper.properties
+$distributionUrl = (Get-Content -Raw "$scriptDir/.mvn/wrapper/maven-wrapper.properties" | ConvertFrom-StringData).distributionUrl
+if (!$distributionUrl) {
+  Write-Error "cannot read distributionUrl property in $scriptDir/.mvn/wrapper/maven-wrapper.properties"
+}
+
+switch -wildcard -casesensitive ( $($distributionUrl -replace '^.*/','') ) {
+  "maven-mvnd-*" {
+    $USE_MVND = $true
+    $distributionUrl = $distributionUrl -replace '-bin\.[^.]*$',"-windows-amd64.zip"
+    $MVN_CMD = "mvnd.cmd"
+    break
+  }
+  default {
+    $USE_MVND = $false
+    $MVN_CMD = $script -replace '^mvnw','mvn'
+    break
+  }
+}
+
+# apply MVNW_REPOURL and calculate MAVEN_HOME
+# maven home pattern: ~/.m2/wrapper/dists/{apache-maven-<version>,maven-mvnd-<version>-<platform>}/<hash>
+if ($env:MVNW_REPOURL) {
+  $MVNW_REPO_PATTERN = if ($USE_MVND) { "/org/apache/maven/" } else { "/maven/mvnd/" }
+  $distributionUrl = "$env:MVNW_REPOURL$MVNW_REPO_PATTERN$($distributionUrl -replace '^.*'+$MVNW_REPO_PATTERN,'')"
+}
+$distributionUrlName = $distributionUrl -replace '^.*/',''
+$distributionUrlNameMain = $distributionUrlName -replace '\.[^.]*$','' -replace '-bin$',''
+$MAVEN_HOME_PARENT = "$HOME/.m2/wrapper/dists/$distributionUrlNameMain"
+$MAVEN_HOME_NAME = ([System.Security.Cryptography.MD5]::Create().ComputeHash([byte[]][char[]]$distributionUrl) | ForEach-Object {$_.ToString("x2")}) -join ''
+$MAVEN_HOME = "$MAVEN_HOME_PARENT/$MAVEN_HOME_NAME"
+
+if (Test-Path -Path "$MAVEN_HOME" -PathType Container) {
+  Write-Verbose "found existing MAVEN_HOME at $MAVEN_HOME"
+  Write-Output "MVN_CMD=$MAVEN_HOME/bin/$MVN_CMD"
+  exit $?
+}
+
+if (! $distributionUrlNameMain -or ($distributionUrlName -eq $distributionUrlNameMain)) {
+  Write-Error "distributionUrl is not valid, must end with *-bin.zip, but found $distributionUrl"
+}
+
+# prepare tmp dir
+$TMP_DOWNLOAD_DIR_HOLDER = New-TemporaryFile
+$TMP_DOWNLOAD_DIR = New-Item -Itemtype Directory -Path "$TMP_DOWNLOAD_DIR_HOLDER.dir"
+$TMP_DOWNLOAD_DIR_HOLDER.Delete() | Out-Null
+trap {
+  if ($TMP_DOWNLOAD_DIR.Exists) {
+    try { Remove-Item $TMP_DOWNLOAD_DIR -Recurse -Force | Out-Null }
+    catch { Write-Warning "Cannot remove $TMP_DOWNLOAD_DIR" }
+  }
+}
+
+New-Item -Itemtype Directory -Path "$MAVEN_HOME_PARENT" -Force | Out-Null
+
+# Download and Install Apache Maven
+Write-Verbose "Couldn't find MAVEN_HOME, downloading and installing it ..."
+Write-Verbose "Downloading from: $distributionUrl"
+Write-Verbose "Downloading to: $TMP_DOWNLOAD_DIR/$distributionUrlName"
+
+$webclient = New-Object System.Net.WebClient
+if ($env:MVNW_USERNAME -and $env:MVNW_PASSWORD) {
+  $webclient.Credentials = New-Object System.Net.NetworkCredential($env:MVNW_USERNAME, $env:MVNW_PASSWORD)
+}
+[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12
+$webclient.DownloadFile($distributionUrl, "$TMP_DOWNLOAD_DIR/$distributionUrlName") | Out-Null
+
+# If specified, validate the SHA-256 sum of the Maven distribution zip file
+$distributionSha256Sum = (Get-Content -Raw "$scriptDir/.mvn/wrapper/maven-wrapper.properties" | ConvertFrom-StringData).distributionSha256Sum
+if ($distributionSha256Sum) {
+  if ($USE_MVND) {
+    Write-Error "Checksum validation is not supported for maven-mvnd. `nPlease disable validation by removing 'distributionSha256Sum' from your maven-wrapper.properties."
+  }
+  Import-Module $PSHOME\Modules\Microsoft.PowerShell.Utility -Function Get-FileHash
+  if ((Get-FileHash "$TMP_DOWNLOAD_DIR/$distributionUrlName" -Algorithm SHA256).Hash.ToLower() -ne $distributionSha256Sum) {
+    Write-Error "Error: Failed to validate Maven distribution SHA-256, your Maven distribution might be compromised. If you updated your Maven version, you need to update the specified distributionSha256Sum property."
+  }
+}
+
+# unzip and move
+Expand-Archive "$TMP_DOWNLOAD_DIR/$distributionUrlName" -DestinationPath "$TMP_DOWNLOAD_DIR" | Out-Null
+Rename-Item -Path "$TMP_DOWNLOAD_DIR/$distributionUrlNameMain" -NewName $MAVEN_HOME_NAME | Out-Null
+try {
+  Move-Item -Path "$TMP_DOWNLOAD_DIR/$MAVEN_HOME_NAME" -Destination $MAVEN_HOME_PARENT | Out-Null
+} catch {
+  if (! (Test-Path -Path "$MAVEN_HOME" -PathType Container)) {
+    Write-Error "fail to move MAVEN_HOME"
+  }
+} finally {
+  try { Remove-Item $TMP_DOWNLOAD_DIR -Recurse -Force | Out-Null }
+  catch { Write-Warning "Cannot remove $TMP_DOWNLOAD_DIR" }
+}
+
+Write-Output "MVN_CMD=$MAVEN_HOME/bin/$MVN_CMD"

UserService/pom.xml

+<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>org.springframework.boot</groupId>
+		<artifactId>spring-boot-starter-parent</artifactId>
+		<version>3.2.6</version>
+		<relativePath/> <!-- lookup parent from repository -->
+	</parent>
+	<groupId>com.springpoc.userservice</groupId>
+	<artifactId>UserService</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<name>UserService</name>
+	<description>UserService project for Spring Boot webflux</description>
+	<properties>
+		<java.version>17</java.version>
+	</properties>
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-jpa</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-mongodb</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-validation</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+		</dependency>
+	<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<version>3.1.0</version>
+		</dependency>
+		<!-- https://mvnrepository.com/artifact/io.projectreactor.kafka/reactor-kafka -->
+	<dependency>
+			<groupId>io.projectreactor.kafka</groupId>
+			<artifactId>reactor-kafka</artifactId>
+			<version>1.3.23</version>
+		</dependency>
+	<!--	<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<scope>test</scope>
+		</dependency>-->
+	<!--	<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<scope>test</scope>
+		</dependency>-->
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-devtools</artifactId>
+			<scope>runtime</scope>
+			<optional>true</optional>
+		</dependency>
+
+		<dependency>
+			<groupId>org.projectlombok</groupId>
+			<artifactId>lombok</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor.kafka</groupId>
+			<artifactId>reactor-kafka</artifactId>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+				<configuration>
+					<excludes>
+						<exclude>
+							<groupId>org.projectlombok</groupId>
+							<artifactId>lombok</artifactId>
+						</exclude>
+
+					</excludes>
+				</configuration>
+			</plugin>
+		</plugins>
+	</build>
+
+</project>

UserService/src/main/java/com/springpoc/userservice/UserServiceApplication.java

+package com.springpoc.userservice;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class UserServiceApplication {
+	public static void main(String[] args) {
+		SpringApplication.run(UserServiceApplication.class, args);
+	}
+}

UserService/src/main/java/com/springpoc/userservice/config/AppConfig.java

+package com.springpoc.userservice.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.client.RestTemplate;
+
+
+@Configuration
+public class AppConfig {
+
+  @Bean
+    public RestTemplate restTemplate() {
+        return new RestTemplate();
+    }
+}
\ No newline at end of file

UserService/src/main/java/com/springpoc/userservice/config/KafkaConfiguration.java

+package com.springpoc.userservice.config;
+
+import com.springpoc.userservice.model.User;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaConfiguration {
+
+
+
+  @Bean
+    public ProducerFactory producerFactory(){
+        Map<String,Object> config = new HashMap<>();
+
+        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
+        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.springframework.kafka.support.serializer.JsonSerializer.class);
+
+        return new DefaultKafkaProducerFactory<>(config);
+    }
+
+    @Bean
+    public KafkaTemplate<String, User> kafkaTemplate(){
+        return new KafkaTemplate<>(producerFactory());
+    }
+}

UserService/src/main/java/com/springpoc/userservice/config/ReactiveKafkaProducerConfig.java

+/*
+package com.springpoc.userservice.config;
+
+import com.springpoc.userservice.model.User;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.reactive.ReactiveKafkaProducerTemplate;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+import reactor.kafka.sender.KafkaSender;
+import reactor.kafka.sender.SenderOptions;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class ReactiveKafkaProducerConfig {
+
+    @Bean
+    public ReactiveKafkaProducerTemplate<String, User> reactiveKafkaProducer(KafkaProperties kafkaProperties) {
+
+        Map<String, Object> props = new HashMap<>();
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+        props.putAll(kafkaProperties.getProducer().getProperties());
+
+        SenderOptions<String, User> senderOptions = SenderOptions.create(props);
+        return new ReactiveKafkaProducerTemplate<>(senderOptions);
+    }
+}*/

UserService/src/main/java/com/springpoc/userservice/controller/UserController.java

+package com.springpoc.userservice.controller;
+
+import com.springpoc.userservice.dto.UserDTO;
+import com.springpoc.userservice.model.User;
+
+import com.springpoc.userservice.service.UserService;
+import jakarta.validation.Valid;
+import lombok.extern.slf4j.Slf4j;
+import org.bson.types.ObjectId;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.*;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.List;
+import java.util.UUID;
+
+@RestController
+@RequestMapping("/user")
+@Slf4j
+public class UserController {
+
+    @Autowired
+    UserService userService;
+
+    private static Logger logger = LoggerFactory.getLogger(UserController.class);
+
+    @PostMapping("/createUser")
+    public Mono<ResponseEntity<UserDTO>> createUser(@Validated @RequestBody Mono<User> user) {
+        log.info("inside the controller, createUser method ");
+        return userService.createUser(user);
+    }
+
+    @PostMapping("/createAllUsers")
+    Flux<User> createAllUsers(@Validated @RequestBody Flux<User> userFlux){
+        return userService.createAllUsers(userFlux);
+    }
+    @GetMapping("/searchUser/{email}")
+    Mono<ResponseEntity<User>> searchUser(@Validated @PathVariable("email") String email){
+       return userService.findByEmail(email);
+    }
+
+    @PatchMapping("/updateUser")
+    Mono<ResponseEntity<User>> updateUser(@Validated @RequestBody Mono<User> userMono){
+        return userService.updateUser(userMono);
+    }
+
+    @DeleteMapping(path="/delete/{email}", produces= MediaType.APPLICATION_JSON_VALUE)
+    public Mono<String> deleteByEmailId(@Valid @RequestBody String email) {
+        return userService.deleteByEmailId(email);
+    }
+}

UserService/src/main/java/com/springpoc/userservice/dto/UserDTO.java

+package com.springpoc.userservice.dto;
+
+
+import com.springpoc.userservice.model.UserAddress;
+import jakarta.validation.constraints.Email;
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.Pattern;
+import lombok.*;
+import org.bson.types.ObjectId;
+
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+@Data
+@Getter
+@Setter
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class UserDTO {
+
+    private ObjectId id;
+
+    @NotBlank(message = "User first name cannot be blank")
+    private String firstName;
+    @NotBlank(message = "User last name cannot be blank")
+    private String lastName;
+    @NotBlank(message = "Phone number is required")
+    @Pattern(regexp="^\\+(?:[0-9] ?){6,14}[0-9]$", message="Invalid phone number format")
+    private String phNo;
+    @Email(message = "Invalid email format")
+    private String email;
+    private UserAddress address;
+    @NotBlank(message = "User userType cannot be blank")
+    private String userType;
+    @NotBlank(message = "User status cannot be blank")
+    private String userStatus;
+    @NotBlank(message = "created By User cannot be blank")
+    private  String  createdBy;
+    private LocalDateTime createdTime;
+    private String updatedBy;
+    private LocalDateTime updatedTime;
+}

UserService/src/main/java/com/springpoc/userservice/exception/CustomException.java

+package com.springpoc.userservice.exception;
+
+public class CustomException extends RuntimeException{
+    public CustomException(String message) {
+        super(message);
+    }
+}

UserService/src/main/java/com/springpoc/userservice/exception/GlobalExceptionHandler.java

+package com.springpoc.userservice.exception;
+
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.ControllerAdvice;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+import reactor.core.publisher.Mono;
+
+@ControllerAdvice
+public class GlobalExceptionHandler {
+
+    @ExceptionHandler(RuntimeException.class)
+    public Mono<ResponseEntity<String>> handleRuntimeException(RuntimeException e) {
+        return Mono.just(new ResponseEntity<>("A runtime exception occurred: " + e.getMessage(), HttpStatus.INTERNAL_SERVER_ERROR));
+    }
+}

UserService/src/main/java/com/springpoc/userservice/model/User.java

+package com.springpoc.userservice.model;
+
+import jakarta.persistence.*;
+import jakarta.validation.constraints.Email;
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.Pattern;
+import lombok.*;
+import org.bson.types.ObjectId;
+import org.springframework.data.mongodb.core.index.Indexed;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+@Getter
+@Setter
+@ToString
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+@Document(collection = "user-details")
+public class User {
+
+    @Id
+    private ObjectId id;
+
+    @Column
+    @NotBlank(message = "User first name cannot be blank")
+    private String firstName;
+    @Column
+    @NotBlank(message = "User last name cannot be blank")
+    private String lastName;
+
+    @NotBlank(message = "Phone number is required")
+    @Pattern(regexp = "^\\+(?:[0-9] ?){6,14}[0-9]$", message = "Invalid phone number format")
+    @Column
+    private String phNo;
+
+    @Email(message = "Invalid email format")
+    @NotBlank(message = "Email is required")
+    @Column(unique = true)
+    private String email;
+    @Column
+    private UserAddress address;
+    @Column
+    @NotBlank(message = "User userType cannot be blank")
+    private String userType;
+    @Column
+    @NotBlank(message = "User status cannot be blank")
+    private String userStatus;
+    @Column
+    @NotBlank(message = "created By User cannot be blank")
+    private String createdBy;
+    @Column
+    private LocalDateTime createdTime;
+    @Column
+    private String updatedBy;
+    @Column
+    private LocalDateTime updatedTime;
+}
\ No newline at end of file

UserService/src/main/java/com/springpoc/userservice/model/UserAddress.java

+package com.springpoc.userservice.model;
+
+
+public class UserAddress {
+
+    private String city;
+    private String state;
+    private String postalCode;
+    private String country;
+
+    // Constructors
+    public UserAddress() {
+    }
+
+    public UserAddress(String city, String state, String postalCode, String country) {
+        this.city = city;
+        this.state = state;
+        this.postalCode = postalCode;
+        this.country = country;
+    }
+
+    // Getters and setters
+    public String getCity() {
+        return city;
+    }
+
+    public void setCity(String city) {
+        this.city = city;
+    }
+
+    public String getState() {
+        return state;
+    }
+
+    public void setState(String state) {
+        this.state = state;
+    }
+
+    public String getPostalCode() {
+        return postalCode;
+    }
+
+    public void setPostalCode(String postalCode) {
+        this.postalCode = postalCode;
+    }
+
+    public String getCountry() {
+        return country;
+    }
+
+    public void setCountry(String country) {
+        this.country = country;
+    }
+
+    // toString method for displaying address details
+    @Override
+    public String toString() {
+        return "UserAddress{" +
+                ", city='" + city + '\'' +
+                ", state='" + state + '\'' +
+                ", postalCode='" + postalCode + '\'' +
+                ", country='" + country + '\'' +
+                '}';
+    }
+}

UserService/src/main/java/com/springpoc/userservice/repo/UserRepository.java

+package com.springpoc.userservice.repo;
+
+import com.springpoc.userservice.dto.UserDTO;
+import com.springpoc.userservice.model.User;
+import org.bson.types.ObjectId;
+import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
+import org.springframework.http.ResponseEntity;
+import org.springframework.stereotype.Repository;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.UUID;
+
+@Repository
+public interface UserRepository extends ReactiveMongoRepository<User, ObjectId> {
+
+    Mono<User> findByEmail(String email);
+    Mono<String> deleteByEmail(String email);
+
+    User findByUserType(String userType);
+}

UserService/src/main/java/com/springpoc/userservice/service/UserService.java

+package com.springpoc.userservice.service;
+
+import com.springpoc.userservice.dto.UserDTO;
+import com.springpoc.userservice.model.User;
+import org.springframework.http.ResponseEntity;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public interface UserService {
+
+    Mono<ResponseEntity<UserDTO>> createUser(Mono<User> userMono);
+    Mono<ResponseEntity<User>> findByEmail(String email);
+    Mono<ResponseEntity<User>> updateUser(Mono<User> userDTO);
+
+    Flux<User> createAllUsers(Flux<User> userFlux);
+
+    Mono<String> deleteByEmailId(String email);
+}

UserService/src/main/java/com/springpoc/userservice/service/impl/UserServiceImpl.java

+package com.springpoc.userservice.service.impl;
+
+import com.springpoc.userservice.dto.UserDTO;
+
+import com.springpoc.userservice.exception.CustomException;
+import com.springpoc.userservice.model.User;
+import com.springpoc.userservice.repo.UserRepository;
+import com.springpoc.userservice.service.UserService;
+import com.springpoc.userservice.utils.ConstantsUtils;
+import com.springpoc.userservice.utils.Roles;
+
+import com.springpoc.userservice.utils.UsersMapper;
+import jakarta.transaction.Transactional;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.http.*;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.stereotype.Service;
+
+import org.springframework.web.client.RestTemplate;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.concurrent.CompletableFuture;
+
+@Service
+public class UserServiceImpl implements UserService {
+    Roles roles;
+    private Boolean createUserFlag = false;
+    @Autowired
+    UserRepository userRepository;
+    private final RestTemplate restTemplate;
+    private final Logger log = LoggerFactory.getLogger(UserServiceImpl.class);
+    //private final ReactiveKafkaProducerTemplate<String, User> reactiveKafkaProducerTemplate;
+
+    @Autowired
+    private KafkaTemplate<String, User> kafkaTemplate;
+
+    @Value("${spring.kafka.producer.topic}")
+    private String kafkaTopic;
+
+
+    public UserServiceImpl(RestTemplate restTemplate) {
+        this.restTemplate = restTemplate;
+
+        // this.reactiveKafkaProducerTemplate = reactiveKafkaProducerTemplate;
+        //  this.kafkaTopic = kafkaTopic;
+        //  this.kafkaTopic = kafkaTopic1;
+    }
+
+    @Override
+    public Mono<ResponseEntity<UserDTO>> createUser(Mono<User> userMono) {
+        return userMono
+                .filter(user -> (user.getCreatedBy().equalsIgnoreCase(roles.ADMIN.name())
+                        && user.getUserType().equalsIgnoreCase(roles.RCADMIN.name())) ||
+                        (user.getCreatedBy().equalsIgnoreCase(roles.RCADMIN.name())
+                                && user.getUserType().equalsIgnoreCase(roles.MERCHANT.name())))
+                .switchIfEmpty(Mono.error(new IllegalAccessException(ConstantsUtils.notAuthrozedToCreate)))
+                .flatMap(this::notifyLoginServiceToSaveUser)
+                .flatMap(userRepository::save)
+                .map(UsersMapper::mapToUserDTO)
+                .map(ResponseEntity::ok);
+    }
+
+
+    @Override
+    public Flux<User> createAllUsers(Flux<User> userFlux) {
+        return userFlux
+                .filter(user -> (user.getCreatedBy().equalsIgnoreCase(roles.ADMIN.name())
+                        && user.getUserType().equalsIgnoreCase(roles.RCADMIN.name())) ||
+                        (user.getCreatedBy().equalsIgnoreCase(roles.RCADMIN.name())
+                                && user.getUserType().equalsIgnoreCase(roles.MERCHANT.name())))
+                .switchIfEmpty(Mono.error(new IllegalAccessException(ConstantsUtils.notAuthrozedToCreate)))
+                .flatMap(user -> notifyLoginServiceToSaveUser(user))
+                .collectList()
+                //.flatMapMany(Flux::fromIterable)
+                .flatMapMany(userRepository::saveAll);
+
+    }
+
+
+
+   /* private Mono<User> notifyLoginServiceToSaveUser(User user) {
+        HttpEntity<User> requestEntity = new HttpEntity<>(user);
+        // Make the exchange
+        ResponseEntity<String> response = restTemplate.exchange(ConstantsUtils.saveUrl, HttpMethod.POST, requestEntity, String.class);
+        // Check the response status
+        if (response.getStatusCode().is2xxSuccessful()) {
+            return Mono.just(user);  // Return the original UserDTO
+        } else {
+            return Mono.error(new RuntimeException(ConstantsUtils.failToNotify));
+        }
+    }*/
+
+    @Override
+    public Mono<ResponseEntity<User>> findByEmail(String userId) {
+        return userRepository.findByEmail(userId)
+                .switchIfEmpty(Mono.error(new CustomException(ConstantsUtils.userNotExists)))
+                .map(ResponseEntity::ok);
+    }
+
+    @Override
+    public Mono<ResponseEntity<User>> updateUser(Mono<User> userMono) {
+        return userMono
+                .flatMap(user -> userRepository.findById(user.getId())
+                        .flatMap(existingUser -> {
+                            // Update fields from the new user object
+                            existingUser.setId(user.getId());
+                            existingUser.setFirstName(user.getFirstName());
+                            existingUser.setLastName(user.getLastName());
+                            existingUser.setPhNo(user.getPhNo());
+                            existingUser.setEmail(user.getEmail());
+                            existingUser.setAddress(user.getAddress());
+                            existingUser.setUserType(user.getUserType());
+                            existingUser.setUserStatus(user.getUserStatus());
+                            existingUser.setUpdatedBy(user.getUpdatedBy());
+                            existingUser.setUpdatedTime(user.getUpdatedTime());
+                            // Save the updated user document
+                            return userRepository.save(existingUser)
+                                    .flatMap(existingUser1 -> notifyLoginServiceToSaveUser(existingUser1));
+                        }).
+                        map(ResponseEntity::ok)
+                        .switchIfEmpty(Mono.error(new CustomException(ConstantsUtils.userNotExists))));
+    }
+
+    @Transactional
+    @Override
+    public Mono<String> deleteByEmailId(String email) {
+        return userRepository.findByEmail(email)
+
+                .flatMap(user -> {
+                    // Delete user and related login entries
+                    return userRepository.deleteByEmail(email)
+                            .flatMap(email1 -> notifyLoginServiceToDelete(email1))
+                            //   .then(loginRepository.deleteByEmail(email))
+                            .then(Mono.just("User and login entries deleted successfully"));
+                })
+                .switchIfEmpty(Mono.error(new CustomException(ConstantsUtils.userNotExists)));
+    }
+
+    private Mono<String> notifyLoginServiceToDelete(String email) {
+        HttpEntity<String> requestEntity = new HttpEntity<>(email);
+        // Make the exchange
+        ResponseEntity<String> response = this.restTemplate.exchange(ConstantsUtils.deleteUrl, HttpMethod.DELETE, requestEntity, String.class);
+        // Check the response status
+        if (response.getStatusCode().is2xxSuccessful()) {
+            return Mono.just(ConstantsUtils.userDeletedSuccessfully);  // Return the original UserDTO
+        } else {
+            return Mono.error(new RuntimeException(ConstantsUtils.failToNotify));
+        }
+    }
+    public Mono<User> notifyLoginServiceToSaveUser(User user) {
+        log.info("send to topic={}, {}={},", kafkaTopic);
+        CompletableFuture<SendResult<String, User>> completableFuture = kafkaTemplate.send(kafkaTopic, user);
+        return Mono.fromFuture(completableFuture)
+                .map(sendResult -> sendResult.getProducerRecord().value());
+    }
+}

UserService/src/main/java/com/springpoc/userservice/utils/ConstantsUtils.java

+package com.springpoc.userservice.utils;
+
+public class ConstantsUtils {
+    public static final String userNotExists = "User does not exists.";
+    public static final String userExists = "User is already  exists.";
+    public static final String userCreationFail = "User creation failed.";
+
+    public static final String  notAuthrozedToCreate ="User is not authorized to create.";
+
+    public static final String saveUrl = "http://localhost:9093/login/createUserEntryLogin";
+    public static final String deleteUrl = "http://localhost:9093/login/delete/{email}";
+    public static final String saveAllUrl = "http://localhost:9093/login/createAllUsersEntryLogin";
+
+    public static final String failToNotify = "Failed to notify login service";
+    public static final String userDeletedSuccessfully =    "User deleted successfully";
+
+
+}

UserService/src/main/java/com/springpoc/userservice/utils/Roles.java

+package com.springpoc.userservice.utils;
+
+
+public enum Roles {
+    ADMIN("admin"),
+    RCADMIN("rcadmin"),
+    MERCHANT("merchant");
+
+    private final String role;
+
+    Roles(String role) {
+        this.role = role;
+    }
+
+    public String getRole() {
+        return role;
+    }
+}

UserService/src/main/java/com/springpoc/userservice/utils/UserStatus.java

+package com.springpoc.userservice.utils;
+
+public enum UserStatus {
+    ACTIVE("active"),
+    INACTIVE("inactive"),
+    BLOCKED("blocked");
+
+    private final String userStatus;
+
+
+
+    UserStatus(String userStatus) {
+        this.userStatus = userStatus;
+    }
+
+    public String getUserStatus() {
+        return userStatus;
+    }
+
+}

UserService/src/main/java/com/springpoc/userservice/utils/UsersMapper.java

+package com.springpoc.userservice.utils;
+
+import com.springpoc.userservice.dto.UserDTO;
+import com.springpoc.userservice.model.User;
+
+public class UsersMapper {
+    public static User mapToUser(UserDTO userDTO) {
+        return User.builder()
+                .id(userDTO.getId())
+                .firstName(userDTO.getFirstName())
+                .lastName(userDTO.getLastName())
+                .phNo(userDTO.getPhNo())
+                .email(userDTO.getEmail())
+                .address(userDTO.getAddress())
+                .userType(userDTO.getUserType())
+                .userStatus(UserStatus.INACTIVE.getUserStatus())
+                .createdBy(userDTO.getCreatedBy())
+                .createdTime(userDTO.getCreatedTime())
+                .updatedBy(userDTO.getUpdatedBy())
+                .updatedTime(userDTO.getUpdatedTime())
+                .build();
+    }
+
+    public static UserDTO mapToUserDTO(User user) {
+        return UserDTO.builder()
+               .id(user.getId())
+                .firstName(user.getFirstName())
+                .lastName(user.getLastName())
+                .phNo(user.getPhNo())
+                .email(user.getEmail())
+                .address(user.getAddress())
+                .userType(user.getUserType())
+                .userStatus(UserStatus.INACTIVE.getUserStatus())
+                .createdBy(user.getCreatedBy())
+                .createdTime(user.getCreatedTime())
+                .updatedBy(user.getUpdatedBy())
+                .updatedTime(user.getUpdatedTime())
+                .build();
+    }
+}

UserService/src/main/resources/application.properties

+spring.application.name=userservice
+server.port =9091
+#data source configuration
+
+spring.data.mongodb.host=localhost
+spring.data.mongodb.port=27017
+spring.data.mongodb.database=epepocdb
+spring.data.mongodb.url = mongodb://localhost:27017
+spring.autoconfigure.exclude=org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration
+myapp.entry-with-special-characters=@~!%&=""
+
+
+
+# Kafka Consumer Configuration
+spring.kafka.consumer.bootstrap-servers=localhost:9092
+spring.kafka.consumer.group-id=group_id
+spring.kafka.producer.topic = create-users-topic
+spring.kafka.consumer.auto-offset-reset=earliest
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
+
+# Kafka Consumer Retry Configuration
+spring.kafka.consumer.retry.max-attempts=3
+spring.kafka.consumer.retry.backoff.max-interval=1000
+spring.kafka.consumer.retry.backoff.initial-interval=1000

UserService/src/test/java/com/springpoc/userservice/UserServiceApplicationTests.java

+package com.springpoc.userservice;
+
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+
+@SpringBootTest
+@Disabled
+class UserServiceApplicationTests {
+
+	@Test
+	void contextLoads() {
+	}
+
+}