Adding autoconfiguration for a JdbcCursorItemReader
This commit adds autoconfiguration for a JdbcCursorItemReader to the single step batch job starter. Fixed build Updates updated to clean db between tests
This commit is contained in:
Michael Minella
committed by
Glenn Renfro
Glenn Renfro
parent
468fab1b42
commit
15eec87b68
@@ -0,0 +1,108 @@
|
||||
/*
|
||||
* Copyright 2020-2020 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
|
||||
*
|
||||
* https://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 org.springframework.cloud.task.batch.autoconfigure.jdbc;
|
||||
|
||||
import java.sql.ResultSet;
|
||||
import java.sql.SQLException;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
|
||||
import org.springframework.batch.item.database.JdbcCursorItemReader;
|
||||
import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
|
||||
import org.springframework.boot.autoconfigure.batch.BatchAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.jdbc.core.PreparedStatementSetter;
|
||||
import org.springframework.jdbc.core.RowMapper;
|
||||
|
||||
/**
|
||||
* @author Michael Minella
|
||||
* @since 2.3
|
||||
*/
|
||||
@Configuration
|
||||
@EnableConfigurationProperties(JdbcCursorItemReaderProperties.class)
|
||||
@AutoConfigureAfter(BatchAutoConfiguration.class)
|
||||
@ConditionalOnProperty(prefix = "spring.batch.job.jdbccursorreader", name = "name")
|
||||
public class JdbcCursorItemReaderAutoConfiguration {
|
||||
|
||||
private final JdbcCursorItemReaderProperties properties;
|
||||
|
||||
private final DataSource dataSource;
|
||||
|
||||
@Autowired(required = false)
|
||||
private PreparedStatementSetter preparedStatementSetter;
|
||||
|
||||
@Autowired(required = false)
|
||||
private RowMapper<Map<Object, Object>> rowMapper;
|
||||
|
||||
public JdbcCursorItemReaderAutoConfiguration(
|
||||
JdbcCursorItemReaderProperties properties, DataSource dataSource) {
|
||||
this.properties = properties;
|
||||
this.dataSource = dataSource;
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public JdbcCursorItemReader<Map<Object, Object>> itemReader() {
|
||||
return new JdbcCursorItemReaderBuilder<Map<Object, Object>>()
|
||||
.name(this.properties.getName())
|
||||
.currentItemCount(this.properties.getCurrentItemCount())
|
||||
.dataSource(this.dataSource)
|
||||
.driverSupportsAbsolute(this.properties.isDriverSupportsAbsolute())
|
||||
.fetchSize(this.properties.getFetchSize())
|
||||
.ignoreWarnings(this.properties.isIgnoreWarnings())
|
||||
.maxItemCount(this.properties.getMaxItemCount())
|
||||
.maxRows(this.properties.getMaxRows())
|
||||
.queryTimeout(this.properties.getQueryTimeout())
|
||||
.saveState(this.properties.isSaveState()).sql(this.properties.getSql())
|
||||
.rowMapper(this.rowMapper)
|
||||
.preparedStatementSetter(this.preparedStatementSetter)
|
||||
.verifyCursorPosition(this.properties.isVerifyCursorPosition())
|
||||
.useSharedExtendedConnection(
|
||||
this.properties.isUseSharedExtendedConnection())
|
||||
.build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public RowMapper<Map<Object, Object>> rowMapper() {
|
||||
return new MapRowMapper();
|
||||
}
|
||||
|
||||
public static class MapRowMapper implements RowMapper<Map<Object, Object>> {
|
||||
|
||||
@Override
|
||||
public Map<Object, Object> mapRow(ResultSet rs, int rowNum) throws SQLException {
|
||||
Map<Object, Object> item = new HashMap<>(rs.getMetaData().getColumnCount());
|
||||
|
||||
for (int i = 1; i <= rs.getMetaData().getColumnCount(); i++) {
|
||||
item.put(rs.getMetaData().getColumnName(i), rs.getObject(i));
|
||||
}
|
||||
|
||||
return item;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,265 @@
|
||||
/*
|
||||
* Copyright 2020-2020 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
|
||||
*
|
||||
* https://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 org.springframework.cloud.task.batch.autoconfigure.jdbc;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
/**
|
||||
* @author Michael Minella
|
||||
* @since 2.3
|
||||
*/
|
||||
@ConfigurationProperties(prefix = "spring.batch.job.jdbccursorreader")
|
||||
public class JdbcCursorItemReaderProperties {
|
||||
|
||||
private boolean saveState = true;
|
||||
|
||||
private String name;
|
||||
|
||||
private int maxItemCount = Integer.MAX_VALUE;
|
||||
|
||||
private int currentItemCount = 0;
|
||||
|
||||
private int fetchSize;
|
||||
|
||||
private int maxRows;
|
||||
|
||||
private int queryTimeout;
|
||||
|
||||
private boolean ignoreWarnings;
|
||||
|
||||
private boolean verifyCursorPosition;
|
||||
|
||||
private boolean driverSupportsAbsolute;
|
||||
|
||||
private boolean useSharedExtendedConnection;
|
||||
|
||||
private String sql;
|
||||
|
||||
/**
|
||||
* Returns the configured value of if the state of the reader will be persisted.
|
||||
* @return true if the state will be persisted
|
||||
*/
|
||||
public boolean isSaveState() {
|
||||
return this.saveState;
|
||||
}
|
||||
|
||||
/**
|
||||
* Configure if the state of the
|
||||
* {@link org.springframework.batch.item.ItemStreamSupport} should be persisted within
|
||||
* the {@link org.springframework.batch.item.ExecutionContext} for restart purposes.
|
||||
* @param saveState defaults to true
|
||||
*/
|
||||
public void setSaveState(boolean saveState) {
|
||||
this.saveState = saveState;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the configured value of the name used to calculate {@code ExecutionContext}
|
||||
* keys.
|
||||
* @return the name
|
||||
*/
|
||||
public String getName() {
|
||||
return this.name;
|
||||
}
|
||||
|
||||
/**
|
||||
* The name used to calculate the key within the
|
||||
* {@link org.springframework.batch.item.ExecutionContext}. Required if
|
||||
* {@link #setSaveState} is set to true.
|
||||
* @param name name of the reader instance
|
||||
* @see org.springframework.batch.item.ItemStreamSupport#setName(String)
|
||||
*/
|
||||
public void setName(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
/**
|
||||
* The maximum number of items to be read.
|
||||
* @return the configured number of items, defaults to Integer.MAX_VALUE
|
||||
*/
|
||||
public int getMaxItemCount() {
|
||||
return this.maxItemCount;
|
||||
}
|
||||
|
||||
/**
|
||||
* Configure the max number of items to be read.
|
||||
* @param maxItemCount the max items to be read
|
||||
* @see org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader#setMaxItemCount(int)
|
||||
*/
|
||||
public void setMaxItemCount(int maxItemCount) {
|
||||
this.maxItemCount = maxItemCount;
|
||||
}
|
||||
|
||||
/**
|
||||
* Provides the index of the current item.
|
||||
* @return item index
|
||||
*/
|
||||
public int getCurrentItemCount() {
|
||||
return this.currentItemCount;
|
||||
}
|
||||
|
||||
/**
|
||||
* Index for the current item. Also used on restarts to indicate where to start from.
|
||||
* @param currentItemCount current index
|
||||
* @see org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader#setCurrentItemCount(int)
|
||||
*/
|
||||
public void setCurrentItemCount(int currentItemCount) {
|
||||
this.currentItemCount = currentItemCount;
|
||||
}
|
||||
|
||||
/**
|
||||
* Provides the number of items to return each time the cursor fetches from the
|
||||
* server.
|
||||
* @return fetch size
|
||||
*/
|
||||
public int getFetchSize() {
|
||||
return fetchSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the number of items to return each time the cursor fetches from the server.
|
||||
* @param fetchSize the number of items
|
||||
* @see org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder#fetchSize(int)
|
||||
*/
|
||||
public void setFetchSize(int fetchSize) {
|
||||
this.fetchSize = fetchSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* Provides the maximum number of rows to read with this reader.
|
||||
* @return maxiumum number of items
|
||||
*/
|
||||
public int getMaxRows() {
|
||||
return maxRows;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the maximum number of rows to be read with this reader.
|
||||
* @param maxRows maximum number of items
|
||||
* @see org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder#maxRows(int)
|
||||
*/
|
||||
public void setMaxRows(int maxRows) {
|
||||
this.maxRows = maxRows;
|
||||
}
|
||||
|
||||
/**
|
||||
* Provides the time in milliseconds for the query to timeout.
|
||||
* @return milliseconds for the timeout
|
||||
*/
|
||||
public int getQueryTimeout() {
|
||||
return queryTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the time in milliseconds for the query to timeout.
|
||||
* @param queryTimeout milliseconds
|
||||
* @see org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder#queryTimeout(int)
|
||||
*/
|
||||
public void setQueryTimeout(int queryTimeout) {
|
||||
this.queryTimeout = queryTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* Provides if SQL warnings should be ignored.
|
||||
* @return true if warnings should be ignored
|
||||
*/
|
||||
public boolean isIgnoreWarnings() {
|
||||
return ignoreWarnings;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets if SQL warnings should be ignored.
|
||||
* @param ignoreWarnings indicator if the warnings should be ignored
|
||||
* @see org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder#ignoreWarnings(boolean)
|
||||
*/
|
||||
public void setIgnoreWarnings(boolean ignoreWarnings) {
|
||||
this.ignoreWarnings = ignoreWarnings;
|
||||
}
|
||||
|
||||
/**
|
||||
* Indicates if the cursor's position should be validated with each item read (to
|
||||
* confirm that the RowMapper has not moved the cursor's location).
|
||||
* @return true if the position should be validated
|
||||
*/
|
||||
public boolean isVerifyCursorPosition() {
|
||||
return verifyCursorPosition;
|
||||
}
|
||||
|
||||
/**
|
||||
* Provides if the cursor's position should be validated with each item read.
|
||||
* @param verifyCursorPosition true if the position should be validated
|
||||
* @see org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder#verifyCursorPosition(boolean)
|
||||
*/
|
||||
public void setVerifyCursorPosition(boolean verifyCursorPosition) {
|
||||
this.verifyCursorPosition = verifyCursorPosition;
|
||||
}
|
||||
|
||||
/**
|
||||
* Provides if the driver supports absolute positioning of a cursor.
|
||||
* @return true if the driver supports absolute positioning
|
||||
*/
|
||||
public boolean isDriverSupportsAbsolute() {
|
||||
return driverSupportsAbsolute;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets if the driver supports absolute positioning of a cursor.
|
||||
* @param driverSupportsAbsolute true if the driver supports absolute positioning
|
||||
* @see org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder#driverSupportsAbsolute(boolean)
|
||||
*/
|
||||
public void setDriverSupportsAbsolute(boolean driverSupportsAbsolute) {
|
||||
this.driverSupportsAbsolute = driverSupportsAbsolute;
|
||||
}
|
||||
|
||||
/**
|
||||
* Provides if the the connection used for the cursor is being used by all other
|
||||
* processing, therefor part of the same transaction.
|
||||
* @return true if the connection is shared beyond this query
|
||||
*/
|
||||
public boolean isUseSharedExtendedConnection() {
|
||||
return useSharedExtendedConnection;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets if the the connection used for the cursor is being used by all other
|
||||
* processing, therefor part of the same transaction.
|
||||
* @param useSharedExtendedConnection true if the connection is shared beyond this
|
||||
* query
|
||||
* @see org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder#useSharedExtendedConnection(boolean)
|
||||
*/
|
||||
public void setUseSharedExtendedConnection(boolean useSharedExtendedConnection) {
|
||||
this.useSharedExtendedConnection = useSharedExtendedConnection;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the SQL query to be executed.
|
||||
* @return the SQL query
|
||||
*/
|
||||
public String getSql() {
|
||||
return sql;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the SQL query to be executed.
|
||||
* @param sql the query
|
||||
* @see org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder#sql(String)
|
||||
*/
|
||||
public void setSql(String sql) {
|
||||
this.sql = sql;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -2,4 +2,5 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration=org.springframewo
|
||||
org.springframework.cloud.task.batch.autoconfigure.RangeConverter,\
|
||||
org.springframework.cloud.task.batch.autoconfigure.SingleStepJobAutoConfiguration,\
|
||||
org.springframework.cloud.task.batch.autoconfigure.FlatFileItemWriterAutoConfiguration, \
|
||||
org.springframework.cloud.task.batch.autoconfigure.jdbc.JdbcItemWriterAutoConfiguration
|
||||
org.springframework.cloud.task.batch.autoconfigure.jdbc.JdbcItemWriterAutoConfiguration, \
|
||||
org.springframework.cloud.task.batch.autoconfigure.jdbc.JdbcCursorItemReaderAutoConfiguration
|
||||
|
||||
@@ -0,0 +1,371 @@
|
||||
/*
|
||||
* Copyright 2020-2020 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
|
||||
*
|
||||
* https://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 org.springframework.cloud.task.batch.autoconfigure.jdbc;
|
||||
|
||||
import java.sql.SQLException;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
|
||||
import org.h2.tools.Server;
|
||||
import org.junit.jupiter.api.AfterAll;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.batch.core.Job;
|
||||
import org.springframework.batch.core.JobExecution;
|
||||
import org.springframework.batch.core.JobParameters;
|
||||
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
|
||||
import org.springframework.batch.core.explore.JobExplorer;
|
||||
import org.springframework.batch.core.launch.JobLauncher;
|
||||
import org.springframework.batch.item.database.JdbcCursorItemReader;
|
||||
import org.springframework.batch.item.support.ListItemWriter;
|
||||
import org.springframework.batch.item.util.ExecutionContextUserSupport;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigurations;
|
||||
import org.springframework.boot.autoconfigure.batch.BatchAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
|
||||
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
|
||||
import org.springframework.cloud.task.batch.autoconfigure.SingleStepJobAutoConfiguration;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.core.io.ClassPathResource;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.core.RowMapper;
|
||||
import org.springframework.jdbc.datasource.DriverManagerDataSource;
|
||||
import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
import org.springframework.util.SocketUtils;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
/**
|
||||
* @author Michael Minella
|
||||
*/
|
||||
public class JdbcCursorItemReaderAutoConfigurationTests {
|
||||
|
||||
private final static String DATASOURCE_URL;
|
||||
|
||||
private final static String DATASOURCE_USER_NAME = "SA";
|
||||
|
||||
private final static String DATASOURCE_USER_PASSWORD = "''";
|
||||
|
||||
private final static String DATASOURCE_DRIVER_CLASS_NAME = "org.h2.Driver";
|
||||
|
||||
private static int randomPort;
|
||||
|
||||
static {
|
||||
randomPort = SocketUtils.findAvailableTcpPort();
|
||||
DATASOURCE_URL = "jdbc:h2:tcp://localhost:" + randomPort
|
||||
+ "/mem:dataflow;DB_CLOSE_DELAY=-1;" + "DB_CLOSE_ON_EXIT=FALSE";
|
||||
}
|
||||
|
||||
@AfterAll
|
||||
public static void clearDB() {
|
||||
DriverManagerDataSource dataSource = new DriverManagerDataSource();
|
||||
dataSource.setDriverClassName(DATASOURCE_DRIVER_CLASS_NAME);
|
||||
dataSource.setUrl(DATASOURCE_URL);
|
||||
dataSource.setUsername(DATASOURCE_USER_NAME);
|
||||
dataSource.setPassword(DATASOURCE_USER_PASSWORD);
|
||||
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
|
||||
jdbcTemplate.execute("TRUNCATE TABLE item");
|
||||
jdbcTemplate.execute("DROP TABLE BATCH_JOB_EXECUTION CASCADE");
|
||||
jdbcTemplate.execute("DROP TABLE BATCH_JOB_INSTANCE CASCADE");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testIntegration() {
|
||||
ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner()
|
||||
.withUserConfiguration(BaseConfiguration.class,
|
||||
TaskLauncherConfiguration.class)
|
||||
.withConfiguration(
|
||||
AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class,
|
||||
BatchAutoConfiguration.class,
|
||||
SingleStepJobAutoConfiguration.class,
|
||||
JdbcCursorItemReaderAutoConfiguration.class))
|
||||
.withPropertyValues("spring.batch.job.jobName=integrationJob",
|
||||
"spring.batch.job.stepName=step1", "spring.batch.job.chunkSize=5",
|
||||
"spring.batch.job.jdbccursorreader.name=fooReader",
|
||||
"spring.batch.job.jdbccursorreader.sql=select item_name from item");
|
||||
|
||||
applicationContextRunner.run((context) -> {
|
||||
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
|
||||
|
||||
Job job = context.getBean(Job.class);
|
||||
|
||||
JobExecution jobExecution = jobLauncher.run(job, new JobParameters());
|
||||
|
||||
JobExplorer jobExplorer = context.getBean(JobExplorer.class);
|
||||
|
||||
while (jobExplorer.getJobExecution(jobExecution.getJobId()).isRunning()) {
|
||||
Thread.sleep(1000);
|
||||
}
|
||||
|
||||
List<Map<Object, Object>> items = context.getBean(ListItemWriter.class)
|
||||
.getWrittenItems();
|
||||
|
||||
assertThat(items.size()).isEqualTo(3);
|
||||
assertThat(items.get(0).get("ITEM_NAME")).isEqualTo("foo");
|
||||
assertThat(items.get(1).get("ITEM_NAME")).isEqualTo("bar");
|
||||
assertThat(items.get(2).get("ITEM_NAME")).isEqualTo("baz");
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCustomRowMapper() {
|
||||
ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner()
|
||||
.withUserConfiguration(RowMapperConfiguration.class,
|
||||
TaskLauncherConfiguration.class)
|
||||
.withConfiguration(
|
||||
AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class,
|
||||
BatchAutoConfiguration.class,
|
||||
SingleStepJobAutoConfiguration.class,
|
||||
JdbcCursorItemReaderAutoConfiguration.class))
|
||||
.withPropertyValues("spring.batch.job.jobName=rowMapperJob",
|
||||
"spring.batch.job.stepName=step1", "spring.batch.job.chunkSize=5",
|
||||
"spring.batch.job.jdbccursorreader.name=fooReader",
|
||||
"spring.batch.job.jdbccursorreader.sql=select * from item");
|
||||
|
||||
applicationContextRunner.run((context) -> {
|
||||
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
|
||||
|
||||
Job job = context.getBean(Job.class);
|
||||
|
||||
JobExecution jobExecution = jobLauncher.run(job, new JobParameters());
|
||||
|
||||
JobExplorer jobExplorer = context.getBean(JobExplorer.class);
|
||||
|
||||
while (jobExplorer.getJobExecution(jobExecution.getJobId()).isRunning()) {
|
||||
Thread.sleep(1000);
|
||||
}
|
||||
|
||||
List<Map<Object, Object>> items = context.getBean(ListItemWriter.class)
|
||||
.getWrittenItems();
|
||||
|
||||
assertThat(items.size()).isEqualTo(3);
|
||||
assertThat(items.get(0).get("item")).isEqualTo("foo");
|
||||
assertThat(items.get(1).get("item")).isEqualTo("bar");
|
||||
assertThat(items.get(2).get("item")).isEqualTo("baz");
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRoseyScenario() {
|
||||
final ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner()
|
||||
.withUserConfiguration(BaseConfiguration.class,
|
||||
TaskLauncherConfiguration.class)
|
||||
.withConfiguration(
|
||||
AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class,
|
||||
BatchAutoConfiguration.class,
|
||||
SingleStepJobAutoConfiguration.class,
|
||||
JdbcCursorItemReaderAutoConfiguration.class))
|
||||
.withPropertyValues("spring.batch.job.jobName=roseyJob",
|
||||
"spring.batch.job.stepName=step1", "spring.batch.job.chunkSize=5",
|
||||
"spring.batch.job.jdbccursorreader.saveState=false",
|
||||
"spring.batch.job.jdbccursorreader.name=fooReader",
|
||||
"spring.batch.job.jdbccursorreader.maxItemCount=15",
|
||||
"spring.batch.job.jdbccursorreader.currentItemCount=2",
|
||||
"spring.batch.job.jdbccursorreader.fetchSize=4",
|
||||
"spring.batch.job.jdbccursorreader.maxRows=6",
|
||||
"spring.batch.job.jdbccursorreader.queryTimeout=8",
|
||||
"spring.batch.job.jdbccursorreader.ignoreWarnings=true",
|
||||
"spring.batch.job.jdbccursorreader.verifyCursorPosition=true",
|
||||
"spring.batch.job.jdbccursorreader.driverSupportsAbsolute=true",
|
||||
"spring.batch.job.jdbccursorreader.useSharedExtendedConnection=true",
|
||||
"spring.batch.job.jdbccursorreader.sql=select * from foo");
|
||||
|
||||
applicationContextRunner.run((context) -> {
|
||||
|
||||
JdbcCursorItemReader<Map<Object, Object>> itemReader = context
|
||||
.getBean(JdbcCursorItemReader.class);
|
||||
|
||||
validateBean(itemReader);
|
||||
});
|
||||
}
|
||||
|
||||
private void validateBean(JdbcCursorItemReader itemReader) {
|
||||
assertThat(itemReader.getSql()).isEqualTo("select * from foo");
|
||||
assertThat(itemReader.getDataSource()).isNotNull();
|
||||
assertThat((Boolean) ReflectionTestUtils.getField(itemReader, "saveState"))
|
||||
.isFalse();
|
||||
assertThat(
|
||||
ReflectionTestUtils.getField(
|
||||
(ExecutionContextUserSupport) ReflectionTestUtils
|
||||
.getField(itemReader, "executionContextUserSupport"),
|
||||
"name")).isEqualTo("fooReader");
|
||||
assertThat((Integer) ReflectionTestUtils.getField(itemReader, "maxItemCount"))
|
||||
.isEqualTo(15);
|
||||
assertThat((Integer) ReflectionTestUtils.getField(itemReader, "currentItemCount"))
|
||||
.isEqualTo(2);
|
||||
assertThat((Integer) ReflectionTestUtils.getField(itemReader, "fetchSize"))
|
||||
.isEqualTo(4);
|
||||
assertThat((Integer) ReflectionTestUtils.getField(itemReader, "maxRows"))
|
||||
.isEqualTo(6);
|
||||
assertThat((Integer) ReflectionTestUtils.getField(itemReader, "queryTimeout"))
|
||||
.isEqualTo(8);
|
||||
assertThat((Boolean) ReflectionTestUtils.getField(itemReader, "ignoreWarnings"))
|
||||
.isTrue();
|
||||
assertThat((Boolean) ReflectionTestUtils.getField(itemReader,
|
||||
"verifyCursorPosition")).isTrue();
|
||||
assertThat((Boolean) ReflectionTestUtils.getField(itemReader,
|
||||
"driverSupportsAbsolute")).isTrue();
|
||||
assertThat((Boolean) ReflectionTestUtils.getField(itemReader,
|
||||
"useSharedExtendedConnection")).isTrue();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testNoName() {
|
||||
final ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner()
|
||||
.withUserConfiguration(BaseConfiguration.class,
|
||||
TaskLauncherConfiguration.class)
|
||||
.withConfiguration(
|
||||
AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class,
|
||||
BatchAutoConfiguration.class,
|
||||
SingleStepJobAutoConfiguration.class,
|
||||
JdbcCursorItemReaderAutoConfiguration.class))
|
||||
.withPropertyValues("spring.batch.job.jobName=noNameJob",
|
||||
"spring.batch.job.stepName=step1",
|
||||
"spring.batch.job.chunkSize=5");
|
||||
|
||||
assertThatThrownBy(() -> {
|
||||
runTest(applicationContextRunner);
|
||||
}).isInstanceOf(IllegalStateException.class)
|
||||
.hasMessageContaining("UnsatisfiedDependencyException");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSqlName() {
|
||||
final ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner()
|
||||
.withUserConfiguration(BaseConfiguration.class,
|
||||
TaskLauncherConfiguration.class)
|
||||
.withConfiguration(
|
||||
AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class,
|
||||
BatchAutoConfiguration.class,
|
||||
SingleStepJobAutoConfiguration.class,
|
||||
JdbcCursorItemReaderAutoConfiguration.class))
|
||||
.withPropertyValues("spring.batch.job.jobName=job",
|
||||
"spring.batch.job.stepName=step1", "spring.batch.job.chunkSize=5",
|
||||
"spring.batch.job.jdbccursorreader.name=fooReader");
|
||||
|
||||
assertThatThrownBy(() -> {
|
||||
runTest(applicationContextRunner);
|
||||
}).isInstanceOf(IllegalStateException.class)
|
||||
.hasMessageContaining("UnsatisfiedDependencyException");
|
||||
}
|
||||
|
||||
private void runTest(ApplicationContextRunner applicationContextRunner) {
|
||||
applicationContextRunner.run((context) -> {
|
||||
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
|
||||
|
||||
Job job = context.getBean(Job.class);
|
||||
|
||||
JobExecution jobExecution = jobLauncher.run(job, new JobParameters());
|
||||
|
||||
JobExplorer jobExplorer = context.getBean(JobExplorer.class);
|
||||
|
||||
while (jobExplorer.getJobExecution(jobExecution.getJobId()).isRunning()) {
|
||||
Thread.sleep(1000);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Configuration
|
||||
public static class TaskLauncherConfiguration {
|
||||
|
||||
private static Server defaultServer;
|
||||
|
||||
@Bean
|
||||
public Server initH2TCPServer() {
|
||||
Server server = null;
|
||||
try {
|
||||
if (defaultServer == null) {
|
||||
server = Server.createTcpServer("-ifNotExists", "-tcp",
|
||||
"-tcpAllowOthers", "-tcpPort", String.valueOf(randomPort))
|
||||
.start();
|
||||
defaultServer = server;
|
||||
DriverManagerDataSource dataSource = new DriverManagerDataSource();
|
||||
dataSource.setDriverClassName(DATASOURCE_DRIVER_CLASS_NAME);
|
||||
dataSource.setUrl(DATASOURCE_URL);
|
||||
dataSource.setUsername(DATASOURCE_USER_NAME);
|
||||
dataSource.setPassword(DATASOURCE_USER_PASSWORD);
|
||||
ClassPathResource setupResource = new ClassPathResource(
|
||||
"schema-h2.sql");
|
||||
ResourceDatabasePopulator resourceDatabasePopulator = new ResourceDatabasePopulator(
|
||||
setupResource);
|
||||
resourceDatabasePopulator.setContinueOnError(true);
|
||||
resourceDatabasePopulator.execute(dataSource);
|
||||
|
||||
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
|
||||
jdbcTemplate.execute("TRUNCATE TABLE item");
|
||||
jdbcTemplate.execute("INSERT INTO item VALUES ('foo')");
|
||||
jdbcTemplate.execute("INSERT INTO item VALUES ('bar')");
|
||||
jdbcTemplate.execute("INSERT INTO item VALUES ('baz')");
|
||||
}
|
||||
}
|
||||
catch (SQLException e) {
|
||||
throw new IllegalStateException(e);
|
||||
}
|
||||
return defaultServer;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DataSource dataSource() {
|
||||
DriverManagerDataSource dataSource = new DriverManagerDataSource();
|
||||
dataSource.setDriverClassName(DATASOURCE_DRIVER_CLASS_NAME);
|
||||
dataSource.setUrl(DATASOURCE_URL);
|
||||
dataSource.setUsername(DATASOURCE_USER_NAME);
|
||||
dataSource.setPassword(DATASOURCE_USER_PASSWORD);
|
||||
return dataSource;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@EnableBatchProcessing
|
||||
@Configuration
|
||||
public static class BaseConfiguration {
|
||||
|
||||
@Bean
|
||||
public ListItemWriter<Map<Object, Object>> itemWriter() {
|
||||
return new ListItemWriter<>();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@EnableBatchProcessing
|
||||
@Configuration
|
||||
public static class RowMapperConfiguration {
|
||||
|
||||
@Bean
|
||||
public RowMapper<Map<Object, Object>> rowMapper() {
|
||||
return (rs, rowNum) -> {
|
||||
Map<Object, Object> item = new HashMap<>();
|
||||
|
||||
item.put("item", rs.getString("item_name"));
|
||||
|
||||
return item;
|
||||
};
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ListItemWriter<Map<Object, Object>> itemWriter() {
|
||||
return new ListItemWriter<>();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
CREATE TABLE item
|
||||
CREATE TABLE IF NOT EXISTS item
|
||||
(
|
||||
item_name varchar(55)
|
||||
);
|
||||
|
||||
Reference in New Issue
Block a user