BATCHADM-63: add integration test for JDBC remote chunking
This commit is contained in:
David Syer
committed by
Michael Minella
Michael Minella
parent
b4092611d7
commit
b5611e051d
@@ -0,0 +1,89 @@
|
||||
package org.springframework.batch.integration.chunk;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
import java.util.Collections;
|
||||
|
||||
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.JobParameter;
|
||||
import org.springframework.batch.core.JobParameters;
|
||||
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.integration.Message;
|
||||
import org.springframework.integration.core.PollableChannel;
|
||||
import org.springframework.test.context.ContextConfiguration;
|
||||
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
|
||||
|
||||
@ContextConfiguration
|
||||
@RunWith(SpringJUnit4ClassRunner.class)
|
||||
public class RemoteChunkFaultTolerantStepJdbcIntegrationTests {
|
||||
|
||||
@Autowired
|
||||
private JobLauncher jobLauncher;
|
||||
|
||||
@Autowired
|
||||
private Job job;
|
||||
|
||||
@Autowired
|
||||
private PollableChannel replies;
|
||||
|
||||
@Before
|
||||
public void drain() {
|
||||
Message<?> message = replies.receive(100L);
|
||||
while (message!=null) {
|
||||
// System.err.println(message);
|
||||
message = replies.receive(100L);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFailedStep() throws Exception {
|
||||
JobExecution jobExecution = jobLauncher.run(job, new JobParameters(Collections.singletonMap("item.three",
|
||||
new JobParameter("unsupported"))));
|
||||
assertEquals(BatchStatus.FAILED, jobExecution.getStatus());
|
||||
StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next();
|
||||
assertEquals(9, stepExecution.getReadCount());
|
||||
// In principle the write count could be more than 2 and less than 9...
|
||||
assertEquals(7, stepExecution.getWriteCount());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFailedStepOnError() throws Exception {
|
||||
JobExecution jobExecution = jobLauncher.run(job, new JobParameters(Collections.singletonMap("item.three",
|
||||
new JobParameter("error"))));
|
||||
assertEquals(BatchStatus.FAILED, jobExecution.getStatus());
|
||||
StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next();
|
||||
assertEquals(9, stepExecution.getReadCount());
|
||||
// In principle the write count could be more than 2 and less than 9...
|
||||
assertEquals(7, stepExecution.getWriteCount());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSunnyDayFaultTolerant() throws Exception {
|
||||
JobExecution jobExecution = jobLauncher.run(job, new JobParameters(Collections.singletonMap("item.three",
|
||||
new JobParameter("3"))));
|
||||
assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
|
||||
StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next();
|
||||
assertEquals(9, stepExecution.getReadCount());
|
||||
assertEquals(9, stepExecution.getWriteCount());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSkipsInWriter() throws Exception {
|
||||
JobExecution jobExecution = jobLauncher.run(job, new JobParametersBuilder().addString("item.three", "fail")
|
||||
.addLong("run.id", 1L).toJobParameters());
|
||||
assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
|
||||
StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next();
|
||||
assertEquals(9, stepExecution.getReadCount());
|
||||
assertEquals(7, stepExecution.getWriteCount());
|
||||
// The whole chunk gets skipped...
|
||||
assertEquals(2, stepExecution.getWriteSkipCount());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
# Default database platform is HSQLDB:
|
||||
batch.jdbc.driver=org.hsqldb.jdbcDriver
|
||||
batch.jdbc.url=jdbc:hsqldb:mem:testdb;sql.enforce_strict_size=true
|
||||
# Override and use this one in for a separate server process so you can inspect
|
||||
# the results (or add it to system properties with -D to override at run time).
|
||||
# batch.jdbc.url=jdbc:hsqldb:hsql://localhost:9005/samples
|
||||
batch.jdbc.user=sa
|
||||
batch.jdbc.password=
|
||||
batch.jdbc.testWhileIdle=false
|
||||
batch.jdbc.validationQuery=
|
||||
batch.data.source.init=true
|
||||
batch.database.incrementer.class=org.springframework.jdbc.support.incrementer.HsqlMaxValueIncrementer
|
||||
batch.schema.script=classpath*:/org/springframework/batch/core/schema-hsqldb.sql
|
||||
batch.drop.script=classpath*:/org/springframework/batch/core/schema-drop-hsqldb.sql
|
||||
integration.schema.script=classpath*:/org/springframework/integration/jdbc/schema-hsqldb.sql
|
||||
@@ -0,0 +1,117 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns:beans="http://www.springframework.org/schema/beans" xmlns:integration="http://www.springframework.org/schema/integration"
|
||||
xmlns:batch="http://www.springframework.org/schema/batch" xmlns:context="http://www.springframework.org/schema/context"
|
||||
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
|
||||
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
|
||||
http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
|
||||
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
|
||||
http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms-2.0.xsd
|
||||
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
|
||||
http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.1.xsd
|
||||
http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
|
||||
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd">
|
||||
|
||||
<job id="job" xmlns="http://www.springframework.org/schema/batch">
|
||||
<step id="step">
|
||||
<tasklet>
|
||||
<chunk reader="reader" writer="writer" commit-interval="2" skip-limit="2">
|
||||
<skippable-exception-classes>
|
||||
<include class="java.lang.IllegalStateException" />
|
||||
</skippable-exception-classes>
|
||||
</chunk>
|
||||
</tasklet>
|
||||
</step>
|
||||
</job>
|
||||
|
||||
<bean id="reader" class="org.springframework.batch.integration.chunk.TestItemReader" scope="step">
|
||||
<property name="items">
|
||||
<list>
|
||||
<value>1</value>
|
||||
<value>2</value>
|
||||
<value>#{jobParameters['item.three']}</value>
|
||||
<value>4</value>
|
||||
<value>5</value>
|
||||
<value>6</value>
|
||||
<value>7</value>
|
||||
<value>8</value>
|
||||
<value>9</value>
|
||||
</list>
|
||||
</property>
|
||||
</bean>
|
||||
|
||||
<bean id="writer" class="org.springframework.batch.integration.chunk.TestItemWriter" />
|
||||
|
||||
<bean id="chunkWriter" class="org.springframework.batch.integration.chunk.ChunkMessageChannelItemWriter" scope="step">
|
||||
<property name="messagingOperations" ref="messagingGateway" />
|
||||
<property name="replyChannel" ref="replies" />
|
||||
<property name="maxWaitTimeouts" value="10" />
|
||||
</bean>
|
||||
|
||||
<bean id="messagingGateway" class="org.springframework.integration.core.MessagingTemplate">
|
||||
<property name="defaultChannel" ref="requests" />
|
||||
<property name="receiveTimeout" value="1000" />
|
||||
</bean>
|
||||
|
||||
<bean class="org.springframework.beans.factory.config.CustomScopeConfigurer">
|
||||
<property name="scopes">
|
||||
<map>
|
||||
<entry key="thread">
|
||||
<bean class="org.springframework.context.support.SimpleThreadScope" />
|
||||
</entry>
|
||||
</map>
|
||||
</property>
|
||||
</bean>
|
||||
|
||||
<int-jdbc:message-store id="messageStore" data-source="dataSource"/>
|
||||
|
||||
<integration:channel id="requests">
|
||||
<integration:queue message-store="messageStore" />
|
||||
</integration:channel>
|
||||
<integration:channel id="replies">
|
||||
<integration:queue message-store="messageStore" />
|
||||
</integration:channel>
|
||||
<integration:service-activator input-channel="requests" output-channel="replies" ref="chunkHandler">
|
||||
<integration:poller>
|
||||
<integration:interval-trigger interval="100" />
|
||||
<integration:transactional />
|
||||
</integration:poller>
|
||||
</integration:service-activator>
|
||||
|
||||
<bean id="chunkHandler" class="org.springframework.batch.integration.chunk.RemoteChunkHandlerFactoryBean">
|
||||
<property name="chunkWriter" ref="chunkWriter" />
|
||||
<property name="step" ref="step" />
|
||||
</bean>
|
||||
|
||||
<bean id="dataSource" class="org.apache.commons.dbcp.BasicDataSource">
|
||||
<property name="driverClassName" value="${batch.jdbc.driver}" />
|
||||
<property name="url" value="${batch.jdbc.url}" />
|
||||
<property name="username" value="${batch.jdbc.user}" />
|
||||
<property name="password" value="${batch.jdbc.password}" />
|
||||
<property name="testWhileIdle" value="${batch.jdbc.testWhileIdle}" />
|
||||
<property name="validationQuery" value="${batch.jdbc.validationQuery}" />
|
||||
</bean>
|
||||
|
||||
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
|
||||
<property name="dataSource" ref="dataSource" />
|
||||
</bean>
|
||||
|
||||
<context:property-placeholder location="classpath:config-${ENVIRONMENT:hsql}.properties"
|
||||
system-properties-mode="OVERRIDE" ignore-resource-not-found="true" ignore-unresolvable="true" />
|
||||
|
||||
<!-- Initialise the database if enabled: -->
|
||||
<jdbc:initialize-database data-source="dataSource" enabled="${batch.data.source.init}"
|
||||
ignore-failures="DROPS">
|
||||
<jdbc:script location="${batch.drop.script}" />
|
||||
<jdbc:script location="${batch.schema.script}" />
|
||||
<jdbc:script location="${integration.schema.script}" />
|
||||
</jdbc:initialize-database>
|
||||
|
||||
<batch:job-repository id="jobRepository" />
|
||||
|
||||
<bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
|
||||
<property name="jobRepository" ref="jobRepository" />
|
||||
</bean>
|
||||
|
||||
</beans>
|
||||
Reference in New Issue
Block a user