Error Handling in SI 2.0M7

[spodxx]

Hello

I'm not sure if this is by design or me having a understanding problem. When I setup a transformer listening to messages from a publish-subscribe-channel, aren't exceptions supposed to be propagated to the "errorChannel" as described in the reference manual?
Currently that doesn't seem to be the case.

I'm getting the following stack trace when I throw a exception deliberately in my transformer:

Exception in thread "main" org.springframework.integration.transformer.MessageTransformationException: org.springframework.integration.MessageHandlingException: java.lang.NullPointerException: test
	at org.springframework.integration.transformer.MessageTransformingHandler.handleRequestMessage(MessageTransformingHandler.java:71)
	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:102)
	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:77)
	at org.springframework.integration.dispatcher.BroadcastingDispatcher.invokeHandler(BroadcastingDispatcher.java:121)
	at org.springframework.integration.dispatcher.BroadcastingDispatcher.dispatch(BroadcastingDispatcher.java:112)
	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:44)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:157)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:128)
	at org.springframework.integration.core.MessagingTemplate.doSend(MessagingTemplate.java:374)
	at org.springframework.integration.core.MessagingTemplate.send(MessagingTemplate.java:219)
	at org.springframework.integration.aop.MessagePublishingInterceptor.publishMessage(MessagePublishingInterceptor.java:147)
	at org.springframework.integration.aop.MessagePublishingInterceptor.invoke(MessagePublishingInterceptor.java:116)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:172)
	at org.springframework.aop.framework.Cglib2AopProxy$DynamicAdvisedInterceptor.intercept(Cglib2AopProxy.java:621)
	at com.komdat.adwords.springframework.ReportingServiceFacade$$EnhancerByCGLIB$$cc1b41da.requestAndPublishReport(<generated>)
	at example.GetReport.getKeywordReport(GetKeywordReport.java:67)
	at example.GetReport.main(GetKeywordReport.java:46)

Caused by: org.springframework.integration.MessageHandlingException: java.lang.NullPointerException: test
	at org.springframework.integration.handler.MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:76)
	at org.springframework.integration.transformer.AbstractMessageProcessingTransformer.transform(AbstractMessageProcessingTransformer.java:56)
	at org.springframework.integration.transformer.MessageTransformingHandler.handleRequestMessage(MessageTransformingHandler.java:65)
	... 16 more

Currently the only solution I could come up with was to programatically create a error message and send it directly to the error channel. However, as the transformer is required to return a value I still need to throw the exception to stop message propagation which causes the stack trace shown above.

@Autowired
@Qualifier("errorChannel")
private PublishSubscribeChannel errorChannel;

public String transform(Report report) {
   ...
   Exception e = new Exception("test");
   errorChannel.send(new ErrorMessage(e));
   throw e;
}

[Mark Fisher]

Exceptions should be sent to the error channel if the point where they occur is within an asynchronous part of the message flow. However, if you are sending to a direct channel (sync) and that is dispatching to the Transformer, then the Exception can be thrown directly to the caller, since it's the same call stack.

Does that clarify the situation?... or are you in fact within an asynchronous flow at the point where the Exception is being thrown?

[spodxx]

So far I'm only using publish-subscribe channels which are, in my understanding, asynchronous.

I just realized that I'm able to catch the exception just where I started to call the @publisher annotated object that will create the very first message. But I don't really know what to do with it at this point in my code. I'd rather have the exception handled in errorChannel.

This is what my programm flow looks like (simplified):

class ReportDownload {

  @Autowired reportService

  main() {
    try {
       reportService.requestAndPublishReport()
    } catch(MessageTransformationException e) {
      // caught! but what to do?
    } 
  }
}

class ReportSerivce {

  @Publisher(channel="reportChannel")
  requestAndPublishReport(report) {
    return new Report()
  }
}

class ReportToTmpFileTransformer {

  public File transform(Report report) {
     
      throw new VerySpecificExceptionFromTransformer();
  }
}

applicationContext.xml:

<publish-subscribe-channel id="reportChannel"/>
<publish-subscribe-channel id="errorChannel"/>

<transformer ref="reportToTmpFileTransformer" 
  input-channel="reportChannel" 
  output-channel="reportFileOutputChannel"/>

<transformer ref="exceptionToTextTransformer" 
  input-channel="errorChannel" 
  output-channel="reportFileOutputChannel"/>

<file:outbound-channel-adapter id="reportFileOutputChannel"
  directory="/var/tmp/" 
  delete-source-files="true"/>

[Mark Fisher]

The publish-subscribe-channel is not actually asynchronous by default. However, if that's what you want, you can add a "task-executor" reference to it. Then it will send asynchronously (return immediately), and any Exceptions that occur in that downstream async flow should be sent to the errorChannel.

HTH,
mark

[spodxx]

Thanks for your help! I've added changed the following and now its working just as aspected.

<beans:bean id="pooled-executor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"/>
	
<publish-subscribe-channel id="reportChannel" task-executor="pooled-executor"/>

[Mark Fisher]

Excellent. Thanks for reporting back!