Spring-boot – How to create queue in rabbitmq from spring-boot with rabbitmq docker

dockerrabbitmqspring-amqpspring-bootspring-rabbit

I've lost an entire day to trying to get the spring-amqp example project up and running against a docker version of rabbitmq. I'm just running the standard rabbitmq docker. While I have no trouble connecting, I always get exceptions related to creating a queue, and I have tried every possible variant, at this point.

I've tried to just declare the queues in my configuration like the sample project. I've tried explicitly configuring a RabbitAdmin. I've tried explicitly configuring the whole autoconfigured mess. I've created a new user in rabbitmq and explicitly assigned permissions. I've tried just using queue names without actually declaring queue beans. I've tried injecting queues and RabbitAdmin into other beans, just to force them to be created.

No matter what I attempt, I get the following errors:

2017-04-24 17:42:19.709  WARN 37360 --- [cTaskExecutor-1] o.s.a.r.listener.BlockingQueueConsumer   : Failed to declare queue:"incoming"
2017-04-24 17:42:19.715  WARN 37360 --- [cTaskExecutor-1] o.s.a.r.listener.BlockingQueueConsumer   : Queue declaration failed; retries left=3

org.springframework.amqp.rabbit.listener.BlockingQueueConsumer$DeclarationException: Failed to declare queue(s):["incoming"]
    at org.springframework.amqp.rabbit.listener.BlockingQueueConsumer.attemptPassiveDeclarations(BlockingQueueConsumer.java:621) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.BlockingQueueConsumer.start(BlockingQueueConsumer.java:520) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1382) [spring-rabbit-1.7.1.RELEASE.jar:na]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_121]
Caused by: java.io.IOException: null
    at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:105) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:101) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:123) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.ChannelN.queueDeclarePassive(ChannelN.java:992) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.ChannelN.queueDeclarePassive(ChannelN.java:50) ~[amqp-client-4.0.2.jar:4.0.2]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_121]
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_121]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_121]
    at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_121]
    at org.springframework.amqp.rabbit.connection.CachingConnectionFactory$CachedChannelInvocationHandler.invoke(CachingConnectionFactory.java:955) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at com.sun.proxy.$Proxy77.queueDeclarePassive(Unknown Source) ~[na:na]
    at org.springframework.amqp.rabbit.listener.BlockingQueueConsumer.attemptPassiveDeclarations(BlockingQueueConsumer.java:600) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    ... 3 common frames omitted
Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no queue '"incoming"' in vhost '/', class-id=50, method-id=10)
    at com.rabbitmq.utility.ValueOrException.getValue(ValueOrException.java:66) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue(BlockingValueOrException.java:32) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply(AMQChannel.java:366) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:229) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:117) ~[amqp-client-4.0.2.jar:4.0.2]
    ... 12 common frames omitted
Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no queue '"incoming"' in vhost '/', class-id=50, method-id=10)
    at com.rabbitmq.client.impl.ChannelN.asyncShutdown(ChannelN.java:505) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.ChannelN.processAsync(ChannelN.java:336) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel.handleCompleteInboundCommand(AMQChannel.java:143) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel.handleFrame(AMQChannel.java:90) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQConnection.readFrame(AMQConnection.java:634) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQConnection.access$300(AMQConnection.java:47) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:572) ~[amqp-client-4.0.2.jar:4.0.2]
    ... 1 common frames omitted

It is actually pretty difficult to determine if it is failing to connect to the queue after not noticing that it didn't create it or if it is failing in the initial creation.

There is no mention in any of the docker-specific documentation that implies that queues need to be explicitly created when running inside a docker container, and digging around inside the container, I find no config at all, which I believe implies that it should be allowed to create queues.

Not also that listing queues reveals an empty list, so this is not a problem with mismatched queue configuration from the queue I'm attempting to configure.

In an attempt to make everything completely explicit, I've got the following configuration class:

@SpringBootApplication(exclude={DataSourceAutoConfiguration.class})
@EnableRabbit
public class Application {

  @Bean
  @ConfigurationProperties("spring.datasource")
  public DataSource dataSource() {
    return DataSourceBuilder.create().build();
  }

  @Autowired
  public ConnectionFactory connectionFactory;

  @Bean
  public RabbitAdmin rabbitAdmin() {
    return new RabbitAdmin(connectionFactory);
  }

  @Bean
  public Queue incomingQueue() {
    Queue queue = new Queue("${ep.service.baseline.listen.rabbitq.name}", true, false, false, null);
    return queue;
  }

  @Bean
  public Queue outboundQueue() {
    return new Queue("${ep.service.baseline.send.rabbitq.name}", true, false, false, null);
  }

  @Bean
  TopicExchange exchange() {
    return new TopicExchange("default-exchange", true, false);
  }

  @Bean
  Binding incomingBinding(@Qualifier("incomingQueue") Queue queue, TopicExchange exchange) {
    return BindingBuilder.bind(queue).to(exchange).with(queue.getName());
  }
  @Bean
  Binding outboundBinding(@Qualifier("outboundQueue") Queue queue, TopicExchange exchange) {
    return BindingBuilder.bind(queue).to(exchange).with(queue.getName());
  }
  public static void main(String[] args) throws Exception {

    SpringApplication.run(Application.class, args);
  }
}

and my listener is declared as follows:

@RabbitListener(queues = "${ep.service.baseline.listen.rabbitq.name}")
@Component
public class IncomingMessageProcessor {
  private final Logger log = LoggerFactory.getLogger(IncomingMessageProcessor.class);

  @RabbitHandler
  public void process(@Payload String msg) {
    log.info("received message: {}", msg);
  }
}

And I receive the following errors:

2017-04-24 18:06:54.551 ERROR 38157 --- [           main] o.s.boot.SpringApplication               : Application startup failed

org.springframework.context.ApplicationContextException: Failed to start bean 'org.springframework.amqp.rabbit.config.internalRabbitListenerEndpointRegistry'; nested exception is org.springframework.amqp.AmqpIllegalStateException: Fatal exception on listener startup
    at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:178) ~[spring-context-4.3.7.RELEASE.jar:4.3.7.RELEASE]
    at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:50) ~[spring-context-4.3.7.RELEASE.jar:4.3.7.RELEASE]
    at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:348) ~[spring-context-4.3.7.RELEASE.jar:4.3.7.RELEASE]
    at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:151) ~[spring-context-4.3.7.RELEASE.jar:4.3.7.RELEASE]
    at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:114) ~[spring-context-4.3.7.RELEASE.jar:4.3.7.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:879) ~[spring-context-4.3.7.RELEASE.jar:4.3.7.RELEASE]
    at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.finishRefresh(EmbeddedWebApplicationContext.java:144) ~[spring-boot-1.5.2.RELEASE.jar:1.5.2.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:545) ~[spring-context-4.3.7.RELEASE.jar:4.3.7.RELEASE]
    at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.refresh(EmbeddedWebApplicationContext.java:122) ~[spring-boot-1.5.2.RELEASE.jar:1.5.2.RELEASE]
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:737) [spring-boot-1.5.2.RELEASE.jar:1.5.2.RELEASE]
    at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:370) [spring-boot-1.5.2.RELEASE.jar:1.5.2.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:314) [spring-boot-1.5.2.RELEASE.jar:1.5.2.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1162) [spring-boot-1.5.2.RELEASE.jar:1.5.2.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1151) [spring-boot-1.5.2.RELEASE.jar:1.5.2.RELEASE]
    at com.ep.service.baseline.Application.main(Application.java:65) [classes/:na]
Caused by: org.springframework.amqp.AmqpIllegalStateException: Fatal exception on listener startup
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doStart(SimpleMessageListenerContainer.java:846) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.start(AbstractMessageListenerContainer.java:552) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry.startIfNecessary(RabbitListenerEndpointRegistry.java:279) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry.start(RabbitListenerEndpointRegistry.java:235) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:175) ~[spring-context-4.3.7.RELEASE.jar:4.3.7.RELEASE]
    ... 14 common frames omitted
Caused by: org.springframework.amqp.rabbit.listener.QueuesNotAvailableException: Cannot prepare queue for listener. Either the queue doesn't exist or the broker will not allow us to use it.
    at org.springframework.amqp.rabbit.listener.BlockingQueueConsumer.start(BlockingQueueConsumer.java:548) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1382) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at java.lang.Thread.run(Thread.java:745) ~[na:1.8.0_121]
Caused by: org.springframework.amqp.rabbit.listener.BlockingQueueConsumer$DeclarationException: Failed to declare queue(s):["incoming"]
    at org.springframework.amqp.rabbit.listener.BlockingQueueConsumer.attemptPassiveDeclarations(BlockingQueueConsumer.java:621) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.BlockingQueueConsumer.start(BlockingQueueConsumer.java:520) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    ... 2 common frames omitted
Caused by: java.io.IOException: null
    at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:105) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:101) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:123) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.ChannelN.queueDeclarePassive(ChannelN.java:992) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.ChannelN.queueDeclarePassive(ChannelN.java:50) ~[amqp-client-4.0.2.jar:4.0.2]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_121]
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_121]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_121]
    at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_121]
    at org.springframework.amqp.rabbit.connection.CachingConnectionFactory$CachedChannelInvocationHandler.invoke(CachingConnectionFactory.java:955) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at com.sun.proxy.$Proxy97.queueDeclarePassive(Unknown Source) ~[na:na]
    at org.springframework.amqp.rabbit.listener.BlockingQueueConsumer.attemptPassiveDeclarations(BlockingQueueConsumer.java:600) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    ... 3 common frames omitted
Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no queue '"incoming"' in vhost '/', class-id=50, method-id=10)
    at com.rabbitmq.utility.ValueOrException.getValue(ValueOrException.java:66) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue(BlockingValueOrException.java:32) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply(AMQChannel.java:366) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:229) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:117) ~[amqp-client-4.0.2.jar:4.0.2]
    ... 12 common frames omitted
Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no queue '"incoming"' in vhost '/', class-id=50, method-id=10)
    at com.rabbitmq.client.impl.ChannelN.asyncShutdown(ChannelN.java:505) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.ChannelN.processAsync(ChannelN.java:336) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel.handleCompleteInboundCommand(AMQChannel.java:143) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel.handleFrame(AMQChannel.java:90) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQConnection.readFrame(AMQConnection.java:634) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQConnection.access$300(AMQConnection.java:47) ~[amqp-client-4.0.2.jar:4.0.2]
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:572) ~[amqp-client-4.0.2.jar:4.0.2]
    ... 1 common frames omitted

The logs from rabbit look like this:

=INFO REPORT==== 24-Apr-2017::23:19:55 ===
accepting AMQP connection <0.917.0> (172.17.0.1:42196 -> 172.17.0.3:5672)

=INFO REPORT==== 24-Apr-2017::23:19:55 ===
Connection <0.917.0> (172.17.0.1:42196 -> 172.17.0.3:5672) has a client-provided name: rabbitConnectionFactory#0

=INFO REPORT==== 24-Apr-2017::23:19:55 ===
connection <0.917.0> (172.17.0.1:42196 -> 172.17.0.3:5672 - rabbitConnectionFactory#0): user 'guest' authenticated and granted access to vhost '/'

=ERROR REPORT==== 24-Apr-2017::23:19:57 ===
Channel error on connection <0.917.0> (172.17.0.1:42196 -> 172.17.0.3:5672, vhost: '/', user: 'guest'), channel 1:
operation queue.declare caused a channel exception not_found: no queue '"incoming"' in vhost '/'

=ERROR REPORT==== 24-Apr-2017::23:20:02 ===
Channel error on connection <0.917.0> (172.17.0.1:42196 -> 172.17.0.3:5672, vhost: '/', user: 'guest'), channel 1:
operation queue.declare caused a channel exception not_found: no queue '"incoming"' in vhost '/'

Best Answer

It's not related to docker, or the location of the broker at all.

It looks like you have some spurious "" around your queue name...

reply-code=404, reply-text=NOT_FOUND - no queue '"incoming"' in vhost '/'

(Note the " within the ').

What is in this property ep.service.baseline.listen.rabbitq.name ?

Also, this won't work...

Queue queue = new Queue("${ep.service.baseline.listen.rabbitq.name}", true, false, false, null);

You can't use property placeholders there. That will create a queue named ${ep.service.baseline.listen.rabbitq.name} - rabbit is very liberal with its naming rules.

You would need to use...

@Value("${ep.service.baseline.listen.rabbitq.name}")
private String queueName;

@Bean
public Queue incomingQueue() {
    Queue queue = new Queue(this.queueName, true, false, false, null);
    return queue;
}

DEBUG logging is always your friend; you will see all the queue declarations logged.

Related Solutions

Docker – How is Docker different from a virtual machine

Docker originally used LinuX Containers (LXC), but later switched to runC (formerly known as libcontainer), which runs in the same operating system as its host. This allows it to share a lot of the host operating system resources. Also, it uses a layered filesystem (AuFS) and manages networking.

AuFS is a layered file system, so you can have a read only part and a write part which are merged together. One could have the common parts of the operating system as read only (and shared amongst all of your containers) and then give each container its own mount for writing.

So, let's say you have a 1 GB container image; if you wanted to use a full VM, you would need to have 1 GB x number of VMs you want. With Docker and AuFS you can share the bulk of the 1 GB between all the containers and if you have 1000 containers you still might only have a little over 1 GB of space for the containers OS (assuming they are all running the same OS image).

A full virtualized system gets its own set of resources allocated to it, and does minimal sharing. You get more isolation, but it is much heavier (requires more resources). With Docker you get less isolation, but the containers are lightweight (require fewer resources). So you could easily run thousands of containers on a host, and it won't even blink. Try doing that with Xen, and unless you have a really big host, I don't think it is possible.

A full virtualized system usually takes minutes to start, whereas Docker/LXC/runC containers take seconds, and often even less than a second.

There are pros and cons for each type of virtualized system. If you want full isolation with guaranteed resources, a full VM is the way to go. If you just want to isolate processes from each other and want to run a ton of them on a reasonably sized host, then Docker/LXC/runC seems to be the way to go.

For more information, check out this set of blog posts which do a good job of explaining how LXC works.

Why is deploying software to a docker image (if that's the right term) easier than simply deploying to a consistent production environment?

Deploying a consistent production environment is easier said than done. Even if you use tools like Chef and Puppet, there are always OS updates and other things that change between hosts and environments.

Docker gives you the ability to snapshot the OS into a shared image, and makes it easy to deploy on other Docker hosts. Locally, dev, qa, prod, etc.: all the same image. Sure you can do this with other tools, but not nearly as easily or fast.

This is great for testing; let's say you have thousands of tests that need to connect to a database, and each test needs a pristine copy of the database and will make changes to the data. The classic approach to this is to reset the database after every test either with custom code or with tools like Flyway - this can be very time-consuming and means that tests must be run serially. However, with Docker you could create an image of your database and run up one instance per test, and then run all the tests in parallel since you know they will all be running against the same snapshot of the database. Since the tests are running in parallel and in Docker containers they could run all on the same box at the same time and should finish much faster. Try doing that with a full VM.

From comments...

Interesting! I suppose I'm still confused by the notion of "snapshot[ting] the OS". How does one do that without, well, making an image of the OS?

Well, let's see if I can explain. You start with a base image, and then make your changes, and commit those changes using docker, and it creates an image. This image contains only the differences from the base. When you want to run your image, you also need the base, and it layers your image on top of the base using a layered file system: as mentioned above, Docker uses AuFS. AuFS merges the different layers together and you get what you want; you just need to run it. You can keep adding more and more images (layers) and it will continue to only save the diffs. Since Docker typically builds on top of ready-made images from a registry, you rarely have to "snapshot" the whole OS yourself.

Docker – How to get a Docker container’s IP address from the host

The --format option of inspect comes to the rescue.

Modern Docker client syntax is:

docker inspect -f '{{range.NetworkSettings.Networks}}{{.IPAddress}}{{end}}' container_name_or_id

Old Docker client syntax is:

docker inspect --format '{{ .NetworkSettings.IPAddress }}' container_name_or_id

These commands will return the Docker container's IP address.

As mentioned in the comments: if you are on Windows, use double quotes " instead of single quotes ' around the curly braces.

Best Answer

Related Solutions

Docker – How is Docker different from a virtual machine

Docker – How to get a Docker container’s IP address from the host

Related Topic