Issue with Gateway when using in combination with Integration flow or a channel

[sirimamilla]

When Spring Integration Java DSL gateway is used in combination with Channel or Subflow, success case works perfectly fine. But when ever error occurs in the subflow or flow called by channel, below error is thrown and the caller will not get any response after that.
Error:

Reply message received but the receiving thread has already received a reply:ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: HTTP request execution failed for URI [http://mwgtc-tibla59u.nam.nsroot.net:16250/private/v1/rewards/memberships?cardNumber=0004547502000000635]; nested exception is org.springframework.web.client.HttpServerErrorException$InternalServerError: 500 Internal Server Error, failedMessage=GenericMessage [payload=com.citi.gcg.nga.cache.esbcahe.CacheElements$CardListingCredit@5637d1de, headers={businesscode=GCB, gatherResultChannel=f22c8e4a-0c6d-4817-a0db-ad4e217af4dd:2, sequenceDetails=[[b11d796e-9bf7-a0d9-1b4b-776873ab0a42, 1, 1]], errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@3f8377dd, sequenceSize=1, countrycode=SG, jms_destination=Queue[sample.queue], jms_timestamp=1548837920152, sid=3b9f7209-113c-4adf-91ec-a3970810f414MBK, ReqPayload=com.citi.gcg.rewards.pwp.rewardsinquiry.request.RewardsInquiryRequest@36d4e584, jmsInboundMessageHandlerReplyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@f1302bd, jms_replyTo=Queue[$TMP$.JNDI_MWDIT2_G2C_01.1C7C5B6D250614E50.40], RestFlowOrignalErrorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@3f8377dd, correlationId=b6446b8f-c456-1bce-0a7b-cb085c303f67, CardListingCredit=com.citi.gcg.nga.cache.esbcahe.CacheElements$CardListingCredit@5637d1de, id=d5d7aa98-524d-5d5c-bc1e-ef68913c393d, contentType=application/json;charset=UTF-8, jms_expiration=0, jms_messageId=ID:JNDI_MWDIT2_G2C_01.1C7C5B6D250614E4F:40, timestamp=1548837923242, sequenceNumber=1, CustomerCache=com.citi.gcg.nga.cache.customer.Customer@e10aaa6, priority=4, replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@3f8377dd, CardCache=com.citi.gcg.nga.cache.esbcahe.CacheElements@f4e8265, jms_redelivered=false, jms_deliveryMode=2, OriginalReplyChannel=jmsInboundResponseHandler.input, channelName=rewardsInquiry.input, SessionCache=com.citi.gcg.nga.cache.account.Accounts@42af56b8, channelid=MBK}], headers={id=c656bffb-dbb9-df3d-01a3-d5bfc36d8aa3, timestamp=1548837923966}]

[garyrussell]

You need to show your flow definition together with any relevant supporting code/configuration.

[artembilan]

This more sounds like a StackOverfow question.
We don't handle them here. More over this project is only on the critical bug fixes support and everybody is advised to upgrade to Spring Integration 5.x.

[sirimamilla]

Hi @artembilan / @garyrussell / @dturanski ,

Good Afternoon, I am using SI 5.1.1

I have created a project to reproduce the error with gateway with minimal code.

Sample Project with test cases: Test case "gatewayTestError" hangs indefinitely.
Gatewaytest.zip

I still see the below error when testing the Gateway:

2019-02-04 15:14:13.918  WARN 9344 --- [Pool-1-worker-1] cMessagingTemplate$TemporaryReplyChannel : Reply message received but the receiving thread has exited due to an exception while sending the request message:ErrorMessage [payload=org.springframework.integration.transformer.MessageTransformationException: Failed to transform Message; nested exception is org.springframework.messaging.MessageHandlingException: nested exception is java.lang.RuntimeException: Throw Error, failedMessage=GenericMessage [payload=Error, headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@79841c00, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@79841c00, id=2c82c314-cf24-023a-0435-6c3e11e17db5, timestamp=1549264453914}], failedMessage=GenericMessage [payload=Error, headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@79841c00, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@79841c00, id=2c82c314-cf24-023a-0435-6c3e11e17db5, timestamp=1549264453914}], headers={id=72249c7c-edd9-39c2-7555-a794a488d161, timestamp=1549264453918}]

[garyrussell]

The DSL was moved to the core project in 5.x; moving this issue there.

[artembilan]

I have reproduced an issue with this configuration:

	@MessagingGateway
	public interface GatewayService {

		@Gateway(requestChannel = "executorChannel")
		public String echoGateway(String msg);

	}

	@Bean
	public MessageChannel executorChannel() {
		return new ExecutorChannel(Executors.newWorkStealingPool());
	}

	@Bean
	@ServiceActivator(inputChannel = "executorChannel")
	public GatewayMessageHandler gatewayMessageHandler() {
		GatewayMessageHandler gatewayMessageHandler = new GatewayMessageHandler();
		gatewayMessageHandler.setRequestChannelName("serviceChannel");
		return gatewayMessageHandler;
	}

	@ServiceActivator(inputChannel = "serviceChannel")
	public void handleWithError() {
		throw new RuntimeException("intentional");
	}

Looks like the culprit is exactly that ExecutorChannel and error handling process in the MessagePublishingErrorHandler.
When we re-throw an exception from downstream flow (having its TemporaryReplyChannel marked with the hasSendFailed), we end up with the errorChannel selected from the failed message in the exception.
And that failedMessage is already not actual to the current state of the flow and we get an errorChannel from headers related to the downstream flow.

Looks like we have to rebuild an exception with really an actual requestMessage to re-throw. This way we would bring into the MessagePublishingErrorHandler a proper errorChannel and replyChannel related to the current gateway.

[artembilan]

I found a workaround like this:

@Bean
  public IntegrationFlow gatewayInnterflow(){
    return f->f.channel(channels -> channels.executor(Executors.newWorkStealingPool()))
        .gateway("gatewayTransformer.input", e -> e.errorChannel("executorErrorChannel"));
  }

  @ServiceActivator(inputChannel = "executorErrorChannel")
  public Throwable handleExecutorError(Exception ex) {
    return ex.getCause().getCause();
  }

Pay attention to the e -> e.errorChannel("executorErrorChannel") configuration and what is inside a @ServiceActivator for that channel.

This way we return an original exception which is thrown to upstream to be wrapped to the proper MessageHandlingException before sending as an ErrorMessage to the MessagePublishingErrorHandler for the mentioned ExecutorChannel.
This way it resolves a proper errorChannel for the upstream gateway.

I wonder if we should treat this as a normal logic in case of nested gateway and an ExecutorChannel upstream...

[sirimamilla]

Hi Artam,

Above workaround resolved the issue. Thank you very much for the workaround.

We need to see how to fix the issue in gateway.

Thanks,
Jayadev

[sirimamilla]

Hi Artam,

We have found similar issue when we use nested ScatterGather. I will create a separate issue with sample project for that. In Scatter Gather with nesting ScatterGather even the Success case is failing.

Thanks,
Jayadev

[artembilan]

i wonder if the real problem is here, in the AbstractMessageHandler.handleMessage():

throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message,
			() -> "error occurred in message handler [" + this + "]", e);

In previous version (4.x) we didn't check for an exception and just threw a one with the current requestMessage as a context.
Now we just re-throw a downstream exception, where its failedMessage might be processed already, like this gateway use-case and may be the mentioned Scatter-Gather one.
I might be wrong - just an idea how these issues may correlated...
Although it doesn't sound like this logic is an absurd: it really makes sense to have a final message which caused the target problem.
Only need to take care about proper propagation exactly in places where we perform request-reply.

Yes, @sirimamilla , that would be great to have some sample to play with.

@garyrussell , WDYT?