Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Replace anonymous types with lambda expressions or method references

Browse Source
This commit is contained in:
Mahmoud Ben Hassine
2023-07-04 18:49:31 +02:00
parent 07154909a1
commit 96f8d6bd35
34 changed files with 405 additions and 695 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/JobRegistryBackgroundJobRunner.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2022 the original author or authors.
* Copyright 2006-2023 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.
@@ -200,16 +200,13 @@ public class JobRegistryBackgroundJobRunner {
if (logger.isInfoEnabled()) {
logger.info("Starting job registry in parent context from XML at: [" + args[0] + "]");
}
new Thread(new Runnable() {
@Override
public void run() {
try {
launcher.run();
}
catch (RuntimeException e) {
errors.add(e);
throw e;
}
new Thread(() -> {
try {
launcher.run();
}
catch (RuntimeException e) {
errors.add(e);
throw e;
}
}).start();

12
spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/SystemCommandTasklet.java
View File

@@ -17,7 +17,6 @@
package org.springframework.batch.core.step.tasklet;
import java.io.File;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import org.apache.commons.logging.Log;
@@ -100,14 +99,9 @@ public class SystemCommandTasklet implements StepExecutionListener, StoppableTas
@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
FutureTask<Integer> systemCommandTask = new FutureTask<>(new Callable<>() {
@Override
public Integer call() throws Exception {
Process process = commandRunner.exec(cmdArray, environmentParams, workingDirectory);
return process.waitFor();
}
FutureTask<Integer> systemCommandTask = new FutureTask<>(() -> {
Process process = commandRunner.exec(cmdArray, environmentParams, workingDirectory);
return process.waitFor();
});
long t0 = System.currentTimeMillis();

6
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/annotation/InlineDataSourceDefinitionTests.java
View File

@@ -61,9 +61,9 @@ class InlineDataSourceDefinitionTests {
@Bean
public Job job(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
return new JobBuilder("job", jobRepository)
.start(new StepBuilder("step", jobRepository).tasklet((contribution, chunkContext) -> {
return RepeatStatus.FINISHED;
}, transactionManager).build())
.start(new StepBuilder("step", jobRepository)
.tasklet((contribution, chunkContext) -> RepeatStatus.FINISHED, transactionManager)
.build())
.build();
}

4
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/support/DefaultBatchConfigurationTests.java
View File

@@ -122,9 +122,7 @@ class DefaultBatchConfigurationTests {
@Bean
public Step myStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
Tasklet myTasklet = (contribution, chunkContext) -> {
return RepeatStatus.FINISHED;
};
Tasklet myTasklet = (contribution, chunkContext) -> RepeatStatus.FINISHED;
return new StepBuilder("myStep", jobRepository).tasklet(myTasklet, transactionManager).build();
}

4
spring-batch-core/src/test/java/org/springframework/batch/core/listener/MulticasterBatchListenerTests.java
View File

@@ -230,7 +230,7 @@ class MulticasterBatchListenerTests {
@Test
void testBeforeReadFails() {
error = true;
Exception exception = assertThrows(StepListenerFailedException.class, () -> multicast.beforeRead());
Exception exception = assertThrows(StepListenerFailedException.class, multicast::beforeRead);
String message = exception.getCause().getMessage();
assertEquals("listener error", message, "Wrong message: " + message);
assertEquals(1, count);
@@ -457,7 +457,7 @@ class MulticasterBatchListenerTests {
StepListener listener = StepListenerFactoryBean.getListener(new AnnotationBasedStepListener());
multicast.register(listener);
Exception exception = assertThrows(StepListenerFailedException.class, () -> multicast.beforeRead());
Exception exception = assertThrows(StepListenerFailedException.class, multicast::beforeRead);
Throwable cause = exception.getCause();
String message = cause.getMessage();
assertTrue(cause instanceof IllegalStateException);

144
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java
View File

@@ -41,7 +41,6 @@ import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.core.step.AbstractStep;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
@@ -90,12 +89,7 @@ class FaultTolerantStepFactoryBeanRetryTests {
JobExecution jobExecution;
private ItemWriter<String> writer = new ItemWriter<>() {
@Override
public void write(Chunk<? extends String> data) throws Exception {
processed.addAll(data.getItems());
}
};
private ItemWriter<String> writer = data -> processed.addAll(data.getItems());
@SuppressWarnings("unchecked")
@BeforeEach
@@ -153,16 +147,13 @@ class FaultTolerantStepFactoryBeanRetryTests {
factory.setJobRepository(repository);
factory.setTransactionManager(new ResourcelessTransactionManager());
ItemWriter<Integer> failingWriter = new ItemWriter<>() {
@Override
public void write(Chunk<? extends Integer> data) throws Exception {
int count = 0;
for (Integer item : data) {
if (count++ == 2) {
throw new Exception("Planned failure in writer");
}
written.add(item);
ItemWriter<Integer> failingWriter = data -> {
int count = 0;
for (Integer item : data) {
if (count++ == 2) {
throw new Exception("Planned failure in writer");
}
written.add(item);
}
};
@@ -203,16 +194,13 @@ class FaultTolerantStepFactoryBeanRetryTests {
void testProcessAllItemsWhenErrorInWriter() throws Exception {
final int RETRY_LIMIT = 3;
final List<String> ITEM_LIST = Arrays.asList("a", "b", "c");
ItemWriter<String> failingWriter = new ItemWriter<>() {
@Override
public void write(Chunk<? extends String> data) throws Exception {
int count = 0;
for (String item : data) {
if (count++ == 2) {
throw new Exception("Planned failure in writer");
}
written.add(item);
ItemWriter<String> failingWriter = data -> {
int count = 0;
for (String item : data) {
if (count++ == 2) {
throw new Exception("Planned failure in writer");
}
written.add(item);
}
};
@@ -249,16 +237,13 @@ class FaultTolerantStepFactoryBeanRetryTests {
@Test
void testNoItemsReprocessedWhenErrorInWriterAndProcessorNotTransactional() throws Exception {
ItemWriter<String> failingWriter = new ItemWriter<>() {
@Override
public void write(Chunk<? extends String> data) throws Exception {
int count = 0;
for (String item : data) {
if (count++ == 2) {
throw new Exception("Planned failure in writer");
}
written.add(item);
ItemWriter<String> failingWriter = data -> {
int count = 0;
for (String item : data) {
if (count++ == 2) {
throw new Exception("Planned failure in writer");
}
written.add(item);
}
};
@@ -359,14 +344,11 @@ class FaultTolerantStepFactoryBeanRetryTests {
reader.setName("foo");
factory.setItemReader(reader);
factory.setStreams(new ItemStream[] { reader });
factory.setItemWriter(new ItemWriter<>() {
@Override
public void write(Chunk<? extends String> chunk) throws Exception {
if (fail && chunk.getItems().contains("e")) {
throw new RuntimeException("Planned failure");
}
processed.addAll(chunk.getItems());
factory.setItemWriter(chunk -> {
if (fail && chunk.getItems().contains("e")) {
throw new RuntimeException("Planned failure");
}
processed.addAll(chunk.getItems());
});
factory.setRetryLimit(0);
Step step = factory.getObject();
@@ -446,15 +428,12 @@ class FaultTolerantStepFactoryBeanRetryTests {
}
};
ItemWriter<String> itemWriter = new ItemWriter<>() {
@Override
public void write(Chunk<? extends String> chunk) throws Exception {
logger.debug("Write Called! Item: [" + chunk.getItems() + "]");
processed.addAll(chunk.getItems());
written.addAll(chunk.getItems());
if (chunk.getItems().contains("b") || chunk.getItems().contains("d")) {
throw new RuntimeException("Write error - planned but recoverable.");
}
ItemWriter<String> itemWriter = chunk -> {
logger.debug("Write Called! Item: [" + chunk.getItems() + "]");
processed.addAll(chunk.getItems());
written.addAll(chunk.getItems());
if (chunk.getItems().contains("b") || chunk.getItems().contains("d")) {
throw new RuntimeException("Write error - planned but recoverable.");
}
};
factory.setItemReader(provider);
@@ -504,15 +483,12 @@ class FaultTolerantStepFactoryBeanRetryTests {
}
};
ItemWriter<String> itemWriter = new ItemWriter<>() {
@Override
public void write(Chunk<? extends String> chunk) throws Exception {
logger.debug("Write Called! Item: [" + chunk + "]");
processed.addAll(chunk.getItems());
written.addAll(chunk.getItems());
if (chunk.getItems().contains("b") || chunk.getItems().contains("d")) {
throw new RuntimeException("Write error - planned but recoverable.");
}
ItemWriter<String> itemWriter = chunk -> {
logger.debug("Write Called! Item: [" + chunk + "]");
processed.addAll(chunk.getItems());
written.addAll(chunk.getItems());
if (chunk.getItems().contains("b") || chunk.getItems().contains("d")) {
throw new RuntimeException("Write error - planned but recoverable.");
}
};
factory.setItemReader(provider);
@@ -556,14 +532,11 @@ class FaultTolerantStepFactoryBeanRetryTests {
return item;
}
};
ItemWriter<String> itemWriter = new ItemWriter<>() {
@Override
public void write(Chunk<? extends String> chunk) throws Exception {
processed.addAll(chunk.getItems());
written.addAll(chunk.getItems());
logger.debug("Write Called! Item: [" + chunk.getItems() + "]");
throw new RuntimeException("Write error - planned but retryable.");
}
ItemWriter<String> itemWriter = chunk -> {
processed.addAll(chunk.getItems());
written.addAll(chunk.getItems());
logger.debug("Write Called! Item: [" + chunk.getItems() + "]");
throw new RuntimeException("Write error - planned but retryable.");
};
factory.setItemReader(provider);
factory.setItemWriter(itemWriter);
@@ -608,14 +581,11 @@ class FaultTolerantStepFactoryBeanRetryTests {
return item;
}
};
ItemWriter<String> itemWriter = new ItemWriter<>() {
@Override
public void write(Chunk<? extends String> chunk) throws Exception {
processed.addAll(chunk.getItems());
written.addAll(chunk.getItems());
logger.debug("Write Called! Item: [" + chunk.getItems() + "]");
throw new RuntimeException("Write error - planned but not skippable.");
}
ItemWriter<String> itemWriter = chunk -> {
processed.addAll(chunk.getItems());
written.addAll(chunk.getItems());
logger.debug("Write Called! Item: [" + chunk.getItems() + "]");
throw new RuntimeException("Write error - planned but not skippable.");
};
factory.setItemReader(provider);
factory.setItemWriter(itemWriter);
@@ -655,14 +625,11 @@ class FaultTolerantStepFactoryBeanRetryTests {
return item;
}
};
ItemWriter<String> itemWriter = new ItemWriter<>() {
@Override
public void write(Chunk<? extends String> chunk) throws Exception {
processed.addAll(chunk.getItems());
written.addAll(chunk.getItems());
logger.debug("Write Called! Item: [" + chunk.getItems() + "]");
throw new RuntimeException("Write error - planned but retryable.");
}
ItemWriter<String> itemWriter = chunk -> {
processed.addAll(chunk.getItems());
written.addAll(chunk.getItems());
logger.debug("Write Called! Item: [" + chunk.getItems() + "]");
throw new RuntimeException("Write error - planned but retryable.");
};
factory.setItemReader(provider);
factory.setItemWriter(itemWriter);
@@ -707,13 +674,10 @@ class FaultTolerantStepFactoryBeanRetryTests {
return item;
}
};
ItemWriter<String> itemWriter = new ItemWriter<>() {
@Override
public void write(Chunk<? extends String> chunk) throws Exception {
processed.addAll(chunk.getItems());
logger.debug("Write Called! Item: [" + chunk.getItems() + "]");
throw new RuntimeException("Write error - planned but retryable.");
}
ItemWriter<String> itemWriter = chunk -> {
processed.addAll(chunk.getItems());
logger.debug("Write Called! Item: [" + chunk.getItems() + "]");
throw new RuntimeException("Write error - planned but retryable.");
};
factory.setItemReader(provider);
factory.setItemWriter(itemWriter);

3
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java
View File

@@ -22,6 +22,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.aopalliance.intercept.Joinpoint;
import org.aopalliance.intercept.MethodInterceptor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -975,7 +976,7 @@ public class FaultTolerantStepFactoryBeanTests {
ProxyFactory proxy = new ProxyFactory();
proxy.setTarget(reader);
proxy.setInterfaces(new Class<?>[] { ItemReader.class, ItemStream.class });
proxy.addAdvice((MethodInterceptor) invocation -> invocation.proceed());
proxy.addAdvice((MethodInterceptor) Joinpoint::proceed);
Object advised = proxy.getProxy();
factory.setItemReader((ItemReader<? extends String>) advised);

2
spring-batch-core/src/test/java/org/springframework/batch/core/step/job/DefaultJobParametersExtractorJobParametersTests.java
View File

@@ -46,7 +46,7 @@ class DefaultJobParametersExtractorJobParametersTests {
@BeforeEach
void setUp() {
DefaultConversionService conversionService = new DefaultConversionService();
conversionService.addConverter(String.class, LocalDate.class, source -> LocalDate.parse(source));
conversionService.addConverter(String.class, LocalDate.class, LocalDate::parse);
this.jobParametersConverter.setConversionService(conversionService);
this.extractor.setJobParametersConverter(this.jobParametersConverter);
}

16
spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncTaskletStepTests.java
View File

@@ -33,7 +33,6 @@ import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.JobRepositorySupport;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemStreamSupport;
@@ -58,15 +57,12 @@ class AsyncTaskletStepTests {
private int throttleLimit = 20;
ItemWriter<String> itemWriter = new ItemWriter<>() {
@Override
public void write(Chunk<? extends String> data) throws Exception {
// Thread.sleep(100L);
logger.info("Items: " + data);
processed.addAll(data.getItems());
if (data.getItems().contains("fail")) {
throw new RuntimeException("Planned");
}
ItemWriter<String> itemWriter = data -> {
// Thread.sleep(100L);
logger.info("Items: " + data);
processed.addAll(data.getItems());
if (data.getItems().contains("fail")) {
throw new RuntimeException("Planned");
}
};

29
spring-batch-core/src/test/java/org/springframework/batch/core/test/repository/JdbcJobRepositoryTests.java
View File

@@ -156,26 +156,23 @@ class JdbcJobRepositoryTests {
}
private JobExecution doConcurrentStart() throws Exception {
new Thread(new Runnable() {
@Override
public void run() {
new Thread(() -> {
try {
JobExecution execution = repository.createJobExecution(job.getName(), new JobParameters());
try {
JobExecution execution = repository.createJobExecution(job.getName(), new JobParameters());
// simulate running execution
execution.setStartTime(LocalDateTime.now());
repository.update(execution);
cacheJobIds(execution);
list.add(execution);
Thread.sleep(1000);
}
catch (Exception e) {
list.add(e);
}
// simulate running execution
execution.setStartTime(LocalDateTime.now());
repository.update(execution);
cacheJobIds(execution);
list.add(execution);
Thread.sleep(1000);
}
catch (Exception e) {
list.add(e);
}
}).start();
Thread.sleep(400);

23
spring-batch-core/src/test/java/org/springframework/batch/core/test/repository/MySQLJdbcJobRepositoryIntegrationTests.java
View File

@@ -47,7 +47,6 @@ import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.convert.support.ConfigurableConversionService;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.core.io.ClassPathResource;
@@ -174,23 +173,15 @@ class MySQLJdbcJobRepositoryIntegrationTests {
public ConfigurableConversionService conversionService() {
DefaultConversionService conversionService = new DefaultConversionService();
final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMddHHmmssSSS");
conversionService.addConverter(String.class, Date.class, new Converter<>() {
@Override
public Date convert(String source) {
try {
return dateFormat.parse(source);
}
catch (ParseException e) {
throw new RuntimeException(e);
}
}
});
conversionService.addConverter(Date.class, String.class, new Converter<>() {
@Override
public String convert(Date source) {
return dateFormat.format(source);
conversionService.addConverter(String.class, Date.class, source -> {
try {
return dateFormat.parse(source);
}
catch (ParseException e) {
throw new RuntimeException(e);
}
});
conversionService.addConverter(Date.class, String.class, dateFormat::format);
return conversionService;
}

19
spring-batch-core/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanIntegrationTests.java
View File

@@ -15,8 +15,6 @@
*/
package org.springframework.batch.core.test.step;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -44,7 +42,6 @@ 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.junit.jupiter.SpringJUnitConfig;
@@ -213,12 +210,8 @@ class FaultTolerantStepFactoryBeanIntegrationTests {
}
public List<String> getCommitted() {
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='written'", new RowMapper<>() {
@Override
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString(1);
}
});
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='written'",
(rs, rowNum) -> rs.getString(1));
}
public void clear() {
@@ -259,12 +252,8 @@ class FaultTolerantStepFactoryBeanIntegrationTests {
}
public List<String> getCommitted() {
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='processed'", new RowMapper<>() {
@Override
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString(1);
}
});
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='processed'",
(rs, rowNum) -> rs.getString(1));
}
public void clear() {

19
spring-batch-core/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackIntegrationTests.java
View File

@@ -15,8 +15,6 @@
*/
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;
@@ -47,7 +45,6 @@ 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.junit.jupiter.SpringJUnitConfig;
@@ -239,12 +236,8 @@ class FaultTolerantStepFactoryBeanRollbackIntegrationTests {
}
public List<String> getCommitted() {
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='written'", new RowMapper<>() {
@Override
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString(1);
}
});
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='written'",
(rs, rowNum) -> rs.getString(1));
}
public void clear() {
@@ -292,12 +285,8 @@ class FaultTolerantStepFactoryBeanRollbackIntegrationTests {
}
public List<String> getCommitted() {
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='processed'", new RowMapper<>() {
@Override
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString(1);
}
});
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='processed'",
(rs, rowNum) -> rs.getString(1));
}
public void clear() {

20
spring-batch-core/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepIntegrationTests.java
View File

@@ -209,22 +209,16 @@ class FaultTolerantStepIntegrationTests {
// Given
ListItemReader<Integer> itemReader = new ListItemReader<>(Arrays.asList(1, 2, 3));
ItemProcessor<Integer, Integer> itemProcessor = new ItemProcessor<>() {
@Override
public Integer process(Integer item) throws Exception {
if (item.equals(2)) {
throw new Exception("Error during process item " + item);
}
return item;
ItemProcessor<Integer, Integer> itemProcessor = item -> {
if (item.equals(2)) {
throw new Exception("Error during process item " + item);
}
return item;
};
ItemWriter<Integer> itemWriter = new ItemWriter<>() {
@Override
public void write(Chunk<? extends Integer> chunk) throws Exception {
if (chunk.getItems().contains(3)) {
throw new Exception("Error during write");
}
ItemWriter<Integer> itemWriter = chunk -> {
if (chunk.getItems().contains(3)) {
throw new Exception("Error during write");
}
};

4
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/transform/RecordFieldExtractor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2022 the original author or authors.
* Copyright 2022-2023 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.
@@ -80,7 +80,7 @@ public class RecordFieldExtractor<T> implements FieldExtractor<T> {
}
private List<String> getRecordComponentNames() {
return Arrays.stream(this.recordComponents).map(recordComponent -> recordComponent.getName()).toList();
return Arrays.stream(this.recordComponents).map(RecordComponent::getName).toList();
}
private void validate(String[] names) {

2
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractFileItemWriter.java
View File

@@ -579,7 +579,7 @@ public abstract class AbstractFileItemWriter<T> extends AbstractItemStreamItemWr
final FileChannel channel = fileChannel;
if (transactional) {
TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(channel,
() -> closeStream());
this::closeStream);
writer.setEncoding(encoding);
writer.setForceSync(forceSync);

3
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java
View File

@@ -481,8 +481,7 @@ public class StaxEventItemWriter<T> extends AbstractItemStreamItemWriter<T>
try {
final FileChannel channel = fileChannel;
if (transactional) {
TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(channel,
() -> closeStream());
TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(channel, this::closeStream);
writer.setEncoding(encoding);
writer.setForceSync(forceSync);

10
spring-batch-infrastructure/src/test/java/org/springframework/batch/container/jms/BatchMessageListenerContainer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2021 the original author or authors.
* Copyright 2006-2023 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.
@@ -65,13 +65,7 @@ public class BatchMessageListenerContainer extends DefaultMessageListenerContain
private Advice[] advices = new Advice[0];
private ContainerDelegate delegate = new ContainerDelegate() {
@Override
public boolean receiveAndExecute(Object invoker, Session session, MessageConsumer consumer)
throws JMSException {
return BatchMessageListenerContainer.super.receiveAndExecute(invoker, session, consumer);
}
};
private ContainerDelegate delegate = BatchMessageListenerContainer.super::receiveAndExecute;
private ContainerDelegate proxy = delegate;

90
spring-batch-infrastructure/src/test/java/org/springframework/batch/container/jms/BatchMessageListenerContainerIntegrationTests.java
View File

@@ -38,7 +38,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.policy.NeverRetryPolicy;
import org.springframework.retry.support.DefaultRetryState;
import org.springframework.retry.support.RetryTemplate;
@@ -86,15 +85,12 @@ class BatchMessageListenerContainerIntegrationTests {
@Test
void testSendAndReceive() throws Exception {
container.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message msg) {
try {
processed.add(((TextMessage) msg).getText());
}
catch (JMSException e) {
throw new IllegalStateException(e);
}
container.setMessageListener((MessageListener) msg -> {
try {
processed.add(((TextMessage) msg).getText());
}
catch (JMSException e) {
throw new IllegalStateException(e);
}
});
container.initializeProxy();
@@ -110,17 +106,14 @@ class BatchMessageListenerContainerIntegrationTests {
@Test
void testFailureAndRepresent() throws Exception {
container.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message msg) {
try {
processed.add(((TextMessage) msg).getText());
}
catch (JMSException e) {
throw new IllegalStateException(e);
}
throw new RuntimeException("planned failure for represent: " + msg);
container.setMessageListener((MessageListener) msg -> {
try {
processed.add(((TextMessage) msg).getText());
}
catch (JMSException e) {
throw new IllegalStateException(e);
}
throw new RuntimeException("planned failure for represent: " + msg);
});
container.initializeProxy();
container.start();
@@ -134,39 +127,30 @@ class BatchMessageListenerContainerIntegrationTests {
void testFailureAndRecovery() throws Exception {
final RetryTemplate retryTemplate = new RetryTemplate();
retryTemplate.setRetryPolicy(new NeverRetryPolicy());
container.setMessageListener(new MessageListener() {
@Override
public void onMessage(final Message msg) {
try {
RetryCallback<Message, Exception> callback = new RetryCallback<>() {
@Override
public Message doWithRetry(RetryContext context) throws Exception {
try {
processed.add(((TextMessage) msg).getText());
}
catch (JMSException e) {
throw new IllegalStateException(e);
}
throw new RuntimeException("planned failure: " + msg);
}
};
RecoveryCallback<Message> recoveryCallback = new RecoveryCallback<>() {
@Override
public Message recover(RetryContext context) {
try {
recovered.add(((TextMessage) msg).getText());
}
catch (JMSException e) {
throw new IllegalStateException(e);
}
return msg;
}
};
retryTemplate.execute(callback, recoveryCallback, new DefaultRetryState(msg.getJMSMessageID()));
}
catch (Exception e) {
throw (RuntimeException) e;
}
container.setMessageListener((MessageListener) msg -> {
try {
RetryCallback<Message, Exception> callback = context -> {
try {
processed.add(((TextMessage) msg).getText());
}
catch (JMSException e) {
throw new IllegalStateException(e);
}
throw new RuntimeException("planned failure: " + msg);
};
RecoveryCallback<Message> recoveryCallback = context -> {
try {
recovered.add(((TextMessage) msg).getText());
}
catch (JMSException e) {
throw new IllegalStateException(e);
}
return msg;
};
retryTemplate.execute(callback, recoveryCallback, new DefaultRetryState(msg.getJMSMessageID()));
}
catch (Exception e) {
throw (RuntimeException) e;
}
});
container.initializeProxy();

20
spring-batch-infrastructure/src/test/java/org/springframework/batch/container/jms/BatchMessageListenerContainerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2022 the original author or authors.
* Copyright 2006-2023 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.
@@ -50,10 +50,7 @@ class BatchMessageListenerContainerTests {
template.setCompletionPolicy(new SimpleCompletionPolicy(2));
container = getContainer(template);
container.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message arg0) {
}
container.setMessageListener((MessageListener) arg0 -> {
});
Session session = mock(Session.class);
@@ -147,14 +144,11 @@ class BatchMessageListenerContainerTests {
private boolean doTestWithException(final Throwable t, boolean expectRollback, int expectGetTransactionCount)
throws JMSException, IllegalAccessException {
container.setAcceptMessagesWhileStopping(true);
container.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message arg0) {
if (t instanceof RuntimeException)
throw (RuntimeException) t;
else
throw (Error) t;
}
container.setMessageListener((MessageListener) arg0 -> {
if (t instanceof RuntimeException)
throw (RuntimeException) t;
else
throw (Error) t;
});
Session session = mock(Session.class);

2
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcBatchItemWriterNamedParameterTests.java
View File

@@ -148,7 +148,7 @@ public class JdbcBatchItemWriterNamedParameterTests {
mapWriter.setSql(sql);
mapWriter.setJdbcTemplate(namedParameterJdbcOperations);
mapWriter.setItemSqlParameterSourceProvider(item -> new MapSqlParameterSource(item));
mapWriter.setItemSqlParameterSourceProvider(MapSqlParameterSource::new);
mapWriter.afterPropertiesSet();
ArgumentCaptor<SqlParameterSource[]> captor = ArgumentCaptor.forClass(SqlParameterSource[].class);

16
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/TransactionAwareListItemReaderTests.java
View File

@@ -24,7 +24,6 @@ import org.junit.jupiter.api.Test;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.batch.support.transaction.TransactionAwareProxyFactory;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
@@ -70,16 +69,13 @@ class TransactionAwareListItemReaderTests {
void testTransactionalExhausted() {
PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
final List<Object> taken = new ArrayList<>();
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
@Override
public Void doInTransaction(TransactionStatus status) {
Object next = reader.read();
while (next != null) {
taken.add(next);
next = reader.read();
}
return null;
new TransactionTemplate(transactionManager).execute((TransactionCallback<Void>) status -> {
Object next = reader.read();
while (next != null) {
taken.add(next);
next = reader.read();
}
return null;
});
assertEquals(3, taken.size());
assertEquals("a", taken.get(0));

79
spring-batch-infrastructure/src/test/java/org/springframework/batch/jms/ExternalRetryInBatchTests.java
View File

@@ -20,8 +20,6 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.repeat.RepeatCallback;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.RepeatSynchronizationManager;
@@ -32,14 +30,12 @@ import org.springframework.jms.core.JmsTemplate;
import org.springframework.lang.Nullable;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.DefaultRetryState;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.test.jdbc.JdbcTestUtils;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
@@ -114,60 +110,45 @@ class ExternalRetryInBatchTests {
// *internal* retry policy.
for (int i = 0; i < 4; i++) {
try {
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
@Override
public Void doInTransaction(TransactionStatus status) {
try {
new TransactionTemplate(transactionManager).execute((TransactionCallback<Void>) status -> {
try {
repeatTemplate.iterate(new RepeatCallback() {
repeatTemplate.iterate(context -> {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
final String item = provider.read();
final String item = provider.read();
if (item == null) {
return RepeatStatus.FINISHED;
}
if (item == null) {
return RepeatStatus.FINISHED;
}
RetryCallback<String, Exception> callback = context12 -> {
// No need for transaction here: the whole
// batch will roll
// back. When it comes back for recovery this
// code is not
// executed...
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)",
list.size(), item);
throw new RuntimeException("Rollback!");
};
RetryCallback<String, Exception> callback = new RetryCallback<>() {
@Override
public String doWithRetry(RetryContext context) throws Exception {
// No need for transaction here: the whole
// batch will roll
// back. When it comes back for recovery this
// code is not
// executed...
jdbcTemplate.update(
"INSERT into T_BARS (id,name,foo_date) values (?,?,null)",
list.size(), item);
throw new RuntimeException("Rollback!");
}
};
RecoveryCallback<String> recoveryCallback = context1 -> {
// aggressive commit on a recovery
RepeatSynchronizationManager.setCompleteOnly();
recovered.add(item);
return item;
};
RecoveryCallback<String> recoveryCallback = new RecoveryCallback<>() {
@Override
public String recover(RetryContext context) {
// aggressive commit on a recovery
RepeatSynchronizationManager.setCompleteOnly();
recovered.add(item);
return item;
}
};
retryTemplate.execute(callback, recoveryCallback, new DefaultRetryState(item));
retryTemplate.execute(callback, recoveryCallback, new DefaultRetryState(item));
return RepeatStatus.CONTINUABLE;
return RepeatStatus.CONTINUABLE;
});
return null;
}
});
return null;
}
catch (Exception e) {
throw new RuntimeException(e.getMessage(), e);
}
}
catch (Exception e) {
throw new RuntimeException(e.getMessage(), e);
}
});
}

30
spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/jms/AsynchronousTests.java
View File

@@ -18,9 +18,7 @@ package org.springframework.batch.repeat.jms;
import java.util.ArrayList;
import java.util.List;
import jakarta.jms.JMSException;
import jakarta.jms.Message;
import jakarta.jms.Session;
import jakarta.jms.TextMessage;
import org.junit.jupiter.api.AfterEach;
@@ -97,13 +95,10 @@ class AsynchronousTests {
assertInitialState();
container.setMessageListener(new SessionAwareMessageListener<>() {
@Override
public void onMessage(Message message, Session session) throws JMSException {
list.add(message.toString());
String text = ((TextMessage) message).getText();
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), text);
}
container.setMessageListener((SessionAwareMessageListener<Message>) (message, session) -> {
list.add(message.toString());
String text = ((TextMessage) message).getText();
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), text);
});
container.initializeProxy();
@@ -131,16 +126,13 @@ class AsynchronousTests {
// Prevent us from being overwhelmed after rollback
container.setRecoveryInterval(500);
container.setMessageListener(new SessionAwareMessageListener<>() {
@Override
public void onMessage(Message message, Session session) throws JMSException {
list.add(message.toString());
final String text = ((TextMessage) message).getText();
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), text);
// This causes the DB to rollback but not the message
if (text.equals("bar")) {
throw new RuntimeException("Rollback!");
}
container.setMessageListener((SessionAwareMessageListener<Message>) (message, session) -> {
list.add(message.toString());
final String text = ((TextMessage) message).getText();
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), text);
// This causes the DB to rollback but not the message
if (text.equals("bar")) {
throw new RuntimeException("Rollback!");
}
});

99
spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/jms/SynchronousTests.java
View File

@@ -25,12 +25,9 @@ import java.util.List;
import jakarta.jms.ConnectionFactory;
import jakarta.jms.JMSException;
import jakarta.jms.Session;
import org.junit.jupiter.api.Test;
import org.springframework.batch.repeat.RepeatCallback;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.beans.factory.annotation.Autowired;
@@ -98,14 +95,11 @@ class SynchronousTests implements ApplicationContextAware {
assertInitialState();
repeatTemplate.iterate(new RepeatCallback() {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
String text = (String) jmsTemplate.receiveAndConvert("queue");
list.add(text);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), text);
return RepeatStatus.continueIf(text != null);
}
repeatTemplate.iterate(context -> {
String text = (String) jmsTemplate.receiveAndConvert("queue");
list.add(text);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), text);
return RepeatStatus.continueIf(text != null);
});
int count = JdbcTestUtils.countRowsInTable(jdbcTemplate, "T_BARS");
@@ -126,23 +120,16 @@ class SynchronousTests implements ApplicationContextAware {
assertInitialState();
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
@Override
public Void doInTransaction(org.springframework.transaction.TransactionStatus status) {
repeatTemplate.iterate(new RepeatCallback() {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
String text = (String) jmsTemplate.receiveAndConvert("queue");
list.add(text);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(),
text);
return RepeatStatus.continueIf(text != null);
}
});
// force rollback...
status.setRollbackOnly();
return null;
}
new TransactionTemplate(transactionManager).execute((TransactionCallback<Void>) status -> {
repeatTemplate.iterate(context -> {
String text = (String) jmsTemplate.receiveAndConvert("queue");
list.add(text);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), text);
return RepeatStatus.continueIf(text != null);
});
// force rollback...
status.setRollbackOnly();
return null;
});
String text = "";
@@ -174,42 +161,32 @@ class SynchronousTests implements ApplicationContextAware {
assertInitialState();
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
@Override
public Void doInTransaction(org.springframework.transaction.TransactionStatus status) {
new TransactionTemplate(transactionManager).execute((TransactionCallback<Void>) status -> {
repeatTemplate.iterate(new RepeatCallback() {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
String text = (String) txJmsTemplate.receiveAndConvert("queue");
list.add(text);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(),
text);
return RepeatStatus.continueIf(text != null);
}
});
// Simulate a message system failure before the main transaction
// commits...
txJmsTemplate.execute(new SessionCallback<Void>() {
@Override
public Void doInJms(Session session) throws JMSException {
try {
assertTrue(session instanceof SessionProxy, "Not a SessionProxy - wrong spring version?");
((SessionProxy) session).getTargetSession().rollback();
}
catch (JMSException e) {
throw e;
}
catch (Exception e) {
// swallow it
}
return null;
}
});
repeatTemplate.iterate(context -> {
String text = (String) txJmsTemplate.receiveAndConvert("queue");
list.add(text);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), text);
return RepeatStatus.continueIf(text != null);
});
// Simulate a message system failure before the main transaction
// commits...
txJmsTemplate.execute((SessionCallback<Void>) session -> {
try {
assertTrue(session instanceof SessionProxy, "Not a SessionProxy - wrong spring version?");
((SessionProxy) session).getTargetSession().rollback();
}
catch (JMSException e) {
throw e;
}
catch (Exception e) {
// swallow it
}
return null;
}
});
return null;
});
String text = "";

57
spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/listener/RepeatListenerTests.java
View File

@@ -21,7 +21,6 @@ import java.util.List;
import org.junit.jupiter.api.Test;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.batch.repeat.RepeatCallback;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.RepeatListener;
import org.springframework.batch.repeat.support.RepeatTemplate;
@@ -51,12 +50,9 @@ class RepeatListenerTests {
calls.add("2");
}
} });
template.iterate(new RepeatCallback() {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
count++;
return RepeatStatus.continueIf(count <= 1);
}
template.iterate(context -> {
count++;
return RepeatStatus.continueIf(count <= 1);
});
// 2 calls: the second time there is no processing
// (despite the fact that the callback returned null and batch was
@@ -77,12 +73,9 @@ class RepeatListenerTests {
context.setCompleteOnly();
}
});
template.iterate(new RepeatCallback() {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
count++;
return RepeatStatus.FINISHED;
}
template.iterate(context -> {
count++;
return RepeatStatus.FINISHED;
});
assertEquals(0, count);
// ... but the interceptor before() was called:
@@ -104,12 +97,9 @@ class RepeatListenerTests {
calls.add("2");
}
} });
template.iterate(new RepeatCallback() {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
count++;
return RepeatStatus.continueIf(count <= 1);
}
template.iterate(context -> {
count++;
return RepeatStatus.continueIf(count <= 1);
});
// 2 calls to the callback, and the second one had no processing...
assertEquals(2, count);
@@ -133,12 +123,9 @@ class RepeatListenerTests {
context.setCompleteOnly();
}
} });
template.iterate(new RepeatCallback() {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
count++;
return RepeatStatus.CONTINUABLE;
}
template.iterate(context -> {
count++;
return RepeatStatus.CONTINUABLE;
});
assertEquals(0, count);
assertEquals("[1, 2]", calls.toString());
@@ -154,13 +141,10 @@ class RepeatListenerTests {
calls.add("1");
}
});
template.iterate(new RepeatCallback() {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
count++;
context.setCompleteOnly();
return RepeatStatus.FINISHED;
}
template.iterate(context -> {
count++;
context.setCompleteOnly();
return RepeatStatus.FINISHED;
});
assertEquals(1, count);
assertEquals("[1]", calls.toString());
@@ -181,12 +165,9 @@ class RepeatListenerTests {
calls.add("2");
}
} });
template.iterate(new RepeatCallback() {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
count++;
return RepeatStatus.continueIf(count < 2);
}
template.iterate(context -> {
count++;
return RepeatStatus.continueIf(count < 2);
});
// Test that more than one call comes in to the callback...
assertEquals(2, count);

10
spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/SimpleRepeatTemplateTests.java
View File

@@ -29,7 +29,6 @@ import java.util.ArrayList;
import java.util.List;
import org.junit.jupiter.api.Test;
import org.springframework.batch.repeat.RepeatCallback;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.RepeatException;
import org.springframework.batch.repeat.RepeatListener;
@@ -115,12 +114,9 @@ class SimpleRepeatTemplateTests extends AbstractTradeBatchTests {
return context;
}
});
template.iterate(new RepeatCallback() {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
count++;
return RepeatStatus.continueIf(count < 1);
}
template.iterate(context1 -> {
count++;
return RepeatStatus.continueIf(count < 1);
});
assertEquals(1, count);

43
spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateAsynchronousTests.java
View File

@@ -38,7 +38,6 @@ import org.springframework.batch.repeat.RepeatCallback;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.batch.repeat.callback.NestedRepeatCallback;
import org.springframework.batch.repeat.exception.ExceptionHandler;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
@@ -84,17 +83,9 @@ class TaskExecutorRepeatTemplateAsynchronousTests extends AbstractTradeBatchTest
taskExecutor.setConcurrencyLimit(2);
template.setTaskExecutor(taskExecutor);
template.setExceptionHandler(new ExceptionHandler() {
@Override
public void handleException(RepeatContext context, Throwable throwable) throws Throwable {
count++;
}
});
template.iterate(new RepeatCallback() {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
throw new IllegalStateException("foo!");
}
template.setExceptionHandler((context, throwable) -> count++);
template.iterate(context -> {
throw new IllegalStateException("foo!");
});
assertTrue(count >= 1, "Too few attempts: " + count);
@@ -140,18 +131,15 @@ class TaskExecutorRepeatTemplateAsynchronousTests extends AbstractTradeBatchTest
final String threadName = Thread.currentThread().getName();
final Set<String> threadNames = new HashSet<>();
final RepeatCallback callback = new RepeatCallback() {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
assertNotSame(threadName, Thread.currentThread().getName());
threadNames.add(Thread.currentThread().getName());
Thread.sleep(100);
Trade item = provider.read();
if (item != null) {
processor.write(Chunk.of(item));
}
return RepeatStatus.continueIf(item != null);
final RepeatCallback callback = context -> {
assertNotSame(threadName, Thread.currentThread().getName());
threadNames.add(Thread.currentThread().getName());
Thread.sleep(100);
Trade item = provider.read();
if (item != null) {
processor.write(Chunk.of(item));
}
return RepeatStatus.continueIf(item != null);
};
template.iterate(callback);
@@ -232,12 +220,9 @@ class TaskExecutorRepeatTemplateAsynchronousTests extends AbstractTradeBatchTest
return super.doInIteration(context);
}
};
RepeatCallback jobCallback = new RepeatCallback() {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
stepTemplate.iterate(stepCallback);
return RepeatStatus.FINISHED;
}
RepeatCallback jobCallback = context -> {
stepTemplate.iterate(stepCallback);
return RepeatStatus.FINISHED;
};
jobTemplate.iterate(jobCallback);

17
spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/ThrottleLimitResultQueueTests.java
View File

@@ -59,17 +59,14 @@ class ThrottleLimitResultQueueTests {
@Test
void testThrottleLimit() throws Exception {
queue.expect();
new Thread(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(100L);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
queue.put("foo");
new Thread(() -> {
try {
Thread.sleep(100L);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
queue.put("foo");
}).start();
long t0 = System.currentTimeMillis();
queue.expect();

79
spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/jms/ExternalRetryTests.java
View File

@@ -31,14 +31,11 @@ import org.springframework.jms.core.JmsTemplate;
import org.springframework.lang.Nullable;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.support.DefaultRetryState;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.test.jdbc.JdbcTestUtils;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -95,20 +92,17 @@ class ExternalRetryTests {
assertInitialState();
final ItemWriter<Object> writer = new ItemWriter<>() {
@Override
public void write(final Chunk<?> texts) {
final ItemWriter<Object> writer = texts -> {
for (Object text : texts) {
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), text);
if (list.size() == 1) {
throw new RuntimeException("Rollback!");
}
for (Object text : texts) {
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), text);
if (list.size() == 1) {
throw new RuntimeException("Rollback!");
}
}
};
Exception exception = assertThrows(Exception.class,
@@ -130,23 +124,17 @@ class ExternalRetryTests {
// Client of retry template has to take care of rollback. This would
// be a message listener container in the MDP case.
new TransactionTemplate(transactionManager).execute(new TransactionCallback<>() {
@Override
public Object doInTransaction(TransactionStatus status) {
try {
final String item = provider.read();
RetryCallback<Object, Exception> callback = new RetryCallback<>() {
@Override
public Object doWithRetry(RetryContext context) throws Exception {
writer.write(Chunk.of(item));
return null;
}
};
return retryTemplate.execute(callback, new DefaultRetryState(item));
}
catch (Exception e) {
throw new RuntimeException(e.getMessage(), e);
}
new TransactionTemplate(transactionManager).execute(status -> {
try {
final String item = provider.read();
RetryCallback<Object, Exception> callback = context -> {
writer.write(Chunk.of(item));
return null;
};
return retryTemplate.execute(callback, new DefaultRetryState(item));
}
catch (Exception e) {
throw new RuntimeException(e.getMessage(), e);
}
});
@@ -169,35 +157,26 @@ class ExternalRetryTests {
assertInitialState();
final String item = provider.read();
final RetryCallback<String, Exception> callback = new RetryCallback<>() {
@Override
public String doWithRetry(RetryContext context) throws Exception {
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), item);
throw new RuntimeException("Rollback!");
}
final RetryCallback<String, Exception> callback = context -> {
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), item);
throw new RuntimeException("Rollback!");
};
final RecoveryCallback<String> recoveryCallback = new RecoveryCallback<>() {
@Override
public String recover(RetryContext context) {
recovered.add(item);
return item;
}
final RecoveryCallback<String> recoveryCallback = context -> {
recovered.add(item);
return item;
};
String result = "start";
for (int i = 0; i < 4; i++) {
try {
result = new TransactionTemplate(transactionManager).execute(new TransactionCallback<>() {
@Override
public String doInTransaction(TransactionStatus status) {
try {
return retryTemplate.execute(callback, recoveryCallback, new DefaultRetryState(item));
}
catch (Exception e) {
throw new RuntimeException(e.getMessage(), e);
}
result = new TransactionTemplate(transactionManager).execute(status -> {
try {
return retryTemplate.execute(callback, recoveryCallback, new DefaultRetryState(item));
}
catch (Exception e) {
throw new RuntimeException(e.getMessage(), e);
}
});
}

154
spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/jms/SynchronousTests.java
View File

@@ -23,14 +23,12 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.test.context.transaction.AfterTransaction;
import org.springframework.test.context.transaction.BeforeTransaction;
import org.springframework.test.jdbc.JdbcTestUtils;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
@@ -109,28 +107,21 @@ public class SynchronousTests {
final String text = (String) jmsTemplate.receiveAndConvert("queue");
assertNotNull(text);
retryTemplate.execute(new RetryCallback<String, Exception>() {
@Override
public String doWithRetry(RetryContext status) throws Exception {
retryTemplate.execute((RetryCallback<String, Exception>) status -> {
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
transactionTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_NESTED);
return transactionTemplate.execute(new TransactionCallback<>() {
@Override
public String doInTransaction(TransactionStatus status) {
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
transactionTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_NESTED);
return transactionTemplate.execute(status1 -> {
list.add(text);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(),
text);
if (list.size() == 1) {
throw new RuntimeException("Rollback!");
}
return text;
list.add(text);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), text);
if (list.size() == 1) {
throw new RuntimeException("Rollback!");
}
return text;
}
});
});
}
});
// Verify the state after transactional processing is complete
@@ -162,29 +153,22 @@ public class SynchronousTests {
final String item = (String) provider.read();
retryTemplate.execute(new RetryCallback<String, Exception>() {
@Override
public String doWithRetry(RetryContext context) throws Exception {
retryTemplate.execute((RetryCallback<String, Exception>) context -> {
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
transactionTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_NESTED);
return transactionTemplate.execute(new TransactionCallback<>() {
@Override
public String doInTransaction(TransactionStatus status) {
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
transactionTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_NESTED);
return transactionTemplate.execute(status -> {
list.add(item);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(),
item);
if (list.size() == 1) {
throw new RuntimeException("Rollback!");
}
list.add(item);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), item);
if (list.size() == 1) {
throw new RuntimeException("Rollback!");
}
return item;
return item;
}
});
});
}
});
// Verify the state after transactional processing is complete
@@ -219,47 +203,38 @@ public class SynchronousTests {
TransactionTemplate outerTxTemplate = new TransactionTemplate(transactionManager);
outerTxTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_REQUIRES_NEW);
outerTxTemplate.execute(new TransactionCallback<Void>() {
@Override
public Void doInTransaction(TransactionStatus outerStatus) {
outerTxTemplate.execute((TransactionCallback<Void>) outerStatus -> {
final String text = (String) jmsTemplate.receiveAndConvert("queue");
final String text = (String) jmsTemplate.receiveAndConvert("queue");
try {
retryTemplate.execute(new RetryCallback<String, Exception>() {
@Override
public String doWithRetry(RetryContext status) throws Exception {
try {
retryTemplate.execute((RetryCallback<String, Exception>) status -> {
TransactionTemplate nestedTxTemplate = new TransactionTemplate(transactionManager);
nestedTxTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_NESTED);
return nestedTxTemplate.execute(new TransactionCallback<>() {
@Override
public String doInTransaction(TransactionStatus nestedStatus) {
TransactionTemplate nestedTxTemplate = new TransactionTemplate(transactionManager);
nestedTxTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_NESTED);
return nestedTxTemplate.execute(nestedStatus -> {
list.add(text);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)",
list.size(), text);
return text;
list.add(text);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(),
text);
return text;
}
});
}
});
}
catch (Exception e) {
throw new RuntimeException(e);
}
// The nested database transaction has committed...
int count = JdbcTestUtils.countRowsInTable(jdbcTemplate, "T_BARS");
assertEquals(1, count);
// force rollback...
outerStatus.setRollbackOnly();
return null;
});
}
catch (Exception e) {
throw new RuntimeException(e);
}
// The nested database transaction has committed...
int count = JdbcTestUtils.countRowsInTable(jdbcTemplate, "T_BARS");
assertEquals(1, count);
// force rollback...
outerStatus.setRollbackOnly();
return null;
});
// Verify the state after transactional processing is complete
@@ -284,32 +259,25 @@ public class SynchronousTests {
assertInitialState();
retryTemplate.execute(new RetryCallback<String, Exception>() {
@Override
public String doWithRetry(RetryContext status) throws Exception {
retryTemplate.execute((RetryCallback<String, Exception>) status -> {
// use REQUIRES_NEW so that the retry executes in its own transaction
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
transactionTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_REQUIRES_NEW);
return transactionTemplate.execute(new TransactionCallback<>() {
@Override
public String doInTransaction(TransactionStatus status) {
// use REQUIRES_NEW so that the retry executes in its own transaction
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
transactionTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_REQUIRES_NEW);
return transactionTemplate.execute(status1 -> {
// The receive is inside the retry and the
// transaction...
final String text = (String) jmsTemplate.receiveAndConvert("queue");
list.add(text);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(),
text);
if (list.size() == 1) {
throw new RuntimeException("Rollback!");
}
return text;
// The receive is inside the retry and the
// transaction...
final String text = (String) jmsTemplate.receiveAndConvert("queue");
list.add(text);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), text);
if (list.size() == 1) {
throw new RuntimeException("Rollback!");
}
return text;
}
});
});
}
});
// Verify the state after transactional processing is complete

56
spring-batch-samples/src/test/java/org/springframework/batch/sample/HibernateFailureJobFunctionalTests.java
View File

@@ -34,12 +34,10 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.UncategorizedSQLException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowCallbackHandler;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.orm.hibernate5.HibernateJdbcException;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.test.jdbc.JdbcTestUtils;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
@@ -123,17 +121,8 @@ class HibernateFailureJobFunctionalTests {
*/
protected void validatePreConditions() {
ensureState();
creditsBeforeUpdate = new TransactionTemplate(transactionManager).execute(new TransactionCallback<>() {
@Override
public List<BigDecimal> doInTransaction(TransactionStatus status) {
return jdbcTemplate.query(ALL_CUSTOMERS, new RowMapper<>() {
@Override
public BigDecimal mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getBigDecimal(CREDIT_COLUMN);
}
});
}
});
creditsBeforeUpdate = new TransactionTemplate(transactionManager)
.execute(status -> jdbcTemplate.query(ALL_CUSTOMERS, (rs, rowNum) -> rs.getBigDecimal(CREDIT_COLUMN)));
}
/*
@@ -141,16 +130,12 @@ class HibernateFailureJobFunctionalTests {
* customer table and reading the expected defaults.
*/
private void ensureState() {
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
@Override
public Void doInTransaction(TransactionStatus status) {
JdbcTestUtils.deleteFromTables(jdbcTemplate, "CUSTOMER");
for (String customer : customers) {
jdbcTemplate.update(customer);
}
return null;
new TransactionTemplate(transactionManager).execute((TransactionCallback<Void>) status -> {
JdbcTestUtils.deleteFromTables(jdbcTemplate, "CUSTOMER");
for (String customer : customers) {
jdbcTemplate.update(customer);
}
return null;
});
}
@@ -160,25 +145,22 @@ class HibernateFailureJobFunctionalTests {
protected void validatePostConditions() {
final List<BigDecimal> matches = new ArrayList<>();
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
@Override
public Void doInTransaction(TransactionStatus status) {
jdbcTemplate.query(ALL_CUSTOMERS, new RowCallbackHandler() {
private int i = 0;
new TransactionTemplate(transactionManager).execute((TransactionCallback<Void>) status -> {
jdbcTemplate.query(ALL_CUSTOMERS, new RowCallbackHandler() {
private int i = 0;
@Override
public void processRow(ResultSet rs) throws SQLException {
final BigDecimal creditBeforeUpdate = creditsBeforeUpdate.get(i++);
final BigDecimal expectedCredit = creditBeforeUpdate.add(CREDIT_INCREASE);
if (expectedCredit.equals(rs.getBigDecimal(CREDIT_COLUMN))) {
matches.add(rs.getBigDecimal(ID_COLUMN));
}
@Override
public void processRow(ResultSet rs) throws SQLException {
final BigDecimal creditBeforeUpdate = creditsBeforeUpdate.get(i++);
final BigDecimal expectedCredit = creditBeforeUpdate.add(CREDIT_INCREASE);
if (expectedCredit.equals(rs.getBigDecimal(CREDIT_COLUMN))) {
matches.add(rs.getBigDecimal(ID_COLUMN));
}
}
});
});
return null;
}
return null;
});
assertEquals((creditsBeforeUpdate.size() - 1), matches.size());

2
spring-batch-samples/src/test/java/org/springframework/batch/sample/jmx/JobExecutionNotificationPublisherTests.java
View File

@@ -39,7 +39,7 @@ class JobExecutionNotificationPublisherTests {
void testRepeatOperationsOpenUsed() {
final List<Notification> list = new ArrayList<>();
publisher.setNotificationPublisher(notification -> list.add(notification));
publisher.setNotificationPublisher(list::add);
publisher.onApplicationEvent(new SimpleMessageApplicationEvent(this, "foo"));
assertEquals(1, list.size());

7
spring-batch-test/src/test/java/org/springframework/batch/test/StepScopeAnnotatedListenerIntegrationTests.java
View File

@@ -35,7 +35,6 @@ import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
@@ -158,11 +157,7 @@ class StepScopeAnnotatedListenerIntegrationTests {
@Bean
public ItemWriter<String> writer() {
return new ItemWriter<>() {
@Override
public void write(Chunk<? extends String> items) throws Exception {
}
return items -> {
};
}