Add an example for bulk imports with SpringBatch

jeffgbutler · jeffgbutler · commit bf66bae8302c · 2019-10-14T13:50:22.000-05:00
diff --git a/src/test/java/examples/springbatch/bulkinsert/BulkInsertConfiguration.java b/src/test/java/examples/springbatch/bulkinsert/BulkInsertConfiguration.java
@@ -0,0 +1,121 @@
+/**
+ *    Copyright 2016-2019 the original author or authors.
+ *
+ *    Licensed 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
+ *
+ *       http://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.
+ */
+package examples.springbatch.bulkinsert;
+
+import static examples.springbatch.mapper.PersonDynamicSqlSupport.firstName;
+import static examples.springbatch.mapper.PersonDynamicSqlSupport.forPagingTest;
+import static examples.springbatch.mapper.PersonDynamicSqlSupport.lastName;
+
+import javax.sql.DataSource;
+
+import org.apache.ibatis.session.SqlSessionFactory;
+import org.mybatis.dynamic.sql.insert.InsertDSL;
+import org.mybatis.dynamic.sql.render.RenderingStrategies;
+import org.mybatis.spring.SqlSessionFactoryBean;
+import org.mybatis.spring.annotation.MapperScan;
+import org.mybatis.spring.batch.MyBatisBatchItemWriter;
+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.core.launch.support.RunIdIncrementer;
+import org.springframework.batch.item.ItemProcessor;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
+import org.springframework.transaction.PlatformTransactionManager;
+
+import examples.springbatch.common.PersonRecord;
+import examples.springbatch.mapper.PersonDynamicSqlSupport;
+import examples.springbatch.mapper.PersonMapper;
+
+@EnableBatchProcessing
+@Configuration
+@ComponentScan("examples.springbatch.bulkinsert")
+@ComponentScan("examples.springbatch.common")
+@MapperScan("examples.springbatch.mapper")
+public class BulkInsertConfiguration {
+    
+    @Autowired
+    private JobBuilderFactory jobBuilderFactory;
+    
+    @Autowired
+    private StepBuilderFactory stepBuilderFactory;
+    
+    @Bean
+    public DataSource dataSource() {
+        return new EmbeddedDatabaseBuilder()
+                .setType(EmbeddedDatabaseType.HSQL)
+                .addScript("classpath:/org/springframework/batch/core/schema-drop-hsqldb.sql")
+                .addScript("classpath:/org/springframework/batch/core/schema-hsqldb.sql")
+                .addScript("classpath:/examples/springbatch/schema.sql")
+                .build();
+    }
+    
+    @Bean
+    public SqlSessionFactory sqlSessionFactory(DataSource dataSource) throws Exception {
+        SqlSessionFactoryBean sessionFactory = new SqlSessionFactoryBean();
+        sessionFactory.setDataSource(dataSource);
+        return sessionFactory.getObject();
+    }
+    
+    @Bean
+    public PlatformTransactionManager transactionManager(DataSource dataSource) {
+        return new DataSourceTransactionManager(dataSource);
+    }
+
+    @Bean
+    public MyBatisBatchItemWriter<PersonRecord> writer(SqlSessionFactory sqlSessionFactory) {
+        MyBatisBatchItemWriter<PersonRecord> writer = new MyBatisBatchItemWriter<>();
+        writer.setSqlSessionFactory(sqlSessionFactory);
+        
+        writer.setItemToParameterConverter(record -> InsertDSL.insert(record)
+                    .into(PersonDynamicSqlSupport.person)
+                    .map(firstName).toProperty("firstName")
+                    .map(lastName).toProperty("lastName")
+                    .map(forPagingTest).toStringConstant("false")
+                    .build()
+                    .render(RenderingStrategies.MYBATIS3));
+        
+        writer.setStatementId(PersonMapper.class.getName() + ".insert");
+        return writer;
+    }
+    
+    @Bean
+    public Step step1(ItemProcessor<PersonRecord, PersonRecord> processor, ItemWriter<PersonRecord> writer) {
+        return stepBuilderFactory.get("step1")
+                .<PersonRecord, PersonRecord>chunk(10)
+                .reader(new TestRecordGenerator())
+                .processor(processor)
+                .writer(writer)
+                .build();
+    }
+
+    @Bean
+    public Job insertRecords(Step step1) {
+        return jobBuilderFactory.get("insertRecords")
+                .incrementer(new RunIdIncrementer())
+                .flow(step1)
+                .end()
+                .build();
+    }
+}
diff --git a/src/test/java/examples/springbatch/bulkinsert/SpringBatchBulkInsertTest.java b/src/test/java/examples/springbatch/bulkinsert/SpringBatchBulkInsertTest.java
@@ -0,0 +1,83 @@
+/**
+ *    Copyright 2016-2019 the original author or authors.
+ *
+ *    Licensed 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
+ *
+ *       http://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.
+ */
+package examples.springbatch.bulkinsert;
+
+import static examples.springbatch.mapper.PersonDynamicSqlSupport.*;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.apache.ibatis.session.SqlSession;
+import org.apache.ibatis.session.SqlSessionFactory;
+import org.junit.jupiter.api.Test;
+import org.mybatis.dynamic.sql.render.RenderingStrategies;
+import org.mybatis.dynamic.sql.select.CountDSL;
+import org.mybatis.dynamic.sql.select.render.SelectStatementProvider;
+import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.test.JobLauncherTestUtils;
+import org.springframework.batch.test.context.SpringBatchTest;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import examples.springbatch.mapper.PersonMapper;
+
+@SpringBatchTest
+@SpringJUnitConfig(classes=BulkInsertConfiguration.class)
+public class SpringBatchBulkInsertTest {
+
+    @Autowired
+    private JobLauncherTestUtils jobLauncherTestUtils;
+
+    @Autowired
+    private SqlSessionFactory sqlSessionFactory;
+
+    @Test
+    public void testThatRowsAreInserted() throws Exception {
+        // starting condition
+        assertThat(rowCount()).isEqualTo(0);
+
+        JobExecution execution = jobLauncherTestUtils.launchJob();
+        assertThat(execution.getExitStatus()).isEqualTo(ExitStatus.COMPLETED);
+        assertThat(numberOfRowsProcessed(execution)).isEqualTo(TestRecordGenerator.recordCount());
+
+        // ending condition
+        assertThat(rowCount()).isEqualTo(TestRecordGenerator.recordCount());
+    }
+
+    private int numberOfRowsProcessed(JobExecution jobExecution) {
+        return jobExecution.getStepExecutions().stream()
+                .map(StepExecution::getExecutionContext)
+                .mapToInt(this::getRowCount)
+                .sum();
+    }
+    
+    private int getRowCount(ExecutionContext executionContext) {
+        return executionContext.getInt("row_count", 0);
+    }
+    
+    private long rowCount() throws Exception {
+        try (SqlSession sqlSession = sqlSessionFactory.openSession()) {
+            PersonMapper personMapper = sqlSession.getMapper(PersonMapper.class);
+
+            SelectStatementProvider selectStatement = CountDSL.countFrom(person)
+                    .build()
+                    .render(RenderingStrategies.MYBATIS3);
+
+            return personMapper.count(selectStatement);
+        }
+    }
+}
diff --git a/src/test/java/examples/springbatch/bulkinsert/TestRecordGenerator.java b/src/test/java/examples/springbatch/bulkinsert/TestRecordGenerator.java
@@ -0,0 +1,47 @@
+/**
+ *    Copyright 2016-2019 the original author or authors.
+ *
+ *    Licensed 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
+ *
+ *       http://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.
+ */
+package examples.springbatch.bulkinsert;
+
+import org.springframework.batch.item.ItemReader;
+
+import examples.springbatch.common.PersonRecord;
+
+public class TestRecordGenerator implements ItemReader<PersonRecord> {
+    
+    private int index = 0;
+    
+    private static PersonRecord[] testRecords = {
+            new PersonRecord("Fred", "Flintstone"),
+            new PersonRecord("Wilma", "Flintstone"),
+            new PersonRecord("Pebbles", "Flintstone"),
+            new PersonRecord("Barney", "Rubble"),
+            new PersonRecord("Betty", "Rubble"),
+            new PersonRecord("Bamm Bamm", "Rubble")
+    };
+
+    @Override
+    public PersonRecord read() {
+        if (index < testRecords.length) {
+            return(testRecords[index++]);
+        } else {
+            return null;
+        }
+    }
+    
+    public static int recordCount() {
+        return testRecords.length;
+    }
+}
diff --git a/src/test/java/examples/springbatch/common/PersonProcessor.java b/src/test/java/examples/springbatch/common/PersonProcessor.java
@@ -24,15 +24,15 @@
 import org.springframework.stereotype.Component;
 
 @Component
-public class PersonProcessor implements ItemProcessor<Person, Person> {
+public class PersonProcessor implements ItemProcessor<PersonRecord, PersonRecord> {
     
     private ExecutionContext executionContext;
 
     @Override
-    public Person process(Person person) throws Exception {
+    public PersonRecord process(PersonRecord person) throws Exception {
         incrementRowCount();
         
-        Person transformed = new Person();
+        PersonRecord transformed = new PersonRecord();
         transformed.setId(person.getId());
         transformed.setFirstName(person.getFirstName().toUpperCase());
         transformed.setLastName(person.getLastName().toUpperCase());
diff --git a/src/test/java/examples/springbatch/common/PersonRecord.java b/src/test/java/examples/springbatch/common/PersonRecord.java
@@ -15,11 +15,20 @@
  */
 package examples.springbatch.common;
 
-public class Person {
+public class PersonRecord {
     private Integer id;
     private String firstName;
     private String lastName;
 
+    public PersonRecord() {
+        super();
+    }
+    
+    public PersonRecord(String firstName, String lastName) {
+        this.firstName = firstName;
+        this.lastName = lastName;
+    }
+    
     public Integer getId() {
         return id;
     }
diff --git a/src/test/java/examples/springbatch/common/UpdateStatementConvertor.java b/src/test/java/examples/springbatch/common/UpdateStatementConvertor.java
@@ -25,10 +25,10 @@
 import org.springframework.stereotype.Component;
 
 @Component
-public class UpdateStatementConvertor implements Converter<Person, UpdateStatementProvider> {
+public class UpdateStatementConvertor implements Converter<PersonRecord, UpdateStatementProvider> {
 
     @Override
-    public UpdateStatementProvider convert(Person source) {
+    public UpdateStatementProvider convert(PersonRecord source) {
         return UpdateDSL.update(person)
                 .set(firstName).equalTo(source::getFirstName)
                 .set(lastName).equalTo(source::getLastName)
diff --git a/src/test/java/examples/springbatch/cursor/CursorReaderBatchConfiguration.java b/src/test/java/examples/springbatch/cursor/CursorReaderBatchConfiguration.java
@@ -48,7 +48,7 @@
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
 import org.springframework.transaction.PlatformTransactionManager;
 
-import examples.springbatch.common.Person;
+import examples.springbatch.common.PersonRecord;
 import examples.springbatch.mapper.PersonMapper;
 
 @EnableBatchProcessing
@@ -87,34 +87,34 @@ public PlatformTransactionManager transactionManager(DataSource dataSource) {
     }
 
     @Bean
-    public MyBatisCursorItemReader<Person> reader(SqlSessionFactory sqlSessionFactory) {
+    public MyBatisCursorItemReader<PersonRecord> reader(SqlSessionFactory sqlSessionFactory) {
         SelectStatementProvider selectStatement =  SpringBatchUtility.selectForCursor(person.allColumns())
                 .from(person)
                 .where(lastName, isEqualTo("flintstone"))
                 .build()
                 .render();
         
-        MyBatisCursorItemReader<Person> reader = new MyBatisCursorItemReader<>();
+        MyBatisCursorItemReader<PersonRecord> reader = new MyBatisCursorItemReader<>();
         reader.setQueryId(PersonMapper.class.getName() + ".selectMany");
         reader.setSqlSessionFactory(sqlSessionFactory);
         reader.setParameterValues(SpringBatchUtility.toParameterValues(selectStatement));
         return reader;
     }
     
     @Bean
-    public MyBatisBatchItemWriter<Person> writer(SqlSessionFactory sqlSessionFactory,
-            Converter<Person, UpdateStatementProvider> convertor) {
-        MyBatisBatchItemWriter<Person> writer = new MyBatisBatchItemWriter<>();
+    public MyBatisBatchItemWriter<PersonRecord> writer(SqlSessionFactory sqlSessionFactory,
+            Converter<PersonRecord, UpdateStatementProvider> convertor) {
+        MyBatisBatchItemWriter<PersonRecord> writer = new MyBatisBatchItemWriter<>();
         writer.setSqlSessionFactory(sqlSessionFactory);
         writer.setItemToParameterConverter(convertor);
         writer.setStatementId(PersonMapper.class.getName() + ".update");
         return writer;
     }
     
     @Bean
-    public Step step1(ItemReader<Person> reader, ItemProcessor<Person, Person> processor, ItemWriter<Person> writer) {
+    public Step step1(ItemReader<PersonRecord> reader, ItemProcessor<PersonRecord, PersonRecord> processor, ItemWriter<PersonRecord> writer) {
         return stepBuilderFactory.get("step1")
-                .<Person, Person>chunk(10)
+                .<PersonRecord, PersonRecord>chunk(10)
                 .reader(reader)
                 .processor(processor)
                 .writer(writer)
diff --git a/src/test/java/examples/springbatch/mapper/PersonMapper.java b/src/test/java/examples/springbatch/mapper/PersonMapper.java
@@ -18,17 +18,19 @@
 import java.util.List;
 import java.util.Map;
 
+import org.apache.ibatis.annotations.InsertProvider;
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Result;
 import org.apache.ibatis.annotations.Results;
 import org.apache.ibatis.annotations.SelectProvider;
 import org.apache.ibatis.annotations.UpdateProvider;
+import org.mybatis.dynamic.sql.insert.render.InsertStatementProvider;
 import org.mybatis.dynamic.sql.select.render.SelectStatementProvider;
 import org.mybatis.dynamic.sql.update.render.UpdateStatementProvider;
 import org.mybatis.dynamic.sql.util.SqlProviderAdapter;
 import org.mybatis.dynamic.sql.util.springbatch.SpringBatchProviderAdapter;
 
-import examples.springbatch.common.Person;
+import examples.springbatch.common.PersonRecord;
 
 @Mapper
 public interface PersonMapper {
@@ -39,11 +41,14 @@ public interface PersonMapper {
         @Result(column="first_name", property="firstName"),
         @Result(column="last_name", property="lastName")
     })
-    List<Person> selectMany(Map<String, Object> parameterValues);
+    List<PersonRecord> selectMany(Map<String, Object> parameterValues);
 
     @UpdateProvider(type=SqlProviderAdapter.class, method="update")
     int update(UpdateStatementProvider updateStatement);
 
     @SelectProvider(type=SqlProviderAdapter.class, method="select")
     long count(SelectStatementProvider selectStatement);
+    
+    @InsertProvider(type=SqlProviderAdapter.class, method="insert")
+    int insert(InsertStatementProvider<PersonRecord> insertStatement);
 }
diff --git a/src/test/java/examples/springbatch/paging/PagingReaderBatchConfiguration.java b/src/test/java/examples/springbatch/paging/PagingReaderBatchConfiguration.java
