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

RESOLVED - issue BATCH-1301: ItemStream is not being opened correctly for multi-threaded Step when scope="step"

Browse Source 
Fixed bug in unit test and tidied up context management
This commit is contained in:
dsyer
2009-06-21 08:52:47 +00:00
parent 528b5f6d9c
commit 1a92c5479a
2 changed files with 28 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepSynchronizationManager.java
View File

@@ -72,7 +72,9 @@ public class StepSynchronizationManager {
if (getCurrent().isEmpty()) {
return null;
}
return contexts.get(getCurrent().peek());
synchronized (contexts) {
return contexts.get(getCurrent().peek());
}
}
/**
@@ -89,10 +91,13 @@ public class StepSynchronizationManager {
return null;
}
getCurrent().push(stepExecution);
StepContext context = contexts.get(stepExecution);
if (context == null) {
context = new StepContext(stepExecution);
contexts.put(stepExecution, context);
StepContext context;
synchronized (contexts) {
context = contexts.get(stepExecution);
if (context == null) {
context = new StepContext(stepExecution);
contexts.put(stepExecution, context);
}
}
increment();
return context;
@@ -119,7 +124,9 @@ public class StepSynchronizationManager {
if (current != null) {
int remaining = counts.get(current).decrementAndGet();
if (remaining <= 0) {
contexts.remove(current);
synchronized (contexts) {
contexts.remove(current);
}
}
}
}
@@ -127,10 +134,13 @@ public class StepSynchronizationManager {
private static void increment() {
StepExecution current = getCurrent().peek();
if (current != null) {
AtomicInteger count = counts.get(current);
if (count == null) {
count = new AtomicInteger();
counts.put(current, count);
AtomicInteger count;
synchronized (counts) {
count = counts.get(current);
if (count == null) {
count = new AtomicInteger();
counts.put(current, count);
}
}
count.incrementAndGet();
}

14
spring-batch-core/src/test/java/org/springframework/batch/core/scope/AsyncStepScopeIntegrationTests.java
View File

@@ -78,14 +78,14 @@ public class AsyncStepScopeIntegrationTests implements BeanFactoryAware {
for (int i = 0; i < 12; i++) {
final String value = "foo" + i;
final Long id = 123L+i;
final Long id = 123L + i;
FutureTask<String> task = new FutureTask<String>(new Callable<String>() {
public String call() throws Exception {
StepExecution stepExecution = new StepExecution(value, new JobExecution(0L), id);
ExecutionContext executionContext = stepExecution.getExecutionContext();
executionContext.put("foo", value);
StepContext context = StepSynchronizationManager.register(stepExecution);
logger.debug("Registered: "+context.getStepExecutionContext());
logger.debug("Registered: " + context.getStepExecutionContext());
try {
return simple.getName();
}
@@ -117,13 +117,13 @@ public class AsyncStepScopeIntegrationTests implements BeanFactoryAware {
assertEquals("foo", simple.getName());
for (int i = 0; i < 12; i++) {
final String value = "foo"+i;
final String value = "foo" + i;
FutureTask<String> task = new FutureTask<String>(new Callable<String>() {
public String call() throws Exception {
ExecutionContext executionContext = stepExecution.getExecutionContext();
executionContext.put("foo", value);
StepContext context = StepSynchronizationManager.register(stepExecution);
logger.debug("Registered: "+context.getStepExecutionContext());
logger.debug("Registered: " + context.getStepExecutionContext());
try {
return simple.getName();
}
@@ -135,8 +135,6 @@ public class AsyncStepScopeIntegrationTests implements BeanFactoryAware {
tasks.add(task);
taskExecutor.execute(task);
}
StepSynchronizationManager.close();
int i = 0;
for (FutureTask<String> task : tasks) {
@@ -144,6 +142,10 @@ public class AsyncStepScopeIntegrationTests implements BeanFactoryAware {
i++;
}
// Don't close the outer scope until all tasks are finished. This should
// always be the case if using an AbstractStep
StepSynchronizationManager.close();
}
}