Added Exception handling code and lombok changes

pom.xml

@@ -62,6 +62,14 @@
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-data-mongodb</artifactId>
 		</dependency>
+
+		<!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
+		<dependency>
+			<groupId>org.projectlombok</groupId>
+			<artifactId>lombok</artifactId>
+			<scope>provided</scope>
+		</dependency>
+
 	</dependencies>
 
 	<build>

src/main/java/com/poc/kafka/mongo/controller/EmployeeController.java

 package com.poc.kafka.mongo.controller;
 
 import java.util.List;
-import java.util.concurrent.ExecutionException;
+import java.util.Optional;
 
 import javax.validation.Valid;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
-import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.DeleteMapping;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -18,6 +17,7 @@ import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
+import com.poc.kafka.mongo.exception.EmployeeException;
 import com.poc.kafka.mongo.model.Emp;
 import com.poc.kafka.mongo.service.EmployeeService;
 
@@ -29,8 +29,9 @@ public class EmployeeController {
 	private EmployeeService employeeService;
 
 	@PostMapping(produces = "application/json", consumes = "application/json", path = "/produce")
-	public void publishEmployee(@Valid @RequestBody Emp emp) throws InterruptedException, ExecutionException {
+	public ResponseEntity<String> publishEmployee(@Valid @RequestBody Emp emp)  {
 		employeeService.send(emp);
+		return new ResponseEntity<>("Employee details published", HttpStatus.OK);
 	}
 
 	@GetMapping(produces = "application/json", consumes = "application/json", path = "/empId/{empId}")

src/main/java/com/poc/kafka/mongo/exception/EmployeeException.java

+package com.poc.kafka.mongo.exception;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
+
+@Setter
+@Getter
+@AllArgsConstructor
+public class EmployeeException extends RuntimeException {
+
+	private String errorCode;
+	private String errorDescription;
+	private String message;
+	private static final long serialVersionUID = 1L;
+
+	public EmployeeException(String exception) {
+		super(exception);
+	}
+
+	public EmployeeException(String errorCode, String errorDescription) {
+		super();
+		this.errorCode = errorCode;
+		this.errorDescription = errorDescription;
+	}
+
+}

src/main/java/com/poc/kafka/mongo/exception/GlobalExceptionHandler.java

+package com.poc.kafka.mongo.exception;
+
+import java.util.Date;
+
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.validation.FieldError;
+import org.springframework.web.HttpMediaTypeNotSupportedException;
+import org.springframework.web.HttpRequestMethodNotSupportedException;
+import org.springframework.web.bind.MethodArgumentNotValidException;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+import org.springframework.web.bind.annotation.RestControllerAdvice;
+import org.springframework.web.context.request.WebRequest;
+import org.springframework.web.multipart.support.MissingServletRequestPartException;
+import org.springframework.web.servlet.NoHandlerFoundException;
+import org.springframework.web.servlet.mvc.method.annotation.ResponseEntityExceptionHandler;
+
+import com.poc.kafka.mongo.model.ErrorDetails;
+import com.poc.kafka.mongo.util.ErrorEnum;
+
+@RestControllerAdvice
+public class GlobalExceptionHandler extends ResponseEntityExceptionHandler {
+
+	@ExceptionHandler(EmployeeException.class)
+	public final ResponseEntity<ErrorDetails> handleUserNotFoundException(EmployeeException ex, WebRequest request) {
+		ErrorDetails error = new ErrorDetails(new Date(), ex.getMessage(), request.getDescription(true),ex.getErrorDescription(),ex.getErrorCode());
+		if(error.getErrorCode().equalsIgnoreCase(ErrorEnum.EC_2.name())) {
+			return new ResponseEntity<>(error, HttpStatus.INTERNAL_SERVER_ERROR);
+		}else {
+			return new ResponseEntity<>(error, HttpStatus.NOT_FOUND);
+		}
+		
+	}
+
+	@Override
+	protected ResponseEntity<Object> handleHttpRequestMethodNotSupported(HttpRequestMethodNotSupportedException ex,
+			HttpHeaders headers, HttpStatus status, WebRequest request) {
+
+		ErrorDetails error = new ErrorDetails(new Date(), ex.getMessage(), request.getDescription(true));
+		return new ResponseEntity<>(error, HttpStatus.METHOD_NOT_ALLOWED);
+	}
+
+	@Override
+	protected ResponseEntity<Object> handleHttpMediaTypeNotSupported(HttpMediaTypeNotSupportedException ex,
+			HttpHeaders headers, HttpStatus status, WebRequest request) {
+		ErrorDetails error = new ErrorDetails(new Date(), ex.getMessage(), request.getDescription(true));
+		return new ResponseEntity<>(error, HttpStatus.UNSUPPORTED_MEDIA_TYPE);
+	}
+
+	@Override
+	protected ResponseEntity<Object> handleNoHandlerFoundException(NoHandlerFoundException ex, HttpHeaders headers,
+			HttpStatus status, WebRequest request) {
+		ErrorDetails error = new ErrorDetails(new Date(), ex.getMessage(), request.getDescription(true));
+		return new ResponseEntity<>(error, HttpStatus.NOT_FOUND);
+	}
+
+	@Override
+	protected ResponseEntity<Object> handleMissingServletRequestPart(MissingServletRequestPartException ex,
+			HttpHeaders headers, HttpStatus status, WebRequest request) {
+		ErrorDetails error = new ErrorDetails(new Date(), ex.getMessage(), request.getDescription(true));
+		return new ResponseEntity<>(error, HttpStatus.BAD_REQUEST);
+	}
+
+	/**
+	 * For handling spring validation errors
+	 * 
+	 */
+	protected ResponseEntity<Object> handleMethodArgumentNotValid(MethodArgumentNotValidException ex,
+			HttpHeaders headers, HttpStatus status, WebRequest request) {
+		FieldError ers = ex.getBindingResult().getFieldError();
+		ErrorDetails error = new ErrorDetails(new Date(), ers.getDefaultMessage(), request.getDescription(true));
+		return new ResponseEntity<>(error, HttpStatus.BAD_REQUEST);
+	}
+
+}

src/main/java/com/poc/kafka/mongo/model/ErrorDetails.java

+package com.poc.kafka.mongo.model;
+
+import java.util.Date;
+
+import org.springframework.lang.Nullable;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class ErrorDetails {
+	private Date timestamp;
+	private String message;
+	private String details;
+	private String errorDescription;
+	@Nullable private String errorCode;
+	
+	public ErrorDetails(Date timestamp, String message, String details){
+		this.timestamp=timestamp;
+		this.message=message;
+		this.details=details;
+	}
+
+}
\ No newline at end of file

src/main/java/com/poc/kafka/mongo/service/EmployeeService.java

@@ -5,8 +5,16 @@ import java.util.List;
 import com.poc.kafka.mongo.model.Emp;
 
 public interface EmployeeService {
-
+	/**
+	 * 
+	 * @param e
+	 */
 	public void send(Emp e);
+
+	/**
+	 * 
+	 * @param message
+	 */
 	public void receive(String message);
 
 	/**
@@ -15,34 +23,40 @@ public interface EmployeeService {
 	 * @return
 	 */
 	public Emp findByEmpId(Long empId);
+
 	/**
 	 * 
 	 * @param id
 	 * @return
 	 */
 	public Emp findById(String id);
+
 	/**
 	 * 
 	 * @return
 	 */
 	public List<Emp> findAll();
+
 	/**
 	 * 
 	 * @param emp
 	 * @return
 	 */
 	public Emp save(Emp emp);
+
 	/**
 	 * 
 	 * @param emp
 	 */
 	public void saveOrUpdate(Emp emp);
+
 	/**
 	 * 
 	 * @param sal
 	 * @return
 	 */
 	public List<Emp> findBySalary(Double sal);
+
 	/**
 	 * 
 	 * @param empId
@@ -50,5 +64,4 @@ public interface EmployeeService {
 	 */
 	public Long deleteByEmpId(Long empId);
 
-
 }

src/main/java/com/poc/kafka/mongo/service/EmployeeServiceImpl.java

@@ -17,10 +17,13 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
 
 import com.google.gson.Gson;
 import com.poc.kafka.mongo.entity.Employee;
+import com.poc.kafka.mongo.exception.EmployeeException;
 import com.poc.kafka.mongo.mapper.EmployeeMapper;
 import com.poc.kafka.mongo.model.Emp;
 import com.poc.kafka.mongo.repository.EmployeeMongoRepository;
 import com.poc.kafka.mongo.repository.EmployeeRepository;
+import com.poc.kafka.mongo.util.ErrorEnum;
+
 @Service
 public class EmployeeServiceImpl implements EmployeeService {
 	private static final Logger logger = LoggerFactory.getLogger(EmployeeServiceImpl.class);
@@ -39,7 +42,6 @@ public class EmployeeServiceImpl implements EmployeeService {
 	@Autowired
 	private EmployeeRepository employeeRepository;
 
-
 	@Override
 	public void send(Emp emp) {
 
@@ -49,6 +51,8 @@ public class EmployeeServiceImpl implements EmployeeService {
 			@Override
 			public void onFailure(Throwable ex) {
 				logger.info("Unable to send message=[ {} ] due to : {}", request, ex.getMessage());
+				throw new EmployeeException(ErrorEnum.EC_2.name(), ErrorEnum.getDesription(ErrorEnum.EC_2.name()),
+						ex.getMessage());
 			}
 
 			@Override
@@ -61,65 +65,123 @@ public class EmployeeServiceImpl implements EmployeeService {
 	@Override
 	@KafkaListener(topics = { "${spring.kafka.topics}" })
 	public void receive(String emp) {
-		logger.info("Kafka event consumed is:{} " , emp);
+		logger.info("Kafka event consumed is:{} ", emp);
 		Emp e = gson.fromJson(emp, Emp.class);
-		logger.info("converted value:{} " , e.toString());
+		logger.info("converted value:{} ", e.toString());
 		/**
 		 * To send to mongo database after listening the record from kafka
 		 */
 		saveOrUpdate(e);
 	}
-	
+
 	@Override
 	public Emp findByEmpId(Long empId) {
-		return (Emp)employeeMapper.mapEntityToPojo(employeeMongoRepository.findByEmpId(empId));
+		try {
+			Object e = employeeMapper.mapEntityToPojo(employeeMongoRepository.findByEmpId(empId));
+			if (e == null) {
+				throw new EmployeeException(ErrorEnum.EC_1.name(), ErrorEnum.getDesription(ErrorEnum.EC_1.name()));
+			}
+			return (Emp) e;
+		} catch (EmployeeException e) {
+			throw e;
+		} catch (Exception e) {
+			throw new EmployeeException(ErrorEnum.EC_2.name(), ErrorEnum.getDesription(ErrorEnum.EC_2.name()),
+					e.getMessage());
+		}
 	}
 
 	@Override
 	public Emp findById(String id) {
-		return (Emp)employeeMapper.mapEntityToPojo(employeeMongoRepository.findById(id).orElse(new Employee()));
+		try {
+			Object e = employeeMapper.mapEntityToPojo(employeeMongoRepository.findById(id));
+			if (e == null) {
+				throw new EmployeeException(ErrorEnum.EC_1.name(), ErrorEnum.getDesription(ErrorEnum.EC_1.name()));
+			}
+			return (Emp) e;
+		} catch (EmployeeException e) {
+			throw e;
+		} catch (Exception e) {
+			throw new EmployeeException(ErrorEnum.EC_2.name(), ErrorEnum.getDesription(ErrorEnum.EC_2.name()),
+					e.getMessage());
+		}
 	}
 
 	@Override
 	public List<Emp> findAll() {
-		List<Employee> list= employeeMongoRepository.findAll();
-		List<Emp> empList= list.stream()
-				.map(e->employeeMapper.mapEntityToPojo(e))
-				.map(e->(Emp)e)
-				.collect(Collectors.toList());
-		return empList;
+		List<Employee> list = employeeMongoRepository.findAll();
+		try {
+			if (list.isEmpty()) {
+				throw new EmployeeException(ErrorEnum.EC_1.name(), ErrorEnum.getDesription(ErrorEnum.EC_1.name()));
+			}
+			return list.stream().map(e -> employeeMapper.mapEntityToPojo(e)).map(e -> (Emp) e)
+					.collect(Collectors.toList());
+		} catch (EmployeeException e) {
+			throw e;
+		} catch (Exception e) {
+			throw new EmployeeException(ErrorEnum.EC_2.name(), ErrorEnum.getDesription(ErrorEnum.EC_2.name()),
+					e.getMessage());
+		}
+
 	}
 
 	@Override
 	public Emp save(Emp emp) {
 		Employee employee = (Employee) employeeMapper.mapPojoToEntity(emp);
-		return (Emp)employeeMapper.mapEntityToPojo(employeeMongoRepository.save(employee));
+		try {
+			return (Emp) employeeMapper.mapEntityToPojo(employeeMongoRepository.save(employee));
+		} catch (Exception e) {
+			throw new EmployeeException(ErrorEnum.EC_2.name(), ErrorEnum.getDesription(ErrorEnum.EC_2.name()),
+					e.getMessage());
+		}
+
 	}
 
 	@Override
 	public void saveOrUpdate(Emp emp) {
 		Employee employee = (Employee) employeeMapper.mapPojoToEntity(emp);
-		employeeRepository.saveOrUpdate(employee);
+		try {
+			employeeRepository.saveOrUpdate(employee);
+		} catch (Exception e) {
+			throw new EmployeeException(ErrorEnum.EC_2.name(), ErrorEnum.getDesription(ErrorEnum.EC_2.name()),
+					e.getMessage());
+		}
 
 	}
 
 	@Override
 	public List<Emp> findBySalary(Double sal) {
-		List<Emp> empList= null;
+		List<Emp> empList = null;
 		try (Stream<Employee> stream = employeeMongoRepository.findBySalary(sal)) {
 			List<Employee> list = stream.collect(Collectors.toList());
-			empList= list.stream()
-					.map(e->(Emp)employeeMapper.mapEntityToPojo(e))
-					.collect(Collectors.toList());
+			if (list.isEmpty()) {
+				throw new EmployeeException(ErrorEnum.EC_1.name(), ErrorEnum.getDesription(ErrorEnum.EC_1.name()));
+			}
+			empList = list.stream().map(e -> (Emp) employeeMapper.mapEntityToPojo(e)).collect(Collectors.toList());
+		} catch (EmployeeException e) {
+			throw e;
+		} catch (Exception e) {
+			throw new EmployeeException(ErrorEnum.EC_2.name(), ErrorEnum.getDesription(ErrorEnum.EC_2.name()),
+					e.getMessage());
 		}
 		return empList;
 	}
 
 	@Override
 	public Long deleteByEmpId(Long empId) {
-		return employeeMongoRepository.deleteByEmployeeId(empId);
-	}
-
+		try {
+			Long id = employeeMongoRepository.deleteByEmployeeId(empId);
+			if (id == 0) {
+				throw new EmployeeException(ErrorEnum.EC_1.name(), ErrorEnum.getDesription(ErrorEnum.EC_1.name()));
+			}
+			logger.info("Deleted emp id:{}", id);
+			return id;
+		} catch (EmployeeException e) {
+			throw e;
+		} catch (Exception e) {
+			throw new EmployeeException(ErrorEnum.EC_2.name(), ErrorEnum.getDesription(ErrorEnum.EC_2.name()),
+					e.getMessage());
+		}
 
+	}
 
 }

src/main/java/com/poc/kafka/mongo/util/ErrorEnum.java

+package com.poc.kafka.mongo.util;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public enum ErrorEnum {
+	EC_1("Record not found"), EC_2("Unexpected Exception");
+
+	private String errDes;
+
+	private ErrorEnum(String s) {
+
+		errDes = s;
+	}
+
+	private static final Map<String, String> BY_LABEL = new HashMap<>();
+
+	static {
+		for (ErrorEnum e : values()) {
+			BY_LABEL.put(e.name(), e.errDes);
+		}
+	}
+	
+	public static String getDesription(String name) {
+		return BY_LABEL.get(name);
+	}
+}