From 7e05fcbe3f7796fec44db648cf25a27f6e0fdbbe Mon Sep 17 00:00:00 2001
From: Shubhra <shubhi.srvstv@gmail.com>
Date: Sun, 9 Feb 2020 18:16:20 +0530
Subject: [PATCH] BAEL-3298 Adding tests and refactoring

---
 .../src/main/java/org/baeldung/batch/App.java |   2 +
 .../org/baeldung/batch/SpringBatchConfig.java |  37 +-----
 .../batch/SpringBatchRetryConfig.java         | 102 +++++++++++++++++
 .../batch/service/RetryItemProcessor.java     |  22 +++-
 .../SpringBatchRetryIntegrationTest.java      | 105 ++++++++++++++++++
 .../output/batchRetry/retryOutput.xml         |   1 +
 spring-batch/xml/retryOutput.xml              |   1 +
 7 files changed, 230 insertions(+), 40 deletions(-)
 create mode 100644 spring-batch/src/main/java/org/baeldung/batch/SpringBatchRetryConfig.java
 create mode 100644 spring-batch/src/test/java/org/baeldung/batch/SpringBatchRetryIntegrationTest.java
 create mode 100644 spring-batch/src/test/resources/output/batchRetry/retryOutput.xml
 create mode 100644 spring-batch/xml/retryOutput.xml

diff --git a/spring-batch/src/main/java/org/baeldung/batch/App.java b/spring-batch/src/main/java/org/baeldung/batch/App.java
index 634f0a9014..764ef72a35 100644
--- a/spring-batch/src/main/java/org/baeldung/batch/App.java
+++ b/spring-batch/src/main/java/org/baeldung/batch/App.java
@@ -18,6 +18,8 @@ public class App {
         final AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
         context.register(SpringConfig.class);
         context.register(SpringBatchConfig.class);
+        context.register(SpringBatchRetryConfig.class);
+
         context.refresh();
 
         // Spring xml config
diff --git a/spring-batch/src/main/java/org/baeldung/batch/SpringBatchConfig.java b/spring-batch/src/main/java/org/baeldung/batch/SpringBatchConfig.java
index 90eb038052..07dd65bcfd 100644
--- a/spring-batch/src/main/java/org/baeldung/batch/SpringBatchConfig.java
+++ b/spring-batch/src/main/java/org/baeldung/batch/SpringBatchConfig.java
@@ -1,17 +1,14 @@
 package org.baeldung.batch;
 
-import org.apache.http.conn.ConnectTimeoutException;
 import org.baeldung.batch.model.Transaction;
 import org.baeldung.batch.service.CustomItemProcessor;
 import org.baeldung.batch.service.CustomSkipPolicy;
 import org.baeldung.batch.service.MissingUsernameException;
 import org.baeldung.batch.service.NegativeAmountException;
 import org.baeldung.batch.service.RecordFieldSetMapper;
-import org.baeldung.batch.service.RetryItemProcessor;
 import org.baeldung.batch.service.SkippingItemProcessor;
 import org.springframework.batch.core.Job;
 import org.springframework.batch.core.Step;
-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.item.ItemProcessor;
@@ -26,15 +23,12 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
 import org.springframework.core.io.Resource;
-import org.springframework.dao.DeadlockLoserDataAccessException;
 import org.springframework.oxm.Marshaller;
 import org.springframework.oxm.jaxb.Jaxb2Marshaller;
 
 import java.text.ParseException;
-@Configuration
-@EnableBatchProcessing
+
 public class SpringBatchConfig {
     @Autowired
     private JobBuilderFactory jobBuilderFactory;
@@ -75,11 +69,6 @@ public class SpringBatchConfig {
         return new SkippingItemProcessor();
     }
 
-    @Bean
-    public ItemProcessor<Transaction, Transaction> retryItemProcessor() {
-        return new RetryItemProcessor();
-    }
-
     @Bean
     public ItemWriter<Transaction> itemWriter(Marshaller marshaller) {
         StaxEventItemWriter<Transaction> itemWriter = new StaxEventItemWriter<>();
@@ -128,22 +117,6 @@ public class SpringBatchConfig {
                 .build();
     }
 
-    @Bean
-    public Step retryStep(@Qualifier("retryItemProcessor") ItemProcessor<Transaction, Transaction> processor,
-        ItemWriter<Transaction> writer) throws ParseException {
-        return stepBuilderFactory
-            .get("retryStep")
-            .<Transaction, Transaction>chunk(10)
-            .reader(itemReader(inputCsv))
-            .processor(processor)
-            .writer(writer)
-            .faultTolerant()
-            .retryLimit(3)
-            .retry(ConnectTimeoutException.class)
-            .retry(DeadlockLoserDataAccessException.class)
-            .build();
-    }
-
     @Bean(name = "skippingBatchJob")
     public Job skippingJob(@Qualifier("skippingStep") Step skippingStep) {
         return jobBuilderFactory
@@ -152,14 +125,6 @@ public class SpringBatchConfig {
                 .build();
     }
 
-    @Bean(name = "retryBatchJob")
-    public Job retryJob(@Qualifier("retryStep") Step retryStep) {
-        return jobBuilderFactory
-            .get("retryBatchJob")
-            .start(retryStep)
-            .build();
-    }
-
     @Bean
     public Step skipPolicyStep(@Qualifier("skippingItemProcessor") ItemProcessor<Transaction, Transaction> processor,
                                ItemWriter<Transaction> writer) throws ParseException {
diff --git a/spring-batch/src/main/java/org/baeldung/batch/SpringBatchRetryConfig.java b/spring-batch/src/main/java/org/baeldung/batch/SpringBatchRetryConfig.java
new file mode 100644
index 0000000000..65a75c2324
--- /dev/null
+++ b/spring-batch/src/main/java/org/baeldung/batch/SpringBatchRetryConfig.java
@@ -0,0 +1,102 @@
+package org.baeldung.batch;
+
+import org.apache.http.conn.ConnectTimeoutException;
+import org.baeldung.batch.model.Transaction;
+import org.baeldung.batch.service.RecordFieldSetMapper;
+import org.baeldung.batch.service.RetryItemProcessor;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.Step;
+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.item.ItemProcessor;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.item.UnexpectedInputException;
+import org.springframework.batch.item.file.FlatFileItemReader;
+import org.springframework.batch.item.file.mapping.DefaultLineMapper;
+import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
+import org.springframework.batch.item.xml.StaxEventItemWriter;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.Resource;
+import org.springframework.dao.DeadlockLoserDataAccessException;
+import org.springframework.oxm.Marshaller;
+import org.springframework.oxm.jaxb.Jaxb2Marshaller;
+
+import java.text.ParseException;
+
+@Configuration
+@EnableBatchProcessing
+public class SpringBatchRetryConfig {
+    @Autowired
+    private JobBuilderFactory jobBuilderFactory;
+
+    @Autowired
+    private StepBuilderFactory stepBuilderFactory;
+
+    @Value("input/record.csv")
+    private Resource inputCsv;
+
+    @Value("file:xml/retryOutput.xml")
+    private Resource outputXml;
+
+    public ItemReader<Transaction> itemReader(Resource inputData) throws UnexpectedInputException, ParseException {
+        FlatFileItemReader<Transaction> reader = new FlatFileItemReader<>();
+        DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
+        String[] tokens = { "username", "userid", "transactiondate", "amount" };
+        tokenizer.setNames(tokens);
+        reader.setResource(inputData);
+        DefaultLineMapper<Transaction> lineMapper = new DefaultLineMapper<>();
+        lineMapper.setLineTokenizer(tokenizer);
+        lineMapper.setFieldSetMapper(new RecordFieldSetMapper());
+        reader.setLinesToSkip(1);
+        reader.setLineMapper(lineMapper);
+        return reader;
+    }
+
+    @Bean
+    public ItemProcessor<Transaction, Transaction> retryItemProcessor() {
+        return new RetryItemProcessor();
+    }
+
+    @Bean
+    public ItemWriter<Transaction> itemWriter(Marshaller marshaller) {
+        StaxEventItemWriter<Transaction> itemWriter = new StaxEventItemWriter<>();
+        itemWriter.setMarshaller(marshaller);
+        itemWriter.setRootTagName("transactionRecord");
+        itemWriter.setResource(outputXml);
+        return itemWriter;
+    }
+
+    @Bean
+    public Marshaller marshaller() {
+        Jaxb2Marshaller marshaller = new Jaxb2Marshaller();
+        marshaller.setClassesToBeBound(Transaction.class);
+        return marshaller;
+    }
+
+    @Bean
+    public Step retryStep(
+        @Qualifier("retryItemProcessor")
+            ItemProcessor<Transaction, Transaction> processor, ItemWriter<Transaction> writer) throws ParseException {
+        return stepBuilderFactory.get("retryStep").<Transaction, Transaction>chunk(10).reader(itemReader(inputCsv))
+            .processor(processor)
+            .writer(writer)
+            .faultTolerant()
+            .retryLimit(3)
+            .retry(ConnectTimeoutException.class)
+            .retry(DeadlockLoserDataAccessException.class)
+            .build();
+    }
+
+    @Bean(name = "retryBatchJob")
+    public Job retryJob(
+        @Qualifier("retryStep")
+            Step retryStep) {
+        return jobBuilderFactory.get("retryBatchJob").start(retryStep).build();
+    }
+}
diff --git a/spring-batch/src/main/java/org/baeldung/batch/service/RetryItemProcessor.java b/spring-batch/src/main/java/org/baeldung/batch/service/RetryItemProcessor.java
index 8eb279caf0..736f778424 100644
--- a/spring-batch/src/main/java/org/baeldung/batch/service/RetryItemProcessor.java
+++ b/spring-batch/src/main/java/org/baeldung/batch/service/RetryItemProcessor.java
@@ -5,7 +5,10 @@ import org.apache.http.client.config.RequestConfig;
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.util.EntityUtils;
 import org.baeldung.batch.model.Transaction;
+import org.codehaus.jettison.json.JSONException;
+import org.codehaus.jettison.json.JSONObject;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.batch.item.ItemProcessor;
@@ -16,18 +19,29 @@ public class RetryItemProcessor implements ItemProcessor<Transaction, Transactio
 
     private static final Logger LOGGER = LoggerFactory.getLogger(RetryItemProcessor.class);
 
+    private CloseableHttpClient client;
+
+    public RetryItemProcessor() {
+        final RequestConfig config = RequestConfig.custom().setConnectTimeout(2 * 1000).build();
+        client = HttpClientBuilder.create().setDefaultRequestConfig(config).build();
+    }
+
     @Override
-    public Transaction process(Transaction transaction) throws IOException {
+    public Transaction process(Transaction transaction) throws IOException, JSONException {
         LOGGER.info("Attempting to process user with id={}", transaction.getUserId());
+        System.out.println("Attempting to process user with id=" + transaction.getUserId());
         HttpResponse response = fetchMoreUserDetails(transaction.getUserId());
+
         //parse user's age and postCode from response and update transaction
+        String result = EntityUtils.toString(response.getEntity());
+        JSONObject userObject = new JSONObject(result);
+        transaction.setAge(Integer.parseInt(userObject.getString("age")));
+        transaction.setPostCode(userObject.getString("postCode"));
+
         return transaction;
     }
 
     private HttpResponse fetchMoreUserDetails(int id) throws IOException {
-        final RequestConfig config = RequestConfig.custom().setConnectTimeout(2 * 1000).build();
-        final CloseableHttpClient client = HttpClientBuilder.create().setDefaultRequestConfig(config).build();
-
         final HttpGet request = new HttpGet("http://www.baeldung.com:81/user/" + id);
         return client.execute(request);
     }
diff --git a/spring-batch/src/test/java/org/baeldung/batch/SpringBatchRetryIntegrationTest.java b/spring-batch/src/test/java/org/baeldung/batch/SpringBatchRetryIntegrationTest.java
new file mode 100644
index 0000000000..f788650c5e
--- /dev/null
+++ b/spring-batch/src/test/java/org/baeldung/batch/SpringBatchRetryIntegrationTest.java
@@ -0,0 +1,105 @@
+package org.baeldung.batch;
+
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.conn.ConnectTimeoutException;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.baeldung.batch.service.RetryItemProcessor;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobInstance;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.JobParametersBuilder;
+import org.springframework.batch.test.AssertFile;
+import org.springframework.batch.test.JobLauncherTestUtils;
+import org.springframework.batch.test.context.SpringBatchTest;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.core.io.FileSystemResource;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.annotation.DirtiesContext.ClassMode;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.TestExecutionListeners;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.support.DependencyInjectionTestExecutionListener;
+import org.springframework.test.context.support.DirtiesContextTestExecutionListener;
+
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+@RunWith(SpringRunner.class)
+@SpringBatchTest
+@EnableAutoConfiguration
+@ContextConfiguration(classes = { SpringBatchRetryConfig.class })
+@TestExecutionListeners({ DependencyInjectionTestExecutionListener.class, DirtiesContextTestExecutionListener.class })
+@DirtiesContext(classMode = ClassMode.AFTER_CLASS)
+public class SpringBatchRetryIntegrationTest {
+
+    private static final String TEST_OUTPUT = "xml/retryOutput.xml";
+
+    private static final String EXPECTED_OUTPUT = "src/test/resources/output/batchRetry/retryOutput.xml";
+
+    @Autowired
+    private JobLauncherTestUtils jobLauncherTestUtils;
+
+    @Mock
+    private CloseableHttpClient httpClient;
+
+    @Mock
+    private CloseableHttpResponse httpResponse;
+
+    @InjectMocks
+    @Autowired
+    private RetryItemProcessor retryItemProcessor;
+
+    @Before
+    public void setup() {
+        MockitoAnnotations.initMocks(this);
+    }
+
+    private JobParameters defaultJobParameters() {
+        JobParametersBuilder paramsBuilder = new JobParametersBuilder();
+        paramsBuilder.addString("jobID", String.valueOf(System.currentTimeMillis()));
+        return paramsBuilder.toJobParameters();
+    }
+
+    @Test
+    public void whenEndpointAlwaysFailing_thenJobFails() throws Exception {
+        when(httpClient.execute(any())).thenThrow(new ConnectTimeoutException("Endpoint is down"));
+
+        JobExecution jobExecution = jobLauncherTestUtils.launchJob(defaultJobParameters());
+        JobInstance actualJobInstance = jobExecution.getJobInstance();
+        ExitStatus actualJobExitStatus = jobExecution.getExitStatus();
+
+        assertThat(actualJobInstance.getJobName(), is("retryBatchJob"));
+        assertThat(actualJobExitStatus.getExitCode(), is("FAILED"));
+        assertThat(actualJobExitStatus.getExitDescription(), containsString("org.apache.http.conn.ConnectTimeoutException"));
+    }
+
+    @Test
+    public void whenEndpointFailsTwiceAndPassesThirdTime_thenSuccess() throws Exception {
+        FileSystemResource expectedResult = new FileSystemResource(EXPECTED_OUTPUT);
+        FileSystemResource actualResult = new FileSystemResource(TEST_OUTPUT);
+
+        //fails for first two calls and passes third time onwards
+        when(httpResponse.getEntity()).thenReturn(new StringEntity("{ \"age\":10, \"postCode\":\"430222\" }"));
+        when(httpClient.execute(any())).thenThrow(new ConnectTimeoutException("Timeout count 1")).thenThrow(new ConnectTimeoutException("Timeout count 2")).thenReturn(httpResponse);
+
+        JobExecution jobExecution = jobLauncherTestUtils.launchJob(defaultJobParameters());
+        JobInstance actualJobInstance = jobExecution.getJobInstance();
+        ExitStatus actualJobExitStatus = jobExecution.getExitStatus();
+
+        assertThat(actualJobInstance.getJobName(), is("retryBatchJob"));
+        assertThat(actualJobExitStatus.getExitCode(), is("COMPLETED"));
+        AssertFile.assertFileEquals(expectedResult, actualResult);
+    }
+}
diff --git a/spring-batch/src/test/resources/output/batchRetry/retryOutput.xml b/spring-batch/src/test/resources/output/batchRetry/retryOutput.xml
new file mode 100644
index 0000000000..31da7bd6b3
--- /dev/null
+++ b/spring-batch/src/test/resources/output/batchRetry/retryOutput.xml
@@ -0,0 +1 @@
+<?xml version="1.0" encoding="UTF-8"?><transactionRecord><transactionRecord><age>10</age><amount>10000.0</amount><postCode>430222</postCode><transactionDate>2015-10-31T00:00:00+05:30</transactionDate><userId>1234</userId><username>devendra</username></transactionRecord><transactionRecord><age>10</age><amount>12321.0</amount><postCode>430222</postCode><transactionDate>2015-12-03T00:00:00+05:30</transactionDate><userId>2134</userId><username>john</username></transactionRecord><transactionRecord><age>10</age><amount>23411.0</amount><postCode>430222</postCode><transactionDate>2015-02-02T00:00:00+05:30</transactionDate><userId>2134</userId><username>robin</username></transactionRecord></transactionRecord>
\ No newline at end of file
diff --git a/spring-batch/xml/retryOutput.xml b/spring-batch/xml/retryOutput.xml
new file mode 100644
index 0000000000..31da7bd6b3
--- /dev/null
+++ b/spring-batch/xml/retryOutput.xml
@@ -0,0 +1 @@
+<?xml version="1.0" encoding="UTF-8"?><transactionRecord><transactionRecord><age>10</age><amount>10000.0</amount><postCode>430222</postCode><transactionDate>2015-10-31T00:00:00+05:30</transactionDate><userId>1234</userId><username>devendra</username></transactionRecord><transactionRecord><age>10</age><amount>12321.0</amount><postCode>430222</postCode><transactionDate>2015-12-03T00:00:00+05:30</transactionDate><userId>2134</userId><username>john</username></transactionRecord><transactionRecord><age>10</age><amount>23411.0</amount><postCode>430222</postCode><transactionDate>2015-02-02T00:00:00+05:30</transactionDate><userId>2134</userId><username>robin</username></transactionRecord></transactionRecord>
\ No newline at end of file