My messaging pipeline is shown below. It all works with one exception: I can't get my processError method to trigger when an exception is thrown from the processFile method defined on the splitter endpoint.

The normal flow is for the processFile to run, return a list of objects with a File object and a Transmission object. The transmission object is routed to the processRules method and the file is moved to the processed directory as defined by the outbound-channel-adapter. That all works during my testing.

If an exception occurs in processFile I want processError called and the file routed to the unprocessed directory defined by the second outbound-channel-adapater. That's the part of the flow I can't working.

My processError method is never called on an exception with the current pipeline. My understanding is the exception will not be pushed if the message handling occurs on the same thread as the send but I didn't think that was the case with my setup.

Grant

Code:
<integration:thread-pool-task-executor id="fileImportPool" core-size="5"/>

<integration:publish-subscribe-channel id="errorChannel" />

<integration:channel id="filesImported">
    <integration:queue />
</integration:channel>

<integration:channel id="rulesToProcess">
    <integration:queue />
</integration:channel>

<integration:poller default="true" task-executor="fileImportPool">
    <integration:interval-trigger interval="1" time-unit="SECONDS" />
</integration:poller>

<file:inbound-channel-adapter
        id="filesToProcess"
        directory="${systemConfig.filesToProcess}"
        filename-pattern="^.*\.xml$" />

<integration:splitter
        id="fileImportSplit"
        ref="importService"
        method="processFile"
        input-channel="filesToProcess"
        output-channel="filesImported" />

<integration:payload-type-router input-channel="filesImported">
    <integration:mapping type="java.io.File" channel="filesProcessed" />
    <integration:mapping type="gov.hhs.acf.cb.nytd.models.Transmission" channel="rulesToProcess" />
</integration:payload-type-router>

<integration:service-activator
        ref="importService"
        method="processRules"
        input-channel="rulesToProcess" />

<integration:service-activator
    ref="importService"
    method="processError"
    input-channel="errorChannel"
    output-channel="filesNotProcessed" />

<file:outbound-channel-adapter
        id="filesProcessed"
        directory="${systemConfig.filesProcessed}"
        delete-source-files="true"/>

<file:outbound-channel-adapter
        id="filesNotProcessed"
        directory="${systemConfig.filesNotProcessed}"
        delete-source-files="true"/>