import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

import javax.persistence.CascadeType;
import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.FetchType;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;
import javax.persistence.JoinColumn;
import javax.persistence.OneToMany;
import javax.validation.constraints.NotBlank;

import org.hibernate.envers.AuditOverride;
import org.hibernate.envers.Audited;
import org.hibernate.envers.NotAudited;

import com.fasterxml.jackson.annotation.JsonManagedReference;

@Entity(name = "PRJ_CUSTOMER")
@Audited
@AuditOverride(forClass = Auditable.class)
public class Customer extends Auditable implements Serializable {

	private static final long serialVersionUID = 1L;

	@Id
	@Column(name = "ID")
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

	@Column(name = "FIRST_NAME")
    private String firstName;

	@Column(name = "LAST_NAME")
    private String lastName;

	@NotBlank(message = "{email.not_empty}")
	@Column(name = "EMAIL_ADDRESS")
    private String emailAddress;

    @OneToMany(fetch = FetchType.EAGER, orphanRemoval = true, targetEntity = Address.class, cascade = CascadeType.ALL)
    @JoinColumn(name = "CUSTOMER_ID")
    @JsonManagedReference
    @NotAudited
    private List<Address> addresses;

	public Long getId() {
		return id;
	}

	public void setId(Long id) {
		this.id = id;
	}

	public String getFirstName() {
		return firstName;
	}

	public void setFirstName(String firstName) {
		this.firstName = firstName;
	}

	public String getLastName() {
		return lastName;
	}

	public void setLastName(String lastName) {
		this.lastName = lastName;
	}

	public String getEmailAddress() {
		return emailAddress;
	}

	public void setEmailAddress(String emailAddress) {
		this.emailAddress = emailAddress;
	}

	public List<Address> getAddresses() {
		return addresses;
	}

	public void setAddresses(List<Address> addresses) {
		if (this.addresses == null) {
			this.addresses = new ArrayList<>();
		} else {
			this.addresses.clear();
		}
		this.addresses.addAll(addresses);
	}

	@Override
	public String toString() {
		return "Customer [id=" + id + ", firstName=" + firstName + ", lastName=" + lastName + ", emailAddress="
				+ emailAddress + ", addresses=" + addresses + "]";
	}
}

import java.io.Serializable;

import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;
import javax.persistence.ManyToOne;

import com.fasterxml.jackson.annotation.JsonBackReference;

@Entity(name = "PRJ_ADDRESS")
public class Address extends Auditable implements Serializable {

	private static final long serialVersionUID = 1L;

	@Id
	@Column(name = "ID")
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
	
	@Column(name = "STREET_ADDRESS")
    private String streetAddress;
	@Column(name = "CITY")
    private String city;
	@Column(name = "STATE_CODE")
    private String stateCode;
	@Column(name = "COUNTRY")
    private  String country;
	@Column(name = "ZIP_CODE")
    private String zipCode;

    @ManyToOne
    @JsonBackReference
    private Customer customer;

	public Long getId() {
		return id;
	}

	public void setId(Long id) {
		this.id = id;
	}

	public String getStreetAddress() {
		return streetAddress;
	}

	public void setStreetAddress(String streetAddress) {
		this.streetAddress = streetAddress;
	}

	public String getCity() {
		return city;
	}

	public void setCity(String city) {
		this.city = city;
	}

	public String getStateCode() {
		return stateCode;
	}

	public void setStateCode(String stateCode) {
		this.stateCode = stateCode;
	}

	public String getCountry() {
		return country;
	}

	public void setCountry(String country) {
		this.country = country;
	}

	public String getZipCode() {
		return zipCode;
	}

	public void setZipCode(String zipCode) {
		this.zipCode = zipCode;
	}

	public Customer getCustomer() {
		return customer;
	}

	public void setCustomer(Customer customer) {
		this.customer = customer;
	}

	@Override
	public String toString() {
		return "Address [id=" + id + ", streetAddress=" + streetAddress + ", city=" + city + ", stateCode=" + stateCode
				+ ", country=" + country + ", zipCode=" + zipCode + ", customer=" + customer + "]";
	}
}

import java.util.List;
import java.util.Optional;

import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.repository.history.RevisionRepository;
import org.springframework.stereotype.Repository;

import com.org.lob.project.repository.entity.Customer;

@Repository
public interface CustomerRepository extends JpaRepository<Customer, Long> {

    //Contains search on either firstname or lastname
    List<Customer> findAllByFirstNameContainingOrLastNameContaining(String firstName, String lastName);

    Optional<Customer> findCustomerByEmailAddress(String email);
}

import java.util.List;
import java.util.Optional;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.dao.EmptyResultDataAccessException;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.stereotype.Service;

import com.org.lob.project.repository.CustomerRepository;
import com.org.lob.project.repository.entity.Customer;

@Service
public class CustomerService {

	private static final Logger LOGGER = LoggerFactory.getLogger(CustomerService.class);

	private CustomerRepository customerRepository;

	public CustomerService(CustomerRepository customerRepository) {
		this.customerRepository = customerRepository;
	}

	public Optional<Customer> getCustomerById(Long customerId) {
		LOGGER.debug("Fetching customer by id: {}", customerId);
		return customerRepository.findById(customerId);
	}

	public Customer create(Customer customer) {
		try {
			LOGGER.debug("Creating a new customer with emailAddress: {}", customer.getEmailAddress());
			return customerRepository.save(customer);
		} catch (DataIntegrityViolationException e) {
			LOGGER.error("Customer already exists with emailAddress: {}", customer.getEmailAddress());
			throw new RuntimeException("Customer already exists with same emailAddress");
		}
	}

	public Customer update(Customer customer) {
		LOGGER.debug("Updating a customer with id: {}", customer.getId());
		Optional<Customer> optionalCustomer = customerRepository.findById(customer.getId());
		if (optionalCustomer.isEmpty()) {
			LOGGER.error("Unable to update customer by id {}", customer.getId());
			throw new RuntimeException("Customer does not exists");
		}
		Customer existingCustomer = optionalCustomer.get();
		existingCustomer.setAddresses(customer.getAddresses());
		existingCustomer.setFirstName(customer.getFirstName());
		existingCustomer.setLastName(customer.getLastName());
		return customerRepository.save(existingCustomer);
	}

	public List<Customer> findByName(String name) {
		return customerRepository.findAllByFirstNameContainingOrLastNameContaining(name, name);
	}

	public Optional<Customer> findByEmail(String email) {
		return customerRepository.findCustomerByEmailAddress(email);
	}

	// Paging implementation of findAll
	public Page<Customer> findAll(Pageable pageable) {
		return customerRepository.findAll(pageable);
	}

	public void deleteCustomer(Long customerId) {
		try {
			customerRepository.deleteById(customerId);
		} catch (EmptyResultDataAccessException e) {
			LOGGER.error("Unable to delete customer by id {}", customerId);
			throw new RuntimeException("Customer does not exists");
		}
	}
}

import static com.org.lob.support.Constants.PATH_VARIABLE_ID;
import static com.org.lob.support.Constants.REQUEST_MAPPING_CUSTOMER;
import static com.org.lob.support.Constants.REQUEST_PARAM_PAGE_NUMBER;

import java.util.Optional;

import javax.validation.Valid;
import javax.validation.constraints.NotBlank;
import javax.validation.constraints.Positive;

import org.hibernate.validator.constraints.Length;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.util.UriComponentsBuilder;

import com.org.lob.project.api.model.ErrorMessage;
import com.org.lob.project.repository.entity.Customer;
import com.org.lob.project.service.CustomerService;

@RestController
@RequestMapping(REQUEST_MAPPING_CUSTOMER)
public class CustomerApi {

	private CustomerService customerService;

	public CustomerApi(CustomerService customerService) {
		this.customerService = customerService;
	}

	@GetMapping(path = "/{id}", produces = MediaType.APPLICATION_JSON_VALUE)
	public ResponseEntity<?> getCustomerDetail(
			@PathVariable(name = PATH_VARIABLE_ID) @Length(min = 1) @Positive Long customerId) {
		Optional<Customer> customer = getCustomerById(customerId);
		if (customer.isEmpty()) {
			return new ResponseEntity<Void>(HttpStatus.NOT_FOUND);
		}
		return ResponseEntity.ok(customer.get());
	}

	@GetMapping(path = "/", produces = MediaType.APPLICATION_JSON_VALUE)
	public ResponseEntity<?> getAllCustomers(
			@RequestParam(name = REQUEST_PARAM_PAGE_NUMBER, required = true) @NotBlank(message = "{page_number.not_empty}") @Positive Integer pageNumber,
			@RequestParam(name = REQUEST_PARAM_PAGE_NUMBER, required = true) @Positive Integer pageSize) {
		try {
			Page<Customer> page = getCustomersPage(pageNumber, pageSize);
			return ResponseEntity.ok(page.getContent());
		} catch (Exception ex) {
			return handleException(ex);
		}
	}

	private Page<Customer> getCustomersPage(Integer pageNumber, Integer pageSize) {
		PageRequest pageRequest = PageRequest.of(pageNumber, pageSize);
		Page<Customer> page = customerService.findAll(pageRequest);
		return page;
	}

	@PostMapping(path = "/", produces = MediaType.APPLICATION_JSON_VALUE, consumes = MediaType.APPLICATION_JSON_VALUE)
	public ResponseEntity<?> createCustomer(@Valid @RequestBody Customer customer, UriComponentsBuilder ucBuilder) {
		try {

			if (customer.getId() != null) {
				return new ResponseEntity<Void>(HttpStatus.CONFLICT);
			}

			Customer createdCustomer = customerService.create(customer);

			return ResponseEntity
					.created(ucBuilder.path(REQUEST_MAPPING_CUSTOMER).buildAndExpand(createdCustomer.getId()).toUri())
					.body(createdCustomer);
		} catch (Exception ex) {
			return handleException(ex);
		}
	}

	@PutMapping(path = "/{id}", produces = MediaType.APPLICATION_JSON_VALUE, consumes = MediaType.APPLICATION_JSON_VALUE)
	public ResponseEntity<?> updateCustomer(
			@PathVariable(name = PATH_VARIABLE_ID) @NotBlank(message = "{id.not_empty}") @Length(min = 1) @Positive Long customerId,
			@RequestBody Customer customer) {
		try {
			Optional<Customer> customerOptional = getCustomerById(customerId);
			if (customerOptional.isEmpty()) {
				return new ResponseEntity<Void>(HttpStatus.NOT_FOUND);
			}
			customer.setId(customerId);
			Customer updatedCustomer = customerService.update(customer);
			return ResponseEntity.ok(updatedCustomer);
		} catch (Exception ex) {
			return handleException(ex);
		}
	}

	@DeleteMapping(path = "/{id}")
	public ResponseEntity<Void> deleteCustomer(
			@PathVariable(name = PATH_VARIABLE_ID) @NotBlank(message = "{id.not_empty}") @Length(min = 1) @Positive Long customerId) {
		Optional<Customer> customer = getCustomerById(customerId);
		if (customer.isEmpty()) {
			return new ResponseEntity<Void>(HttpStatus.NOT_FOUND);
		}
		customerService.deleteCustomer(null);
		return new ResponseEntity<Void>(HttpStatus.OK);
	}

	private Optional<Customer> getCustomerById(Long customerId) {
		return customerService.getCustomerById(customerId);
	}

	private ResponseEntity<ErrorMessage> handleException(Exception ex) {
		ex.printStackTrace();
		ErrorMessage error = new ErrorMessage(HttpStatus.BAD_REQUEST.value(), ex.getMessage());
		return ResponseEntity.badRequest().body(error);
	}
}

		<dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-oxm</artifactId>
        </dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-batch</artifactId>
		</dependency>

import java.net.MalformedURLException;

import javax.sql.DataSource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.configuration.annotation.BatchConfigurer;
import org.springframework.batch.core.configuration.annotation.DefaultBatchConfigurer;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.adapter.ItemWriterAdapter;
import org.springframework.batch.item.xml.StaxEventItemReader;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.UrlResource;
import org.springframework.oxm.jaxb.Jaxb2Marshaller;

import com.org.lob.project.batch.CustomerProcessor;
import com.org.lob.project.batch.model.CustomerData;
import com.org.lob.project.repository.entity.Customer;
import com.org.lob.project.service.CustomerService;
import com.org.lob.support.batch.LoggingJobExecutionListener;
import com.org.lob.support.batch.LoggingStepExecutionListener;

@Configuration
@EnableBatchProcessing
public class BatchConfig {

	private static final Logger LOGGER = LoggerFactory.getLogger(BatchConfig.class);

	private JobBuilderFactory jobBuilderFactory;	
	private StepBuilderFactory stepBuilderFactory;	

	BatchConfig(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
		this.jobBuilderFactory = jobBuilderFactory;
		this.stepBuilderFactory = stepBuilderFactory;
	}

	@Bean
	JobExecutionListener loggingJobExecutionListener() {
		return new LoggingJobExecutionListener();
	}

	@Bean
	StepExecutionListener loggingStepExecutionListener() {
		return new LoggingStepExecutionListener();
	}

	@Bean
	BatchConfigurer batchConfigurer(DataSource dataSource) {
		return new DefaultBatchConfigurer(dataSource);
	}

	@Bean
	Job processJob(Step step1, @Value("${app.batch_process.job.name}") String jobName, JobExecutionListener executionListener) {
		return jobBuilderFactory.get(jobName)
				.incrementer(new RunIdIncrementer())
				.listener(executionListener)
				.flow(step1)
				.end()
				.build();
	}

	@Bean
	Step step1(@Value("${app.batch_process.step1.name}") String stepName, StaxEventItemReader<CustomerData> reader, CustomerProcessor processor, ItemWriterAdapter<Customer> writer, StepExecutionListener stepExecutionListener) {
		return stepBuilderFactory.get(stepName)
				.listener(stepExecutionListener)
				.<CustomerData, Customer>chunk(10)
					.reader(reader)
					.processor(processor)
					.faultTolerant()
					//.skipPolicy(skip())
					.writer(writer)
				.build();
	}

	@Bean
	@StepScope
	StaxEventItemReader<CustomerData> customerDataReader(@Value("#{jobParameters['fileName']}") String file) throws MalformedURLException {

		LOGGER.info("StaxEventItemReader:fileName: {}", file);

		StaxEventItemReader<CustomerData> reader = new StaxEventItemReader<>();
		reader.setResource(new UrlResource(file));
		reader.setFragmentRootElementNames(new String[] { "customer" });
		reader.setUnmarshaller(newCustomerDataMarshaller());
		return reader;
	}

	private Jaxb2Marshaller newCustomerDataMarshaller() {
		Jaxb2Marshaller marshaller = new Jaxb2Marshaller();
		marshaller.setClassesToBeBound(CustomerData.class);
		return marshaller;
	}

	@Bean
	CustomerProcessor customerProcessor() {
		return new CustomerProcessor();
	}

	@Bean
	ItemWriterAdapter<Customer> customerItemWriter(CustomerService customerService) {
		ItemWriterAdapter<Customer> writer = new ItemWriterAdapter<>();
		writer.setTargetObject(customerService);
		writer.setTargetMethod("create");
		return writer;
	}
}

	@Bean
	BatchConfigurer batchConfigurer(DataSource dataSource) {
		return new DefaultBatchConfigurer(dataSource);
	}

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.listener.JobExecutionListenerSupport;

public class LoggingJobExecutionListener extends JobExecutionListenerSupport {

	private static final Logger LOGGER = LoggerFactory.getLogger(LoggingJobExecutionListener.class);

	@Override
	public void beforeJob(JobExecution jobExecution) {
		LOGGER.info("Executing Job {} ", jobExecution.getJobInstance().getJobName());
	}

	@Override
	public void afterJob(JobExecution jobExecution) {
		LOGGER.info("Finished Job {} ", jobExecution.getJobInstance().getJobName(), jobExecution.getStatus());
		super.afterJob(jobExecution);
	}
}

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.listener.StepExecutionListenerSupport;

public class LoggingStepExecutionListener extends StepExecutionListenerSupport {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(LoggingStepExecutionListener.class);

	@Override
	public void beforeStep(StepExecution stepExecution) {
		LOGGER.info("Executing Step {}", stepExecution.getStepName());
	}
	
	@Override
	public ExitStatus afterStep(StepExecution stepExecution) {
		LOGGER.info("Finished Step {} with staus {}", stepExecution.getStepName(), stepExecution.getExitStatus());
		return stepExecution.getExitStatus();
	}

}

import java.util.List;
import java.util.stream.Collectors;

import org.springframework.batch.item.ItemProcessor;

import com.org.lob.project.batch.model.AddressData;
import com.org.lob.project.batch.model.CustomerData;
import com.org.lob.project.repository.entity.Address;
import com.org.lob.project.repository.entity.Customer;

public class CustomerProcessor implements ItemProcessor<CustomerData, Customer>{

	@Override
	public Customer process(CustomerData item) throws Exception {
		Customer customer = new Customer();
		customer.setId(item.getId());
		customer.setEmailAddress(item.getEmailAddress());
		customer.setFirstName(item.getFirstName());
		customer.setLastName(item.getLastName());
		customer.setAddresses(buildAddress(item.getAddresses()));
		return customer;
	}

	private List<Address> buildAddress(List<AddressData> addresses) {
		return addresses.stream().map(ad -> buildAddrss(ad)).collect(Collectors.toList());
	}

	private Address buildAddrss(AddressData ad) {
		Address address = new Address();
		address.setCity(ad.getCity());
		address.setStateCode(ad.getStateCode());
		address.setCountry(ad.getCountry());
		address.setId(ad.getId());
		address.setStateCode(ad.getStateCode());
		address.setStreetAddress(ad.getStreetAddress());
		
		return address;
	}
}

import java.util.ArrayList;
import java.util.List;

import javax.xml.bind.annotation.XmlAccessType;
import javax.xml.bind.annotation.XmlAccessorType;
import javax.xml.bind.annotation.XmlElement;
import javax.xml.bind.annotation.XmlElementWrapper;
import javax.xml.bind.annotation.XmlRootElement;

@XmlRootElement(name = "customer")
@XmlAccessorType(XmlAccessType.FIELD)
public class CustomerData {

	private Long id;
	private String firstName;
	private String lastName;
	private String emailAddress;

	@XmlElementWrapper(name = "addresses")
	@XmlElement(name = "address")
	private List<AddressData> addresses;

	public Long getId() {
		return id;
	}

	public void setId(Long id) {
		this.id = id;
	}

	public String getFirstName() {
		return firstName;
	}

	public void setFirstName(String firstName) {
		this.firstName = firstName;
	}

	public String getLastName() {
		return lastName;
	}

	public void setLastName(String lastName) {
		this.lastName = lastName;
	}

	public String getEmailAddress() {
		return emailAddress;
	}

	public void setEmailAddress(String emailAddress) {
		this.emailAddress = emailAddress;
	}

	public List<AddressData> getAddresses() {
		return addresses;
	}

	public void setAddresses(List<AddressData> addresses) {
		if (this.addresses == null) {
			this.addresses = new ArrayList<>();
		} else {
			this.addresses.clear();
		}
		this.addresses.addAll(addresses);
	}

}

import javax.xml.bind.annotation.XmlAccessType;
import javax.xml.bind.annotation.XmlAccessorType;
import javax.xml.bind.annotation.XmlElement;
import javax.xml.bind.annotation.XmlRootElement;
import javax.xml.bind.annotation.XmlType;

@XmlRootElement(name="address")
@XmlAccessorType(XmlAccessType.FIELD)
@XmlType(propOrder = { "id", "zipCode", "stateCode", "country", "streetAddress", "city"})
public class AddressData {

    private Long id;

    @XmlElement(name = "street-address")
    private String streetAddress;

    private String city;

    @XmlElement(name = "state-code")
    private String stateCode;

    private  String country;

    @XmlElement(name = "zip-code")
    private String zipCode;
    
    public Long getId() {
		return id;
	}

	public void setId(Long id) {
		this.id = id;
	}

	public String getStreetAddress() {
		return streetAddress;
	}

	public void setStreetAddress(String streetAddress) {
		this.streetAddress = streetAddress;
	}

	public String getCity() {
		return city;
	}

	public void setCity(String city) {
		this.city = city;
	}

	public String getStateCode() {
		return stateCode;
	}

	public void setStateCode(String stateCode) {
		this.stateCode = stateCode;
	}

	public String getCountry() {
		return country;
	}

	public void setCountry(String country) {
		this.country = country;
	}

	public String getZipCode() {
		return zipCode;
	}

	public void setZipCode(String zipCode) {
		this.zipCode = zipCode;
	}
}

import java.util.Date;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.JobParametersInvalidException;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.org.lob.project.messaging.model.BatchProcessEvent;

@Component
public class BatchProcessMQConsumer {

	private static final Logger LOGGER = LoggerFactory.getLogger(BatchProcessMQConsumer.class);

	private ObjectMapper mapper;
	private JobLauncher jobLauncher;
	private Job job;

	public BatchProcessMQConsumer(ObjectMapper mapper, JobLauncher jobLauncher, Job job) {
		this.mapper = mapper;
		this.jobLauncher = jobLauncher;
		this.job = job;
	}

	@RabbitListener(queues = "#{'${rabbitmq.batch_process.triggered.queue}'}")
	public void consumeMessage(String message) {
		try {
			LOGGER.trace("Message received: {} ", message);

			JobExecution execution = launchJob(create(message));

			LOGGER.debug("Message processed successfully with status {} ", execution.getExitStatus());
		} catch (Exception e) {
			LOGGER.error("Unnable to process the Message", e);
		}
	}

	private BatchProcessEvent create(String message) throws JsonProcessingException {
		return this.mapper.readValue(message, BatchProcessEvent.class);
	}

	private JobExecution launchJob(BatchProcessEvent batchProcessEvent) throws JobExecutionAlreadyRunningException, JobRestartException,
			JobInstanceAlreadyCompleteException, JobParametersInvalidException {
		JobParameters params = jobParams(batchProcessEvent);
		return this.jobLauncher.run(this.job, params);
	}

	private JobParameters jobParams(BatchProcessEvent batchProcessEvent) {
		return new JobParametersBuilder()
				.addString("JobID", String.valueOf(System.currentTimeMillis()))
				.addString("fileName", batchProcessEvent.getFilePath())
				.addDate("launchDate", new Date())
				.toJobParameters();
	}
}

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<customers>
	<customer>
		<firstName>firstname_batch</firstName>
		<lastName>lastName_batch</lastName>
		<emailAddress>[email protected]</emailAddress>
		<addresses>
			<address>
				<zip-code>ZC</zip-code>
				<state-code>SC</state-code>
				<country>copuntyhr</country>
				<street-address>Str Adr</street-address>
				<city>city</city>
			</address>
			<address>
				<zip-code>ZC1</zip-code>
				<state-code>SC1</state-code>
				<country>copuntyhr1</country>
				<street-address>Str Adr1</street-address>
				<city>city1</city>
			</address>
		</addresses>
	</customer>

</customers>

content_type = text/plain
content_encoding = UTF-8

{"id":1, "path": "E:\\eWorkspaces\\job\\customer.xml"}

        <dependency>
		    <groupId>org.springframework.data</groupId>
		    <artifactId>spring-data-envers</artifactId>
		</dependency>

Hibernate: 
    
    create table prj_address (
       id bigint not null auto_increment,
        created_by varchar(25),
        create_date_time timestamp default '2021-06-10 20:47:05.967394',
        modified_by varchar(25),
        modified_date_time timestamp default '2021-06-10 20:47:05.967394',
        city varchar(255),
        country varchar(255),
        state_code varchar(255),
        street_address varchar(255),
        zip_code varchar(255),
        customer_id bigint,
        primary key (id)
    ) engine=InnoDB
Hibernate: 
    
    create table prj_customer (
       id bigint not null auto_increment,
        created_by varchar(25),
        create_date_time timestamp default '2021-06-10 20:47:05.967394',
        modified_by varchar(25),
        modified_date_time timestamp default '2021-06-10 20:47:05.967394',
        email_address varchar(255),
        first_name varchar(255),
        last_name varchar(255),
        primary key (id)
    ) engine=InnoDB
Hibernate: 
    
    create table prj_customer_aud (
       id bigint not null,
        rev integer not null,
        revtype tinyint,
        email_address varchar(255),
        first_name varchar(255),
        last_name varchar(255),
        primary key (id, rev)
    ) engine=InnoDB
Hibernate: 
    
    create table revinfo (
       rev integer not null auto_increment,
        revtstmp bigint,
        primary key (rev)
    ) engine=InnoDB
Hibernate: 
    
    alter table prj_address 
       add constraint FKk5h0gt5brecxp11or1okmsslr 
       foreign key (customer_id) 
       references prj_customer (id)
Hibernate: 
    
    alter table prj_customer_aud 
       add constraint FK119fx2iv1tqbhvjopxr2f6age 
       foreign key (rev) 
       references revinfo (rev)

curl --location --request POST 'http://localhost:8091/api/v1/customer/' \
--header 'Content-Type: application/json' \
--data-raw '{
    "firstName": "Nadeem",
    "lastName": "Mohammad",
    "emailAddress" : "[email protected]",
    "addresses": [
        {
            "streetAddress": "afd",
            "city": "afds",
            "stateCode": "asdf",
            "country": "sfdds",
            "zipCode": "432423"
        },
        {
            "streetAddress": "asdf",
            "city": "ffff",
            "stateCode": "sdf",
            "country": "gfdg",
            "zipCode": "444"
        }
    ]
}'

curl --location --request PUT 'http://localhost:8091/api/v1/customer/1' \
--header 'Content-Type: application/json' \
--data-raw '{

    "firstName": "Nadeem1",
    "lastName": "Mohammad",
    "emailAddress" : "[email protected]",
    "addresses": [
        {
            "streetAddress": "afd",
            "city": "afds",
            "stateCode": "asdf",
            "country": "sfdds",
            "zipCode": "432423"
        },
        {
            "streetAddress": "asdf",
            "city": "ffff",
            "stateCode": "sdf",
            "country": "gfdg",
            "zipCode": "444"
        }
    ]
}'

curl --location --request DELETE 'http://localhost:8091/api/v1/customer/1' \
--header 'Accept: application/json' \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJtbmFkZWVtIiwibmFtZSI6Ik1vaGFtbWFkIE5hZGVlbSIsImlhdCI6MTUxNjIzOTAyMiwiZXhwIjoxNjczMDQyNDU3NjI5fQ.gT8xVbDGJJQK8tsID6CCdBZCftnUVn7Tud-v1NAOHhw'

Hibernate: delete from PRJ_ADDRESS where ID=?
Hibernate: delete from PRJ_ADDRESS where ID=?
Hibernate: delete from PRJ_CUSTOMER where ID=?
Hibernate: insert into REVINFO (REVTSTMP) values (?)
Hibernate: insert into PRJ_CUSTOMER_AUD (REVISION_TYPE, CREATED_BY, CREATE_DATE_TIME, MODIFIED_BY, MODIFIED_DATE_TIME, EMAIL_ADDRESS, FIRST_NAME, LAST_NAME, ID, REVISION_ID) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)

spring.jpa.properties.org.hibernate.envers.audit_table_suffix=_AUD
spring.jpa.properties.org.hibernate.envers.revision_field_name=REVISION_ID
spring.jpa.properties.org.hibernate.envers.revision_type_field_name=REVISION_TYPE