[Spring AMQP] CorrelationId mismatch when using RPC

[kswat]

Community Support Policy

• I have read RabbitMQ's Community Support Policy
• I agree to provide all relevant information (versions, logs, rabbitmq-diagnostics output, detailed reproduction steps)

RabbitMQ version used

4.0.3

Erlang version used

26.2.x

Operating system (distribution) used

Windows

How is RabbitMQ deployed?

Community Docker image

rabbitmq-diagnostics status output

See https://www.rabbitmq.com/docs/cli to learn how to use rabbitmq-diagnostics

[Client] Sending request with CorrelationId: bb6a3029-2239-46df-9812-197b58afbe32
(Body:'5' MessageProperties [headers={}, correlationId=bb6a3029-2239-46df-9812-197b58afbe32, contentType=text/plain, contentEncoding=UTF-8, contentLength=1, deliveryMode=PERSISTENT, priority=0, deliveryTag=0])
[Client] Response is null!

Server/processor logs (profile)

(Body:'5' MessageProperties [headers={}, correlationId=1, replyTo=rpc.reply.queue, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=rpc.topic.exchange, receivedRoutingKey=rpc.request.key, deliveryTag=1, consumerTag=amq.ctag-wmyqvfNlOvavftea7xp14g, consumerQueue=rpc.request.queue])
[Server] Received request: 5
[Server] Computed result: 5
correlationId : 1

Logs from node 1 (with sensitive values edited out)

See https://www.rabbitmq.com/docs/logging to learn how to collect logs

[Client] Sending request with CorrelationId: bb6a3029-2239-46df-9812-197b58afbe32
(Body:'5' MessageProperties [headers={}, correlationId=bb6a3029-2239-46df-9812-197b58afbe32, contentType=text/plain, contentEncoding=UTF-8, contentLength=1, deliveryMode=PERSISTENT, priority=0, deliveryTag=0])
[Client] Response is null!

Logs from node 2 (if applicable, with sensitive values edited out)

See https://www.rabbitmq.com/docs/logging to learn how to collect logs

(Body:'5' MessageProperties [headers={}, correlationId=1, replyTo=rpc.reply.queue, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=rpc.topic.exchange, receivedRoutingKey=rpc.request.key, deliveryTag=1, consumerTag=amq.ctag-wmyqvfNlOvavftea7xp14g, consumerQueue=rpc.request.queue])
[Server] Received request: 5
[Server] Computed result: 5
correlationId : 1

rabbitmq.conf

See https://www.rabbitmq.com/docs/configure#config-location to learn how to find rabbitmq.conf file location

package com.example;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;

@Configuration
public class Config {

    public static final String RPC_EXCHANGE = "rpc.topic.exchange";
    public static final String RPC_REQUEST_QUEUE = "rpc.request.queue";
    public static final String RPC_REPLY_QUEUE = "rpc.reply.queue";

    @Bean
    public TopicExchange rpcExchange() {
        return new TopicExchange(RPC_EXCHANGE);
    }

    @Bean
    public Queue requestQueue() {
        return new Queue(RPC_REQUEST_QUEUE);
    }

    @Bean
    public Queue replyQueue() {
        return new Queue(RPC_REPLY_QUEUE);
    }

    @Bean
//    @Profile("client")
    public Binding requestBinding(Queue requestQueue, TopicExchange rpcExchange) {
        return BindingBuilder.bind(requestQueue).to(rpcExchange).with("rpc.request.#");
    }

    @Profile("client")
    private static class ServerConfig {
        @Bean
        RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
            RabbitTemplate template = new RabbitTemplate(connectionFactory);
            template.setReplyAddress(RPC_REPLY_QUEUE);
            template.setReplyTimeout(6000);
            template.setMessageConverter(new SimpleMessageConverter()); //Change 1 for NumberFormatException
            return template;
        }

        @Bean
        SimpleMessageListenerContainer replyContainer(ConnectionFactory connectionFactory, RabbitTemplate rabbitTemplate) {
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
            container.setConnectionFactory(connectionFactory);
            container.setQueueNames(RPC_REPLY_QUEUE);
            container.setMessageListener(rabbitTemplate);
            return container;
        }
    }
}

Steps to deploy RabbitMQ cluster

docker-compose up -d

Steps to reproduce the behavior in question

https://github.com/kswat/rabbit-topic-rpc/tree/master

Clone
Build mvn clean package

Spin docker :
docker-compose up -d

Run using profiles
-Dspring.profiles.active=client

and

-Dspring.profiles.active=server

Clean up:
stop applications

docker-compose down

advanced.config

See https://www.rabbitmq.com/docs/configure#config-location to learn how to find advanced.config file location

# PASTE advanced.config HERE, BETWEEN BACKTICKS

Application code

https://github.com/kswat/rabbit-topic-rpc/tree/master

Kubernetes deployment file

# Relevant parts of K8S deployment that demonstrate how RabbitMQ is deployed
# PASTE YAML HERE, BETWEEN BACKTICKS

What problem are you trying to solve?

Spring Boot - RPC CorrelationId with fixed reply queue. TopicExchange

[michaelklishin]

RabbitMQ does not generate or modify the correlation_id of basic.publish content properties. It is provided by applications and passed along to other applications.

This belongs to a Spring AMQP forum, and please provide an executable example instead of a single short sweet line ("RPC CorrelationId with fixed reply queue. TopicExchange").

[kswat]

The source is provided . github link included

[michaelklishin]

Like I said, RabbitMQ does not generate or modify correlation_id, what you are looking at is a behavior of your app or Spring AMQP. We maintain RabbitMQ and its Java client, not Spring AMQP.