Merge spring-batch-core-tests in spring-batch-core

This commit is contained in:
Mahmoud Ben Hassine
2020-12-04 16:03:58 +01:00
parent bf612650ea
commit 236ee88dcf
72 changed files with 1602 additions and 2125 deletions

View File

@@ -299,10 +299,14 @@ project('spring-batch-core') {
testCompile "javax.inject:javax.inject:$javaxInjectVersion"
testCompile "org.hsqldb:hsqldb:$hsqldbVersion"
testCompile "com.h2database:h2:$h2databaseVersion"
testCompile "org.apache.derby:derby:$derbyVersion"
testCompile "commons-io:commons-io:$commonsIoVersion"
testCompile "org.apache.commons:commons-dbcp2:$commonsDdbcpVersion"
testCompile "junit:junit:${junitVersion}"
testCompile "org.hamcrest:hamcrest-library:$hamcrestVersion"
testCompile "org.testcontainers:mysql:$testcontainersVersion"
testCompile "mysql:mysql-connector-java:$mysqlVersion"
optional "com.ibm.jbatch:com.ibm.jbatch-tck-spi:$jbatchTckSpi"
optional "com.thoughtworks.xstream:xstream:$xstreamVersion"
optional "org.aspectj:aspectjrt:$aspectjVersion"
@@ -417,44 +421,6 @@ project('spring-batch-docs') {
description = 'Spring Batch Docs'
}
project('spring-batch-core-tests') {
description = 'Spring Batch Core Tests'
project.tasks.findByPath("artifactoryPublish")?.enabled = false
dependencies {
compile project(":spring-batch-core")
compile project(":spring-batch-infrastructure")
compile "org.apache.commons:commons-dbcp2:$commonsDdbcpVersion"
compile "org.springframework:spring-jdbc:$springVersion"
compile "org.springframework.retry:spring-retry:$springRetryVersion"
compile "org.springframework:spring-tx:$springVersion"
compile "org.springframework:spring-aop:$springVersion"
testCompile "org.hsqldb:hsqldb:$hsqldbVersion"
testCompile "commons-io:commons-io:$commonsIoVersion"
testCompile "org.apache.derby:derby:$derbyVersion"
testCompile "junit:junit:${junitVersion}"
testCompile "org.hamcrest:hamcrest-library:$hamcrestVersion"
testCompile "org.apache.logging.log4j:log4j-api:$log4jVersion"
testCompile "org.apache.logging.log4j:log4j-core:$log4jVersion"
testCompile "org.springframework:spring-test:$springVersion"
testCompile "org.springframework:spring-jdbc:$springVersion"
testCompile "org.testcontainers:mysql:$testcontainersVersion"
testCompile "mysql:mysql-connector-java:$mysqlVersion"
runtime "org.postgresql:postgresql:$postgresqlVersion"
runtime "javax.batch:javax.batch-api:$javaxBatchApiVersion"
optional "org.aspectj:aspectjrt:$aspectjVersion"
optional "org.aspectj:aspectjweaver:$aspectjVersion"
optional "org.springframework.ldap:spring-ldap-core:$springLdapVersion"
optional "org.springframework.ldap:spring-ldap-core-tiger:$springLdapVersion"
optional "org.springframework.ldap:spring-ldap-ldif-core:$springLdapVersion"
}
test {
enabled = project.hasProperty('alltests')
}
}
//Domain for batch job testing
project('spring-batch-test') {
description = 'Spring Batch Test'

View File

@@ -1,299 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beansProjectDescription>
<version>1</version>
<pluginVersion><![CDATA[3.3.0.201307091516-RELEASE]]></pluginVersion>
<configSuffixes>
<configSuffix><![CDATA[xml]]></configSuffix>
</configSuffixes>
<enableImports><![CDATA[false]]></enableImports>
<configs>
<config>src/test/resources/data-source-context.xml</config>
<config>src/test/resources/simple-job-launcher-context.xml</config>
<config>src/test/resources/org/springframework/batch/core/test/step/SplitJobMapRepositoryIntegrationTests-context.xml</config>
<config>src/main/resources/META-INF/batch/footballJob.xml</config>
<config>src/main/resources/META-INF/batch/footballSkipJob.xml</config>
<config>src/main/resources/META-INF/batch/parallelJob.xml</config>
<config>src/main/resources/META-INF/batch/timeoutJob.xml</config>
</configs>
<configSets>
<configSet>
<name><![CDATA[repositoryTest]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[beanWrapper]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[delegating]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[football]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[infiniteLoop]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[multiline]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[parallel]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[restart]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[retry]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[trade]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[batchUpdate]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[quartzLauncher]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[compositeItemWriter]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[adhocLoop]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[staging]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[common]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[jmxLauncher]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[headerFooter]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[customerFilterJob]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[hibernate]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[multiResourceJob]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[skipSample]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[iosample-hibernate]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[iosample-jdbc]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[iosample-multiline]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[iosample-multiRecordType]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[iosample-jpa]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[data-source-contex]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[partition]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[multilineOrder]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
<configSet>
<name><![CDATA[restartFile]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
<profiles>
</profiles>
</configSet>
</configSets>
</beansProjectDescription>

View File

@@ -1,12 +0,0 @@
log4j.rootCategory=INFO, stdout
log4j.appender.stdout=org.apache.logging.log4j.core.appender.ConsoleAppender
log4j.appender.stdout.layout=org.apache.logging.log4j.core.layout.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{1}:%L - %m%n
log4j.category.org.apache.activemq=ERROR
# log4j.category.org.springframework=DEBUG
log4j.category.org.springframework.jdbc=DEBUG
log4j.category.org.springframework.jms=DEBUG
log4j.category.org.springframework.batch=DEBUG
log4j.category.org.springframework.retry=DEBUG

View File

@@ -1,165 +0,0 @@
/*
* Copyright 2006-2007 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 test.jdbc.datasource;
import java.io.IOException;
import java.util.List;
import java.util.Arrays;
import javax.sql.DataSource;
import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.io.Resource;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
public class DataSourceInitializer implements InitializingBean, DisposableBean {
private Resource[] initScripts;
private Resource destroyScript;
private DataSource dataSource;
private boolean initialize = false;
private Log logger = LogFactory.getLog(getClass());
private static boolean initialized = false;
public void setInitialize(boolean initialize) {
this.initialize = initialize;
}
@Override
public void destroy() throws Exception {
if (!initialized) {
return;
}
try {
if (destroyScript!=null) {
doExecuteScript(destroyScript);
initialized = false;
}
}
catch (Exception e) {
if (logger.isDebugEnabled()) {
logger.warn("Could not execute destroy script [" + destroyScript + "]", e);
}
else {
logger.warn("Could not execute destroy script [" + destroyScript + "]");
}
}
}
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(dataSource, "A DataSource is required");
logger.info("Initializing with scripts: "+Arrays.asList(initScripts));
if (!initialized && initialize) {
try {
doExecuteScript(destroyScript);
}
catch (Exception e) {
logger.debug("Could not execute destroy script [" + destroyScript + "]", e);
}
if (initScripts != null) {
for (int i = 0; i < initScripts.length; i++) {
Resource initScript = initScripts[i];
logger.info("Executing init script: "+initScript);
doExecuteScript(initScript);
}
}
initialized = true;
}
}
private void doExecuteScript(final Resource scriptResource) {
if (scriptResource == null || !scriptResource.exists())
return;
TransactionTemplate transactionTemplate = new TransactionTemplate(new DataSourceTransactionManager(dataSource));
transactionTemplate.execute(new TransactionCallback<Void>() {
@Override
@SuppressWarnings("unchecked")
public Void doInTransaction(TransactionStatus status) {
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
String[] scripts;
try {
scripts = StringUtils.delimitedListToStringArray(stripComments(IOUtils.readLines(scriptResource
.getInputStream(), "UTF-8")), ";");
}
catch (IOException e) {
throw new BeanInitializationException("Cannot load script from [" + scriptResource + "]", e);
}
for (int i = 0; i < scripts.length; i++) {
String script = scripts[i].trim();
if (StringUtils.hasText(script)) {
try {
jdbcTemplate.execute(scripts[i]);
} catch (DataAccessException e) {
if (!script.toUpperCase().startsWith("DROP")) {
throw e;
}
}
}
}
return null;
}
});
}
private String stripComments(List<String> list) {
StringBuilder buffer = new StringBuilder();
for (String line : list) {
if (!line.startsWith("//") && !line.startsWith("--")) {
buffer.append(line).append("\n");
}
}
return buffer.toString();
}
public Class<DataSource> getObjectType() {
return DataSource.class;
}
public void setInitScripts(Resource[] initScripts) {
this.initScripts = initScripts;
}
public void setDestroyScript(Resource destroyScript) {
this.destroyScript = destroyScript;
}
public void setDataSource(DataSource dataSource) {
this.dataSource = dataSource;
}
}

View File

@@ -1,15 +0,0 @@
log4j.rootCategory=WARN, stdout
log4j.appender.stdout=org.apache.logging.log4j.core.appender.ConsoleAppender
log4j.appender.stdout.layout=org.apache.logging.log4j.core.layout.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{1}:%L - %m%n
log4j.category.org.apache.activemq=ERROR
# log4j.category.org.springframework=DEBUG
log4j.category.org.springframework.jdbc=INFO
log4j.category.org.springframework.context=INFO
log4j.category.org.springframework.jms=INFO
# log4j.category.org.springframework.batch=INFO
log4j.category.org.springframework.batch.core.test=INFO
log4j.category.org.springframework.retry=INFO
# log4j.category.org.springframework.beans.factory.config=TRACE

View File

@@ -29,7 +29,7 @@ import org.springframework.batch.core.job.flow.FlowJob;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.test.namespace.config.DummyNamespaceHandler;
import org.springframework.batch.core.test.namespace.config.DummyNamespaceHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.test.context.ContextConfiguration;

View File

@@ -41,7 +41,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "/simple-job-launcher-context.xml", "/META-INF/batch/footballJob.xml" })
@ContextConfiguration(locations = {"/simple-job-launcher-context.xml", "/META-INF/batch/footballJob.xml"})
public class FootballJobIntegrationTests extends AbstractIntegrationTests {
/** Logger */

View File

@@ -43,7 +43,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "/simple-job-launcher-context.xml", "/META-INF/batch/footballSkipJob.xml" })
@ContextConfiguration(locations = {"/simple-job-launcher-context.xml", "/META-INF/batch/footballSkipJob.xml"})
public class FootballJobSkipIntegrationTests extends AbstractIntegrationTests {
/** Logger */

View File

@@ -1,79 +1,79 @@
/*
* Copyright 2006-2009 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.batch.core.test.football;
import static org.junit.Assert.assertEquals;
import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.jdbc.JdbcTestUtils;
/**
* @author Dave Syer
*
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "/simple-job-launcher-context.xml", "/META-INF/batch/parallelJob.xml" })
public class ParallelJobIntegrationTests {
/** Logger */
private final Log logger = LogFactory.getLog(getClass());
@Autowired
private JobLauncher jobLauncher;
private JdbcTemplate jdbcTemplate;
@Autowired
private Job job;
@Autowired
public void setDataSource(DataSource dataSource) {
this.jdbcTemplate = new JdbcTemplate(dataSource);
}
@Before
public void clear() {
JdbcTestUtils.deleteFromTables(jdbcTemplate, "PLAYER_SUMMARY", "GAMES", "PLAYERS");
}
@Test
public void testLaunchJob() throws Exception {
JobExecution execution = jobLauncher.run(job, new JobParametersBuilder().toJobParameters());
assertEquals(BatchStatus.COMPLETED, execution.getStatus());
for (StepExecution stepExecution : execution.getStepExecutions()) {
logger.info("Processed: "+stepExecution);
}
}
}
/*
* Copyright 2006-2009 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.batch.core.test.football;
import static org.junit.Assert.assertEquals;
import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.jdbc.JdbcTestUtils;
/**
* @author Dave Syer
*
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = {"/simple-job-launcher-context.xml", "/META-INF/batch/parallelJob.xml"})
public class ParallelJobIntegrationTests {
/** Logger */
private final Log logger = LogFactory.getLog(getClass());
@Autowired
private JobLauncher jobLauncher;
private JdbcTemplate jdbcTemplate;
@Autowired
private Job job;
@Autowired
public void setDataSource(DataSource dataSource) {
this.jdbcTemplate = new JdbcTemplate(dataSource);
}
@Before
public void clear() {
JdbcTestUtils.deleteFromTables(jdbcTemplate, "PLAYER_SUMMARY", "GAMES", "PLAYERS");
}
@Test
public void testLaunchJob() throws Exception {
JobExecution execution = jobLauncher.run(job, new JobParametersBuilder().toJobParameters());
assertEquals(BatchStatus.COMPLETED, execution.getStatus());
for (StepExecution stepExecution : execution.getStepExecutions()) {
logger.info("Processed: "+stepExecution);
}
}
}

View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.batch.core.test.football;
package org.springframework.batch.core.test.football.domain;
import java.io.Serializable;

View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.batch.core.test.football;
package org.springframework.batch.core.test.football.domain;
import java.io.Serializable;

View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.batch.core.test.football;
package org.springframework.batch.core.test.football.domain;
/**

View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.batch.core.test.football;
package org.springframework.batch.core.test.football.domain;
/**

View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.batch.sample.domain.football.internal;
package org.springframework.batch.core.test.football.internal;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

View File

@@ -14,9 +14,9 @@
* limitations under the License.
*/
package org.springframework.batch.sample.domain.football.internal;
package org.springframework.batch.core.test.football.internal;
import org.springframework.batch.core.test.football.Game;
import org.springframework.batch.core.test.football.domain.Game;
import org.springframework.batch.item.file.mapping.FieldSetMapper;
import org.springframework.batch.item.file.transform.FieldSet;

View File

@@ -14,11 +14,11 @@
* limitations under the License.
*/
package org.springframework.batch.sample.domain.football.internal;
package org.springframework.batch.core.test.football.internal;
import java.util.List;
import org.springframework.batch.core.test.football.Game;
import org.springframework.batch.core.test.football.domain.Game;
import org.springframework.batch.item.ItemWriter;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;

View File

@@ -14,10 +14,10 @@
* limitations under the License.
*/
package org.springframework.batch.sample.domain.football.internal;
package org.springframework.batch.core.test.football.internal;
import org.springframework.batch.core.test.football.Player;
import org.springframework.batch.core.test.football.PlayerDao;
import org.springframework.batch.core.test.football.domain.Player;
import org.springframework.batch.core.test.football.domain.PlayerDao;
import org.springframework.jdbc.core.namedparam.BeanPropertySqlParameterSource;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;

View File

@@ -14,11 +14,11 @@
* limitations under the License.
*/
package org.springframework.batch.sample.domain.football.internal;
package org.springframework.batch.core.test.football.internal;
import java.util.List;
import org.springframework.batch.core.test.football.PlayerSummary;
import org.springframework.batch.core.test.football.domain.PlayerSummary;
import org.springframework.batch.item.ItemWriter;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;

View File

@@ -14,9 +14,9 @@
* limitations under the License.
*/
package org.springframework.batch.sample.domain.football.internal;
package org.springframework.batch.core.test.football.internal;
import org.springframework.batch.core.test.football.Player;
import org.springframework.batch.core.test.football.domain.Player;
import org.springframework.batch.item.file.mapping.FieldSetMapper;
import org.springframework.batch.item.file.transform.FieldSet;

View File

@@ -14,12 +14,12 @@
* limitations under the License.
*/
package org.springframework.batch.sample.domain.football.internal;
package org.springframework.batch.core.test.football.internal;
import java.util.List;
import org.springframework.batch.core.test.football.Player;
import org.springframework.batch.core.test.football.PlayerDao;
import org.springframework.batch.core.test.football.domain.Player;
import org.springframework.batch.core.test.football.domain.PlayerDao;
import org.springframework.batch.item.ItemWriter;
public class PlayerItemWriter implements ItemWriter<Player> {

View File

@@ -13,16 +13,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.sample.domain.football.internal;
package org.springframework.batch.core.test.football.internal;
import java.sql.ResultSet;
import java.sql.SQLException;
import org.springframework.batch.core.test.football.PlayerSummary;
import org.springframework.batch.core.test.football.domain.PlayerSummary;
import org.springframework.jdbc.core.RowMapper;
/**
* RowMapper used to map a ResultSet to a {@link org.springframework.batch.core.test.football.PlayerSummary}
* RowMapper used to map a ResultSet to a {@link org.springframework.batch.core.test.football.domain.PlayerSummary}
*
* @author Lucas Ward
* @author Mahmoud Ben Hassine

View File

@@ -13,16 +13,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.sample.domain.football.internal;
package org.springframework.batch.core.test.football.internal;
import java.sql.ResultSet;
import java.sql.SQLException;
import org.springframework.batch.core.test.football.PlayerSummary;
import org.springframework.batch.core.test.football.domain.PlayerSummary;
import org.springframework.jdbc.core.RowMapper;
/**
* RowMapper used to map a ResultSet to a {@link org.springframework.batch.core.test.football.PlayerSummary}
* RowMapper used to map a ResultSet to a {@link org.springframework.batch.core.test.football.domain.PlayerSummary}
*
* @author Lucas Ward
* @author Mahmoud Ben Hassine

View File

@@ -40,7 +40,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.Assert;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "/simple-job-launcher-context.xml", "/applicationContext-test1.xml"})
@ContextConfiguration(locations = {"/simple-job-launcher-context.xml", "/applicationContext-test1.xml"})
public class LdifReaderTests {
private Resource expected;

View File

@@ -42,7 +42,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.Assert;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "/simple-job-launcher-context.xml", "/applicationContext-test2.xml"})
@ContextConfiguration(locations = {"/simple-job-launcher-context.xml", "/applicationContext-test2.xml"})
public class MappingLdifReaderTests {
private static Logger log = LoggerFactory.getLogger(MappingLdifReaderTests.class);

View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.batch.test.namespace.config;
package org.springframework.batch.core.test.namespace.config;
import java.util.Random;

View File

@@ -1,142 +1,142 @@
/*
* Copyright 2006-2009 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.batch.core.test.repository;
import static org.junit.Assert.assertEquals;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.junit.Test;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.dao.MapExecutionContextDao;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;
/**
* @author Dave Syer
*
*/
public class ConcurrentMapExecutionContextDaoTests {
private MapExecutionContextDao dao = new MapExecutionContextDao();
private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
@Test
public void testSaveUpdate() throws Exception {
StepExecution stepExecution = new StepExecution("step", new JobExecution(11L));
stepExecution.setId(123L);
stepExecution.getExecutionContext().put("foo", "bar");
dao.saveExecutionContext(stepExecution);
ExecutionContext executionContext = dao.getExecutionContext(stepExecution);
assertEquals("bar", executionContext.get("foo"));
}
@Test
public void testTransactionalSaveUpdate() throws Exception {
final StepExecution stepExecution = new StepExecution("step", new JobExecution(11L));
stepExecution.setId(123L);
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
@Override
public Void doInTransaction(TransactionStatus status) {
stepExecution.getExecutionContext().put("foo", "bar");
dao.saveExecutionContext(stepExecution);
return null;
}
});
ExecutionContext executionContext = dao.getExecutionContext(stepExecution);
assertEquals("bar", executionContext.get("foo"));
}
@Test
public void testConcurrentTransactionalSaveUpdate() throws Exception {
ExecutorService executor = Executors.newFixedThreadPool(3);
CompletionService<StepExecution> completionService = new ExecutorCompletionService<>(executor);
final int outerMax = 10;
final int innerMax = 100;
for (int i = 0; i < outerMax; i++) {
final StepExecution stepExecution1 = new StepExecution("step", new JobExecution(11L));
stepExecution1.setId(123L + i);
final StepExecution stepExecution2 = new StepExecution("step", new JobExecution(11L));
stepExecution2.setId(1234L + i);
completionService.submit(new Callable<StepExecution>() {
@Override
public StepExecution call() throws Exception {
for (int i = 0; i < innerMax; i++) {
String value = "bar" + i;
saveAndAssert(stepExecution1, value);
}
return stepExecution1;
}
});
completionService.submit(new Callable<StepExecution>() {
@Override
public StepExecution call() throws Exception {
for (int i = 0; i < innerMax; i++) {
String value = "spam" + i;
saveAndAssert(stepExecution2, value);
}
return stepExecution2;
}
});
completionService.take().get();
completionService.take().get();
}
executor.shutdown();
}
private void saveAndAssert(final StepExecution stepExecution, final String value) {
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
@Override
public Void doInTransaction(TransactionStatus status) {
stepExecution.getExecutionContext().put("foo", value);
dao.saveExecutionContext(stepExecution);
return null;
}
});
ExecutionContext executionContext = dao.getExecutionContext(stepExecution);
Assert.state(executionContext != null, "Lost insert: null executionContext at value=" + value);
String foo = executionContext.getString("foo");
Assert.state(value.equals(foo), "Lost update: wrong value=" + value + " (found " + foo + ") for id="
+ stepExecution.getId());
}
}
/*
* Copyright 2006-2009 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.batch.core.test.repository;
import static org.junit.Assert.assertEquals;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.junit.Test;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.dao.MapExecutionContextDao;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;
/**
* @author Dave Syer
*
*/
public class ConcurrentMapExecutionContextDaoTests {
private MapExecutionContextDao dao = new MapExecutionContextDao();
private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
@Test
public void testSaveUpdate() throws Exception {
StepExecution stepExecution = new StepExecution("step", new JobExecution(11L));
stepExecution.setId(123L);
stepExecution.getExecutionContext().put("foo", "bar");
dao.saveExecutionContext(stepExecution);
ExecutionContext executionContext = dao.getExecutionContext(stepExecution);
assertEquals("bar", executionContext.get("foo"));
}
@Test
public void testTransactionalSaveUpdate() throws Exception {
final StepExecution stepExecution = new StepExecution("step", new JobExecution(11L));
stepExecution.setId(123L);
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
@Override
public Void doInTransaction(TransactionStatus status) {
stepExecution.getExecutionContext().put("foo", "bar");
dao.saveExecutionContext(stepExecution);
return null;
}
});
ExecutionContext executionContext = dao.getExecutionContext(stepExecution);
assertEquals("bar", executionContext.get("foo"));
}
@Test
public void testConcurrentTransactionalSaveUpdate() throws Exception {
ExecutorService executor = Executors.newFixedThreadPool(3);
CompletionService<StepExecution> completionService = new ExecutorCompletionService<>(executor);
final int outerMax = 10;
final int innerMax = 100;
for (int i = 0; i < outerMax; i++) {
final StepExecution stepExecution1 = new StepExecution("step", new JobExecution(11L));
stepExecution1.setId(123L + i);
final StepExecution stepExecution2 = new StepExecution("step", new JobExecution(11L));
stepExecution2.setId(1234L + i);
completionService.submit(new Callable<StepExecution>() {
@Override
public StepExecution call() throws Exception {
for (int i = 0; i < innerMax; i++) {
String value = "bar" + i;
saveAndAssert(stepExecution1, value);
}
return stepExecution1;
}
});
completionService.submit(new Callable<StepExecution>() {
@Override
public StepExecution call() throws Exception {
for (int i = 0; i < innerMax; i++) {
String value = "spam" + i;
saveAndAssert(stepExecution2, value);
}
return stepExecution2;
}
});
completionService.take().get();
completionService.take().get();
}
executor.shutdown();
}
private void saveAndAssert(final StepExecution stepExecution, final String value) {
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
@Override
public Void doInTransaction(TransactionStatus status) {
stepExecution.getExecutionContext().put("foo", value);
dao.saveExecutionContext(stepExecution);
return null;
}
});
ExecutionContext executionContext = dao.getExecutionContext(stepExecution);
Assert.state(executionContext != null, "Lost insert: null executionContext at value=" + value);
String foo = executionContext.getString("foo");
Assert.state(value.equals(foo), "Lost update: wrong value=" + value + " (found " + foo + ") for id="
+ stepExecution.getId());
}
}

View File

@@ -47,7 +47,7 @@ import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "/simple-job-launcher-context.xml" })
@ContextConfiguration(locations = {"/simple-job-launcher-context.xml"})
public class JdbcJobRepositoryTests extends AbstractIntegrationTests {
private JobSupport job;

View File

@@ -24,6 +24,7 @@ import com.mysql.cj.jdbc.MysqlDataSource;
import org.junit.Assert;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.testcontainers.containers.MySQLContainer;
@@ -46,7 +47,6 @@ import org.springframework.beans.factory.annotation.Autowired;
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.datasource.init.ResourceDatabasePopulator;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
@@ -56,6 +56,7 @@ import org.springframework.test.context.junit4.SpringRunner;
*/
@RunWith(SpringRunner.class)
@ContextConfiguration
@Ignore("Temporarily ignored until integration tests profile is created")
public class MySQLJdbcJobRepositoryTests {
@ClassRule

View File

@@ -1,288 +1,288 @@
/*
* Copyright 2010-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
*
* 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.batch.core.test.step;
import static org.junit.Assert.assertEquals;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.jdbc.JdbcTestUtils;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.Assert;
/**
* Tests for {@link FaultTolerantStepFactoryBean}.
*/
@ContextConfiguration(locations = "/simple-job-launcher-context.xml")
@RunWith(SpringJUnit4ClassRunner.class)
public class FaultTolerantStepFactoryBeanIntegrationTests {
private static final int MAX_COUNT = 1000;
private final Log logger = LogFactory.getLog(getClass());
private FaultTolerantStepFactoryBean<String, String> factory;
private SkipProcessorStub processor;
private SkipWriterStub writer;
private JobExecution jobExecution;
private StepExecution stepExecution;
@Autowired
private DataSource dataSource;
@Autowired
private JobRepository repository;
@Autowired
private PlatformTransactionManager transactionManager;
@Before
public void setUp() throws Exception {
writer = new SkipWriterStub(dataSource);
processor = new SkipProcessorStub(dataSource);
factory = new FaultTolerantStepFactoryBean<>();
factory.setBeanName("stepName");
factory.setTransactionManager(transactionManager);
factory.setJobRepository(repository);
factory.setCommitInterval(3);
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(3);
taskExecutor.setMaxPoolSize(6);
taskExecutor.setQueueCapacity(0);
taskExecutor.afterPropertiesSet();
factory.setTaskExecutor(taskExecutor);
JdbcTestUtils.deleteFromTables(new JdbcTemplate(dataSource), "ERROR_LOG");
}
@Test
public void testUpdatesNoRollback() throws Exception {
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
writer.write(Arrays.asList("foo", "bar"));
processor.process("spam");
assertEquals(3, JdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG"));
writer.clear();
processor.clear();
assertEquals(0, JdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG"));
}
@Test
public void testMultithreadedSunnyDay() throws Throwable {
jobExecution = repository.createJobExecution("vanillaJob", new JobParameters());
for (int i = 0; i < MAX_COUNT; i++) {
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
SkipReaderStub reader = new SkipReaderStub();
reader.clear();
reader.setItems("1", "2", "3", "4", "5");
factory.setItemReader(reader);
writer.clear();
factory.setItemWriter(writer);
processor.clear();
factory.setItemProcessor(processor);
assertEquals(0, JdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG"));
try {
Step step = factory.getObject();
stepExecution = jobExecution.createStepExecution(factory.getName());
repository.add(stepExecution);
step.execute(stepExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
List<String> committed = new ArrayList<>(writer.getCommitted());
Collections.sort(committed);
assertEquals("[1, 2, 3, 4, 5]", committed.toString());
List<String> processed = new ArrayList<>(processor.getCommitted());
Collections.sort(processed);
assertEquals("[1, 2, 3, 4, 5]", processed.toString());
assertEquals(0, stepExecution.getSkipCount());
}
catch (Throwable e) {
logger.info("Failed on iteration " + i + " of " + MAX_COUNT);
throw e;
}
}
}
private static class SkipReaderStub implements ItemReader<String> {
private String[] items;
private int counter = -1;
public SkipReaderStub() throws Exception {
super();
}
public void setItems(String... items) {
Assert.isTrue(counter < 0, "Items cannot be set once reading has started");
this.items = items;
}
public void clear() {
counter = -1;
}
@Nullable
@Override
public synchronized String read() throws Exception, UnexpectedInputException, ParseException {
counter++;
if (counter >= items.length) {
return null;
}
String item = items[counter];
return item;
}
}
private static class SkipWriterStub implements ItemWriter<String> {
private List<String> written = new ArrayList<>();
private Collection<String> failures = Collections.emptySet();
private JdbcTemplate jdbcTemplate;
public SkipWriterStub(DataSource dataSource) {
jdbcTemplate = new JdbcTemplate(dataSource);
}
public List<String> getCommitted() {
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='written'",
new RowMapper<String>() {
@Override
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString(1);
}
});
}
public void clear() {
written.clear();
jdbcTemplate.update("DELETE FROM ERROR_LOG where STEP_NAME='written'");
}
@Override
public void write(List<? extends String> items) throws Exception {
for (String item : items) {
written.add(item);
jdbcTemplate.update("INSERT INTO ERROR_LOG (MESSAGE, STEP_NAME) VALUES (?, ?)", item, "written");
checkFailure(item);
}
}
private void checkFailure(String item) {
if (failures.contains(item)) {
throw new RuntimeException("Planned failure");
}
}
}
private static class SkipProcessorStub implements ItemProcessor<String, String> {
private final Log logger = LogFactory.getLog(getClass());
private List<String> processed = new ArrayList<>();
private JdbcTemplate jdbcTemplate;
/**
* @param dataSource
*/
public SkipProcessorStub(DataSource dataSource) {
jdbcTemplate = new JdbcTemplate(dataSource);
}
public List<String> getCommitted() {
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='processed'",
new RowMapper<String>() {
@Override
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString(1);
}
});
}
public void clear() {
processed.clear();
jdbcTemplate.update("DELETE FROM ERROR_LOG where STEP_NAME='processed'");
}
@Nullable
@Override
public String process(String item) throws Exception {
processed.add(item);
logger.debug("Processed item: "+item);
jdbcTemplate.update("INSERT INTO ERROR_LOG (MESSAGE, STEP_NAME) VALUES (?, ?)", item, "processed");
return item;
}
}
}
/*
* Copyright 2010-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
*
* 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.batch.core.test.step;
import static org.junit.Assert.assertEquals;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.jdbc.JdbcTestUtils;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.Assert;
/**
* Tests for {@link FaultTolerantStepFactoryBean}.
*/
@ContextConfiguration(locations = "/simple-job-launcher-context.xml")
@RunWith(SpringJUnit4ClassRunner.class)
public class FaultTolerantStepFactoryBeanIntegrationTests {
private static final int MAX_COUNT = 1000;
private final Log logger = LogFactory.getLog(getClass());
private FaultTolerantStepFactoryBean<String, String> factory;
private SkipProcessorStub processor;
private SkipWriterStub writer;
private JobExecution jobExecution;
private StepExecution stepExecution;
@Autowired
private DataSource dataSource;
@Autowired
private JobRepository repository;
@Autowired
private PlatformTransactionManager transactionManager;
@Before
public void setUp() throws Exception {
writer = new SkipWriterStub(dataSource);
processor = new SkipProcessorStub(dataSource);
factory = new FaultTolerantStepFactoryBean<>();
factory.setBeanName("stepName");
factory.setTransactionManager(transactionManager);
factory.setJobRepository(repository);
factory.setCommitInterval(3);
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(3);
taskExecutor.setMaxPoolSize(6);
taskExecutor.setQueueCapacity(0);
taskExecutor.afterPropertiesSet();
factory.setTaskExecutor(taskExecutor);
JdbcTestUtils.deleteFromTables(new JdbcTemplate(dataSource), "ERROR_LOG");
}
@Test
public void testUpdatesNoRollback() throws Exception {
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
writer.write(Arrays.asList("foo", "bar"));
processor.process("spam");
assertEquals(3, JdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG"));
writer.clear();
processor.clear();
assertEquals(0, JdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG"));
}
@Test
public void testMultithreadedSunnyDay() throws Throwable {
jobExecution = repository.createJobExecution("vanillaJob", new JobParameters());
for (int i = 0; i < MAX_COUNT; i++) {
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
SkipReaderStub reader = new SkipReaderStub();
reader.clear();
reader.setItems("1", "2", "3", "4", "5");
factory.setItemReader(reader);
writer.clear();
factory.setItemWriter(writer);
processor.clear();
factory.setItemProcessor(processor);
assertEquals(0, JdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG"));
try {
Step step = factory.getObject();
stepExecution = jobExecution.createStepExecution(factory.getName());
repository.add(stepExecution);
step.execute(stepExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
List<String> committed = new ArrayList<>(writer.getCommitted());
Collections.sort(committed);
assertEquals("[1, 2, 3, 4, 5]", committed.toString());
List<String> processed = new ArrayList<>(processor.getCommitted());
Collections.sort(processed);
assertEquals("[1, 2, 3, 4, 5]", processed.toString());
assertEquals(0, stepExecution.getSkipCount());
}
catch (Throwable e) {
logger.info("Failed on iteration " + i + " of " + MAX_COUNT);
throw e;
}
}
}
private static class SkipReaderStub implements ItemReader<String> {
private String[] items;
private int counter = -1;
public SkipReaderStub() throws Exception {
super();
}
public void setItems(String... items) {
Assert.isTrue(counter < 0, "Items cannot be set once reading has started");
this.items = items;
}
public void clear() {
counter = -1;
}
@Nullable
@Override
public synchronized String read() throws Exception, UnexpectedInputException, ParseException {
counter++;
if (counter >= items.length) {
return null;
}
String item = items[counter];
return item;
}
}
private static class SkipWriterStub implements ItemWriter<String> {
private List<String> written = new ArrayList<>();
private Collection<String> failures = Collections.emptySet();
private JdbcTemplate jdbcTemplate;
public SkipWriterStub(DataSource dataSource) {
jdbcTemplate = new JdbcTemplate(dataSource);
}
public List<String> getCommitted() {
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='written'",
new RowMapper<String>() {
@Override
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString(1);
}
});
}
public void clear() {
written.clear();
jdbcTemplate.update("DELETE FROM ERROR_LOG where STEP_NAME='written'");
}
@Override
public void write(List<? extends String> items) throws Exception {
for (String item : items) {
written.add(item);
jdbcTemplate.update("INSERT INTO ERROR_LOG (MESSAGE, STEP_NAME) VALUES (?, ?)", item, "written");
checkFailure(item);
}
}
private void checkFailure(String item) {
if (failures.contains(item)) {
throw new RuntimeException("Planned failure");
}
}
}
private static class SkipProcessorStub implements ItemProcessor<String, String> {
private final Log logger = LogFactory.getLog(getClass());
private List<String> processed = new ArrayList<>();
private JdbcTemplate jdbcTemplate;
/**
* @param dataSource
*/
public SkipProcessorStub(DataSource dataSource) {
jdbcTemplate = new JdbcTemplate(dataSource);
}
public List<String> getCommitted() {
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='processed'",
new RowMapper<String>() {
@Override
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString(1);
}
});
}
public void clear() {
processed.clear();
jdbcTemplate.update("DELETE FROM ERROR_LOG where STEP_NAME='processed'");
}
@Nullable
@Override
public String process(String item) throws Exception {
processed.add(item);
logger.debug("Processed item: "+item);
jdbcTemplate.update("INSERT INTO ERROR_LOG (MESSAGE, STEP_NAME) VALUES (?, ?)", item, "processed");
return item;
}
}
}

View File

@@ -1,320 +1,320 @@
/*
* Copyright 2010-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
*
* 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.batch.core.test.step;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.jdbc.JdbcTestUtils;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.Assert;
import static org.junit.Assert.assertEquals;
/**
* Tests for {@link FaultTolerantStepFactoryBean}.
*/
@ContextConfiguration(locations = "/simple-job-launcher-context.xml")
@RunWith(SpringJUnit4ClassRunner.class)
public class FaultTolerantStepFactoryBeanRollbackIntegrationTests {
private static final int MAX_COUNT = 1000;
private final Log logger = LogFactory.getLog(getClass());
private FaultTolerantStepFactoryBean<String, String> factory;
private SkipProcessorStub processor;
private SkipWriterStub writer;
private JobExecution jobExecution;
private StepExecution stepExecution;
@Autowired
private DataSource dataSource;
@Autowired
private JobRepository repository;
@Autowired
private PlatformTransactionManager transactionManager;
@Before
public void setUp() throws Exception {
writer = new SkipWriterStub(dataSource);
processor = new SkipProcessorStub(dataSource);
factory = new FaultTolerantStepFactoryBean<>();
factory.setBeanName("stepName");
factory.setTransactionManager(transactionManager);
factory.setJobRepository(repository);
factory.setCommitInterval(3);
factory.setSkipLimit(10);
JdbcTestUtils.deleteFromTables(new JdbcTemplate(dataSource), "ERROR_LOG");
}
@Test
public void testUpdatesNoRollback() throws Exception {
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
writer.write(Arrays.asList("foo", "bar"));
processor.process("spam");
assertEquals(3, JdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG"));
writer.clear();
processor.clear();
assertEquals(0, JdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG"));
}
@Test
public void testMultithreadedSkipInWriter() throws Throwable {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(3);
taskExecutor.setMaxPoolSize(6);
taskExecutor.setQueueCapacity(0);
taskExecutor.afterPropertiesSet();
factory.setTaskExecutor(taskExecutor);
@SuppressWarnings("unchecked")
Map<Class<? extends Throwable>, Boolean> skippable = getExceptionMap(Exception.class);
factory.setSkippableExceptionClasses(skippable);
jobExecution = repository.createJobExecution("skipJob", new JobParameters());
for (int i = 0; i < MAX_COUNT; i++) {
if (i % 100 == 0) {
logger.info("Starting step: " + i);
}
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
assertEquals(0, JdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG"));
try {
SkipReaderStub reader = new SkipReaderStub();
reader.clear();
reader.setItems("1", "2", "3", "4", "5");
factory.setItemReader(reader);
writer.clear();
factory.setItemWriter(writer);
processor.clear();
factory.setItemProcessor(processor);
writer.setFailures("1", "2", "3", "4", "5");
Step step = factory.getObject();
stepExecution = jobExecution.createStepExecution(factory.getName());
repository.add(stepExecution);
step.execute(stepExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
assertEquals("[]", writer.getCommitted().toString());
assertEquals("[]", processor.getCommitted().toString());
List<String> processed = new ArrayList<>(processor.getProcessed());
Collections.sort(processed);
assertEquals("[1, 1, 2, 2, 3, 3, 4, 4, 5, 5]", processed.toString());
assertEquals(5, stepExecution.getSkipCount());
}
catch (Throwable e) {
logger.info("Failed on iteration " + i + " of " + MAX_COUNT);
throw e;
}
}
}
@SuppressWarnings("unchecked")
private Map<Class<? extends Throwable>, Boolean> getExceptionMap(Class<? extends Throwable>... args) {
Map<Class<? extends Throwable>, Boolean> map = new HashMap<>();
for (Class<? extends Throwable> arg : args) {
map.put(arg, true);
}
return map;
}
private static class SkipReaderStub implements ItemReader<String> {
private String[] items;
private int counter = -1;
public SkipReaderStub() throws Exception {
super();
}
public void setItems(String... items) {
Assert.isTrue(counter < 0, "Items cannot be set once reading has started");
this.items = items;
}
public void clear() {
counter = -1;
}
@Nullable
@Override
public synchronized String read() throws Exception, UnexpectedInputException, ParseException {
counter++;
if (counter >= items.length) {
return null;
}
String item = items[counter];
return item;
}
}
private static class SkipWriterStub implements ItemWriter<String> {
private List<String> written = new CopyOnWriteArrayList<>();
private Collection<String> failures = Collections.emptySet();
private JdbcTemplate jdbcTemplate;
public SkipWriterStub(DataSource dataSource) {
jdbcTemplate = new JdbcTemplate(dataSource);
}
public void setFailures(String... failures) {
this.failures = Arrays.asList(failures);
}
public List<String> getCommitted() {
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='written'",
new RowMapper<String>() {
@Override
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString(1);
}
});
}
public void clear() {
written.clear();
jdbcTemplate.update("DELETE FROM ERROR_LOG where STEP_NAME='written'");
}
@Override
public void write(List<? extends String> items) throws Exception {
for (String item : items) {
written.add(item);
jdbcTemplate.update("INSERT INTO ERROR_LOG (MESSAGE, STEP_NAME) VALUES (?, ?)", item, "written");
checkFailure(item);
}
}
private void checkFailure(String item) {
if (failures.contains(item)) {
throw new RuntimeException("Planned failure");
}
}
}
private static class SkipProcessorStub implements ItemProcessor<String, String> {
private final Log logger = LogFactory.getLog(getClass());
private List<String> processed = new CopyOnWriteArrayList<>();
private JdbcTemplate jdbcTemplate;
/**
* @param dataSource
*/
public SkipProcessorStub(DataSource dataSource) {
jdbcTemplate = new JdbcTemplate(dataSource);
}
/**
* @return the processed
*/
public List<String> getProcessed() {
return processed;
}
public List<String> getCommitted() {
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='processed'",
new RowMapper<String>() {
@Override
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString(1);
}
});
}
public void clear() {
processed.clear();
jdbcTemplate.update("DELETE FROM ERROR_LOG where STEP_NAME='processed'");
}
@Nullable
@Override
public String process(String item) throws Exception {
processed.add(item);
logger.debug("Processed item: " + item);
jdbcTemplate.update("INSERT INTO ERROR_LOG (MESSAGE, STEP_NAME) VALUES (?, ?)", item, "processed");
return item;
}
}
}
/*
* Copyright 2010-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
*
* 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.batch.core.test.step;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.jdbc.JdbcTestUtils;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.Assert;
import static org.junit.Assert.assertEquals;
/**
* Tests for {@link FaultTolerantStepFactoryBean}.
*/
@ContextConfiguration(locations = "/simple-job-launcher-context.xml")
@RunWith(SpringJUnit4ClassRunner.class)
public class FaultTolerantStepFactoryBeanRollbackIntegrationTests {
private static final int MAX_COUNT = 1000;
private final Log logger = LogFactory.getLog(getClass());
private FaultTolerantStepFactoryBean<String, String> factory;
private SkipProcessorStub processor;
private SkipWriterStub writer;
private JobExecution jobExecution;
private StepExecution stepExecution;
@Autowired
private DataSource dataSource;
@Autowired
private JobRepository repository;
@Autowired
private PlatformTransactionManager transactionManager;
@Before
public void setUp() throws Exception {
writer = new SkipWriterStub(dataSource);
processor = new SkipProcessorStub(dataSource);
factory = new FaultTolerantStepFactoryBean<>();
factory.setBeanName("stepName");
factory.setTransactionManager(transactionManager);
factory.setJobRepository(repository);
factory.setCommitInterval(3);
factory.setSkipLimit(10);
JdbcTestUtils.deleteFromTables(new JdbcTemplate(dataSource), "ERROR_LOG");
}
@Test
public void testUpdatesNoRollback() throws Exception {
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
writer.write(Arrays.asList("foo", "bar"));
processor.process("spam");
assertEquals(3, JdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG"));
writer.clear();
processor.clear();
assertEquals(0, JdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG"));
}
@Test
public void testMultithreadedSkipInWriter() throws Throwable {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(3);
taskExecutor.setMaxPoolSize(6);
taskExecutor.setQueueCapacity(0);
taskExecutor.afterPropertiesSet();
factory.setTaskExecutor(taskExecutor);
@SuppressWarnings("unchecked")
Map<Class<? extends Throwable>, Boolean> skippable = getExceptionMap(Exception.class);
factory.setSkippableExceptionClasses(skippable);
jobExecution = repository.createJobExecution("skipJob", new JobParameters());
for (int i = 0; i < MAX_COUNT; i++) {
if (i % 100 == 0) {
logger.info("Starting step: " + i);
}
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
assertEquals(0, JdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG"));
try {
SkipReaderStub reader = new SkipReaderStub();
reader.clear();
reader.setItems("1", "2", "3", "4", "5");
factory.setItemReader(reader);
writer.clear();
factory.setItemWriter(writer);
processor.clear();
factory.setItemProcessor(processor);
writer.setFailures("1", "2", "3", "4", "5");
Step step = factory.getObject();
stepExecution = jobExecution.createStepExecution(factory.getName());
repository.add(stepExecution);
step.execute(stepExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
assertEquals("[]", writer.getCommitted().toString());
assertEquals("[]", processor.getCommitted().toString());
List<String> processed = new ArrayList<>(processor.getProcessed());
Collections.sort(processed);
assertEquals("[1, 1, 2, 2, 3, 3, 4, 4, 5, 5]", processed.toString());
assertEquals(5, stepExecution.getSkipCount());
}
catch (Throwable e) {
logger.info("Failed on iteration " + i + " of " + MAX_COUNT);
throw e;
}
}
}
@SuppressWarnings("unchecked")
private Map<Class<? extends Throwable>, Boolean> getExceptionMap(Class<? extends Throwable>... args) {
Map<Class<? extends Throwable>, Boolean> map = new HashMap<>();
for (Class<? extends Throwable> arg : args) {
map.put(arg, true);
}
return map;
}
private static class SkipReaderStub implements ItemReader<String> {
private String[] items;
private int counter = -1;
public SkipReaderStub() throws Exception {
super();
}
public void setItems(String... items) {
Assert.isTrue(counter < 0, "Items cannot be set once reading has started");
this.items = items;
}
public void clear() {
counter = -1;
}
@Nullable
@Override
public synchronized String read() throws Exception, UnexpectedInputException, ParseException {
counter++;
if (counter >= items.length) {
return null;
}
String item = items[counter];
return item;
}
}
private static class SkipWriterStub implements ItemWriter<String> {
private List<String> written = new CopyOnWriteArrayList<>();
private Collection<String> failures = Collections.emptySet();
private JdbcTemplate jdbcTemplate;
public SkipWriterStub(DataSource dataSource) {
jdbcTemplate = new JdbcTemplate(dataSource);
}
public void setFailures(String... failures) {
this.failures = Arrays.asList(failures);
}
public List<String> getCommitted() {
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='written'",
new RowMapper<String>() {
@Override
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString(1);
}
});
}
public void clear() {
written.clear();
jdbcTemplate.update("DELETE FROM ERROR_LOG where STEP_NAME='written'");
}
@Override
public void write(List<? extends String> items) throws Exception {
for (String item : items) {
written.add(item);
jdbcTemplate.update("INSERT INTO ERROR_LOG (MESSAGE, STEP_NAME) VALUES (?, ?)", item, "written");
checkFailure(item);
}
}
private void checkFailure(String item) {
if (failures.contains(item)) {
throw new RuntimeException("Planned failure");
}
}
}
private static class SkipProcessorStub implements ItemProcessor<String, String> {
private final Log logger = LogFactory.getLog(getClass());
private List<String> processed = new CopyOnWriteArrayList<>();
private JdbcTemplate jdbcTemplate;
/**
* @param dataSource
*/
public SkipProcessorStub(DataSource dataSource) {
jdbcTemplate = new JdbcTemplate(dataSource);
}
/**
* @return the processed
*/
public List<String> getProcessed() {
return processed;
}
public List<String> getCommitted() {
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='processed'",
new RowMapper<String>() {
@Override
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString(1);
}
});
}
public void clear() {
processed.clear();
jdbcTemplate.update("DELETE FROM ERROR_LOG where STEP_NAME='processed'");
}
@Nullable
@Override
public String process(String item) throws Exception {
processed.add(item);
logger.debug("Processed item: " + item);
jdbcTemplate.update("INSERT INTO ERROR_LOG (MESSAGE, STEP_NAME) VALUES (?, ?)", item, "processed");
return item;
}
}
}

View File

@@ -1,259 +1,259 @@
/*
* Copyright 2010-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
*
* 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.batch.core.test.step;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.Assert;
import static org.junit.Assert.assertEquals;
/**
* Tests for {@link FaultTolerantStepFactoryBean}.
*/
public class MapRepositoryFaultTolerantStepFactoryBeanRollbackTests {
private static final int MAX_COUNT = 1000;
private final Log logger = LogFactory.getLog(getClass());
private FaultTolerantStepFactoryBean<String, String> factory;
private SkipReaderStub reader;
private SkipProcessorStub processor;
private SkipWriterStub writer;
private JobExecution jobExecution;
private StepExecution stepExecution;
private JobRepository repository;
private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
@SuppressWarnings("unchecked")
@Before
public void setUp() throws Exception {
reader = new SkipReaderStub();
writer = new SkipWriterStub();
processor = new SkipProcessorStub();
factory = new FaultTolerantStepFactoryBean<>();
factory.setTransactionManager(transactionManager);
factory.setBeanName("stepName");
factory.setCommitInterval(3);
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(3);
taskExecutor.setMaxPoolSize(6);
taskExecutor.setQueueCapacity(0);
taskExecutor.afterPropertiesSet();
factory.setTaskExecutor(taskExecutor);
factory.setSkipLimit(10);
factory.setSkippableExceptionClasses(getExceptionMap(Exception.class));
}
@Test
public void testUpdatesNoRollback() throws Exception {
writer.write(Arrays.asList("foo", "bar"));
processor.process("spam");
assertEquals(2, writer.getWritten().size());
assertEquals(1, processor.getProcessed().size());
writer.clear();
processor.clear();
assertEquals(0, processor.getProcessed().size());
}
@Test
public void testMultithreadedSkipInWrite() throws Throwable {
for (int i = 0; i < MAX_COUNT; i++) {
if (i%100==0) {
logger.info("Starting step: "+i);
repository = new MapJobRepositoryFactoryBean(transactionManager).getObject();
factory.setJobRepository(repository);
jobExecution = repository.createJobExecution("vanillaJob", new JobParameters());
}
reader.clear();
reader.setItems("1", "2", "3", "4", "5");
factory.setItemReader(reader);
writer.clear();
factory.setItemWriter(writer);
processor.clear();
factory.setItemProcessor(processor);
writer.setFailures("1", "2", "3", "4", "5");
try {
Step step = factory.getObject();
stepExecution = jobExecution.createStepExecution(factory.getName());
repository.add(stepExecution);
step.execute(stepExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
assertEquals(5, stepExecution.getSkipCount());
List<String> processed = new ArrayList<>(processor.getProcessed());
Collections.sort(processed);
assertEquals("[1, 1, 2, 2, 3, 3, 4, 4, 5, 5]", processed.toString());
}
catch (Throwable e) {
logger.info("Failed on iteration " + i + " of " + MAX_COUNT);
throw e;
}
}
}
private static class SkipReaderStub implements ItemReader<String> {
private String[] items;
private int counter = -1;
public SkipReaderStub() throws Exception {
super();
}
public void setItems(String... items) {
Assert.isTrue(counter < 0, "Items cannot be set once reading has started");
this.items = items;
}
public void clear() {
counter = -1;
}
@Nullable
@Override
public synchronized String read() throws Exception {
counter++;
if (counter >= items.length) {
return null;
}
return items[counter];
}
}
private static class SkipWriterStub implements ItemWriter<String> {
private final Log logger = LogFactory.getLog(getClass());
private List<String> written = new CopyOnWriteArrayList<>();
private Collection<String> failures = Collections.emptySet();
public void setFailures(String... failures) {
this.failures = Arrays.asList(failures);
}
public List<String> getWritten() {
return written;
}
public void clear() {
written.clear();
}
@Override
public void write(List<? extends String> items) throws Exception {
for (String item : items) {
logger.trace("Writing: "+item);
written.add(item);
checkFailure(item);
}
}
private void checkFailure(String item) {
if (failures.contains(item)) {
throw new RuntimeException("Planned failure");
}
}
}
private static class SkipProcessorStub implements ItemProcessor<String, String> {
private final Log logger = LogFactory.getLog(getClass());
private List<String> processed = new CopyOnWriteArrayList<>();
public List<String> getProcessed() {
return processed;
}
public void clear() {
processed.clear();
}
@Nullable
@Override
public String process(String item) throws Exception {
processed.add(item);
logger.debug("Processed item: "+item);
return item;
}
}
@SuppressWarnings("unchecked")
private Map<Class<? extends Throwable>, Boolean> getExceptionMap(Class<? extends Throwable>... args) {
Map<Class<? extends Throwable>, Boolean> map = new HashMap<>();
for (Class<? extends Throwable> arg : args) {
map.put(arg, true);
}
return map;
}
}
/*
* Copyright 2010-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
*
* 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.batch.core.test.step;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.Assert;
import static org.junit.Assert.assertEquals;
/**
* Tests for {@link FaultTolerantStepFactoryBean}.
*/
public class MapRepositoryFaultTolerantStepFactoryBeanRollbackTests {
private static final int MAX_COUNT = 1000;
private final Log logger = LogFactory.getLog(getClass());
private FaultTolerantStepFactoryBean<String, String> factory;
private SkipReaderStub reader;
private SkipProcessorStub processor;
private SkipWriterStub writer;
private JobExecution jobExecution;
private StepExecution stepExecution;
private JobRepository repository;
private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
@SuppressWarnings("unchecked")
@Before
public void setUp() throws Exception {
reader = new SkipReaderStub();
writer = new SkipWriterStub();
processor = new SkipProcessorStub();
factory = new FaultTolerantStepFactoryBean<>();
factory.setTransactionManager(transactionManager);
factory.setBeanName("stepName");
factory.setCommitInterval(3);
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(3);
taskExecutor.setMaxPoolSize(6);
taskExecutor.setQueueCapacity(0);
taskExecutor.afterPropertiesSet();
factory.setTaskExecutor(taskExecutor);
factory.setSkipLimit(10);
factory.setSkippableExceptionClasses(getExceptionMap(Exception.class));
}
@Test
public void testUpdatesNoRollback() throws Exception {
writer.write(Arrays.asList("foo", "bar"));
processor.process("spam");
assertEquals(2, writer.getWritten().size());
assertEquals(1, processor.getProcessed().size());
writer.clear();
processor.clear();
assertEquals(0, processor.getProcessed().size());
}
@Test
public void testMultithreadedSkipInWrite() throws Throwable {
for (int i = 0; i < MAX_COUNT; i++) {
if (i%100==0) {
logger.info("Starting step: "+i);
repository = new MapJobRepositoryFactoryBean(transactionManager).getObject();
factory.setJobRepository(repository);
jobExecution = repository.createJobExecution("vanillaJob", new JobParameters());
}
reader.clear();
reader.setItems("1", "2", "3", "4", "5");
factory.setItemReader(reader);
writer.clear();
factory.setItemWriter(writer);
processor.clear();
factory.setItemProcessor(processor);
writer.setFailures("1", "2", "3", "4", "5");
try {
Step step = factory.getObject();
stepExecution = jobExecution.createStepExecution(factory.getName());
repository.add(stepExecution);
step.execute(stepExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
assertEquals(5, stepExecution.getSkipCount());
List<String> processed = new ArrayList<>(processor.getProcessed());
Collections.sort(processed);
assertEquals("[1, 1, 2, 2, 3, 3, 4, 4, 5, 5]", processed.toString());
}
catch (Throwable e) {
logger.info("Failed on iteration " + i + " of " + MAX_COUNT);
throw e;
}
}
}
private static class SkipReaderStub implements ItemReader<String> {
private String[] items;
private int counter = -1;
public SkipReaderStub() throws Exception {
super();
}
public void setItems(String... items) {
Assert.isTrue(counter < 0, "Items cannot be set once reading has started");
this.items = items;
}
public void clear() {
counter = -1;
}
@Nullable
@Override
public synchronized String read() throws Exception {
counter++;
if (counter >= items.length) {
return null;
}
return items[counter];
}
}
private static class SkipWriterStub implements ItemWriter<String> {
private final Log logger = LogFactory.getLog(getClass());
private List<String> written = new CopyOnWriteArrayList<>();
private Collection<String> failures = Collections.emptySet();
public void setFailures(String... failures) {
this.failures = Arrays.asList(failures);
}
public List<String> getWritten() {
return written;
}
public void clear() {
written.clear();
}
@Override
public void write(List<? extends String> items) throws Exception {
for (String item : items) {
logger.trace("Writing: "+item);
written.add(item);
checkFailure(item);
}
}
private void checkFailure(String item) {
if (failures.contains(item)) {
throw new RuntimeException("Planned failure");
}
}
}
private static class SkipProcessorStub implements ItemProcessor<String, String> {
private final Log logger = LogFactory.getLog(getClass());
private List<String> processed = new CopyOnWriteArrayList<>();
public List<String> getProcessed() {
return processed;
}
public void clear() {
processed.clear();
}
@Nullable
@Override
public String process(String item) throws Exception {
processed.add(item);
logger.debug("Processed item: "+item);
return item;
}
}
@SuppressWarnings("unchecked")
private Map<Class<? extends Throwable>, Boolean> getExceptionMap(Class<? extends Throwable>... args) {
Map<Class<? extends Throwable>, Boolean> map = new HashMap<>();
for (Class<? extends Throwable> arg : args) {
map.put(arg, true);
}
return map;
}
}

View File

@@ -1,240 +1,240 @@
/*
* Copyright 2010-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
*
* 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.batch.core.test.step;
import static org.junit.Assert.assertEquals;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.Assert;
/**
* Tests for {@link FaultTolerantStepFactoryBean}.
*/
public class MapRepositoryFaultTolerantStepFactoryBeanTests {
private static final int MAX_COUNT = 1000;
private final Log logger = LogFactory.getLog(getClass());
private FaultTolerantStepFactoryBean<String, String> factory;
private SkipReaderStub reader;
private SkipProcessorStub processor;
private SkipWriterStub writer;
private JobExecution jobExecution;
private StepExecution stepExecution;
private JobRepository repository;
private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
@Before
public void setUp() throws Exception {
reader = new SkipReaderStub();
writer = new SkipWriterStub();
processor = new SkipProcessorStub();
factory = new FaultTolerantStepFactoryBean<>();
factory.setBeanName("stepName");
factory.setTransactionManager(transactionManager);
factory.setCommitInterval(3);
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(3);
taskExecutor.setMaxPoolSize(6);
taskExecutor.setQueueCapacity(0);
taskExecutor.afterPropertiesSet();
factory.setTaskExecutor(taskExecutor);
}
@Test
public void testUpdatesNoRollback() throws Exception {
writer.write(Arrays.asList("foo", "bar"));
processor.process("spam");
assertEquals(2, writer.getWritten().size());
assertEquals(1, processor.getProcessed().size());
writer.clear();
processor.clear();
assertEquals(0, processor.getProcessed().size());
}
@Test
public void testMultithreadedSunnyDay() throws Throwable {
for (int i = 0; i < MAX_COUNT; i++) {
if (i%100==0) {
logger.info("Starting step: "+i);
repository = new MapJobRepositoryFactoryBean(transactionManager).getObject();
factory.setJobRepository(repository);
jobExecution = repository.createJobExecution("vanillaJob", new JobParameters());
}
reader.clear();
reader.setItems("1", "2", "3", "4", "5");
factory.setItemReader(reader);
writer.clear();
factory.setItemWriter(writer);
processor.clear();
factory.setItemProcessor(processor);
try {
Step step = factory.getObject();
stepExecution = jobExecution.createStepExecution(factory.getName());
repository.add(stepExecution);
step.execute(stepExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
List<String> committed = new ArrayList<>(writer.getWritten());
Collections.sort(committed);
assertEquals("[1, 2, 3, 4, 5]", committed.toString());
List<String> processed = new ArrayList<>(processor.getProcessed());
Collections.sort(processed);
assertEquals("[1, 2, 3, 4, 5]", processed.toString());
assertEquals(0, stepExecution.getSkipCount());
}
catch (Throwable e) {
logger.info("Failed on iteration " + i + " of " + MAX_COUNT);
throw e;
}
}
}
private static class SkipReaderStub implements ItemReader<String> {
private String[] items;
private int counter = -1;
public SkipReaderStub() throws Exception {
super();
}
public void setItems(String... items) {
Assert.isTrue(counter < 0, "Items cannot be set once reading has started");
this.items = items;
}
public void clear() {
counter = -1;
}
@Nullable
@Override
public synchronized String read() throws Exception, UnexpectedInputException, ParseException {
counter++;
if (counter >= items.length) {
return null;
}
String item = items[counter];
return item;
}
}
private static class SkipWriterStub implements ItemWriter<String> {
private List<String> written = new CopyOnWriteArrayList<>();
private Collection<String> failures = Collections.emptySet();
public List<String> getWritten() {
return written;
}
public void clear() {
written.clear();
}
@Override
public void write(List<? extends String> items) throws Exception {
for (String item : items) {
written.add(item);
checkFailure(item);
}
}
private void checkFailure(String item) {
if (failures.contains(item)) {
throw new RuntimeException("Planned failure");
}
}
}
private static class SkipProcessorStub implements ItemProcessor<String, String> {
private final Log logger = LogFactory.getLog(getClass());
private List<String> processed = new CopyOnWriteArrayList<>();
public List<String> getProcessed() {
return processed;
}
public void clear() {
processed.clear();
}
@Nullable
@Override
public String process(String item) throws Exception {
processed.add(item);
logger.debug("Processed item: "+item);
return item;
}
}
}
/*
* Copyright 2010-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
*
* 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.batch.core.test.step;
import static org.junit.Assert.assertEquals;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.Assert;
/**
* Tests for {@link FaultTolerantStepFactoryBean}.
*/
public class MapRepositoryFaultTolerantStepFactoryBeanTests {
private static final int MAX_COUNT = 1000;
private final Log logger = LogFactory.getLog(getClass());
private FaultTolerantStepFactoryBean<String, String> factory;
private SkipReaderStub reader;
private SkipProcessorStub processor;
private SkipWriterStub writer;
private JobExecution jobExecution;
private StepExecution stepExecution;
private JobRepository repository;
private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
@Before
public void setUp() throws Exception {
reader = new SkipReaderStub();
writer = new SkipWriterStub();
processor = new SkipProcessorStub();
factory = new FaultTolerantStepFactoryBean<>();
factory.setBeanName("stepName");
factory.setTransactionManager(transactionManager);
factory.setCommitInterval(3);
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(3);
taskExecutor.setMaxPoolSize(6);
taskExecutor.setQueueCapacity(0);
taskExecutor.afterPropertiesSet();
factory.setTaskExecutor(taskExecutor);
}
@Test
public void testUpdatesNoRollback() throws Exception {
writer.write(Arrays.asList("foo", "bar"));
processor.process("spam");
assertEquals(2, writer.getWritten().size());
assertEquals(1, processor.getProcessed().size());
writer.clear();
processor.clear();
assertEquals(0, processor.getProcessed().size());
}
@Test
public void testMultithreadedSunnyDay() throws Throwable {
for (int i = 0; i < MAX_COUNT; i++) {
if (i%100==0) {
logger.info("Starting step: "+i);
repository = new MapJobRepositoryFactoryBean(transactionManager).getObject();
factory.setJobRepository(repository);
jobExecution = repository.createJobExecution("vanillaJob", new JobParameters());
}
reader.clear();
reader.setItems("1", "2", "3", "4", "5");
factory.setItemReader(reader);
writer.clear();
factory.setItemWriter(writer);
processor.clear();
factory.setItemProcessor(processor);
try {
Step step = factory.getObject();
stepExecution = jobExecution.createStepExecution(factory.getName());
repository.add(stepExecution);
step.execute(stepExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
List<String> committed = new ArrayList<>(writer.getWritten());
Collections.sort(committed);
assertEquals("[1, 2, 3, 4, 5]", committed.toString());
List<String> processed = new ArrayList<>(processor.getProcessed());
Collections.sort(processed);
assertEquals("[1, 2, 3, 4, 5]", processed.toString());
assertEquals(0, stepExecution.getSkipCount());
}
catch (Throwable e) {
logger.info("Failed on iteration " + i + " of " + MAX_COUNT);
throw e;
}
}
}
private static class SkipReaderStub implements ItemReader<String> {
private String[] items;
private int counter = -1;
public SkipReaderStub() throws Exception {
super();
}
public void setItems(String... items) {
Assert.isTrue(counter < 0, "Items cannot be set once reading has started");
this.items = items;
}
public void clear() {
counter = -1;
}
@Nullable
@Override
public synchronized String read() throws Exception, UnexpectedInputException, ParseException {
counter++;
if (counter >= items.length) {
return null;
}
String item = items[counter];
return item;
}
}
private static class SkipWriterStub implements ItemWriter<String> {
private List<String> written = new CopyOnWriteArrayList<>();
private Collection<String> failures = Collections.emptySet();
public List<String> getWritten() {
return written;
}
public void clear() {
written.clear();
}
@Override
public void write(List<? extends String> items) throws Exception {
for (String item : items) {
written.add(item);
checkFailure(item);
}
}
private void checkFailure(String item) {
if (failures.contains(item)) {
throw new RuntimeException("Planned failure");
}
}
}
private static class SkipProcessorStub implements ItemProcessor<String, String> {
private final Log logger = LogFactory.getLog(getClass());
private List<String> processed = new CopyOnWriteArrayList<>();
public List<String> getProcessed() {
return processed;
}
public void clear() {
processed.clear();
}
@Nullable
@Override
public String process(String item) throws Exception {
processed.add(item);
logger.debug("Processed item: "+item);
return item;
}
}
}

View File

@@ -1,101 +1,101 @@
/*
* Copyright 2006-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
*
* 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.batch.core.test.step;
import static org.junit.Assert.assertEquals;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.lang.Nullable;
/**
* @author Dave Syer
*
*/
public class SplitJobMapRepositoryIntegrationTests {
private static final int MAX_COUNT = 1000;
/** Logger */
private final Log logger = LogFactory.getLog(getClass());
@SuppressWarnings("resource")
@Test
public void testMultithreadedSplit() throws Throwable {
JobLauncher jobLauncher = null;
Job job = null;
ClassPathXmlApplicationContext context = null;
for (int i = 0; i < MAX_COUNT; i++) {
if (i % 100 == 0) {
if (context!=null) {
context.close();
}
logger.info("Starting job: " + i);
context = new ClassPathXmlApplicationContext(getClass().getSimpleName()
+ "-context.xml", getClass());
jobLauncher = context.getBean("jobLauncher", JobLauncher.class);
job = context.getBean("job", Job.class);
}
try {
JobExecution execution = jobLauncher.run(job, new JobParametersBuilder().addLong("count", new Long(i))
.toJobParameters());
assertEquals(BatchStatus.COMPLETED, execution.getStatus());
}
catch (Throwable e) {
logger.info("Failed on iteration " + i + " of " + MAX_COUNT);
throw e;
}
}
}
public static class CountingTasklet implements Tasklet {
private int maxCount = 10;
private AtomicInteger count = new AtomicInteger(0);
@Nullable
@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
contribution.incrementReadCount();
contribution.incrementWriteCount(1);
return RepeatStatus.continueIf(count.incrementAndGet() < maxCount);
}
}
}
/*
* Copyright 2006-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
*
* 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.batch.core.test.step;
import static org.junit.Assert.assertEquals;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.lang.Nullable;
/**
* @author Dave Syer
*
*/
public class SplitJobMapRepositoryIntegrationTests {
private static final int MAX_COUNT = 1000;
/** Logger */
private final Log logger = LogFactory.getLog(getClass());
@SuppressWarnings("resource")
@Test
public void testMultithreadedSplit() throws Throwable {
JobLauncher jobLauncher = null;
Job job = null;
ClassPathXmlApplicationContext context = null;
for (int i = 0; i < MAX_COUNT; i++) {
if (i % 100 == 0) {
if (context!=null) {
context.close();
}
logger.info("Starting job: " + i);
context = new ClassPathXmlApplicationContext(getClass().getSimpleName()
+ "-context.xml", getClass());
jobLauncher = context.getBean("jobLauncher", JobLauncher.class);
job = context.getBean("job", Job.class);
}
try {
JobExecution execution = jobLauncher.run(job, new JobParametersBuilder().addLong("count", new Long(i))
.toJobParameters());
assertEquals(BatchStatus.COMPLETED, execution.getStatus());
}
catch (Throwable e) {
logger.info("Failed on iteration " + i + " of " + MAX_COUNT);
throw e;
}
}
}
public static class CountingTasklet implements Tasklet {
private int maxCount = 10;
private AtomicInteger count = new AtomicInteger(0);
@Nullable
@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
contribution.incrementReadCount();
contribution.incrementWriteCount(1);
return RepeatStatus.continueIf(count.incrementAndGet() < maxCount);
}
}
}

View File

@@ -1,33 +1,33 @@
/*
* Copyright 2014 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.batch.core.test.timeout;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.ItemWriter;
public class LoggingItemWriter<T> implements ItemWriter<T> {
protected Log logger = LogFactory.getLog(LoggingItemWriter.class);
@Override
public void write(List<? extends T> items) throws Exception {
logger.info(items);
}
}
/*
* Copyright 2014 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.batch.core.test.timeout;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.ItemWriter;
public class LoggingItemWriter<T> implements ItemWriter<T> {
protected Log logger = LogFactory.getLog(LoggingItemWriter.class);
@Override
public void write(List<? extends T> items) throws Exception {
logger.info(items);
}
}

View File

@@ -1,36 +1,36 @@
/*
* Copyright 2014-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
*
* 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.batch.core.test.timeout;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.lang.Nullable;
public class SleepingItemProcessor<I> implements ItemProcessor<I, I> {
private long millisToSleep;
@Nullable
@Override
public I process(I item) throws Exception {
Thread.sleep(millisToSleep);
return item;
}
public void setMillisToSleep(long millisToSleep) {
this.millisToSleep = millisToSleep;
}
}
/*
* Copyright 2014-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
*
* 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.batch.core.test.timeout;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.lang.Nullable;
public class SleepingItemProcessor<I> implements ItemProcessor<I, I> {
private long millisToSleep;
@Nullable
@Override
public I process(I item) throws Exception {
Thread.sleep(millisToSleep);
return item;
}
public void setMillisToSleep(long millisToSleep) {
this.millisToSleep = millisToSleep;
}
}

View File

@@ -1,40 +1,40 @@
/*
* Copyright 2014-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
*
* 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.batch.core.test.timeout;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.lang.Nullable;
public class SleepingTasklet implements Tasklet {
private long millisToSleep;
@Nullable
@Override
public RepeatStatus execute(StepContribution contribution,
ChunkContext chunkContext) throws Exception {
Thread.sleep(millisToSleep);
return RepeatStatus.FINISHED;
}
public void setMillisToSleep(long millisToSleep) {
this.millisToSleep = millisToSleep;
}
}
/*
* Copyright 2014-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
*
* 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.batch.core.test.timeout;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.lang.Nullable;
public class SleepingTasklet implements Tasklet {
private long millisToSleep;
@Nullable
@Override
public RepeatStatus execute(StepContribution contribution,
ChunkContext chunkContext) throws Exception {
Thread.sleep(millisToSleep);
return RepeatStatus.FINISHED;
}
public void setMillisToSleep(long millisToSleep) {
this.millisToSleep = millisToSleep;
}
}

View File

@@ -35,7 +35,7 @@ import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "/simple-job-launcher-context.xml", "/META-INF/batch/timeoutJob.xml" })
@ContextConfiguration(locations = {"/simple-job-launcher-context.xml", "/META-INF/batch/timeoutJob.xml"})
public class TimeoutJobIntegrationTests extends AbstractIntegrationTests {
/** Logger */

View File

@@ -30,19 +30,19 @@
</tasklet>
</step>
<bean id="playerWriter" class="org.springframework.batch.sample.domain.football.internal.PlayerItemWriter">
<bean id="playerWriter" class="org.springframework.batch.core.test.football.internal.PlayerItemWriter">
<property name="playerDao">
<bean class="org.springframework.batch.sample.domain.football.internal.JdbcPlayerDao">
<bean class="org.springframework.batch.core.test.football.internal.JdbcPlayerDao">
<property name="dataSource" ref="dataSource" />
</bean>
</property>
</bean>
<bean id="gameWriter" class="org.springframework.batch.sample.domain.football.internal.JdbcGameDao">
<bean id="gameWriter" class="org.springframework.batch.core.test.football.internal.JdbcGameDao">
<property name="dataSource" ref="dataSource" />
</bean>
<bean id="summaryWriter" class="org.springframework.batch.sample.domain.football.internal.JdbcPlayerSummaryDao">
<bean id="summaryWriter" class="org.springframework.batch.core.test.football.internal.JdbcPlayerSummaryDao">
<property name="dataSource" ref="dataSource" />
</bean>
@@ -56,7 +56,7 @@
</bean>
</property>
<property name="fieldSetMapper">
<bean class="org.springframework.batch.sample.domain.football.internal.PlayerFieldSetMapper" />
<bean class="org.springframework.batch.core.test.football.internal.PlayerFieldSetMapper" />
</property>
</bean>
</property>
@@ -72,7 +72,7 @@
</bean>
</property>
<property name="fieldSetMapper">
<bean class="org.springframework.batch.sample.domain.football.internal.GameFieldSetMapper" />
<bean class="org.springframework.batch.core.test.football.internal.GameFieldSetMapper" />
</property>
</bean>
</property>
@@ -82,7 +82,7 @@
<property name="dataSource" ref="dataSource" />
<property name="verifyCursorPosition" value="${batch.verify.cursor.position}" />
<property name="rowMapper">
<bean class="org.springframework.batch.sample.domain.football.internal.PlayerSummaryMapper" />
<bean class="org.springframework.batch.core.test.football.internal.PlayerSummaryMapper" />
</property>
<property name="sql">
<value>

View File

@@ -48,19 +48,19 @@
</tasklet>
</step>
<bean id="playerWriter" class="org.springframework.batch.sample.domain.football.internal.PlayerItemWriter">
<bean id="playerWriter" class="org.springframework.batch.core.test.football.internal.PlayerItemWriter">
<property name="playerDao">
<bean class="org.springframework.batch.sample.domain.football.internal.JdbcPlayerDao">
<bean class="org.springframework.batch.core.test.football.internal.JdbcPlayerDao">
<property name="dataSource" ref="dataSource" />
</bean>
</property>
</bean>
<bean id="gameWriter" class="org.springframework.batch.sample.domain.football.internal.JdbcGameDao">
<bean id="gameWriter" class="org.springframework.batch.core.test.football.internal.JdbcGameDao">
<property name="dataSource" ref="dataSource" />
</bean>
<bean id="summaryWriter" class="org.springframework.batch.sample.domain.football.internal.JdbcPlayerSummaryDao">
<bean id="summaryWriter" class="org.springframework.batch.core.test.football.internal.JdbcPlayerSummaryDao">
<property name="dataSource" ref="dataSource" />
</bean>
@@ -74,7 +74,7 @@
</bean>
</property>
<property name="fieldSetMapper">
<bean class="org.springframework.batch.sample.domain.football.internal.PlayerFieldSetMapper" />
<bean class="org.springframework.batch.core.test.football.internal.PlayerFieldSetMapper" />
</property>
</bean>
</property>
@@ -91,7 +91,7 @@
</bean>
</property>
<property name="fieldSetMapper">
<bean class="org.springframework.batch.sample.domain.football.internal.GameFieldSetMapper" />
<bean class="org.springframework.batch.core.test.football.internal.GameFieldSetMapper" />
</property>
</bean>
</property>
@@ -101,7 +101,7 @@
<property name="dataSource" ref="dataSource" />
<property name="verifyCursorPosition" value="${batch.verify.cursor.position}" />
<property name="rowMapper">
<bean class="org.springframework.batch.sample.domain.football.internal.PlayerSummaryMapper" />
<bean class="org.springframework.batch.core.test.football.internal.PlayerSummaryMapper" />
</property>
<property name="sql">
<value>

View File

@@ -36,19 +36,19 @@
<property name="queueCapacity" value="0"/>
</bean>
<bean id="playerWriter" class="org.springframework.batch.sample.domain.football.internal.PlayerItemWriter">
<bean id="playerWriter" class="org.springframework.batch.core.test.football.internal.PlayerItemWriter">
<property name="playerDao">
<bean class="org.springframework.batch.sample.domain.football.internal.JdbcPlayerDao">
<bean class="org.springframework.batch.core.test.football.internal.JdbcPlayerDao">
<property name="dataSource" ref="dataSource" />
</bean>
</property>
</bean>
<bean id="gameWriter" class="org.springframework.batch.sample.domain.football.internal.JdbcGameDao">
<bean id="gameWriter" class="org.springframework.batch.core.test.football.internal.JdbcGameDao">
<property name="dataSource" ref="dataSource" />
</bean>
<bean id="summaryWriter" class="org.springframework.batch.sample.domain.football.internal.JdbcPlayerSummaryDao">
<bean id="summaryWriter" class="org.springframework.batch.core.test.football.internal.JdbcPlayerSummaryDao">
<property name="dataSource" ref="dataSource" />
</bean>
@@ -62,7 +62,7 @@
</bean>
</property>
<property name="fieldSetMapper">
<bean class="org.springframework.batch.sample.domain.football.internal.PlayerFieldSetMapper" />
<bean class="org.springframework.batch.core.test.football.internal.PlayerFieldSetMapper" />
</property>
</bean>
</property>
@@ -79,7 +79,7 @@
</bean>
</property>
<property name="fieldSetMapper">
<bean class="org.springframework.batch.sample.domain.football.internal.GameFieldSetMapper" />
<bean class="org.springframework.batch.core.test.football.internal.GameFieldSetMapper" />
</property>
</bean>
</property>
@@ -89,7 +89,7 @@
<property name="dataSource" ref="dataSource" />
<property name="verifyCursorPosition" value="${batch.verify.cursor.position}" />
<property name="rowMapper">
<bean class="org.springframework.batch.sample.domain.football.internal.PlayerSummaryMapper" />
<bean class="org.springframework.batch.core.test.football.internal.PlayerSummaryMapper" />
</property>
<property name="sql">
<value>

View File

@@ -1 +1 @@
http\://www.springframework.org/schema/batch/test=org.springframework.batch.test.namespace.config.DummyNamespaceHandler
http\://www.springframework.org/schema/batch/test=org.springframework.batch.core.test.namespace.config.DummyNamespaceHandler

View File

@@ -1 +1 @@
http\://www.springframework.org/schema/batch/test/test.xsd=org/springframework/batch/test/namespace/config/test.xsd
http\://www.springframework.org/schema/batch/test/test.xsd=org/springframework/batch/core/test/namespace/config/test.xsd

View File

@@ -9,6 +9,7 @@ batch.jdbc.user=sa
batch.jdbc.password=
batch.jdbc.testWhileIdle=false
batch.jdbc.validationQuery=
batch.drop.script=classpath:/org/springframework/batch/core/schema-drop-hsqldb.sql
batch.schema.script=classpath:/org/springframework/batch/core/schema-hsqldb.sql
batch.business.schema.script=classpath:/business-schema-hsqldb.sql
batch.data.source.init=true
@@ -17,3 +18,4 @@ batch.database.incrementer.parent=columnIncrementerParent
batch.grid.size=2
batch.verify.cursor.position=true
batch.jdbc.validationQuery=SELECT 1 FROM INFORMATION_SCHEMA.SYSTEM_USERS
batch.table.prefix=BATCH_

View File

@@ -7,7 +7,6 @@
<!-- Initialise the database before every test case: -->
<bean id="dataSourceInitializer" class="test.jdbc.datasource.DataSourceInitializer">
<property name="dataSource" ref="dataSource"/>
<property name="initialize" value="${batch.data.source.init}"/>
<property name="initScripts">
<list>
<value>${batch.drop.script}</value>