SpringCloud/spring-cloud-task
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Updated to files to fit the Standard.

Browse Source
This commit is contained in:
Glenn Renfro
2022-07-25 11:41:43 -04:00
parent 39908bb499
commit 76a5d12136
200 changed files with 2839 additions and 4125 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-cloud-task-integration-tests/src/test/java/configuration/JobConfiguration.java
View File

@@ -41,8 +41,7 @@ import org.springframework.context.annotation.Configuration;
*/
@Configuration
@EnableBatchProcessing
@ConditionalOnProperty(prefix = "spring.cloud.task.test", name = "enable-job-configuration",
havingValue = "true")
@ConditionalOnProperty(prefix = "spring.cloud.task.test", name = "enable-job-configuration", havingValue = "true")
public class JobConfiguration {
private static final int DEFAULT_CHUNK_COUNT = 3;
@@ -62,8 +61,7 @@ public class JobConfiguration {
public Step step1() {
return this.stepBuilderFactory.get("step1").tasklet(new Tasklet() {
@Override
public RepeatStatus execute(StepContribution contribution,
ChunkContext chunkContext) throws Exception {
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
System.out.println("Executed");
return RepeatStatus.FINISHED;
}
@@ -72,8 +70,7 @@ public class JobConfiguration {
@Bean
public Step step2() {
return this.stepBuilderFactory.get("step2")
.<String, String>chunk(DEFAULT_CHUNK_COUNT)
return this.stepBuilderFactory.get("step2").<String, String>chunk(DEFAULT_CHUNK_COUNT)
.reader(new ListItemReader<>(Arrays.asList("1", "2", "3", "4", "5", "6")))
.processor(new ItemProcessor<String, String>() {
@Override

9
spring-cloud-task-integration-tests/src/test/java/configuration/JobSkipConfiguration.java
View File

@@ -54,8 +54,7 @@ public class JobSkipConfiguration {
public Step step1() {
return this.stepBuilderFactory.get("step1").tasklet(new Tasklet() {
@Override
public RepeatStatus execute(StepContribution contribution,
ChunkContext chunkContext) throws Exception {
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
System.out.println("Executed");
return RepeatStatus.FINISHED;
}
@@ -64,10 +63,8 @@ public class JobSkipConfiguration {
@Bean
public Step step2() {
return this.stepBuilderFactory.get("step2").chunk(3).faultTolerant()
.skip(IllegalStateException.class).skipLimit(100)
.reader(new SkipItemReader())
.processor(new ItemProcessor<Object, Object>() {
return this.stepBuilderFactory.get("step2").chunk(3).faultTolerant().skip(IllegalStateException.class)
.skipLimit(100).reader(new SkipItemReader()).processor(new ItemProcessor<Object, Object>() {
@Override
public String process(Object item) throws Exception {
return String.valueOf(Integer.parseInt((String) item) * -1);

3
spring-cloud-task-integration-tests/src/test/java/configuration/SkipItemReader.java
View File

@@ -31,8 +31,7 @@ public class SkipItemReader implements ItemReader {
boolean finished = false;
@Override
public Object read() throws Exception, UnexpectedInputException, ParseException,
NonTransientResourceException {
public Object read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
String result = "1";
if (this.failCount < 2) {
this.failCount++;

88
spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/executionid/TaskStartTests.java
View File

@@ -72,8 +72,7 @@ public class TaskStartTests {
private final static int MAX_WAIT_TIME = 5000;
private final static String URL = "maven://io.spring.cloud:"
+ "timestamp-task:jar:1.1.0.RELEASE";
private final static String URL = "maven://io.spring.cloud:" + "timestamp-task:jar:1.1.0.RELEASE";
private final static String DATASOURCE_URL;
@@ -89,8 +88,8 @@ public class TaskStartTests {
static {
randomPort = TestSocketUtils.findAvailableTcpPort();
DATASOURCE_URL = "jdbc:h2:tcp://localhost:" + randomPort
+ "/mem:dataflow;DB_CLOSE_DELAY=-1;" + "DB_CLOSE_ON_EXIT=FALSE";
DATASOURCE_URL = "jdbc:h2:tcp://localhost:" + randomPort + "/mem:dataflow;DB_CLOSE_DELAY=-1;"
+ "DB_CLOSE_ON_EXIT=FALSE";
}
private DataSource dataSource;
@@ -113,8 +112,7 @@ public class TaskStartTests {
@Autowired
public void setDataSource(DataSource dataSource) {
this.dataSource = dataSource;
TaskExecutionDaoFactoryBean factoryBean = new TaskExecutionDaoFactoryBean(
dataSource);
TaskExecutionDaoFactoryBean factoryBean = new TaskExecutionDaoFactoryBean(dataSource);
this.taskExplorer = new SimpleTaskExplorer(factoryBean);
this.taskRepository = new SimpleTaskRepository(factoryBean);
}
@@ -125,8 +123,7 @@ public class TaskStartTests {
this.properties.put("spring.datasource.url", DATASOURCE_URL);
this.properties.put("spring.datasource.username", DATASOURCE_USER_NAME);
this.properties.put("spring.datasource.password", DATASOURCE_USER_PASSWORD);
this.properties.put("spring.datasource.driverClassName",
DATASOURCE_DRIVER_CLASS_NAME);
this.properties.put("spring.datasource.driverClassName", DATASOURCE_DRIVER_CLASS_NAME);
this.properties.put("spring.application.name", TASK_NAME);
this.properties.put("spring.cloud.task.initialize-enabled", "false");
@@ -151,8 +148,7 @@ public class TaskStartTests {
initializer.setDataSource(this.dataSource);
ResourceDatabasePopulator databasePopulator = new ResourceDatabasePopulator();
databasePopulator.addScript(
new ClassPathResource("/org/springframework/cloud/task/schema-h2.sql"));
databasePopulator.addScript(new ClassPathResource("/org/springframework/cloud/task/schema-h2.sql"));
initializer.setDatabasePopulator(databasePopulator);
initializer.afterPropertiesSet();
}
@@ -160,41 +156,33 @@ public class TaskStartTests {
@Test
public void testWithGeneratedTaskExecution() throws Exception {
this.taskRepository.createTaskExecution();
assertThat(this.taskExplorer.getTaskExecutionCount())
.as("Only one row is expected").isEqualTo(1);
assertThat(this.taskExplorer.getTaskExecutionCount()).as("Only one row is expected").isEqualTo(1);
this.applicationContext = getTaskApplication(1).run(new String[0]);
assertThat(waitForDBToBePopulated()).isTrue();
Page<TaskExecution> taskExecutions = this.taskExplorer
.findAll(PageRequest.of(0, 10));
Page<TaskExecution> taskExecutions = this.taskExplorer.findAll(PageRequest.of(0, 10));
TaskExecution te = taskExecutions.iterator().next();
assertThat(taskExecutions.getTotalElements()).as("Only one row is expected")
.isEqualTo(1);
assertThat(taskExecutions.iterator().next().getExitCode().intValue())
.as("return code should be 0").isEqualTo(0);
assertThat(taskExecutions.getTotalElements()).as("Only one row is expected").isEqualTo(1);
assertThat(taskExecutions.iterator().next().getExitCode().intValue()).as("return code should be 0")
.isEqualTo(0);
}
@Test
public void testWithGeneratedTaskExecutionWithName() throws Exception {
final String TASK_EXECUTION_NAME = "PRE-EXECUTION-TEST-NAME";
this.taskRepository.createTaskExecution(TASK_EXECUTION_NAME);
assertThat(this.taskExplorer.getTaskExecutionCount())
.as("Only one row is expected").isEqualTo(1);
assertThat(this.taskExplorer.getTaskExecution(1).getTaskName())
.isEqualTo(TASK_EXECUTION_NAME);
assertThat(this.taskExplorer.getTaskExecutionCount()).as("Only one row is expected").isEqualTo(1);
assertThat(this.taskExplorer.getTaskExecution(1).getTaskName()).isEqualTo(TASK_EXECUTION_NAME);
this.applicationContext = getTaskApplication(1).run(new String[0]);
assertThat(waitForDBToBePopulated()).isTrue();
Page<TaskExecution> taskExecutions = this.taskExplorer
.findAll(PageRequest.of(0, 10));
Page<TaskExecution> taskExecutions = this.taskExplorer.findAll(PageRequest.of(0, 10));
TaskExecution te = taskExecutions.iterator().next();
assertThat(taskExecutions.getTotalElements()).as("Only one row is expected")
.isEqualTo(1);
assertThat(taskExecutions.iterator().next().getExitCode().intValue())
.as("return code should be 0").isEqualTo(0);
assertThat(this.taskExplorer.getTaskExecution(1).getTaskName())
.isEqualTo("batchEvents");
assertThat(taskExecutions.getTotalElements()).as("Only one row is expected").isEqualTo(1);
assertThat(taskExecutions.iterator().next().getExitCode().intValue()).as("return code should be 0")
.isEqualTo(0);
assertThat(this.taskExplorer.getTaskExecution(1).getTaskName()).isEqualTo("batchEvents");
}
@Test
@@ -207,8 +195,7 @@ public class TaskStartTests {
@Test
public void testCompletedTaskExecution() throws Exception {
this.taskRepository.createTaskExecution();
assertThat(this.taskExplorer.getTaskExecutionCount())
.as("Only one row is expected").isEqualTo(1);
assertThat(this.taskExplorer.getTaskExecutionCount()).as("Only one row is expected").isEqualTo(1);
this.taskRepository.completeTaskExecution(1, 0, new Date(), "");
assertThatExceptionOfType(ApplicationContextException.class).isThrownBy(() -> {
this.applicationContext = getTaskApplication(1).run(new String[0]);
@@ -217,26 +204,21 @@ public class TaskStartTests {
@Test
public void testDuplicateTaskExecutionWithSingleInstanceEnabled() throws Exception {
String[] params = { "--spring.cloud.task.single-instance-enabled=true",
"--spring.cloud.task.name=foo" };
String[] params = { "--spring.cloud.task.single-instance-enabled=true", "--spring.cloud.task.name=foo" };
boolean testFailed = false;
try {
this.taskRepository.createTaskExecution();
assertThat(this.taskExplorer.getTaskExecutionCount())
.as("Only one row is expected").isEqualTo(1);
assertThat(this.taskExplorer.getTaskExecutionCount()).as("Only one row is expected").isEqualTo(1);
enableLock("foo");
getTaskApplication(1).run(params);
}
catch (ApplicationContextException taskException) {
assertThat(taskException.getMessage())
.isEqualTo("Failed to start bean 'taskLifecycleListener'; nested "
+ "exception is org.springframework.cloud.task."
+ "listener.TaskExecutionException: Failed to process "
+ "@BeforeTask or @AfterTask annotation because: Task with name \"foo\" is already running.");
assertThat(taskException.getCause().getMessage()).isEqualTo("Failed to process "
+ "@BeforeTask or @AfterTask annotation because: Task with name \"foo\" is already running.");
testFailed = true;
}
assertThat(testFailed).as("Expected TaskExecutionException for because of "
+ "single-instance-enabled is enabled").isTrue();
assertThat(testFailed)
.as("Expected TaskExecutionException for because of " + "single-instance-enabled is enabled").isTrue();
}
@@ -244,8 +226,7 @@ public class TaskStartTests {
public void testDuplicateTaskExecutionWithSingleInstanceDisabled() throws Exception {
this.taskRepository.createTaskExecution();
TaskExecution execution = this.taskRepository.createTaskExecution();
this.taskRepository.startTaskExecution(execution.getExecutionId(), "bar",
new Date(), new ArrayList<>(), "");
this.taskRepository.startTaskExecution(execution.getExecutionId(), "bar", new Date(), new ArrayList<>(), "");
String[] params = { "--spring.cloud.task.name=bar" };
enableLock("bar");
this.applicationContext = getTaskApplication(1).run(params);
@@ -258,8 +239,7 @@ public class TaskStartTests {
ConfigurableEnvironment environment = new StandardEnvironment();
MutablePropertySources propertySources = environment.getPropertySources();
myMap.put("spring.cloud.task.executionid", executionId);
propertySources
.addFirst(new MapPropertySource("EnvrionmentTestPropsource", myMap));
propertySources.addFirst(new MapPropertySource("EnvrionmentTestPropsource", myMap));
myapp.setEnvironment(environment);
return myapp;
}
@@ -267,8 +247,7 @@ public class TaskStartTests {
private boolean tableExists() throws SQLException {
boolean result;
try (Connection conn = this.dataSource.getConnection();
ResultSet res = conn.getMetaData().getTables(null, null, "TASK_EXECUTION",
new String[] { "TABLE" })) {
ResultSet res = conn.getMetaData().getTables(null, null, "TASK_EXECUTION", new String[] { "TABLE" })) {
result = res.next();
}
return result;
@@ -287,11 +266,9 @@ public class TaskStartTests {
}
private void enableLock(String lockKey) {
SimpleJdbcInsert taskLockInsert = new SimpleJdbcInsert(this.dataSource)
.withTableName("TASK_LOCK");
SimpleJdbcInsert taskLockInsert = new SimpleJdbcInsert(this.dataSource).withTableName("TASK_LOCK");
Map<String, Object> taskLockParams = new HashMap<>();
taskLockParams.put("LOCK_KEY",
UUID.nameUUIDFromBytes(lockKey.getBytes()).toString());
taskLockParams.put("LOCK_KEY", UUID.nameUUIDFromBytes(lockKey.getBytes()).toString());
taskLockParams.put("REGION", "DEFAULT");
taskLockParams.put("CLIENT_ID", "aClientID");
taskLockParams.put("CREATED_DATE", new Date());
@@ -308,9 +285,8 @@ public class TaskStartTests {
Server server = null;
try {
if (defaultServer == null) {
server = Server.createTcpServer("-ifNotExists", "-tcp",
"-tcpAllowOthers", "-tcpPort", String.valueOf(randomPort))
.start();
server = Server.createTcpServer("-ifNotExists", "-tcp", "-tcpAllowOthers", "-tcpPort",
String.valueOf(randomPort)).start();
defaultServer = server;
}
}

39
spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/initializer/TaskInitializerTests.java
View File

@@ -47,7 +47,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DriverManagerDataSource;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
@@ -59,8 +58,7 @@ public class TaskInitializerTests {
private final static int MAX_WAIT_TIME = 5000;
private final static String URL = "maven://io.spring.cloud:"
+ "timestamp-task:jar:1.1.0.RELEASE";
private final static String URL = "maven://io.spring.cloud:" + "timestamp-task:jar:1.1.0.RELEASE";
private final static String DATASOURCE_URL;
@@ -76,8 +74,8 @@ public class TaskInitializerTests {
static {
randomPort = TestSocketUtils.findAvailableTcpPort();
DATASOURCE_URL = "jdbc:h2:tcp://localhost:" + randomPort
+ "/mem:dataflow;DB_CLOSE_DELAY=-1;" + "DB_CLOSE_ON_EXIT=FALSE";
DATASOURCE_URL = "jdbc:h2:tcp://localhost:" + randomPort + "/mem:dataflow;DB_CLOSE_DELAY=-1;"
+ "DB_CLOSE_ON_EXIT=FALSE";
}
private DataSource dataSource;
@@ -96,8 +94,7 @@ public class TaskInitializerTests {
@Autowired
public void setDataSource(DataSource dataSource) {
this.dataSource = dataSource;
TaskExecutionDaoFactoryBean factoryBean = new TaskExecutionDaoFactoryBean(
dataSource);
TaskExecutionDaoFactoryBean factoryBean = new TaskExecutionDaoFactoryBean(dataSource);
this.taskExplorer = new SimpleTaskExplorer(factoryBean);
}
@@ -138,13 +135,11 @@ public class TaskInitializerTests {
this.applicationContext = myapp.run(properties);
assertThat(waitForDBToBePopulated()).isTrue();
Page<TaskExecution> taskExecutions = this.taskExplorer
.findAll(PageRequest.of(0, 10));
Page<TaskExecution> taskExecutions = this.taskExplorer.findAll(PageRequest.of(0, 10));
TaskExecution te = taskExecutions.iterator().next();
assertThat(taskExecutions.getTotalElements()).as("Only one row is expected")
.isEqualTo(1);
assertThat(taskExecutions.iterator().next().getExitCode().intValue())
.as("return code should be 0").isEqualTo(0);
assertThat(taskExecutions.getTotalElements()).as("Only one row is expected").isEqualTo(1);
assertThat(taskExecutions.iterator().next().getExitCode().intValue()).as("return code should be 0")
.isEqualTo(0);
}
@Test
@@ -163,20 +158,17 @@ public class TaskInitializerTests {
this.applicationContext = myapp.run(properties);
assertThat(waitForDBToBePopulated()).isTrue();
Page<TaskExecution> taskExecutions = this.taskExplorer
.findAll(PageRequest.of(0, 10));
Page<TaskExecution> taskExecutions = this.taskExplorer.findAll(PageRequest.of(0, 10));
TaskExecution te = taskExecutions.iterator().next();
assertThat(taskExecutions.getTotalElements()).as("Only one row is expected")
.isEqualTo(1);
assertThat(taskExecutions.iterator().next().getExitCode().intValue())
.as("return code should be 0").isEqualTo(0);
assertThat(taskExecutions.getTotalElements()).as("Only one row is expected").isEqualTo(1);
assertThat(taskExecutions.iterator().next().getExitCode().intValue()).as("return code should be 0")
.isEqualTo(0);
}
private boolean tableExists() throws SQLException {
boolean result;
try (Connection conn = this.dataSource.getConnection();
ResultSet res = conn.getMetaData().getTables(null, null, "TASK_EXECUTION",
new String[] { "TABLE" })) {
ResultSet res = conn.getMetaData().getTables(null, null, "TASK_EXECUTION", new String[] { "TABLE" })) {
result = res.next();
}
return result;
@@ -204,9 +196,8 @@ public class TaskInitializerTests {
Server server = null;
try {
if (defaultServer == null) {
server = Server.createTcpServer("-ifNotExists", "-tcp",
"-tcpAllowOthers", "-tcpPort", String.valueOf(randomPort))
.start();
server = Server.createTcpServer("-ifNotExists", "-tcp", "-tcpAllowOthers", "-tcpPort",
String.valueOf(randomPort)).start();
defaultServer = server;
}
}

47
spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/launcher/TaskLauncherSinkTests.java
View File

@@ -56,21 +56,17 @@ import org.springframework.test.context.junit.jupiter.SpringExtension;
import static org.assertj.core.api.Assertions.assertThat;
@ExtendWith(SpringExtension.class)
@SpringBootTest(
classes = { TaskLauncherSinkApplication.class,
TaskLauncherSinkTests.TaskLauncherConfiguration.class },
properties = {
"maven.remote-repositories.repo1.url=https://repo.spring.io/libs-release",
"spring.cloud.stream.function.bindings.taskLauncherSink-in-0=input",
"spring.cloud.stream.bindings.input.destination=taskLauncherSinkExchange" })
@SpringBootTest(classes = { TaskLauncherSinkApplication.class, TaskLauncherSinkTests.TaskLauncherConfiguration.class },
properties = { "maven.remote-repositories.repo1.url=https://repo.spring.io/libs-release",
"spring.cloud.stream.function.bindings.taskLauncherSink-in-0=input",
"spring.cloud.stream.bindings.input.destination=taskLauncherSinkExchange" })
public class TaskLauncherSinkTests {
private final static int WAIT_INTERVAL = 500;
private final static int MAX_WAIT_TIME = 120000;
private final static String URL = "maven://io.spring.cloud:"
+ "timestamp-task:3.0.0-SNAPSHOT";
private final static String URL = "maven://io.spring.cloud:" + "timestamp-task:3.0.0-SNAPSHOT";
private final static String DATASOURCE_URL;
@@ -86,8 +82,8 @@ public class TaskLauncherSinkTests {
static {
randomPort = TestSocketUtils.findAvailableTcpPort();
DATASOURCE_URL = "jdbc:h2:tcp://localhost:" + randomPort
+ "/mem:dataflow;DB_CLOSE_DELAY=-1;" + "DB_CLOSE_ON_EXIT=FALSE";
DATASOURCE_URL = "jdbc:h2:tcp://localhost:" + randomPort + "/mem:dataflow;DB_CLOSE_DELAY=-1;"
+ "DB_CLOSE_ON_EXIT=FALSE";
}
@Autowired
@@ -102,8 +98,7 @@ public class TaskLauncherSinkTests {
@Autowired
public void setDataSource(DataSource dataSource) {
this.dataSource = dataSource;
this.taskExplorer = new SimpleTaskExplorer(
new TaskExecutionDaoFactoryBean(dataSource));
this.taskExplorer = new SimpleTaskExplorer(new TaskExecutionDaoFactoryBean(dataSource));
}
@BeforeEach
@@ -112,8 +107,7 @@ public class TaskLauncherSinkTests {
this.properties.put("spring.datasource.url", DATASOURCE_URL);
this.properties.put("spring.datasource.username", DATASOURCE_USER_NAME);
this.properties.put("spring.datasource.password", DATASOURCE_USER_PASSWORD);
this.properties.put("spring.datasource.driverClassName",
DATASOURCE_DRIVER_CLASS_NAME);
this.properties.put("spring.datasource.driverClassName", DATASOURCE_DRIVER_CLASS_NAME);
this.properties.put("spring.application.name", TASK_NAME);
JdbcTemplate template = new JdbcTemplate(this.dataSource);
@@ -123,8 +117,7 @@ public class TaskLauncherSinkTests {
initializer.setDataSource(this.dataSource);
ResourceDatabasePopulator databasePopulator = new ResourceDatabasePopulator();
databasePopulator.addScript(
new ClassPathResource("/org/springframework/cloud/task/schema-h2.sql"));
databasePopulator.addScript(new ClassPathResource("/org/springframework/cloud/task/schema-h2.sql"));
initializer.setDatabasePopulator(databasePopulator);
initializer.afterPropertiesSet();
@@ -135,20 +128,17 @@ public class TaskLauncherSinkTests {
launchTask(URL);
assertThat(waitForDBToBePopulated()).isTrue();
Page<TaskExecution> taskExecutions = this.taskExplorer
.findAll(PageRequest.of(0, 10));
assertThat(taskExecutions.getTotalElements()).as("Only one row is expected")
.isEqualTo(1);
Page<TaskExecution> taskExecutions = this.taskExplorer.findAll(PageRequest.of(0, 10));
assertThat(taskExecutions.getTotalElements()).as("Only one row is expected").isEqualTo(1);
assertThat(waitForTaskToComplete()).isTrue();
assertThat(taskExecutions.iterator().next().getExitCode().intValue())
.as("return code should be 0").isEqualTo(0);
assertThat(taskExecutions.iterator().next().getExitCode().intValue()).as("return code should be 0")
.isEqualTo(0);
}
private boolean tableExists() throws SQLException {
boolean result;
try (Connection conn = this.dataSource.getConnection();
ResultSet res = conn.getMetaData().getTables(null, null, "TASK_EXECUTION",
new String[] { "TABLE" })) {
ResultSet res = conn.getMetaData().getTables(null, null, "TASK_EXECUTION", new String[] { "TABLE" })) {
result = res.next();
}
return result;
@@ -180,8 +170,7 @@ public class TaskLauncherSinkTests {
}
private void launchTask(String artifactURL) {
TaskLaunchRequest request = new TaskLaunchRequest(artifactURL, null,
this.properties, null, null);
TaskLaunchRequest request = new TaskLaunchRequest(artifactURL, null, this.properties, null, null);
GenericMessage<TaskLaunchRequest> message = new GenericMessage<>(request);
this.streamBridge.send("taskLauncherSinkExchange", message);
}
@@ -201,8 +190,8 @@ public class TaskLauncherSinkTests {
public Server initH2TCPServer() {
Server server;
try {
server = Server.createTcpServer("-ifNotExists", "-tcp", "-tcpAllowOthers",
"-tcpPort", String.valueOf(randomPort)).start();
server = Server.createTcpServer("-ifNotExists", "-tcp", "-tcpAllowOthers", "-tcpPort",
String.valueOf(randomPort)).start();
}
catch (SQLException e) {
throw new IllegalStateException(e);

89
spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/listener/BatchExecutionEventTests.java
View File

@@ -29,7 +29,6 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
@@ -51,6 +50,7 @@ public class BatchExecutionEventTests {
private static final String TASK_NAME = "taskEventTest";
private final ObjectMapper objectMapper = new ObjectMapper();
private ConfigurableApplicationContext applicationContext;
@BeforeEach
@@ -68,45 +68,39 @@ public class BatchExecutionEventTests {
@Test
public void testContext() {
this.applicationContext = new SpringApplicationBuilder()
.sources(TestChannelBinderConfiguration
.getCompleteConfiguration(BatchEventsApplication.class)).web(WebApplicationType.NONE)
.build().run(getCommandLineParams(
"--spring.cloud.stream.bindings.job-execution-events.destination=bazbar"));
.sources(TestChannelBinderConfiguration.getCompleteConfiguration(BatchEventsApplication.class))
.web(WebApplicationType.NONE).build()
.run(getCommandLineParams("--spring.cloud.stream.bindings.job-execution-events.destination=bazbar"));
assertThat(this.applicationContext.getBean("jobExecutionEventsListener"))
.isNotNull();
assertThat(this.applicationContext.getBean("stepExecutionEventsListener"))
.isNotNull();
assertThat(this.applicationContext.getBean("jobExecutionEventsListener")).isNotNull();
assertThat(this.applicationContext.getBean("stepExecutionEventsListener")).isNotNull();
assertThat(this.applicationContext.getBean("chunkEventsListener")).isNotNull();
assertThat(this.applicationContext.getBean("itemReadEventsListener")).isNotNull();
assertThat(this.applicationContext.getBean("itemWriteEventsListener"))
.isNotNull();
assertThat(this.applicationContext.getBean("itemProcessEventsListener"))
.isNotNull();
assertThat(this.applicationContext.getBean("itemWriteEventsListener")).isNotNull();
assertThat(this.applicationContext.getBean("itemProcessEventsListener")).isNotNull();
assertThat(this.applicationContext.getBean("skipEventsListener")).isNotNull();
}
@Test
public void testJobEventListener() throws Exception {
List<Message<byte[]>> result = testListener(
"--spring.cloud.task.batch.events.jobExecutionEventBindingName=foobar", "foobar", 1);
"--spring.cloud.task.batch.events.jobExecutionEventBindingName=foobar", "foobar", 1);
JobExecutionEvent jobExecutionEvent = this.objectMapper.readValue(result.get(0).getPayload(),
JobExecutionEvent.class);
Assertions.assertThat(jobExecutionEvent.getJobInstance().getJobName())
.isEqualTo("job").as("Job name should be job");
JobExecutionEvent.class);
Assertions.assertThat(jobExecutionEvent.getJobInstance().getJobName()).isEqualTo("job")
.as("Job name should be job");
}
@Test
public void testStepEventListener() throws Exception {
final String bindingName = "step-execution-foobar";
List<Message<byte[]>> result = testListener(
"--spring.cloud.task.batch.events.stepExecutionEventBindingName=" + bindingName,
bindingName, 4);
"--spring.cloud.task.batch.events.stepExecutionEventBindingName=" + bindingName, bindingName, 4);
int stepOneCount = 0;
int stepTwoCount = 0;
for (int i = 0; i < 4; i++) {
StepExecutionEvent stepExecutionEvent = this.objectMapper.readValue(result.get(i).getPayload(),
StepExecutionEvent.class);
StepExecutionEvent.class);
if (stepExecutionEvent.getStepName().equals("step1")) {
stepOneCount++;
}
@@ -115,10 +109,8 @@ public class BatchExecutionEventTests {
}
}
assertThat(stepOneCount).as("the number of step1 events did not match")
.isEqualTo(2);
assertThat(stepTwoCount).as("the number of step2 events did not match")
.isEqualTo(2);
assertThat(stepOneCount).as("the number of step1 events did not match").isEqualTo(2);
assertThat(stepTwoCount).as("the number of step2 events did not match").isEqualTo(2);
}
@@ -127,8 +119,7 @@ public class BatchExecutionEventTests {
final String bindingName = "item-execution-foobar";
List<Message<byte[]>> result = testListener(
"--spring.cloud.task.batch.events.itemProcessEventBindingName=" + bindingName,
bindingName, 1);
"--spring.cloud.task.batch.events.itemProcessEventBindingName=" + bindingName, bindingName, 1);
String value = new String(result.get(0).getPayload());
assertThat(value).isEqualTo("item did not equal result after processing");
@@ -139,8 +130,7 @@ public class BatchExecutionEventTests {
final String bindingName = "chunk-events-foobar";
List<Message<byte[]>> result = testListener(
"--spring.cloud.task.batch.events.chunkEventBindingName=" + bindingName,
bindingName, 2);
"--spring.cloud.task.batch.events.chunkEventBindingName=" + bindingName, bindingName, 2);
String value = new String(result.get(0).getPayload());
assertThat(value).isEqualTo("Before Chunk Processing");
value = new String(result.get(1).getPayload());
@@ -152,8 +142,7 @@ public class BatchExecutionEventTests {
final String bindingName = "item-write-events-foobar";
List<Message<byte[]>> result = testListener(
"--spring.cloud.task.batch.events.itemWriteEventBindingName=" + bindingName,
bindingName, 2);
"--spring.cloud.task.batch.events.itemWriteEventBindingName=" + bindingName, bindingName, 2);
String value = new String(result.get(0).getPayload());
assertThat(value).isEqualTo("3 items to be written.");
value = new String(result.get(1).getPayload());
@@ -165,16 +154,12 @@ public class BatchExecutionEventTests {
}
private String[] getCommandLineParams(String sinkChannelParam, boolean enableFailJobConfig) {
String jobConfig = enableFailJobConfig ?
"--spring.cloud.task.test.enable-job-configuration=true" :
"--spring.cloud.task.test.enable-fail-job-configuration=true";
return new String[]{"--spring.cloud.task.closecontext_enable=false",
"--spring.cloud.task.name=" + TASK_NAME,
"--spring.main.web-environment=false",
"--spring.cloud.stream.defaultBinder=rabbit",
"--spring.cloud.stream.bindings.task-events.destination=test",
jobConfig,
"foo=" + UUID.randomUUID(), sinkChannelParam};
String jobConfig = enableFailJobConfig ? "--spring.cloud.task.test.enable-job-configuration=true"
: "--spring.cloud.task.test.enable-fail-job-configuration=true";
return new String[] { "--spring.cloud.task.closecontext_enable=false", "--spring.cloud.task.name=" + TASK_NAME,
"--spring.main.web-environment=false", "--spring.cloud.stream.defaultBinder=rabbit",
"--spring.cloud.stream.bindings.task-events.destination=test", jobConfig, "foo=" + UUID.randomUUID(),
sinkChannelParam };
}
private List<Message<byte[]>> testListener(String channelBinding, String bindingName, int numberToRead) {
@@ -185,14 +170,13 @@ public class BatchExecutionEventTests {
return testListenerForApp(channelBinding, bindingName, numberToRead, BatchSkipEventsApplication.class, false);
}
private List<Message<byte[]>> testListenerForApp(String channelBinding,
String bindingName, int numberToRead, Class clazz, boolean enableFailJobConfig) {
private List<Message<byte[]>> testListenerForApp(String channelBinding, String bindingName, int numberToRead,
Class clazz, boolean enableFailJobConfig) {
List<Message<byte[]>> results = new ArrayList<>();
this.applicationContext = new SpringApplicationBuilder()
.sources(TestChannelBinderConfiguration
.getCompleteConfiguration(clazz)).web(WebApplicationType.NONE)
.build().run(getCommandLineParams(channelBinding, enableFailJobConfig));
.sources(TestChannelBinderConfiguration.getCompleteConfiguration(clazz)).web(WebApplicationType.NONE)
.build().run(getCommandLineParams(channelBinding, enableFailJobConfig));
OutputDestination target = this.applicationContext.getBean(OutputDestination.class);
@@ -207,8 +191,7 @@ public class BatchExecutionEventTests {
final String bindingName = "item-read-events-foobar";
List<Message<byte[]>> result = testListenerSkip(
"--spring.cloud.task.batch.events.itemReadEventBindingName=" + bindingName,
bindingName, 1);
"--spring.cloud.task.batch.events.itemReadEventBindingName=" + bindingName, bindingName, 1);
String exceptionMessage = new String(result.get(0).getPayload());
assertThat(exceptionMessage).isEqualTo("Exception while item was being read");
}
@@ -220,8 +203,7 @@ public class BatchExecutionEventTests {
final String SKIPPING_WRITE_CONTENT = "-1";
final String bindingName = "skip-event-foobar";
List<Message<byte[]>> result = testListenerSkip(
"--spring.cloud.task.batch.events.skipEventBindingName=" + bindingName,
bindingName, 3);
"--spring.cloud.task.batch.events.skipEventBindingName=" + bindingName, bindingName, 3);
int readSkipCount = 0;
int writeSkipCount = 0;
for (int i = 0; i < 3; i++) {
@@ -234,19 +216,20 @@ public class BatchExecutionEventTests {
}
}
assertThat(readSkipCount).as("the number of read skip events did not match")
.isEqualTo(2);
assertThat(writeSkipCount).as("the number of write skip events did not match")
.isEqualTo(1);
assertThat(readSkipCount).as("the number of read skip events did not match").isEqualTo(2);
assertThat(writeSkipCount).as("the number of write skip events did not match").isEqualTo(1);
}
@SpringBootApplication
@Import(JobConfiguration.class)
public static class BatchEventsApplication {
}
@SpringBootApplication
@Import(JobSkipConfiguration.class)
public static class BatchSkipEventsApplication {
}
}

28
spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/listener/TaskEventTests.java
View File

@@ -57,24 +57,21 @@ public class TaskEventTests {
@Test
public void testTaskEventListener() throws Exception {
List<Message<byte[]>> result = testListener(
"--spring.cloud.task.batch.events.itemWriteEventBindingName=task-events",
"task-events", 2);
TaskExecution taskExecution = this.objectMapper.readValue(result.get(0).getPayload(),
TaskExecution.class);
"--spring.cloud.task.batch.events.itemWriteEventBindingName=task-events", "task-events", 2);
TaskExecution taskExecution = this.objectMapper.readValue(result.get(0).getPayload(), TaskExecution.class);
Assertions.assertThat(taskExecution.getTaskName()).isEqualTo(TASK_NAME)
.as(String.format("Task name should be '%s'", TASK_NAME));
taskExecution = this.objectMapper.readValue(result.get(1).getPayload(),
TaskExecution.class);
.as(String.format("Task name should be '%s'", TASK_NAME));
taskExecution = this.objectMapper.readValue(result.get(1).getPayload(), TaskExecution.class);
Assertions.assertThat(taskExecution.getTaskName()).isEqualTo(TASK_NAME)
.as(String.format("Task name should be '%s'", TASK_NAME));
.as(String.format("Task name should be '%s'", TASK_NAME));
}
private List<Message<byte[]>> testListener(String channelBinding, String bindingName, int numberToRead) {
List<Message<byte[]>> results = new ArrayList<>();
this.applicationContext = new SpringApplicationBuilder()
.sources(TestChannelBinderConfiguration
.getCompleteConfiguration(BatchExecutionEventTests.BatchEventsApplication.class)).web(WebApplicationType.NONE).build()
.run(getCommandLineParams(channelBinding));
.sources(TestChannelBinderConfiguration
.getCompleteConfiguration(BatchExecutionEventTests.BatchEventsApplication.class))
.web(WebApplicationType.NONE).build().run(getCommandLineParams(channelBinding));
OutputDestination target = this.applicationContext.getBean(OutputDestination.class);
for (int i = 0; i < numberToRead; i++) {
results.add(target.receive(10000, bindingName));
@@ -83,11 +80,9 @@ public class TaskEventTests {
}
private String[] getCommandLineParams(String sinkChannelParam) {
return new String[]{"--spring.cloud.task.closecontext_enable=false",
"--spring.cloud.task.name=" + TASK_NAME,
"--spring.main.web-environment=false",
"--spring.cloud.stream.defaultBinder=rabbit",
"foo=" + UUID.randomUUID(), sinkChannelParam};
return new String[] { "--spring.cloud.task.closecontext_enable=false", "--spring.cloud.task.name=" + TASK_NAME,
"--spring.main.web-environment=false", "--spring.cloud.stream.defaultBinder=rabbit",
"foo=" + UUID.randomUUID(), sinkChannelParam };
}
@EnableTask
@@ -95,4 +90,5 @@ public class TaskEventTests {
public static class TaskEventsConfiguration {
}
}