Spring Batch Application

Spring-Boot-Batch/springbootbatch/pom.xml

@@ -69,6 +69,16 @@
 			<artifactId>spring-batch-test</artifactId>
 			<scope>test</scope>
 		</dependency>
+        <dependency>
+            <groupId>org.springframework.batch</groupId>
+            <artifactId>spring-batch-core</artifactId>
+            <version>4.2.2.RELEASE</version>
+            <scope>test</scope>
+        </dependency>
+		<dependency>
+			<groupId>org.springframework.batch</groupId>
+			<artifactId>spring-batch-core</artifactId>
+		</dependency>
 	</dependencies>
 
 	<build>

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/SpringBootBatchBasicApplication.java

@@ -4,6 +4,9 @@ import org.springframework.batch.core.configuration.annotation.EnableBatchProces
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 
+/**
+ * @Author - Kunwar Vikas
+ */
 @SpringBootApplication
 @EnableBatchProcessing
 public class SpringBootBatchBasicApplication {

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/config/SpringBatchConfig.java

@@ -1,11 +1,20 @@
 package com.javadevjournal.springbootbatch.config;
 
+import com.javadevjournal.springbootbatch.listener.SPJobExecutionListener;
+import com.javadevjournal.springbootbatch.listener.SPStepListener;
+import com.javadevjournal.springbootbatch.model.Employee;
+import com.javadevjournal.springbootbatch.step.EmployeeItemProcessor;
 import org.springframework.batch.core.Job;
 import org.springframework.batch.core.JobExecutionListener;
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
 import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
 import org.springframework.batch.core.launch.support.RunIdIncrementer;
+import org.springframework.batch.item.file.FlatFileItemReader;
+import org.springframework.batch.item.file.FlatFileItemWriter;
+import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
+import org.springframework.batch.item.file.builder.FlatFileItemWriterBuilder;
+import org.springframework.batch.item.file.transform.PassThroughLineAggregator;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -14,7 +23,12 @@ import com.javadevjournal.springbootbatch.listener.SpringBatchJobCompletionListe
 import com.javadevjournal.springbootbatch.step.SBProcessor;
 import com.javadevjournal.springbootbatch.step.SBReader;
 import com.javadevjournal.springbootbatch.step.SBWriter;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.FileSystemResource;
 
+/**
+ * @Author - Kunwar Vikas
+ */
 @Configuration
 public class SpringBatchConfig {
 
@@ -27,20 +41,45 @@ public class SpringBatchConfig {
 	@Bean
 	public Job processJob() {
 		return jobBuilderFactory.get("javadevjournaljob")
-				.incrementer(new RunIdIncrementer()).listener(listener())
+				.incrementer(new RunIdIncrementer())
+				.listener(new SPJobExecutionListener())
 				.flow(orderStep1()).end().build();
 	}
 
-	@Bean
+
-	public Step orderStep1() {
+    @Bean
-		return stepBuilderFactory.get("step1").<String, String> chunk(1)
+    public Step orderStep1() {
-				.reader(new SBReader()).processor(new SBProcessor())
+        return stepBuilderFactory.get("step1")
-				.writer(new SBWriter()).build();
+                .listener(new SPStepListener())
-	}
+                .<Employee, String>chunk(10).reader(reader())
+                .processor(processor()).writer(writer()).build();
+    }
+
+    @Bean
+    public FlatFileItemReader<Employee> reader() {
+        return new FlatFileItemReaderBuilder<Employee>()
+                .name("employeeItemReader")
+                .resource(new ClassPathResource("csv/employees.csv"))
+                .delimited().names(new String[] {"firstName", "lastName","department"})
+                .targetType(Employee.class).build();
+    }
+
+    @Bean
+    public EmployeeItemProcessor processor() {
+        return new EmployeeItemProcessor();
+    }
+
+    @Bean
+    public FlatFileItemWriter<String> writer() {
+        return new FlatFileItemWriterBuilder<String>()
+                .name("greetingItemWriter")
+                .resource(new FileSystemResource(
+                        "target/output.txt"))
+                .lineAggregator(new PassThroughLineAggregator<>()).build();
+    }
 
 	@Bean
 	public JobExecutionListener listener() {
 		return new SpringBatchJobCompletionListener();
 	}
-
 }

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/controller/SpringBatchJobController.java

@@ -9,6 +9,9 @@ import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
+/**
+ * @Author - Kunwar Vikas
+ */
 @RestController
 public class SpringBatchJobController {
 

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/listener/SPChunkListener.java

@@ -0,0 +1,30 @@
+package com.javadevjournal.springbootbatch.listener;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.batch.core.ChunkListener;
+import org.springframework.batch.core.StepListener;
+import org.springframework.batch.core.scope.context.ChunkContext;
+
+/**
+ * @Author - Kunwar Vikas
+ */
+public class SPChunkListener implements ChunkListener {
+
+    Logger logger = LoggerFactory.getLogger(SPChunkListener.class);
+
+    @Override
+    public void beforeChunk(ChunkContext chunkContext) {
+        logger.info("beforeChunk");
+    }
+
+    @Override
+    public void afterChunk(ChunkContext chunkContext) {
+        logger.info("afterChunk");
+    }
+
+    @Override
+    public void afterChunkError(ChunkContext chunkContext) {
+        logger.info("afterChunkError");
+    }
+}

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/listener/SPItemProcessorListener.java

@@ -0,0 +1,29 @@
+package com.javadevjournal.springbootbatch.listener;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.batch.core.ItemProcessListener;
+
+
+/**
+ * @Author - Kunwar Vikas
+ */
+public class SPItemProcessorListener implements ItemProcessListener<String, Number> {
+
+    Logger logger = LoggerFactory.getLogger(SPItemProcessorListener.class);
+
+    @Override
+    public void beforeProcess(String item) {
+        logger.info("beforeProcess");
+    }
+
+    @Override
+    public void afterProcess(String item, Number result) {
+        logger.info("afterProcess");
+    }
+
+    @Override
+    public void onProcessError(String item, Exception e) {
+        logger.info(" onProcessError");
+    }
+}

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/listener/SPItemReadListener.java

@@ -0,0 +1,28 @@
+package com.javadevjournal.springbootbatch.listener;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.batch.core.ItemReadListener;
+
+/**
+ * @Author - Kunwar Vikas
+ */
+public class SPItemReadListener implements ItemReadListener<String> {
+
+    Logger logger = LoggerFactory.getLogger(SPItemReadListener.class);
+
+    @Override
+    public void beforeRead() {
+        logger.info("Before reading an item");
+    }
+
+    @Override
+    public void afterRead(String item) {
+        logger.info("After reading an item: "+ item.toString());
+    }
+
+    @Override
+    public void onReadError(Exception ex) {
+        logger.error("Error occurred while reading an item!");
+    }
+}

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/listener/SPItemWriteListener.java

@@ -0,0 +1,28 @@
+package com.javadevjournal.springbootbatch.listener;
+
+import com.javadevjournal.springbootbatch.model.User;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.batch.core.ItemWriteListener;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import javax.sql.DataSource;
+import java.util.List;
+
+/**
+ * @Author - Kunwar Vikas
+ */
+public class SPItemWriteListener implements ItemWriteListener<Number> {
+    Logger logger = LoggerFactory.getLogger(SPItemWriteListener.class);
+
+    public void beforeWrite(List items) {
+        logger.info("before write");
+    }
+
+    public void onWriteError(Exception exception, List items) {
+        logger.info("Error occurred when writing items!");
+    }
+    public void afterWrite(List items) {
+        logger.info("after write");
+    }
+}

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/listener/SPJobExecutionListener.java

@@ -0,0 +1,27 @@
+package com.javadevjournal.springbootbatch.listener;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobExecutionListener;
+
+/**
+ * @Author - Kunwar Vikas
+ */
+public class SPJobExecutionListener implements JobExecutionListener {
+
+    Logger logger = LoggerFactory.getLogger(SPJobExecutionListener.class);
+
+    public void beforeJob(JobExecution jobExecution) {
+        logger.info("BEFORE BATCH JOB STARTS");
+    }
+
+    public void afterJob(JobExecution jobExecution) {
+        if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
+            logger.info("BATCH JOB COMPLETED SUCCESSFULLY");
+        }else if(jobExecution.getStatus() == BatchStatus.FAILED){
+            logger.info("BATCH JOB FAILED");
+        }
+    }
+}

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/listener/SPSkipListener.java

@@ -0,0 +1,28 @@
+package com.javadevjournal.springbootbatch.listener;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.batch.core.SkipListener;
+
+/**
+ * @Author - Kunwar Vikas
+ */
+public class SPSkipListener implements SkipListener<String, Number> {
+
+    Logger logger = LoggerFactory.getLogger(SPSkipListener.class);
+
+    @Override
+    public void onSkipInRead(Throwable t) {
+        logger.info("onSkipInRead");
+    }
+
+    @Override
+    public void onSkipInWrite(Number item, Throwable t) {
+        logger.info("onSkipInWrite");
+    }
+
+    @Override
+    public void onSkipInProcess(String item, Throwable t) {
+        logger.info("onWriteError");
+    }
+}

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/listener/SPStepListener.java

@@ -0,0 +1,27 @@
+package com.javadevjournal.springbootbatch.listener;
+
+import com.javadevjournal.springbootbatch.step.SBWriter;
+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.StepExecutionListener;
+
+/**
+ * @Author - Kunwar Vikas
+ */
+public class SPStepListener implements StepExecutionListener {
+
+    Logger logger = LoggerFactory.getLogger(SPStepListener.class);
+
+    @Override
+    public void beforeStep(StepExecution stepExecution) {
+        logger.info("SPStepListener - CALLED BEFORE STEP.");
+    }
+
+    @Override
+    public ExitStatus afterStep(StepExecution stepExecution) {
+        logger.info("SPStepListener - CALLED AFTER STEP.");
+        return ExitStatus.COMPLETED;
+    }
+}

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/listener/SpringBatchJobCompletionListener.java

@@ -1,19 +1,28 @@
 package com.javadevjournal.springbootbatch.listener;
 
-import com.javadevjournal.springbootbatch.step.SBWriter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.listener.JobExecutionListenerSupport;
 
+/**
+ * @Author - Kunwar Vikas
+ */
 public class SpringBatchJobCompletionListener extends JobExecutionListenerSupport {
 	Logger logger = LoggerFactory.getLogger(SpringBatchJobCompletionListener.class);
 
+	@Override
+	public void beforeJob(JobExecution jobExecution) {
+		logger.info("SpringBatchJobCompletionListener - BEFORE BATCH JOB STARTS");
+	}
+
 	@Override
 	public void afterJob(JobExecution jobExecution) {
 		if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
-			logger.info("BATCH JOB COMPLETED SUCCESSFULLY");
+			logger.info("SpringBatchJobCompletionListener - BATCH JOB COMPLETED SUCCESSFULLY");
+		}else if(jobExecution.getStatus() == BatchStatus.FAILED){
+			logger.info("SpringBatchJobCompletionListener - BATCH JOB FAILED");
 		}
 	}
 

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/mapper/UserRowMapper.java

@@ -0,0 +1,19 @@
+package com.javadevjournal.springbootbatch.mapper;
+
+import com.javadevjournal.springbootbatch.model.User;
+import org.springframework.jdbc.core.RowMapper;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+public class UserRowMapper implements RowMapper {
+    public User mapRow(ResultSet rs, int rowNum) throws SQLException {
+        User user = new User();
+        user.setFirstName(rs.getString("firstName"));
+        user.setMiddleName(rs.getString("middleName"));
+        user.setLastName(rs.getString("lastName"));
+        user.setCity(rs.getString("city"));
+        user.setId(rs.getInt("id"));
+        return user;
+    }
+}

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/model/Employee.java

@@ -0,0 +1,13 @@
+package com.javadevjournal.springbootbatch.model;
+
+import lombok.Data;
+
+@Data
+public class Employee {
+  private String firstName;
+  private String lastName;
+  private String department;
+
+  public Employee() {    
+  }
+}

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/model/User.java

@@ -0,0 +1,15 @@
+package com.javadevjournal.springbootbatch.model;
+
+import lombok.Data;
+
+/**
+ * Author Kunwar Vikas
+ */
+@Data
+public class User {
+    private int id;
+    private String firstName;
+    private String middleName;
+    private String lastName;
+    private String  city;
+}

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/repository/SpringBatchBasicRepository.java

@@ -0,0 +1,52 @@
+package com.javadevjournal.springbootbatch.repository;
+
+import org.springframework.batch.core.configuration.annotation.BatchConfigurer;
+import org.springframework.batch.core.explore.JobExplorer;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.batch.core.launch.support.SimpleJobLauncher;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.transaction.PlatformTransactionManager;
+
+import javax.sql.DataSource;
+
+/**
+ * currently not in use, we are using H2 DB
+ * @Author - Kunwar Vikas
+ */
+public class SpringBatchBasicRepository implements BatchConfigurer {
+
+    @Autowired
+    private DataSource dataSource;
+
+    @Autowired
+    private PlatformTransactionManager transactionManager;
+
+    @Override
+    public JobRepository getJobRepository() throws Exception {
+        JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
+        factory.setTablePrefix("SCHEMA_OWNER.BATCH_");
+        factory.setMaxVarCharLength(1200);
+        return factory.getObject();
+    }
+
+    @Override
+    public PlatformTransactionManager getTransactionManager() throws Exception {
+        return new DataSourceTransactionManager();
+    }
+
+    @Override
+    public JobLauncher getJobLauncher() throws Exception {
+        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
+        jobLauncher.setJobRepository(getJobRepository());
+        jobLauncher.afterPropertiesSet();
+        return jobLauncher;
+    }
+
+    @Override
+    public JobExplorer getJobExplorer() throws Exception {
+        return null;
+    }
+}

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/step/EmployeeItemProcessor.java

@@ -0,0 +1,22 @@
+package com.javadevjournal.springbootbatch.step;
+
+import com.javadevjournal.springbootbatch.model.Employee;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.batch.item.ItemProcessor;
+
+public class EmployeeItemProcessor
+    implements ItemProcessor<Employee, String> {
+
+  private static final Logger LOGGER =
+      LoggerFactory.getLogger(EmployeeItemProcessor.class);
+
+  @Override
+  public String process(Employee employee) throws Exception {
+    String greeting = "Hello " + employee.getFirstName() + " "
+        + employee.getLastName() + " from " + employee.getDepartment()+"!";
+
+    LOGGER.info("converting '{}' into '{}'", employee, greeting);
+    return greeting;
+  }
+}

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/step/SBProcessor.java

@@ -2,6 +2,9 @@ package com.javadevjournal.springbootbatch.step;
 
 import org.springframework.batch.item.ItemProcessor;
 
+/**
+ * @Author - Kunwar Vikas
+ */
 public class SBProcessor implements ItemProcessor<String, String> {
 
 	@Override

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/step/SBReader.java

@@ -5,6 +5,9 @@ import org.springframework.batch.item.NonTransientResourceException;
 import org.springframework.batch.item.ParseException;
 import org.springframework.batch.item.UnexpectedInputException;
 
+/**
+ * @Author - Kunwar Vikas
+ */
 public class SBReader implements ItemReader<String> {
 
 	private String[] messages = { "javadevjournal.com",

Spring-Boot-Batch/springbootbatch/src/main/java/com/javadevjournal/springbootbatch/step/SBWriter.java

@@ -6,6 +6,9 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.batch.item.ItemWriter;
 
+/**
+ * @Author - Kunwar Vikas
+ */
 public class SBWriter implements ItemWriter<String> {
 	Logger logger = LoggerFactory.getLogger(SBWriter.class);
 	@Override

Spring-Boot-Batch/springbootbatch/src/main/resources/csv/employees.csv

@@ -0,0 +1,4 @@
+Janardan, Srivastava, FCI
+Chandra, Srivastava, Powergrid
+Vijay, Verma, CDA
+Savita, Srivastava, NTPC

Spring-Boot-Batch/springbootbatch/src/test/java/com/javadevjournal/springbootbatch/SpringBootBatchBasicApplicationTests.java

@@ -3,6 +3,9 @@ package com.javadevjournal.springbootbatch;
 import org.junit.jupiter.api.Test;
 import org.springframework.boot.test.context.SpringBootTest;
 
+/**
+ * @Author - Kunwar Vikas
+ */
 @SpringBootTest
 class SpringBootBatchBasicApplicationTests {