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

Use lambdas and method references where appropriate

Browse Source
This commit is contained in:
Mahmoud Ben Hassine
2023-06-12 16:43:25 +02:00
parent c5b4f1a777
commit 93d911e100
115 changed files with 1260 additions and 2500 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/SplitState.java
View File

@@ -17,7 +17,6 @@ package org.springframework.batch.core.job.flow.support.state;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
@@ -89,12 +88,7 @@ public class SplitState extends AbstractState implements FlowHolder {
for (final Flow flow : flows) {
final FutureTask<FlowExecution> task = new FutureTask<>(new Callable<>() {
@Override
public FlowExecution call() throws Exception {
return flow.start(executor);
}
});
final FutureTask<FlowExecution> task = new FutureTask<>(() -> flow.start(executor));
tasks.add(task);

10
spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/TaskExecutorPartitionHandler.java
View File

@@ -18,7 +18,6 @@ package org.springframework.batch.core.partition.support;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
@@ -129,12 +128,9 @@ public class TaskExecutorPartitionHandler extends AbstractPartitionHandler imple
* @return the task executing the given step
*/
protected FutureTask<StepExecution> createTask(final Step step, final StepExecution stepExecution) {
return new FutureTask<>(new Callable<>() {
@Override
public StepExecution call() throws Exception {
step.execute(stepExecution);
return stepExecution;
}
return new FutureTask<>(() -> {
step.execute(stepExecution);
return stepExecution;
});
}

20
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcExecutionContextDao.java
View File

@@ -37,7 +37,6 @@ import org.springframework.batch.core.repository.ExecutionContextSerializer;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.core.serializer.Serializer;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.PreparedStatementSetter;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.support.lob.DefaultLobHandler;
import org.springframework.jdbc.support.lob.LobHandler;
@@ -289,18 +288,15 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
longContext = null;
}
getJdbcTemplate().update(getQuery(sql), new PreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps) throws SQLException {
ps.setString(1, shortContext);
if (longContext != null) {
lobHandler.getLobCreator().setClobAsString(ps, 2, longContext);
}
else {
ps.setNull(2, getClobTypeToUse());
}
ps.setLong(3, executionId);
getJdbcTemplate().update(getQuery(sql), ps -> {
ps.setString(1, shortContext);
if (longContext != null) {
lobHandler.getLobCreator().setClobAsString(ps, 2, longContext);
}
else {
ps.setNull(2, getClobTypeToUse());
}
ps.setLong(3, executionId);
});
}

44
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcJobExecutionDao.java
View File

@@ -373,12 +373,9 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements
public Set<JobExecution> findRunningJobExecutions(String jobName) {
final Set<JobExecution> result = new HashSet<>();
RowCallbackHandler handler = new RowCallbackHandler() {
@Override
public void processRow(ResultSet rs) throws SQLException {
JobExecutionRowMapper mapper = new JobExecutionRowMapper();
result.add(mapper.mapRow(rs, 0));
}
RowCallbackHandler handler = rs -> {
JobExecutionRowMapper mapper = new JobExecutionRowMapper();
result.add(mapper.mapRow(rs, 0));
};
getJdbcTemplate().query(getQuery(GET_RUNNING_EXECUTIONS), handler, jobName);
@@ -455,27 +452,24 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements
*/
protected JobParameters getJobParameters(Long executionId) {
final Map<String, JobParameter<?>> map = new HashMap<>();
RowCallbackHandler handler = new RowCallbackHandler() {
@Override
public void processRow(ResultSet rs) throws SQLException {
String parameterName = rs.getString("PARAMETER_NAME");
RowCallbackHandler handler = rs -> {
String parameterName = rs.getString("PARAMETER_NAME");
Class<?> parameterType = null;
try {
parameterType = Class.forName(rs.getString("PARAMETER_TYPE"));
}
catch (ClassNotFoundException e) {
throw new RuntimeException(e);
}
String stringValue = rs.getString("PARAMETER_VALUE");
Object typedValue = conversionService.convert(stringValue, parameterType);
boolean identifying = rs.getString("IDENTIFYING").equalsIgnoreCase("Y");
JobParameter<?> jobParameter = new JobParameter(typedValue, parameterType, identifying);
map.put(parameterName, jobParameter);
Class<?> parameterType = null;
try {
parameterType = Class.forName(rs.getString("PARAMETER_TYPE"));
}
catch (ClassNotFoundException e) {
throw new RuntimeException(e);
}
String stringValue = rs.getString("PARAMETER_VALUE");
Object typedValue = conversionService.convert(stringValue, parameterType);
boolean identifying = rs.getString("IDENTIFYING").equalsIgnoreCase("Y");
JobParameter<?> jobParameter = new JobParameter(typedValue, parameterType, identifying);
map.put(parameterName, jobParameter);
};
getJdbcTemplate().query(getQuery(FIND_PARAMS_FROM_ID), handler, executionId);

7
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcJobInstanceDao.java
View File

@@ -221,12 +221,7 @@ public class JdbcJobInstanceDao extends AbstractJdbcBatchMetadataDao implements
*/
@Override
public List<String> getJobNames() {
return getJdbcTemplate().query(getQuery(FIND_JOB_NAMES), new RowMapper<>() {
@Override
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString(1);
}
});
return getJdbcTemplate().query(getQuery(FIND_JOB_NAMES), (rs, rowNum) -> rs.getString(1));
}
/*

16
spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/AbstractJobRepositoryFactoryBean.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.
@@ -19,7 +19,6 @@ package org.springframework.batch.core.repository.support;
import java.util.Properties;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.DefaultPointcutAdvisor;
import org.springframework.aop.support.NameMatchMethodPointcut;
@@ -197,15 +196,12 @@ public abstract class AbstractJobRepositoryFactoryBean implements FactoryBean<Jo
TransactionInterceptor advice = new TransactionInterceptor((TransactionManager) this.transactionManager,
this.transactionAttributeSource);
if (this.validateTransactionState) {
DefaultPointcutAdvisor advisor = new DefaultPointcutAdvisor(new MethodInterceptor() {
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
if (TransactionSynchronizationManager.isActualTransactionActive()) {
throw new IllegalStateException("Existing transaction detected in JobRepository. "
+ "Please fix this and try again (e.g. remove @Transactional annotations from client).");
}
return invocation.proceed();
DefaultPointcutAdvisor advisor = new DefaultPointcutAdvisor((MethodInterceptor) invocation -> {
if (TransactionSynchronizationManager.isActualTransactionActive()) {
throw new IllegalStateException("Existing transaction detected in JobRepository. "
+ "Please fix this and try again (e.g. remove @Transactional annotations from client).");
}
return invocation.proceed();
});
NameMatchMethodPointcut pointcut = new NameMatchMethodPointcut();
pointcut.addMethodName("create*");

8
spring-batch-core/src/main/java/org/springframework/batch/core/scope/BatchScopeSupport.java
View File

@@ -28,7 +28,6 @@ import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.core.Ordered;
import org.springframework.util.Assert;
import org.springframework.util.StringValueResolver;
/**
* ScopeSupport.
@@ -174,12 +173,7 @@ public abstract class BatchScopeSupport implements Scope, BeanFactoryPostProcess
private final boolean scoped;
public Scopifier(BeanDefinitionRegistry registry, String scope, boolean proxyTargetClass, boolean scoped) {
super(new StringValueResolver() {
@Override
public String resolveStringValue(String value) {
return value;
}
});
super(value -> value);
this.registry = registry;
this.proxyTargetClass = proxyTargetClass;
this.scope = scope;

275
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
View File

@@ -215,87 +215,77 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
final I item = iterator.next();
RetryCallback<O, Exception> retryCallback = new RetryCallback<>() {
@Override
public O doWithRetry(RetryContext context) throws Exception {
Timer.Sample sample = BatchMetrics.createTimerSample(meterRegistry);
String status = BatchMetrics.STATUS_SUCCESS;
O output = null;
try {
O cached = (cacheIterator != null && cacheIterator.hasNext()) ? cacheIterator.next() : null;
if (cached != null && !processorTransactional) {
output = cached;
}
else {
output = doProcess(item);
if (output == null) {
data.incrementFilterCount();
}
else if (!processorTransactional && !data.scanning()) {
cache.add(output);
}
}
}
catch (Exception e) {
status = BatchMetrics.STATUS_FAILURE;
if (rollbackClassifier.classify(e)) {
// Default is to rollback unless the classifier
// allows us to continue
throw e;
}
else if (shouldSkip(itemProcessSkipPolicy, e, contribution.getStepSkipCount())) {
// If we are not re-throwing then we should check if
// this is skippable
contribution.incrementProcessSkipCount();
logger.debug("Skipping after failed process with no rollback", e);
// If not re-throwing then the listener will not be
// called in next chunk.
callProcessSkipListener(item, e);
}
else {
// If it's not skippable that's an error in
// configuration - it doesn't make sense to not roll
// back if we are also not allowed to skip
throw new NonSkippableProcessException(
"Non-skippable exception in processor. Make sure any exceptions that do not cause a rollback are skippable.",
e);
}
}
finally {
stopTimer(sample, contribution.getStepExecution(), "item.process", status, "Item processing");
}
if (output == null) {
// No need to re-process filtered items
iterator.remove();
}
return output;
}
};
RecoveryCallback<O> recoveryCallback = new RecoveryCallback<>() {
@Override
public O recover(RetryContext context) throws Exception {
Throwable e = context.getLastThrowable();
if (shouldSkip(itemProcessSkipPolicy, e, contribution.getStepSkipCount())) {
iterator.remove(e);
contribution.incrementProcessSkipCount();
logger.debug("Skipping after failed process", e);
return null;
RetryCallback<O, Exception> retryCallback = context -> {
Timer.Sample sample = BatchMetrics.createTimerSample(meterRegistry);
String status = BatchMetrics.STATUS_SUCCESS;
O output = null;
try {
O cached = (cacheIterator != null && cacheIterator.hasNext()) ? cacheIterator.next() : null;
if (cached != null && !processorTransactional) {
output = cached;
}
else {
if (rollbackClassifier.classify(e)) {
// Default is to rollback unless the classifier
// allows us to continue
throw new RetryException("Non-skippable exception in recoverer while processing", e);
output = doProcess(item);
if (output == null) {
data.incrementFilterCount();
}
else if (!processorTransactional && !data.scanning()) {
cache.add(output);
}
iterator.remove(e);
return null;
}
}
catch (Exception e) {
status = BatchMetrics.STATUS_FAILURE;
if (rollbackClassifier.classify(e)) {
// Default is to rollback unless the classifier
// allows us to continue
throw e;
}
else if (shouldSkip(itemProcessSkipPolicy, e, contribution.getStepSkipCount())) {
// If we are not re-throwing then we should check if
// this is skippable
contribution.incrementProcessSkipCount();
logger.debug("Skipping after failed process with no rollback", e);
// If not re-throwing then the listener will not be
// called in next chunk.
callProcessSkipListener(item, e);
}
else {
// If it's not skippable that's an error in
// configuration - it doesn't make sense to not roll
// back if we are also not allowed to skip
throw new NonSkippableProcessException(
"Non-skippable exception in processor. Make sure any exceptions that do not cause a rollback are skippable.",
e);
}
}
finally {
stopTimer(sample, contribution.getStepExecution(), "item.process", status, "Item processing");
}
if (output == null) {
// No need to re-process filtered items
iterator.remove();
}
return output;
};
RecoveryCallback<O> recoveryCallback = context -> {
Throwable e = context.getLastThrowable();
if (shouldSkip(itemProcessSkipPolicy, e, contribution.getStepSkipCount())) {
iterator.remove(e);
contribution.incrementProcessSkipCount();
logger.debug("Skipping after failed process", e);
return null;
}
else {
if (rollbackClassifier.classify(e)) {
// Default is to rollback unless the classifier
// allows us to continue
throw new RetryException("Non-skippable exception in recoverer while processing", e);
}
iterator.remove(e);
return null;
}
};
O output = batchRetryTemplate.execute(retryCallback, recoveryCallback,
@@ -328,76 +318,68 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
final UserData<O> data = (UserData<O>) inputs.getUserData();
final AtomicReference<RetryContext> contextHolder = new AtomicReference<>();
RetryCallback<Object, Exception> retryCallback = new RetryCallback<>() {
@Override
public Object doWithRetry(RetryContext context) throws Exception {
contextHolder.set(context);
RetryCallback<Object, Exception> retryCallback = context -> {
contextHolder.set(context);
if (!data.scanning()) {
chunkMonitor.setChunkSize(inputs.size());
Timer.Sample sample = BatchMetrics.createTimerSample(meterRegistry);
String status = BatchMetrics.STATUS_SUCCESS;
try {
doWrite(outputs);
}
catch (Exception e) {
status = BatchMetrics.STATUS_FAILURE;
if (rollbackClassifier.classify(e)) {
throw e;
}
/*
* If the exception is marked as no-rollback, we need to override
* that, otherwise there's no way to write the rest of the chunk
* or to honour the skip listener contract.
*/
throw new ForceRollbackForWriteSkipException(
"Force rollback on skippable exception so that skipped item can be located.", e);
}
finally {
stopTimer(sample, contribution.getStepExecution(), "chunk.write", status, "Chunk writing");
}
contribution.incrementWriteCount(outputs.size());
if (!data.scanning()) {
chunkMonitor.setChunkSize(inputs.size());
Timer.Sample sample = BatchMetrics.createTimerSample(meterRegistry);
String status = BatchMetrics.STATUS_SUCCESS;
try {
doWrite(outputs);
}
else {
scan(contribution, inputs, outputs, chunkMonitor, false);
catch (Exception e) {
status = BatchMetrics.STATUS_FAILURE;
if (rollbackClassifier.classify(e)) {
throw e;
}
/*
* If the exception is marked as no-rollback, we need to override
* that, otherwise there's no way to write the rest of the chunk or to
* honour the skip listener contract.
*/
throw new ForceRollbackForWriteSkipException(
"Force rollback on skippable exception so that skipped item can be located.", e);
}
return null;
finally {
stopTimer(sample, contribution.getStepExecution(), "chunk.write", status, "Chunk writing");
}
contribution.incrementWriteCount(outputs.size());
}
else {
scan(contribution, inputs, outputs, chunkMonitor, false);
}
return null;
};
if (!buffering) {
RecoveryCallback<Object> batchRecoveryCallback = new RecoveryCallback<>() {
RecoveryCallback<Object> batchRecoveryCallback = context -> {
@Override
public Object recover(RetryContext context) throws Exception {
Throwable e = context.getLastThrowable();
if (outputs.size() > 1 && !rollbackClassifier.classify(e)) {
throw new RetryException("Invalid retry state during write caused by "
+ "exception that does not classify for rollback: ", e);
}
Throwable e = context.getLastThrowable();
if (outputs.size() > 1 && !rollbackClassifier.classify(e)) {
throw new RetryException("Invalid retry state during write caused by "
+ "exception that does not classify for rollback: ", e);
Chunk<I>.ChunkIterator inputIterator = inputs.iterator();
for (Chunk<O>.ChunkIterator outputIterator = outputs.iterator(); outputIterator.hasNext();) {
inputIterator.next();
outputIterator.next();
checkSkipPolicy(inputIterator, outputIterator, e, contribution, true);
if (!rollbackClassifier.classify(e)) {
throw new RetryException(
"Invalid retry state during recovery caused by exception that does not classify for rollback: ",
e);
}
Chunk<I>.ChunkIterator inputIterator = inputs.iterator();
for (Chunk<O>.ChunkIterator outputIterator = outputs.iterator(); outputIterator.hasNext();) {
inputIterator.next();
outputIterator.next();
checkSkipPolicy(inputIterator, outputIterator, e, contribution, true);
if (!rollbackClassifier.classify(e)) {
throw new RetryException(
"Invalid retry state during recovery caused by exception that does not classify for rollback: ",
e);
}
}
return null;
}
return null;
};
batchRetryTemplate.execute(retryCallback, batchRecoveryCallback,
@@ -406,26 +388,21 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
}
else {
RecoveryCallback<Object> recoveryCallback = new RecoveryCallback<>() {
@Override
public Object recover(RetryContext context) throws Exception {
/*
* If the last exception was not skippable we don't need to do any
* scanning. We can just bomb out with a retry exhausted.
*/
if (!shouldSkip(itemWriteSkipPolicy, context.getLastThrowable(), -1)) {
throw new ExhaustedRetryException(
"Retry exhausted after last attempt in recovery path, but exception is not skippable.",
context.getLastThrowable());
}
inputs.setBusy(true);
data.scanning(true);
scan(contribution, inputs, outputs, chunkMonitor, true);
return null;
RecoveryCallback<Object> recoveryCallback = context -> {
/*
* If the last exception was not skippable we don't need to do any
* scanning. We can just bomb out with a retry exhausted.
*/
if (!shouldSkip(itemWriteSkipPolicy, context.getLastThrowable(), -1)) {
throw new ExhaustedRetryException(
"Retry exhausted after last attempt in recovery path, but exception is not skippable.",
context.getLastThrowable());
}
inputs.setBusy(true);
data.scanning(true);
scan(contribution, inputs, outputs, chunkMonitor, true);
return null;
};
if (logger.isDebugEnabled()) {

53
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.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.
@@ -31,8 +31,6 @@ import org.springframework.batch.core.listener.MulticasterBatchListener;
import org.springframework.batch.core.observability.BatchMetrics;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.repeat.RepeatCallback;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.RepeatOperations;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.lang.Nullable;
@@ -126,34 +124,29 @@ public class SimpleChunkProvider<I> implements ChunkProvider<I> {
public Chunk<I> provide(final StepContribution contribution) throws Exception {
final Chunk<I> inputs = new Chunk<>();
repeatOperations.iterate(new RepeatCallback() {
@Override
public RepeatStatus doInIteration(final RepeatContext context) throws Exception {
I item = null;
Timer.Sample sample = Timer.start(Metrics.globalRegistry);
String status = BatchMetrics.STATUS_SUCCESS;
try {
item = read(contribution, inputs);
}
catch (SkipOverflowException e) {
// read() tells us about an excess of skips by throwing an
// exception
status = BatchMetrics.STATUS_FAILURE;
return RepeatStatus.FINISHED;
}
finally {
stopTimer(sample, contribution.getStepExecution(), status);
}
if (item == null) {
inputs.setEnd();
return RepeatStatus.FINISHED;
}
inputs.add(item);
contribution.incrementReadCount();
return RepeatStatus.CONTINUABLE;
repeatOperations.iterate(context -> {
I item = null;
Timer.Sample sample = Timer.start(Metrics.globalRegistry);
String status = BatchMetrics.STATUS_SUCCESS;
try {
item = read(contribution, inputs);
}
catch (SkipOverflowException e) {
// read() tells us about an excess of skips by throwing an
// exception
status = BatchMetrics.STATUS_FAILURE;
return RepeatStatus.FINISHED;
}
finally {
stopTimer(sample, contribution.getStepExecution(), status);
}
if (item == null) {
inputs.setEnd();
return RepeatStatus.FINISHED;
}
inputs.add(item);
contribution.incrementReadCount();
return RepeatStatus.CONTINUABLE;
});
return inputs;