Skip processor step when retrying a chunk due to an error in the writer [BATCH-2639]

Adelino Rodrigues opened BATCH-2639 and commented

Background

When retrying a chunk due to a failure in the writer,
the read items are by default not read again.

For cases where processors do time consuming activities, it is very useful
to skip the processor step for the items that are already processed.

Question

I found a way to avoid processing again processed items that are ready to be written but it
seems a bit circumvoluted and would be happy to find about a simpler mechanism to achieve the
same goal.

Approach tested with version 2.2.7

1) Adding a step listener

This listener helps caching items in a chunk that have been processed.

public class SkipProcessedChunkListener {

    private static final String CACHE_ATTR_NAME = "ProcessedItems";
    
    /** Step execution context */
    private ExecutionContext stepExecCtx;

    @BeforeChunk
    public void beforeChunk() {
        Map<Object, Object> cache = getCache(this.stepExecCtx);
        
        if (cache == null) {
            cache = new HashMap<>();
            stepExecCtx.put(CACHE_ATTR_NAME, cache);
        }
    }
    
    @SuppressWarnings("unchecked")
    static Map<Object, Object> getCache(ExecutionContext stepExecCtx) {
        return (Map<Object, Object>) stepExecCtx.get(CACHE_ATTR_NAME);
    }

    @AfterChunk
    public void afterChunk() {
        Map<Object, Object> cache = getCache(this.stepExecCtx);
        
        if (cache != null) {
            stepExecCtx.remove(CACHE_ATTR_NAME);
            cache.clear();
        }
    }
    
    @BeforeStep
    public void setStepExecution(StepExecution stepExecution) {
        this.stepExecCtx = stepExecution.getExecutionContext();
    }
}

This listener is created as a bean with a step scope

<bean id="skipProcessedChunkListener"
	scope="step"
	class="SkipProcessedChunkListener"/>

and added to the step configuration;

<batch:step id="step" parent="faultTolerantStep">
	<batch:tasklet>
		<batch:chunk reader="reader" processor="itemProcessor" writer="writer"
				commit-interval="10">
			<!-- details skipped -->
			<batch:listeners>
				<batch:listener ref="skipProcessedChunkListener"/>
			</batch:listeners>
		</batch:chunk>
	</batch:tasklet>
</batch:step>

2) Substituting the itemProcessor

Processed items in a read chunk are cached and discarded when the chunk completes. The normal
item processor is replaced by an item processor (SkipProcessedItemProcessor) that:

• checks if the item has already been processed
• if not delegates the processing to the original itemProcessor

public class SkipProcessedItemProcessor<I, K, O> implements ItemProcessor<I, O> {

    /** Step execution context */
    private ExecutionContext stepExecCtx;

    /** Item processor to invoke when item is not cached */
    private ItemProcessor<I, O> delegate;
    
    /** Function to convert item to key */
    private Function<I, K> toKeyFn;

    /** New instance */
    public SkipProcessedItemProcessor(ItemProcessor<I, O> delegate, Function<I, K> toKeyFn) {
        this.delegate = delegate;
        this.toKeyFn = toKeyFn;
    }

    @SuppressWarnings("unchecked" )
    @Override
    public O process(I item) throws Exception {
        K key = toKeyFn.apply(item);
        Map<Object, Object> cache = SkipProcessedChunkListener.getCache(this.stepExecCtx);

        Object result = cache.get(key);

        if (result == null) {
            result = delegate.process(item);
            cache.put(key, result);
        }
        
        return (O) result;
    }

    @BeforeStep
    public void setStepExecution(StepExecution stepExecution) {
        this.stepExecCtx = stepExecution.getExecutionContext();
    }
}

This processor is created in the step scope:

<bean id="skipProcessedItemProcessor"
		scope="step"
		class="SkipProcessedItemProcessor">
	<constructor-arg ref="itemProcessor"/>
	<constructor-arg ref="itemToKeyFunction"/>
</bean>

This new processor is used instead of the original itemProcessor:

<batch:chunk reader="reader" processor="skipProcessedItemProcessor" writer="writer"
		commit-interval="10">
	<!-- details skipped -->
</batch:chunk>

---

No further details from BATCH-2639

Start with the fault-tolerant chunk configuration and the ItemProcessor, step listener, and ExecutionContext examples described in the issue. Trace how a writer failure causes chunk retry, then define and test the expected behavior: already processed items should skip processing while still being available for the writer.