The AmqpRpcServer looks unconditional for the handler errors #163
Description
Describe the bug
The logic there is like:
.messageHandler(
(ctx, msg) -> {
ctx.accept();
Message reply = handler.handle(context, msg);
if (reply != null && msg.replyTo() != null) {
reply.to(msg.replyTo());
}
Object correlationId = correlationIdExtractor.apply(msg);
reply = replyPostProcessor.apply(reply, correlationId);
if (reply != null) {
sendReply(reply);
}
})
So, ctx.accept() even before pushing message down to the end-user handler.
However there is no guarantee that message is going to be processed properly there.
Right, there could be some error handling logic in there, but who knows if that is really OK for target application to have message as always accepted. The accept before error might just lead to data loss.
Reproduction steps
Just throw an exception from the RPC handler implementation, and see how message is lost!
Expected behavior
The handler on RpcServer's consumer should try-catch target handler errors and (at least) discard such a message for future consideration in the DLQ.
Additional context
I have a logic like this:
S reply;
try {
reply = callback.handle((R) receive);
}
catch (ClassCastException ex) {
StackTraceElement[] trace = ex.getStackTrace();
if (trace[0].getMethodName().equals("handle")
&& Objects.equals(trace[1].getFileName(), "RabbitAmqpTemplate.java")) {
throw new IllegalArgumentException("ReceiveAndReplyCallback '" + callback
+ "' can't handle received object '" + receive + "'", ex);
}
else {
throw ex;
}
}
So, that ClassCastException happens even before(!) end-user code.