Issues migrating a Spring AMQP consumer/producer service to a Spring Stream source

Question

I am migrating a Spring Boot microservice that consumes data from 3 RabbitMQ queues on server A, saves it into Redis and finally produces messages into an exchange in a different RabbitMQ on server B so these messages can be consumed by another microservice. This flow is working fine but I would like to migrate it to Spring Cloud Stream using the RabbitMQ binder. All Spring AMQP configuration is customised in the properties file and no spring property is used to create connections, queues, bindings, etc...

My first idea was setting up two bindings in Spring Cloud Stream, one connected to server A (consumer) and the other connected to server B (producer), and migrate the existing code to a Processor but I discarded it because it seems connection names cannot be set yet if multiple binders are used and I need to add several bindings to consume from server A's queues and bindingRoutingKey property does not support a list of values (I know it can be done programmately as explained here).

So I decided to only refactor the part of code related to the producer to use Spring Cloud Stream over RabbitMQ so the same microservice should consume via Spring AMQP from server A (original code) and should produce into server B via Spring Cloud Stream.

The first issue I found was a NonUniqueBeanDefinitionException in Spring Cloud Stream because a org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory bean was defined twice with handlerMethodFactory and integrationMessageHandlerMethodFactory names.

org.springframework.beans.factory.NoUniqueBeanDefinitionException: No qualifying bean of type 'org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory' available: expected single matching bean but found 2: handlerMethodFactory,integrationMessageHandlerMethodFactory
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.resolveNamedBean(DefaultListableBeanFactory.java:1144)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.resolveBean(DefaultListableBeanFactory.java:411)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.getBean(DefaultListableBeanFactory.java:344)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.getBean(DefaultListableBeanFactory.java:337)
    at org.springframework.context.support.AbstractApplicationContext.getBean(AbstractApplicationContext.java:1123)
    at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.injectAndPostProcessDependencies(StreamListenerAnnotationBeanPostProcessor.java:317)
    at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.afterSingletonsInstantiated(StreamListenerAnnotationBeanPostProcessor.java:113)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:862)
    at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:877)
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:549)
    at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:141)
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:743)
    at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:390)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:312)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1214)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1203)

It seems the former bean is created by Spring AMQP and the latter by Spring Cloud Stream so I created my own primary bean:

@Bean
@Primary
public MessageHandlerMethodFactory messageHandlerMethodFactory() {
    return new DefaultMessageHandlerMethodFactory();
}

Now the application is able to start but the output channel is created by Spring Cloud Stream in server A instead of server B. It seems that Spring Cloud Stream configuration is using the connection created by Spring AMQP instead of using its own configuration.

The configuration of Spring AMQP is this:

@Bean
public SimpleRabbitListenerContainerFactory priceRabbitListenerContainerFactory(
 ConnectionFactory consumerConnectionFactory) {
  return
    getSimpleRabbitListenerContainerFactory(
      consumerConnectionFactory,
      rabbitProperties.getConsumer().getListeners().get(LISTENER_A));
}

@Bean
public SimpleRabbitListenerContainerFactory maxbetRabbitListenerContainerFactory(
  ConnectionFactory consumerConnectionFactory) {
    return
      getSimpleRabbitListenerContainerFactory(
        consumerConnectionFactory,
       rabbitProperties.getConsumer().getListeners().get(LISTENER_B));
}

@Bean
public ConnectionFactory consumerConnectionFactory() throws Exception {
  return
    new CachingConnectionFactory(
      getRabbitConnectionFactoryBean(
        rabbitProperties.getConsumer()
      ).getObject()
    );
}

private SimpleRabbitListenerContainerFactory getSimpleRabbitListenerContainerFactory(
  ConnectionFactory connectionFactory,
  RabbitProperties.ListenerProperties listenerProperties) {
    //return a SimpleRabbitListenerContainerFactory set up from external properties
}

/**
 * Create the AMQ Admin.
 */
@Bean
public AmqpAdmin consumerAmqpAdmin(ConnectionFactory consumerConnectionFactory) {
  return new RabbitAdmin(consumerConnectionFactory);
}

/**
 * Create the map of available queues and declare them in the admin.
 */
@Bean
public Map<String, Queue> queues(AmqpAdmin consumerAmqpAdmin) {
  return
    rabbitProperties.getConsumer().getListeners().entrySet().stream()
      .map(listenerEntry -> {
        Queue queue =
          QueueBuilder
            .nonDurable(listenerEntry.getValue().getQueueName())
            .autoDelete()
            .build();

          consumerAmqpAdmin.declareQueue(queue);

          return new AbstractMap.SimpleEntry<>(listenerEntry.getKey(), queue);
      }).collect(
        Collectors.toMap(
          AbstractMap.SimpleEntry::getKey,
          AbstractMap.SimpleEntry::getValue
        )
      );
}

/**
 * Create the map of available exchanges and declare them in the admin.
 */
@Bean
public Map<String, TopicExchange> exchanges(AmqpAdmin consumerAmqpAdmin) {
  return
    rabbitProperties.getConsumer().getListeners().entrySet().stream()
      .map(listenerEntry -> {
        TopicExchange exchange =
          new TopicExchange(listenerEntry.getValue().getExchangeName());

        consumerAmqpAdmin.declareExchange(exchange);

        return new AbstractMap.SimpleEntry<>(listenerEntry.getKey(), exchange);
      }).collect(
        Collectors.toMap(
          AbstractMap.SimpleEntry::getKey,
          AbstractMap.SimpleEntry::getValue
        )
      );
}

/**
 * Create the list of bindings and declare them in the admin.
 */
@Bean
public List<Binding> bindings(Map<String, Queue> queues, Map<String, TopicExchange> exchanges, AmqpAdmin consumerAmqpAdmin) {
  return
    rabbitProperties.getConsumer().getListeners().keySet().stream()
      .map(listenerName -> {
        Queue queue = queues.get(listenerName);
        TopicExchange exchange = exchanges.get(listenerName);

        return
          rabbitProperties.getConsumer().getListeners().get(listenerName).getKeys().stream()
            .map(bindingKey -> {
              Binding binding = BindingBuilder.bind(queue).to(exchange).with(bindingKey);

              consumerAmqpAdmin.declareBinding(binding);

              return binding;
            }).collect(Collectors.toList());
      }).flatMap(Collection::stream)
      .collect(Collectors.toList());
}

Message listeners are:

@RabbitListener(
  queues="${consumer.listeners.LISTENER_A.queue-name}",
  containerFactory = "priceRabbitListenerContainerFactory"
)
public void handleMessage(Message rawMessage, org.springframework.messaging.Message<ModelPayload> message) {
   // call a service to process the message payload
}

@RabbitListener(
  queues="${consumer.listeners.LISTENER_B.queue-name}",
  containerFactory = "maxbetRabbitListenerContainerFactory"
)
public void handleMessage(Message rawMessage, org.springframework.messaging.Message<ModelPayload> message) {
  // call a service to process the message payload
}

Properties:

#
# Server A config (Spring AMQP)
#
consumer.host=server-a
consumer.username=
consumer.password=
consumer.port=5671
consumer.ssl.enabled=true
consumer.ssl.algorithm=TLSv1.2
consumer.ssl.validate-server-certificate=false
consumer.connection-name=local:microservice-1
consumer.thread-factory.thread-group-name=server-a-consumer
consumer.thread-factory.thread-name-prefix=server-a-consumer-
# LISTENER_A configuration
consumer.listeners.LISTENER_A.queue-name=local.listenerA
consumer.listeners.LISTENER_A.exchange-name=exchangeA
consumer.listeners.LISTENER_A.keys[0]=*.1.*.*
consumer.listeners.LISTENER_A.keys[1]=*.3.*.*
consumer.listeners.LISTENER_A.keys[2]=*.6.*.*
consumer.listeners.LISTENER_A.keys[3]=*.8.*.*
consumer.listeners.LISTENER_A.keys[4]=*.9.*.*
consumer.listeners.LISTENER_A.initial-concurrency=5
consumer.listeners.LISTENER_A.maximum-concurrency=20
consumer.listeners.LISTENER_A.thread-name-prefix=listenerA-consumer-
# LISTENER_B configuration
consumer.listeners.LISTENER_B.queue-name=local.listenerB
consumer.listeners.LISTENER_B.exchange-name=exchangeB
consumer.listeners.LISTENER_B.keys[0]=*.1.*
consumer.listeners.LISTENER_B.keys[1]=*.3.*
consumer.listeners.LISTENER_B.keys[2]=*.6.*
consumer.listeners.LISTENER_B.initial-concurrency=5
consumer.listeners.LISTENER_B.maximum-concurrency=20
consumer.listeners.LISTENER_B.thread-name-prefix=listenerB-consumer-

#
# Server B config (Spring Cloud Stream)
#
spring.rabbitmq.host=server-b
spring.rabbitmq.port=5672
spring.rabbitmq.username=
spring.rabbitmq.password=

spring.cloud.stream.bindings.outbound.destination=microservice-out
spring.cloud.stream.bindings.outbound.group=default

spring.cloud.stream.rabbit.binder.connection-name-prefix=local:microservice

So my question is: is it possible to use in the same Spring Boot application code that consumes data from RabbitMQ via Spring AMQP and produces messages into a different server via Spring Cloud Stream RabbitMQ? If it is, could somebody tell me what I am doing wrong, please?

Spring AMQP version is the one provided by Boot version 2.1.7 (2.1.8-RELEASE) and Spring Cloud Stream version is the one provided by Spring Cloud train Greenwich.SR2 (2.1.3.RELEASE).

EDIT

I was able to make it work configuring the binder via multiple configuration properties instead of the default one. So with this configuration it works:

#
# Server B config (Spring Cloud Stream)
#
spring.cloud.stream.binders.transport-layer.type=rabbit
spring.cloud.stream.binders.transport-layer.environment.spring.rabbitmq.host=server-b
spring.cloud.stream.binders.transport-layer.environment.spring.rabbitmq.port=5672
spring.cloud.stream.binders.transport-layer.environment.spring.rabbitmq.username=
spring.cloud.stream.binders.transport-layer.environment.spring.rabbitmq.password=

spring.cloud.stream.bindings.stream-output.destination=microservice-out
spring.cloud.stream.bindings.stream-output.group=default

Unfortunately it is not possible to set the connection-name yet in multiple binders configuration: A custom ConnectionNameStrategy is ignored if there is a custom binder configuration.

Anyway, I still do not understand why it seems the contexts are "mixed" when using Spring AMQP and Spring Cloud Stream RabbitMQ. It is still necessary to set a primary MessageHandlerMethodFactory bean in order the implementation to work.

EDIT

I found out that the NoUniqueBeanDefinitionException was caused because the microservice itself was creating a ConditionalGenericConverter to be used by Spring AMQP part to deserialize messages from Server A.

I removed it and added some MessageConverters instead. Now the problem is solved and the @Primary bean is no longer necessary.

Gary Russell Gary Russell · Accepted Answer · 2019-08-30T13:23:47

Unrelated, but

consumerAmqpAdmin.declareQueue(queue);

You should never communicate with the broker within a @Bean definition; it is too early in application context lifecycle. It might work but YMMV; also if the broker is not available it will prevent your app from starting.

It's better to define beans of type Declarables containing the lists of queues, channels, bindings and the Admin will automatically declare them when the connection is first opened successfully. See the reference manual.

I have never seen the MessageHandlerFactory problem; Spring AMQP declares no such bean. If you can provide a small sample app that exhibits the behavior, that would be useful.
I'll see if I can find a work around to the connection name issue.

EDIT

I found a work around to the connection name issue; it involves a bit of reflection but it works. I suggest you open a new feature request against the binder to request a mechanism to set the connection name strategy when using multiple binders.

Anyway; here's the work around...

@SpringBootApplication
@EnableBinding(Processor.class)
public class So57725710Application {

    public static void main(String[] args) {
        SpringApplication.run(So57725710Application.class, args);
    }

    @Bean
    public Object connectionNameConfigurer(BinderFactory binderFactory) throws Exception {
        setConnectionName(binderFactory, "rabbit1", "myAppProducerSide");
        setConnectionName(binderFactory, "rabbit2", "myAppConsumerSide");
        return null;
    }

    private void setConnectionName(BinderFactory binderFactory, String binderName,
            String conName) throws Exception {

        binderFactory.getBinder(binderName, MessageChannel.class); // force creation
        @SuppressWarnings("unchecked")
        Map<String, Map.Entry<Binder<?, ?, ?>, ApplicationContext>> binders =
                (Map<String, Entry<Binder<?, ?, ?>, ApplicationContext>>) new DirectFieldAccessor(binderFactory)
                    .getPropertyValue("binderInstanceCache");
        binders.get(binderName)
                .getValue()
                .getBean(CachingConnectionFactory.class).setConnectionNameStrategy(queue -> conName);
    }

    @StreamListener(Processor.INPUT)
    @SendTo(Processor.OUTPUT)
    public String listen(String in) {
        System.out.println(in);
        return in.toUpperCase();
    }

}

and

spring.cloud.stream.binders.rabbit1.type=rabbit
spring.cloud.stream.binders.rabbit1.environment.spring.rabbitmq.host=localhost
spring.cloud.stream.binders.rabbit1.environment.spring.rabbitmq.port=5672
spring.cloud.stream.binders.rabbit1.environment.spring.rabbitmq.username=guest
spring.cloud.stream.binders.rabbit1.environment.spring.rabbitmq.password=guest

spring.cloud.stream.bindings.output.destination=outDest
spring.cloud.stream.bindings.output.producer.required-groups=outQueue
spring.cloud.stream.bindings.output.binder=rabbit1

spring.cloud.stream.binders.rabbit2.type=rabbit
spring.cloud.stream.binders.rabbit2.environment.spring.rabbitmq.host=localhost
spring.cloud.stream.binders.rabbit2.environment.spring.rabbitmq.port=5672
spring.cloud.stream.binders.rabbit2.environment.spring.rabbitmq.username=guest
spring.cloud.stream.binders.rabbit2.environment.spring.rabbitmq.password=guest

spring.cloud.stream.bindings.input.destination=inDest
spring.cloud.stream.bindings.input.group=default
spring.cloud.stream.bindings.input.binder=rabbit2

and

Issues migrating a Spring AMQP consumer/producer service to a Spring Stream source

1 Answers