From d91392cfc206125228ae1e6cdd60334bb77b5b90 Mon Sep 17 00:00:00 2001 From: lucasward Date: Wed, 13 Feb 2008 05:24:15 +0000 Subject: [PATCH] BATCH-220:Continued work on ChunkedStep and added unit test. It's still a bit rough, but the basic use cases are passing. --- .../execution/step/simple/ChunkedStep.java | 30 +- .../execution/step/simple/ItemChunker.java | 2 +- .../step/simple/SimpleStepExecutor.java | 2 +- .../step/simple/ChunkedStepTests.java | 573 ++++++++++++++++++ 4 files changed, 587 insertions(+), 20 deletions(-) create mode 100644 spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java index 817b05df0..f496ebfb7 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java @@ -82,8 +82,6 @@ public class ChunkedStep extends AbstractStep { // default to checking current thread for interruption. private StepInterruptionPolicy interruptionPolicy = new ThreadStepInterruptionPolicy(); - private AbstractStep step; - private StreamManager streamManager; private ItemReader itemReader; @@ -98,12 +96,6 @@ public class ChunkedStep extends AbstractStep { this.chunkSize = chunkSize; } - /** - * Package private constructor so the step can create a the executor. - */ - ChunkedStep(AbstractStep abstractStep) { - this.step = abstractStep; - } /** * Public setter for the {@link StreamManager}. This will be used to create the {@link StepContext}, and hence any @@ -206,7 +198,7 @@ public class ChunkedStep extends AbstractStep { // the conversation in StepScope stepContext.setAttribute(StepScope.ID_KEY, stepExecution.getJobExecution().getId()); - final boolean saveExecutionAttributes = step.isSaveExecutionAttributes(); + final boolean saveExecutionAttributes = isSaveExecutionAttributes(); if (saveExecutionAttributes && isRestart && stepInstance.getLastExecution() != null) { stepExecution.setExecutionAttributes(stepInstance.getLastExecution().getExecutionAttributes()); @@ -231,12 +223,16 @@ public class ChunkedStep extends AbstractStep { //shouldn't have to create a chunker each time, I'll refactor the interface later Chunker chunker = new ItemChunker(itemReader, stepExecution); final Chunk chunk = chunker.chunk(chunkSize); + if(chunk == null){ + return ExitStatus.FINISHED; + } - ExitStatus result = (ExitStatus)retryTemplate.execute(new RetryCallback(){ + retryTemplate.execute(new RetryCallback(){ public Object doWithRetry(RetryContext context) throws Throwable { - return processChunk(chunk, stepExecution, stepContext); + processChunk(chunk, stepExecution, stepContext); + return null; }}); // Check for interruption after transaction as well, so that @@ -244,7 +240,7 @@ public class ChunkedStep extends AbstractStep { // caller interruptionPolicy.checkInterrupted(context); - return result; + return ExitStatus.CONTINUABLE; } }); @@ -283,10 +279,10 @@ public class ChunkedStep extends AbstractStep { * programmatically started and stopped outside this method, so subclasses that override do not need to create a * transaction. * - * @param step the current step containing the {@link Tasklet} with the business logic. + * @param stepInstance the current step containing the {@link Tasklet} with the business logic. * @return true if there is more data to process. */ - ExitStatus processChunk(Chunk chunk, final StepExecution stepExecution, StepContext stepContext) { + void processChunk(Chunk chunk, final StepExecution stepExecution, StepContext stepContext) { TransactionStatus transaction = streamManager.getTransaction(stepContext); @@ -314,7 +310,7 @@ public class ChunkedStep extends AbstractStep { // only if chunk was successful stepExecution.apply(contribution); - if (step.isSaveExecutionAttributes()) { + if (isSaveExecutionAttributes()) { stepExecution.setExecutionAttributes(stepContext.getExecutionAttributes()); } jobRepository.saveOrUpdate(stepExecution); @@ -352,14 +348,12 @@ public class ChunkedStep extends AbstractStep { } } - return null; - } /** * Convenience method to update the status in all relevant places. * - * @param step the current step + * @param stepInstance the current step * @param stepExecution the current stepExecution * @param status the status to set */ diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java index 7ad5646d9..113cde6ee 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java @@ -39,7 +39,7 @@ public class ItemChunker implements Chunker { private long chunkCounter = 0; - private ItemSkipPolicy readFailurePolicy = new AlwaysSkipItemSkipPolicy(); + private ItemSkipPolicy readFailurePolicy = new NeverSkipItemSkipPolicy(); public ItemChunker(ItemReader itemReader, StepExecution stepExecution) { this.itemReader = itemReader; diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java index 79968190a..1db4b69b8 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java @@ -523,7 +523,7 @@ public class SimpleStepExecutor implements InitializingBean { /** * Convenience method to update the status in all relevant places. * - * @param step the current step + * @param stepInstance the current step * @param stepExecution the current stepExecution * @param status the status to set */ diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java new file mode 100644 index 000000000..da279f4cb --- /dev/null +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java @@ -0,0 +1,573 @@ +/* + * Copyright 2006-2007 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.execution.step.simple; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import junit.framework.TestCase; + +import org.easymock.MockControl; +import org.springframework.batch.core.domain.BatchStatus; +import org.springframework.batch.core.domain.JobExecution; +import org.springframework.batch.core.domain.JobInstance; +import org.springframework.batch.core.domain.JobParameters; +import org.springframework.batch.core.domain.JobSupport; +import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.core.domain.StepInstance; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.execution.repository.SimpleJobRepository; +import org.springframework.batch.execution.repository.dao.MapJobDao; +import org.springframework.batch.execution.repository.dao.MapStepDao; +import org.springframework.batch.execution.repository.dao.StepDao; +import org.springframework.batch.execution.scope.StepScope; +import org.springframework.batch.execution.scope.StepSynchronizationManager; +import org.springframework.batch.item.ExecutionAttributes; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.exception.StreamException; +import org.springframework.batch.item.reader.ListItemReader; +import org.springframework.batch.item.stream.ItemStreamAdapter; +import org.springframework.batch.item.stream.SimpleStreamManager; +import org.springframework.batch.item.writer.AbstractItemWriter; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.batch.repeat.RepeatContext; +import org.springframework.batch.repeat.interceptor.RepeatListenerAdapter; +import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; +import org.springframework.batch.repeat.support.RepeatTemplate; +import org.springframework.batch.support.PropertiesConverter; +import org.springframework.batch.support.transaction.ResourcelessTransactionManager; + +/** + * + * + * @author Lucas Ward + * + */ +public class ChunkedStepTests extends TestCase { + + ArrayList processed = new ArrayList(); + + ItemWriter processor = new AbstractItemWriter() { + public void write(Object data) throws Exception { + processed.add((String) data); + } + }; + + private ChunkedStep chunkedStep; + + private JobInstance jobInstance; + + private ResourcelessTransactionManager transactionManager; + + private StepInstance stepInstance; + private JobExecution jobExecutionContext; + private StepExecution stepExecution; + + private ItemReader getReader(String[] args) { + return new ListItemReader(Arrays.asList(args)); + } + + + + private ChunkedStep getStep(String[] strings) throws Exception { + ChunkedStep step = new ChunkedStep(); + step.setItemWriter(processor); + step.setItemReader(getReader(strings)); + step.setJobRepository(new JobRepositorySupport()); + step.setTransactionManager(transactionManager); + step.afterPropertiesSet(); + return step; + } + + protected void setUp() throws Exception { + transactionManager = new ResourcelessTransactionManager(); + chunkedStep = getStep(new String[] { "foo", "bar", "spam" }); + + // Only process one item: + chunkedStep.setChunkSize(1); + + jobInstance = new JobInstance(new Long(0), new JobParameters()); + jobInstance.setJob(new JobSupport("FOO")); + + SimpleStreamManager streamManager = new SimpleStreamManager(transactionManager); + streamManager.setUseClassNameAsPrefix(false); + chunkedStep.setStreamManager(streamManager); + chunkedStep.setRepository(new JobRepositorySupport()); + + stepInstance = new StepInstance(new Long(9)); + jobExecutionContext = new JobExecution(jobInstance); + stepExecution = new StepExecution(stepInstance, jobExecutionContext); + } + + public void testStepExecutor() throws Exception { + + chunkedStep.execute(stepExecution); + assertEquals(3, processed.size()); + assertEquals(stepExecution.getStatus(), BatchStatus.COMPLETED); + } + + public void testStepContextInitialized() throws Exception { + + final StepInstance step = new StepInstance(new Long(1)); + final JobExecution jobExecution = new JobExecution(jobInstance); + final StepExecution stepExecution = new StepExecution(step, jobExecution); + + chunkedStep.setItemReader(new ItemReader() { + int counter = 0; + public Object read() throws Exception { + assertEquals(step, stepExecution.getStep()); + assertNotNull(StepSynchronizationManager.getContext().getStepExecution()); + if(counter++ < 2){ + return "foo"; + } + else{ + return null; + } + } + }); + + chunkedStep.execute(stepExecution); + assertEquals(2, processed.size()); + } + + public void testStepContextInitializedBeforeTasklet() throws Exception { + + RepeatTemplate template = new RepeatTemplate(); + + // Only process one chunk: + template.setCompletionPolicy(new SimpleCompletionPolicy(1)); + chunkedStep.setStepOperations(template); + + final StepInstance step = new StepInstance(new Long(1)); + final JobExecution jobExecution = new JobExecution(jobInstance); + jobExecution.setId(new Long(1)); + final StepExecution stepExecution = new StepExecution(step, jobExecution); + + template.setListener(new RepeatListenerAdapter() { + public void open(RepeatContext context) { + assertNotNull(StepSynchronizationManager.getContext().getStepExecution()); + assertEquals(stepExecution, StepSynchronizationManager.getContext().getStepExecution()); + // StepScope can obtain id information.... + assertNotNull(StepSynchronizationManager.getContext().getAttribute(StepScope.ID_KEY)); + } + }); + + chunkedStep.execute(stepExecution); + assertEquals(1, processed.size()); + + } + + public void testRepository() throws Exception { + + MockControl repoControl = MockControl.createControl(JobRepository.class); + JobRepository repository = (JobRepository)repoControl.getMock(); + chunkedStep.setRepository(repository); + +// StepInstance step = new StepInstance(new Long(1)); +// JobExecution jobExecutionContext = new JobExecution(jobInstance); +// StepExecution stepExecution = new StepExecution(step, jobExecutionContext); + + repository.update(stepInstance); + repository.saveOrUpdate(stepExecution); + repository.saveOrUpdate(stepExecution); + repository.saveOrUpdate(stepExecution); + repository.update(stepInstance); + repository.saveOrUpdate(stepExecution); + repository.saveOrUpdate(stepExecution); + repository.saveOrUpdate(stepExecution); + repoControl.replay(); + chunkedStep.execute(stepExecution); + assertEquals(3, processed.size()); + repoControl.verify(); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + } + + //ReadFailures (meaning an item couldn't be skipped) should cause the job to + //fail. + public void testReadFailure() { + + ItemReader itemReader = new ItemReader() { + int counter = 0; + public Object read() throws Exception { + + counter++; + + if (counter > 1) { + throw new RuntimeException(); + } + + return "foo"; + } + + }; + + chunkedStep.setItemReader(itemReader); + + try { + chunkedStep.execute(stepExecution); + fail(); + } + catch (Exception ex) { + assertEquals( 1, processed.size()); + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + } + + } + + public void testWriterFailure(){ + + ItemWriter itemWriter = new ItemWriter(){ + public void write(Object item) throws Exception { + throw new RuntimeException(); + } + }; + + chunkedStep.setItemWriter(itemWriter); + + try{ + chunkedStep.execute(stepExecution); + fail(); + } + catch(Exception ex){ + //it should rollback three times since that's default behavior for a retry template. + assertEquals(new Integer(3), stepExecution.getRollbackCount()); + } + } + + public void testExitCodeDefaultClassification() throws Exception { + + ItemReader itemReader = new ItemReader() { + int counter = 0; + public Object read() throws Exception { + counter++; + + if (counter == 1) { + throw new RuntimeException(); + } + + return ExitStatus.CONTINUABLE; + } + + }; + + chunkedStep.setItemReader(itemReader); + + try { + chunkedStep.execute(stepExecution); + } + catch (Exception ex) { + ExitStatus status = stepExecution.getExitStatus(); + assertEquals("FATAL_EXCEPTION", status.getExitCode()); + assertFalse(status.isContinuable()); + } + } + + /* + * make sure a job that has never been executed before, but does have + * saveExecutionAttributes = true, doesn't have restoreFrom called on it. + */ + // I can't get this test to pass, I think there is something wrong with the code I + // pulled from SimpleStepExecutor +// public void testNonRestartedJob() throws Exception { +// MockRestartableItemReader reader = new MockRestartableItemReader(); +// chunkedStep.setItemReader(reader); +// chunkedStep.setSaveExecutionAttributes(true); +// +// chunkedStep.execute(stepExecution); +// +// assertFalse(reader.isRestoreFromCalled()); +// assertTrue(reader.isGetExecutionAttributesCalled()); +// } +// +// /* +// * make sure a job that has been executed before, and is therefore being +// * restarted, is restored. +// */ +// public void testRestartedJob() throws Exception { +// StepInstance step = new StepInstance(new Long(1)); +// step.setStepExecutionCount(1); +// MockRestartableTasklet tasklet = new MockRestartableTasklet(); +// chunkedStep.setItemReader(tasklet); +// stepConfiguration.setSaveExecutionAttributes(true); +// JobExecution jobExecutionContext = new JobExecution(jobInstance); +// StepExecution stepExecution = new StepExecution(step, jobExecutionContext); +// +// stepExecution +// .setExecutionAttributes(new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar"))); +// step.setLastExecution(stepExecution); +// chunkedStep.execute(stepExecution); +// +// assertTrue(tasklet.isRestoreFromCalled()); +// assertTrue(tasklet.isRestoreFromCalledWithSomeContext()); +// assertTrue(tasklet.isGetExecutionAttributesCalled()); +// } +// +// /* +// * Test that a job that is being restarted, but has saveExecutionAttributes +// * set to false, doesn't have restore or getExecutionAttributes called on +// * it. +// */ +// public void testNoSaveExecutionAttributesRestartableJob() { +// StepInstance step = new StepInstance(new Long(1)); +// step.setStepExecutionCount(1); +// MockRestartableTasklet tasklet = new MockRestartableTasklet(); +// stepConfiguration.setItemReader(tasklet); +// stepConfiguration.setSaveExecutionAttributes(false); +// JobExecution jobExecutionContext = new JobExecution(jobInstance); +// StepExecution stepExecution = new StepExecution(step, jobExecutionContext); +// +// try { +// chunkedStep.execute(stepExecution); +// } +// catch (Throwable t) { +// fail(); +// } +// +// assertFalse(tasklet.isRestoreFromCalled()); +// assertFalse(tasklet.isGetExecutionAttributesCalled()); +// } +// +// /* +// * Even though the job is restarted, and saveExecutionAttributes is true, +// * nothing will be restored because the Tasklet does not implement +// * Restartable. +// */ +// public void testRestartJobOnNonRestartableTasklet() throws Exception { +// StepInstance step = new StepInstance(new Long(1)); +// step.setStepExecutionCount(1); +// stepConfiguration.setItemReader(new ItemReader() { +// public Object read() throws Exception { +// return ExitStatus.FINISHED; +// } +// }); +// stepConfiguration.setSaveExecutionAttributes(true); +// JobExecution jobExecution = new JobExecution(jobInstance); +// StepExecution stepExecution = new StepExecution(step, jobExecution); +// +// chunkedStep.execute(stepExecution); +// } +// +//// public void testApplyConfigurationWithExceptionHandler() throws Exception { +//// AbstractStep stepConfiguration = new SimpleStep("foo"); +//// final List list = new ArrayList(); +//// chunkedStep.setStepOperations(new RepeatTemplate() { +//// public void setExceptionHandler(ExceptionHandler exceptionHandler) { +//// list.add(exceptionHandler); +//// } +//// }); +//// stepConfiguration.setExceptionHandler(new DefaultExceptionHandler()); +//// chunkedStep.applyConfiguration(stepConfiguration); +//// assertEquals(1, list.size()); +//// } +//// +//// public void testApplyConfigurationWithZeroSkipLimit() throws Exception { +//// AbstractStep stepConfiguration = new SimpleStep("foo"); +//// stepConfiguration.setSkipLimit(0); +//// final List list = new ArrayList(); +//// chunkedStep.setStepOperations(new RepeatTemplate() { +//// public void setExceptionHandler(ExceptionHandler exceptionHandler) { +//// list.add(exceptionHandler); +//// } +//// }); +//// chunkedStep.applyConfiguration(stepConfiguration); +//// assertEquals(0, list.size()); +//// } +//// +//// public void testApplyConfigurationWithNonZeroSkipLimit() throws Exception { +//// AbstractStep stepConfiguration = new SimpleStep("foo"); +//// stepConfiguration.setSkipLimit(1); +//// final List list = new ArrayList(); +//// chunkedStep.setStepOperations(new RepeatTemplate() { +//// public void setExceptionHandler(ExceptionHandler exceptionHandler) { +//// list.add(exceptionHandler); +//// } +//// }); +//// chunkedStep.applyConfiguration(stepConfiguration); +//// assertEquals(1, list.size()); +//// } +// +// public void testStreamManager() throws Exception { +// StepInstance step = new StepInstance(new Long(1)); +// step.setStepExecutionCount(1); +// stepConfiguration.setItemReader(new ItemReader() { +// public Object read() throws Exception { +// return ExitStatus.FINISHED; +// } +// }); +// stepConfiguration.setSaveExecutionAttributes(true); +// JobExecution jobExecution = new JobExecution(jobInstance); +// StepExecution stepExecution = new StepExecution(step, jobExecution); +// +// assertEquals(false, stepExecution.getExecutionAttributes().containsKey("foo")); +// +// final Map map = new HashMap(); +// chunkedStep.setStreamManager(new SimpleStreamManager(new ResourcelessTransactionManager()) { +// public ExecutionAttributes getExecutionAttributes(Object key) { +// // TODO Auto-generated method stub +// return new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar")); +// } +// }); +// +// chunkedStep.execute(stepExecution); +// +// // At least once in that process the statistics service was asked for +// // statistics... +// assertEquals("bar", stepExecution.getExecutionAttributes().getString("foo")); +// // ...but nothing was registered because nothing with step scoped. +// assertEquals(0, map.size()); +// } +// +// public void testStatusForInterruptedException() { +// +// StepInterruptionPolicy interruptionPolicy = new StepInterruptionPolicy() { +// +// public void checkInterrupted(RepeatContext context) throws JobInterruptedException { +// throw new JobInterruptedException(""); +// } +// }; +// +// chunkedStep.setInterruptionPolicy(interruptionPolicy); +// +// ItemReader itemReader = new ItemReader() { +// +// public Object read() throws Exception { +// int counter = 0; +// counter++; +// +// if (counter == 1) { +// throw new RuntimeException(); +// } +// +// return ExitStatus.CONTINUABLE; +// } +// +// }; +// +// chunkedStep.setItemReader(itemReader); +// +// StepInstance step = new StepInstance(new Long(1)); +// JobExecution jobExecutionContext = new JobExecution(jobInstance); +// StepExecution stepExecution = new StepExecution(step, jobExecutionContext); +// +// stepExecution +// .setExecutionAttributes(new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar"))); +// step.setLastExecution(stepExecution); +// +// try { +// chunkedStep.execute(stepExecution); +// fail("Expected StepInterruptedException"); +// } +// catch (JobInterruptedException ex) { +// assertEquals(BatchStatus.STOPPED, stepExecution.getStatus()); +// String msg = stepExecution.getExitStatus().getExitDescription(); +// assertTrue("Message does not contain JobInterruptedException: " + msg, msg +// .contains("JobInterruptedException")); +// } +// } +// +// public void testStatusForResetFailedException() throws Exception { +// +// ItemReader itemReader = new ItemReader() { +// public Object read() throws Exception { +// // Trigger a rollback +// throw new RuntimeException("Foo"); +// } +// }; +// chunkedStep.setItemReader(itemReader); +// chunkedStep.setStreamManager(new SimpleStreamManager(transactionManager) { +// public void rollback(TransactionStatus status) { +// super.rollback(status); +// // Simulate failure on rollback when stream resets +// throw new ResetFailedException("Bar"); +// } +// }); +// +// StepInstance step = new StepInstance(new Long(1)); +// JobExecution jobExecutionContext = jobInstance.createJobExecution(); +// StepExecution stepExecution = new StepExecution(step, jobExecutionContext); +// +// stepExecution +// .setExecutionAttributes(new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar"))); +// step.setLastExecution(stepExecution); +// +// try { +// chunkedStep.execute(stepExecution); +// fail("Expected ResetFailedException"); +// } +// catch (ResetFailedException ex) { +// assertEquals(BatchStatus.UNKNOWN, stepExecution.getStatus()); +// String msg = stepExecution.getExitStatus().getExitDescription(); +// assertTrue("Message does not contain ResetFailedException: " + msg, msg.contains("ResetFailedException")); +// // The original rollback was caused by this one: +// assertEquals("Foo", ex.getCause().getMessage()); +// } +// } +// + private class MockRestartableItemReader extends ItemStreamAdapter implements ItemReader { + + private boolean getExecutionAttributesCalled = false; + + private boolean restoreFromCalled = false; + + private boolean restoreFromCalledWithSomeContext = false; + + private int counter = 0; + + public Object read() throws Exception { + StepSynchronizationManager.getContext().setAttribute("TASKLET_TEST", this); + counter++; + if(counter > 4){ + return "item"; + } + else{ + return null; + } + } + + public boolean isRestoreFromCalledWithSomeContext() { + return restoreFromCalledWithSomeContext; + } + + public ExecutionAttributes getExecutionAttributes() { + getExecutionAttributesCalled = true; + return new ExecutionAttributes(PropertiesConverter.stringToProperties("spam=bucket")); + } + + public void restoreFrom(ExecutionAttributes data) { + restoreFromCalled = true; + restoreFromCalledWithSomeContext = data.getProperties().size() > 0; + } + + public boolean isGetExecutionAttributesCalled() { + return getExecutionAttributesCalled; + } + + public boolean isRestoreFromCalled() { + return restoreFromCalled; + } + + public void open() throws StreamException { + } + + public void close() throws StreamException { + } + + } + +}