tcp-inbound-gateway error-channel assistance

[Wrangler]

Hi,

The way I understand the use of the tcp-inbound-gateway error handling, is thrown exceptions are caught, then sent to the gateway's property "error-channel" as a MessagingException object. I will need to use this object to build a proper reply to go back to the client, and place the reply on the tcp-inbound-gateway "reply-channel".

To implement this I have set the tcp-inbound-gateway property "error-channel " as the "input-channel" on a service-activator. This service-activator has its "output-channel" set to the gateway's "reply-channel".

As a test of this flow, I created a service-activator which all it does throws an exception. The logs show the message is received by the service-activator and it throws the exception. This is as I expected.

With the logs in DEBUG I see the service-activator throws the exception, the message has a preSend on the gateways error channel.

Then there is the message "Dispatcher has no subscribers.", which I believe means I do not have something configured properly.

The attempt to build a reply message out of the MessagingException object does not go to the gateways reply channel, so it does not go back to the client. The gateway times out because of a null reply, and nothing is returned.

2012-04-30 10:46:05,215  WARN [pool-4-thread-3] (MessagingGatewaySupport.java:240) - failure occurred in gateway sendAndReceive
org.springframework.integration.dispatcher.AggregateMessageDeliveryException: All attempts to deliver Message to MessageHandlers failed. Multiple causes:

2012-04-30 10:46:05,215 DEBUG [pool-4-thread-3] (AbstractMessageChannel.java:224) - preSend on channel 'inboundTcpGatewayErrorChannel', message: [Payload=org.springframework.integration.dispatcher.AggregateMessageDeliveryException: All attempts to deliver Message to MessageHandlers failed. Multiple causes:
    Dispatcher has no subscribers.

This tells me I must have a configuration set incorrectly, or perhaps my understanding (in the first paragraph above) is not correct.

(Note: Company policy dictates only relevant snippets be shown)

    <int-ip:tcp-inbound-gateway
            connection-factory="connection-factory"
            request-channel="rawInboundChannel"
            reply-channel="outboundChannel"
            error-channel="inboundTcpGatewayErrorChannel"
            reply-timeout="10000" />

    <int:service-activator id="gatewayErrorHandler"
                           input-channel="inboundTcpGatewayErrorChannel"
                           output-channel="outboundChannel"
                           method="onErrorMessage">
        <bean class="com...DefaultErrorMessageHandler" />
    </int:service-activator>

    <int:service-activator  input-channel="inboundChannel"
                           method="onMessage">
        <bean class="com....ErrorThrowingServiceActivator" />
    </int:service-activator>

[Gary Russell]

A few things to check...

1. If you can upgrade to 2.1.1, the 'Dispatcher Has No Subcribers' message now contains the channel name.
2. If the above is not your actual code, check for a typo in channel names. If there's a typo, the subscriber will automatically create a channel and there'll be no connection. Also, your error-throwing service activator uses inboundChannel, but the gateway sends to rawInboundChannel - is there something else between them?
3. When you forward your error message to the outboundChannel, you must copy the replyChannel header from the failedMessage to your error message. Otherwise the gateway won't be able to correlate the response and it will be dropped. The failedMessage is a property of the ErrorMessage.
4. The AggregateMessageDeliveryException is used when a dispatcher sends messages to multiple endpoints and they all fail. Without seeing more of your configuration, it's hard to say why you are seeing this; my guess is you have multiple subscribers on inboundChannel (or rawInboundChannel); we try to dispatch to your error thrower; that fails so we try the next subscriber and, downstream of that subscriber, there is a channel with no subscriber.

[Wrangler]

A few things to check...
1. If you can upgrade to 2.1.1, the 'Dispatcher Has No Subcribers' message now contains the channel name.

Done. I can see the channel name now. I had a copy paste error.

2. If the above is not your actual code, check for a typo in channel names. If there's a typo, the subscriber will automatically create a channel and there'll be no connection. Also, your error-throwing service activator uses inboundChannel, but the gateway sends to rawInboundChannel - is there something else between them?

I have a header-enricher inbetween those two channels so I can use a wire tap for logging. It works fine.

3. When you forward your error message to the outboundChannel, you must copy the replyChannel header from the failedMessage to your error message. Otherwise the gateway won't be able to correlate the response and it will be dropped. The failedMessage is a property of the ErrorMessage.

I have read this before, and I have an implementation. I do not think I am doing it correctly. Here is the error I see now, is this an indication that the reply channel is not done correctly?

When the TcpInboundGateway shates "null reply recieved for" but shows the proper payload, does this refer to the reply-channel being incorrect?

2012-04-30 12:57:57,138 DEBUG [pool-4-thread-3] (TcpInboundGateway.java:74) - null reply received for [Payload=BAD_TEST java.lang.IllegalStateException: Test of throwing exception.][Headers={timestamp=1335815867088, id=f4a51673-bfbb-4e32-a234-100380270e7a, ip_address=127.0.0.1, ip_connection_seq=1, ip_hostname=localhost, ip_tcp_remote_port=3517, ip_connection_id=localhost:3517:2017f935-f5f5-4b5b-9dd6-58baff6ff2d6}] nothing to send


2012-04-30 12:58:17,082  WARN [pool-4-thread-1] (AbstractConnectionFactory.java:513) - Timing out TcpNioConnection 17001 : localhost:3517:2017f935-f5f5-4b5b-9dd6-58baff6ff2d6

2012-04-30 12:58:47,091 DEBUG [pool-4-thread-1] (AbstractConnectionFactory.java:508) - Removing closed channel

Here is my implementation of the reply-channel coping. I do not see in the logs this being done.

    <int:chain input-channel="errorChannel">
        <!--copy over the original reply channel header-->
        <int:header-enricher>
            <int:reply-channel expression="payload.getFailedMessage().getHeaders().getReplyChannel()" />
        </int:header-enricher>
        <int:service-activator  method="onErrorMessage">
            <bean class="com...DefaultErrorMessageHandler"/>
        </int:service-activator>
    </int:chain>

4. The AggregateMessageDeliveryException is used when a dispatcher sends messages to multiple endpoints and they all fail. Without seeing more of your configuration, it's hard to say why you are seeing this; my guess is you have multiple subscribers on inboundChannel (or rawInboundChannel); we try to dispatch to your error thrower; that fails so we try the next subscriber and, downstream of that subscriber, there is a channel with no subscriber.

I know I have only one subscriber for rawInboundChannel, the header-enricher. My copy paste error was using the inboundChannel (only the router should) instead of the correct routed to channel name.

I think my error now is to configure the reply-channel in the proper place. Should this occur as I have it now with a chain on the "errorChannel", or on the tcp-inbound-gateway property "error-channel="mu-inboundTcpGatewayErrorChannel", or on both channels?

[Gary Russell]

Ok; I see what's happening. I think I misled you.

When we send the ErrorMessage, it gets a new replyChannel header, which is the one that needs to receive the good result; not the reply channel from the original message. The gateway will take care of resolving the reply to the original request. So don't copy the replyChannel header from the failed message.

By far the cleanest approach is to simply remove the output-channel from the gatewayErrorHandler; when there is no output channel, the framework will figure out the correct reply channel to send it to.

You shouldn't use this at all.

        <int:header-enricher>
            <int:reply-channel expression="payload.getFailedMessage().getHeaders().getReplyChannel()" />
        </int:header-enricher>

BTW, if you did need it, you could use...

            <int:reply-channel expression="payload.failedMessage.headers.replyChannel" />

...but you don't.

[Wrangler]

Gary,

Thanks for your assistance. I removed the reply-channel setup, and still did not receive a reply at the client. I looked thru the logs and followed each line, and now have a question.

Since I have header history turned on, the logs show each item the message passes thru from the gateway "request-channel" thru several entities and lastly the gateway "reply-channel". This flow looks correct. I also notices the "replyChannel" header object has the same value for each item in the history.

After the postSend is done on the gateway "reply-channel", I see a log item for a postSend on the gateway "error-channel". I would have thought this channel would have been in the same header history above, instead this history is just the error-channel. I also noticed the "replyChannel" object is different.

I see an indication of why there is no reply to the client. The tcp-inbound-gateway is configured with a deserializer and a serializer. Both work fine on messages when there is no exception.

In the logs with the exception thrown, I see only the deserializer, there are no log entries for the serializer for the response (all it does is a toString() call on the object.

Since this works for normal messages fine, this makes me wonder if the gateway serializer is not called when an exception is thrown (meaning the error-channel) is used.

I attached my IDE to the running Weblogic instance, and verified the serializer is never used.

The client (JMeter) only shows a completion (no response received) because the gateway timeout is reached [ (AbstractConnectionFactory.java:513) - Timing out TcpNioConnection] and the socket is closed [(AbstractConnectionFactory.java:508) - Removing closed channel].

[Gary Russell]

Are you still seeing...

2012-04-30 12:57:57,138 DEBUG [pool-4-thread-3] (TcpInboundGateway.java:74) - null reply received for [Payload=BAD_TEST java.lang.IllegalStateException: Test of throwing exception.][Headers={timestamp=1335815867088, id=f4a51673-bfbb-4e32-a234-100380270e7a, ip_address=127.0.0.1, ip_connection_seq=1, ip_hostname=localhost, ip_tcp_remote_port=3517, ip_connection_id=localhost:3517:2017f935-f5f5-4b5b-9dd6-58baff6ff2d6}] nothing to send

???

The serializer won't be called if we have no data to send.

Please take a look at the tcp-client-server sample; the third test case is doing pretty much the same as you - forces an error that is caught by the inbound gateway, invokes an error flow, and returns a failure indication.

The fact that you can't show me your logs makes it hard for me to help you debug it.

I suggest you run the sample and compare the differences in the logs. The fact that the sample gateway has no explicit reply-channel defined is not material; although you really don't need it unless you have code that explicitly needs it (such as making it a pub-sub channel, or you want to wire tap it).

[Wrangler]

Gary,

I'll take another look at the tcp-client-server sample.

The IllegalStateException is the exception my "errorThrowingServiceActivator" is throwing in order to verify the error handling. The request message makes its way to this service-activator fine.

The thrown exception goes to the "gatewayErrorHandler" which is configured as the gateway error-channel, and adds the exception to the payload, and returns this message.

The entire log is large, and this forum does not allow that many characters. Looks like a forum quick reply does not have an option to attach a file. Here are the log entries from "gatewayErrorHandler" on.

2012-04-30 14:56:20,212 DEBUG [pool-4-thread-3] (AbstractMessageChannel.java:224) - preSend on channel 'outboundChannel', message: [Payload=BAD_TEST00001BAD_TEST0001         0000java.lang.IllegalStateException: Test of throwing exception. properly~][Headers={timestamp=1335822980212, id=44635b65-0d2f-4ac1-91c3-35d241b88aad, history=tcp-inbound-Gateway,rawInboundChannel,enrichWithTransactionId,inboundChannel,copybookNameRouter,inboundBadTestChannel,errorThrowingServiceActivator,outboundChannel, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, ip_address=127.0.0.1, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, ip_hostname=localhost, ip_connection_seq=1, ip_tcp_remote_port=4264, transactionGuid=01675350-6489-43b3-9941-e3ad33d892f5, ip_connection_id=localhost:4264:57f66a96-b13b-4a34-a486-35f0f541e93d}]
2012-04-30 14:56:20,212 DEBUG [pool-4-thread-3] (AbstractMessageChannel.java:224) - preSend on channel 'globalLoggingChannel', message: [Payload=BAD_TEST00001BAD_TEST0001         0000java.lang.IllegalStateException: Test of throwing exception. properly~][Headers={timestamp=1335822980212, id=53d1b519-0648-4f07-aebf-305c6e63d8a7, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, history=tcp-inbound-Gateway,rawInboundChannel,enrichWithTransactionId,inboundChannel,copybookNameRouter,inboundBadTestChannel,errorThrowingServiceActivator,outboundChannel,globalLoggingChannel, ip_address=127.0.0.1, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, ip_connection_seq=1, ip_hostname=localhost, ip_tcp_remote_port=4264, transactionGuid=01675350-6489-43b3-9941-e3ad33d892f5, ip_connection_id=localhost:4264:57f66a96-b13b-4a34-a486-35f0f541e93d}]
2012-04-30 14:56:20,212 DEBUG [pool-4-thread-3] (AbstractMessageHandler.java:67) - org.springframework.integration.handler.LoggingHandler#0 received message: [Payload=BAD_TEST00001BAD_TEST0001         0000java.lang.IllegalStateException: Test of throwing exception. properly~][Headers={timestamp=1335822980212, id=53d1b519-0648-4f07-aebf-305c6e63d8a7, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, history=tcp-inbound-Gateway,rawInboundChannel,enrichWithTransactionId,inboundChannel,copybookNameRouter,inboundBadTestChannel,errorThrowingServiceActivator,outboundChannel,globalLoggingChannel, ip_address=127.0.0.1, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, ip_connection_seq=1, ip_hostname=localhost, ip_tcp_remote_port=4264, transactionGuid=01675350-6489-43b3-9941-e3ad33d892f5, ip_connection_id=localhost:4264:57f66a96-b13b-4a34-a486-35f0f541e93d}]
2012-04-30 14:56:20,212 DEBUG [pool-4-thread-3] (LoggingHandler.java:141) - [Payload=BAD_TEST00001BAD_TEST0001         0000java.lang.IllegalStateException: Test of throwing exception. properly~][Headers={timestamp=1335822980212, id=90f6605c-7732-4e17-bde1-1d798776c93c, history=tcp-inbound-Gateway,rawInboundChannel,enrichWithTransactionId,inboundChannel,copybookNameRouter,inboundBadTestChannel,errorThrowingServiceActivator,outboundChannel,globalLoggingChannel,global-logging-channel-adapter, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, ip_address=127.0.0.1, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, ip_hostname=localhost, ip_connection_seq=1, ip_tcp_remote_port=4264, transactionGuid=01675350-6489-43b3-9941-e3ad33d892f5, ip_connection_id=localhost:4264:57f66a96-b13b-4a34-a486-35f0f541e93d}]
2012-04-30 14:56:20,212 DEBUG [pool-4-thread-3] (AbstractMessageChannel.java:237) - postSend (sent=true) on channel 'globalLoggingChannel', message: [Payload=BAD_TEST00001BAD_TEST0001         0000java.lang.IllegalStateException: Test of throwing exception. properly~][Headers={timestamp=1335822980212, id=53d1b519-0648-4f07-aebf-305c6e63d8a7, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, history=tcp-inbound-Gateway,rawInboundChannel,enrichWithTransactionId,inboundChannel,copybookNameRouter,inboundBadTestChannel,errorThrowingServiceActivator,outboundChannel,globalLoggingChannel, ip_address=127.0.0.1, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, ip_connection_seq=1, ip_hostname=localhost, ip_tcp_remote_port=4264, transactionGuid=01675350-6489-43b3-9941-e3ad33d892f5, ip_connection_id=localhost:4264:57f66a96-b13b-4a34-a486-35f0f541e93d}]
2012-04-30 14:56:20,212 DEBUG [pool-4-thread-3] (AbstractMessageHandler.java:67) - org.springframework.integration.handler.BridgeHandler@6cf5e8 received message: [Payload=BAD_TEST00001BAD_TEST0001         0000java.lang.IllegalStateException: Test of throwing exception. properly~][Headers={timestamp=1335822980212, id=44635b65-0d2f-4ac1-91c3-35d241b88aad, history=tcp-inbound-Gateway,rawInboundChannel,enrichWithTransactionId,inboundChannel,copybookNameRouter,inboundBadTestChannel,errorThrowingServiceActivator,outboundChannel, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, ip_address=127.0.0.1, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, ip_hostname=localhost, ip_connection_seq=1, ip_tcp_remote_port=4264, transactionGuid=01675350-6489-43b3-9941-e3ad33d892f5, ip_connection_id=localhost:4264:57f66a96-b13b-4a34-a486-35f0f541e93d}]
2012-04-30 14:56:20,212 DEBUG [pool-4-thread-3] (AbstractReplyProducingMessageHandler.java:156) - handler 'org.springframework.integration.handler.BridgeHandler@6cf5e8' sending reply Message: [Payload=BAD_TEST00001BAD_TEST0001         0000java.lang.IllegalStateException: Test of throwing exception. properly~][Headers={timestamp=1335822980212, id=44635b65-0d2f-4ac1-91c3-35d241b88aad, history=tcp-inbound-Gateway,rawInboundChannel,enrichWithTransactionId,inboundChannel,copybookNameRouter,inboundBadTestChannel,errorThrowingServiceActivator,outboundChannel, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, ip_address=127.0.0.1, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, ip_hostname=localhost, ip_connection_seq=1, ip_tcp_remote_port=4264, transactionGuid=01675350-6489-43b3-9941-e3ad33d892f5, ip_connection_id=localhost:4264:57f66a96-b13b-4a34-a486-35f0f541e93d}]
2012-04-30 14:56:20,212 DEBUG [pool-4-thread-3] (AbstractMessageChannel.java:237) - postSend (sent=true) on channel 'outboundChannel', message: [Payload=BAD_TEST00001BAD_TEST0001         0000java.lang.IllegalStateException: Test of throwing exception. properly~][Headers={timestamp=1335822980212, id=44635b65-0d2f-4ac1-91c3-35d241b88aad, history=tcp-inbound-Gateway,rawInboundChannel,enrichWithTransactionId,inboundChannel,copybookNameRouter,inboundBadTestChannel,errorThrowingServiceActivator,outboundChannel, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, ip_address=127.0.0.1, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@86ae77, ip_hostname=localhost, ip_connection_seq=1, ip_tcp_remote_port=4264, transactionGuid=01675350-6489-43b3-9941-e3ad33d892f5, ip_connection_id=localhost:4264:57f66a96-b13b-4a34-a486-35f0f541e93d}]
2012-04-30 14:56:20,212 DEBUG [pool-4-thread-3] (AbstractMessageChannel.java:237) - postSend (sent=true) on channel 'inboundTcpGatewayErrorChannel', message: [Payload=org.springframework.integration.MessageHandlingException: java.lang.IllegalStateException: Test of throwing exception. properly][Headers={timestamp=1335822980212, id=b1eca83a-e1cb-427b-b449-113e98334c35, history=inboundTcpGatewayErrorChannel, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@7b3555, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@7b3555}]
2012-04-30 14:56:30,210 DEBUG [pool-4-thread-3] (TcpInboundGateway.java:74) - null reply received for [Payload=BAD_TEST00001BAD_TEST0001         0000java.lang.IllegalStateException: Test of throwing exception. properly~][Headers={timestamp=1335822980149, id=2d8837a8-0e2b-4a65-92a6-82a5d5f5436b, ip_address=127.0.0.1, ip_connection_seq=1, ip_hostname=localhost, ip_tcp_remote_port=4264, ip_connection_id=localhost:4264:57f66a96-b13b-4a34-a486-35f0f541e93d}] nothing to send
2012-04-30 14:56:50,145  WARN [pool-4-thread-1] (AbstractConnectionFactory.java:513) - Timing out TcpNioConnection 17001 : localhost:4264:57f66a96-b13b-4a34-a486-35f0f541e93d
2012-04-30 14:57:20,141 DEBUG [pool-4-thread-1] (AbstractConnectionFactory.java:508) - Removing closed channel

[Gary Russell]

What does your DefaultErrorMessageHandler do?

I see the input message to the error flow; payload ErrorMessage, replyChannel header object @7b3555; it is this object that the gateway is expecting to receive a reply on.

However, the message being sent to the outboundChannel has a different reply channel header (@86ae77 - presumably the one from the original message, but I can't see that in the snippet you posted); so it looks like there is still some manipulation of the replyChannel header going on. The bridge handler sends the message to that channel and nothing is received on the @7b3555 channel.

[Wrangler]

I'll attach the code of the DefaultErrorMessageHandler. The concept of the reply channel in the header is something I am still learning about. I noticed the two different reply channel objects, but I do not yet understand why there are two, or which one needs to go back.

Since the flow seems to be good until this code, please take a look at this code, and let me know if you see a problem with it.

public class DefaultErrorMessageHandler {
    private final Logger logger = Logger.getLogger(getClass());

    public Message<MUMessage> onErrorMessage(ErrorMessage errorMessage) {
        MessagingException messagingException = (MessagingException) errorMessage.getPayload();
        String exceptionMessage = messagingException.getMessage();

        Message<MUMessage> failedMuMessage = (Message<MUMessage>) messagingException.getFailedMessage();
        if (failedMuMessage == null) {
            MUMessage muMessage = new MUMessage();
            try {
                muMessage.setContent(formatResponse(null, exceptionMessage));
            } catch (InvalidMessageFormatException e) {
                logger.error("Unable to build new MUMessage from content. " + errorMessage);
            }
            failedMuMessage = MessageBuilder.withPayload(muMessage).build();
        }


        Message<MUMessage> siMessageWithMUMessage = MessageBuilder.fromMessage(failedMuMessage).build();
        try {
            String payloadContent = formatResponse(siMessageWithMUMessage.getPayload(), exceptionMessage);
            siMessageWithMUMessage.getPayload().setContent(payloadContent);
            //logger.debug("replyChannel: " + failedMuMessage.getHeaders().getReplyChannel());
            return siMessageWithMUMessage;
        } catch (InvalidMessageFormatException e) {
            logger.error("Could not write the body of a payload to let the sender know the message was unroutable: " + errorMessage);
            throw new IllegalStateException("Could not write the body of a payload to let the sender know the message was unroutable");
        }
    }

    /**
     * We need to append the delimiter character so that the calling system can process the delimiter
     */
    private String formatResponse(MUMessage originalMessage, String exceptionMessage) {
        char deliminter = '~';

        if (originalMessage != null) {
            deliminter = originalMessage.getDelimiter();
        }
        return exceptionMessage + deliminter;
    }
}

[Gary Russell]

Here's the smoking gun....

Message<MUMessage> siMessageWithMUMessage = MessageBuilder.fromMessage(failedMuMessage).build();

You are building the error reply message from the failed message (which has the wrong replyChannel header).

You could use

Message<MUMessage> siMessageWithMUMessage = MessageBuilder.fromMessage(failedMuMessage)
                     setReplyChannel(errorMessage.getHeaders().getReplyChannel()).build();

However, you really don't need all that messaging infrastructure in your handler; just have it return the payload you want, and the framework will handle the headers properly for you...

    public MUMessage onErrorMessage(ErrorMessage errorMessage) {
        ...
        return <MUMessage with error content>
    }