Execution of Rabbit message listener failed, and no ErrorHandler has been set. Failed to invoke target method with argument type = [class [B],

Question

I am using spring amqp rabbitmq, and sending messages using

Message message = MessageBuilder
.withBody(item.toString().getBytes())
.setReplyTo("importReply")
.setCorrelationId(item.toString().getBytes()).build();

My message handler is

public class Foundation {   
    public Message importExchange(Message exchange) {
        System.out.println("Command:" + exchange.getBody());        
                Message message = MessageBuilder
                .withBody(exchange.getBody().toString().getBytes()).setCorrelationId(exchange.getMessageProperties().getCorrelationId()                                  .toString().getBytes()).build();

        return message; 
    }
}

I have hooked it using

<rabbit:listener-container
    connection-factory="rabbitConnectionFactory" concurrency="10">
    <rabbit:listener queues="${rabbitmq.import.queue}"
        ref="foundation" method="importExchange" />
    <rabbit:listener queues="${rabbitmq.import.reply.queue}"
        ref="importExchangeItemWriter" method="replyHandler" />
</rabbit:listener-container>

But I am getting below execption

Execution of Rabbit message listener failed, and no ErrorHandler has been set.
org.springframework.amqp.rabbit.listener.ListenerExecutionFailedException: Failed to invoke target method 'importExchange' with argument type = [class [B], value = [{[B@427829d8}]
    at org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter.invokeListenerMethod(MessageListenerAdapter.java:483)
    at org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter.onMessage(MessageListenerAdapter.java:374)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:647)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:573)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$001(SimpleMessageListenerContainer.java:75)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$1.invokeListener(SimpleMessageListenerContainer.java:154)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1111)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:556)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:904)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:888)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$500(SimpleMessageListenerContainer.java:75)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:989)
    at java.lang.Thread.run(Thread.java:744)
Caused by: java.lang.NoSuchMethodException: com.stockopedia.symfony.Foundation.importExchange([B)
    at java.lang.Class.getMethod(Class.java:1665)
    at org.springframework.util.MethodInvoker.prepare(MethodInvoker.java:178)
    at org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter.invokeListenerMethod(MessageListenerAdapter.java:466)
    ... 12 more

and the similar issue is also on producer side replyHandler

public void replyHandler(Message message) {
    System.out.println("In Reply Handler:" + message.getMessageProperties().getCorrelationId());

}

Also, how can I get exception in replyHandler if there is any exception in importExchange ?

Solution 1

Since you are using POJO for message listening you can't get deal with messages.

Your Foundation#importExchange should accept message body (in your case byte[]) and return something appropriate to be for reply message body.

The replyHandler just has to implement MessageListener.

The Framework will do for you the correlation stuff.

Solution 2

Or you can add .setContentType="text/plain" and the converter will be able to convert to String for you.

Comments

[vishal]

Just updated importExchange, I want to send message object back to replyHandler to check correlation id

[Artem Bilan]

Yes, I saw. Just try to follow with my advice

[vishal]

I want to do correlation as the use case is, I want to send all messages first asynchronously without waiting for reply and then wait for replies. I would like my replyHandler to trigger with reply messages, I should know for which message I got reply.

[Artem Bilan]

You do everything correct with correlation and replies. Your issue here is around POJOs. Just make replyHandler as MessageListener and you'll receive reply messages correctly. But your importExchange should not get deal with Message objects.

[vishal]

One small question, Also, how can I get exception in replyHandler if there is any exception in importExchange ?

[Artem Bilan]

Bad thought. You are in the asynchronous messaging: the producer just sends message to the queue and it doesn't worry and doesn't know anything about the consumer. You can hanle exceptions on consumer and send them as messages to another queue.

[vishal]

yeah may be, my producer is my spring batch item writer, to notify spring batch about any failure to stop job, item writer should throw exception.

[vishal]

yeah did same thing. Thanks

[vishal]

after queuing all messages, I am receiving reply in replyHandler, How can I keep my producer waiting after sending all messages and get notified by replyHandler upon getting all replies to release waiting ?

[vishal]

posted separete question here, stackoverflow.com/questions/23294934/…