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

Refactored JobExecution for performance

Browse Source 
In order to make the Map based JobRepository threadsafe, the
JobExecutino was modified to use a `CopyOnWriteArraySet` for the
collection backing the child `StepExecution` collection.  However, this
collection option has bad performance characteristics when used with
large collections.  When using partitioning, it can be common to have a
large number of `StepExecution` instances to add which drastically hurts
start up time.

This commit remove the use of the `CopyOnWriteArraySet` for the
`JobExecution#stepExecutions` and replaces it with a `LinkedHashSet`
wrapped via `Collection.synchronizedSet`.

Resolves BATCH-2384
This commit is contained in:
Michael Minella
2017-11-02 12:38:35 -05:00
parent 9322fe6fae
commit fa286dc1e9
1 changed files with 9 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java
View File

@@ -1,5 +1,5 @@
/* /*
* Copyright 2006-2013 the original author or authors. * Copyright 2006-2017 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -23,10 +23,10 @@ import java.util.Collection;
import java.util.Collections; import java.util.Collections;
import java.util.Date; import java.util.Date;
import java.util.HashSet; import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List; import java.util.List;
import java.util.Set; import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ExecutionContext;
@@ -44,7 +44,7 @@ public class JobExecution extends Entity {
private JobInstance jobInstance; private JobInstance jobInstance;
private volatile Collection<StepExecution> stepExecutions = new CopyOnWriteArraySet<StepExecution>(); private volatile Collection<StepExecution> stepExecutions = Collections.synchronizedSet(new LinkedHashSet<>());
private volatile BatchStatus status = BatchStatus.STARTING; private volatile BatchStatus status = BatchStatus.STARTING;
@@ -60,7 +60,7 @@ public class JobExecution extends Entity {
private volatile ExecutionContext executionContext = new ExecutionContext(); private volatile ExecutionContext executionContext = new ExecutionContext();
private transient volatile List<Throwable> failureExceptions = new CopyOnWriteArrayList<Throwable>(); private transient volatile List<Throwable> failureExceptions = new CopyOnWriteArrayList<>();
private final String jobConfigurationName; private final String jobConfigurationName;
@@ -212,7 +212,7 @@ public class JobExecution extends Entity {
* @return the step executions that were registered * @return the step executions that were registered
*/ */
public Collection<StepExecution> getStepExecutions() { public Collection<StepExecution> getStepExecutions() {
return Collections.unmodifiableList(new ArrayList<StepExecution>(stepExecutions)); return Collections.unmodifiableList(new ArrayList<>(stepExecutions));
} }
/** /**
@@ -299,7 +299,7 @@ public class JobExecution extends Entity {
/** /**
* Package private method for re-constituting the step executions from * Package private method for re-constituting the step executions from
* existing instances. * existing instances.
* @param stepExecution * @param stepExecution execution to be added
*/ */
void addStepExecution(StepExecution stepExecution) { void addStepExecution(StepExecution stepExecution) {
stepExecutions.add(stepExecution); stepExecutions.add(stepExecution);
@@ -346,12 +346,12 @@ public class JobExecution extends Entity {
*/ */
public synchronized List<Throwable> getAllFailureExceptions() { public synchronized List<Throwable> getAllFailureExceptions() {
Set<Throwable> allExceptions = new HashSet<Throwable>(failureExceptions); Set<Throwable> allExceptions = new HashSet<>(failureExceptions);
for (StepExecution stepExecution : stepExecutions) { for (StepExecution stepExecution : stepExecutions) {
allExceptions.addAll(stepExecution.getFailureExceptions()); allExceptions.addAll(stepExecution.getFailureExceptions());
} }
return new ArrayList<Throwable>(allExceptions); return new ArrayList<>(allExceptions);
} }
/** /**
@@ -365,7 +365,7 @@ public class JobExecution extends Entity {
*/ */
private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException { private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
stream.defaultReadObject(); stream.defaultReadObject();
failureExceptions = new ArrayList<Throwable>(); failureExceptions = new ArrayList<>();
} }
/* /*
@@ -390,5 +390,4 @@ public class JobExecution extends Entity {
this.stepExecutions.addAll(stepExecutions); this.stepExecutions.addAll(stepExecutions);
} }
} }
} }