[BUG] Default echange consume failed, when broker restart.

Open
#596 0 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Assessment

Difficulty
4/5
Estimated time
3-5 days
Newbie friendliness
35/100
Issue type
Bug
Clarity
Mostly clear
Activity status
Stale
Tech stack
java

Research direction

Start with the supplied Java producer and consumer around queueDeclare and basicConsume, then reproduce unloading vhost-test and restarting the producer. Investigate the default-exchange queue behavior during broker restart or namespace migration; done means the consumer resumes receiving messages after the unload, as expected.

Written by the indexing model from the issue text.

Description

type/bug

Describe the bug
The queue corresponding to the default Exchange cannot consume messages when the broker restarts or the namespace migrates to another broker.

To Reproduce
Steps to reproduce the behavior:

  1. producer code:
  ConnectionFactory connectionFactory = new ConnectionFactory();
  connectionFactory.setHost("127.0.0.1");
  connectionFactory.setPort(5682);
  connectionFactory.setVirtualHost("vhost-test");
  Connection connection = null;
  Channel channel = null;
  try {
      connection = connectionFactory.newConnection();
      channel = connection.createChannel();
      channel.queueDeclare("queue-test", true, true, true, null);
      for (int i = 0; i < 100; i++) {
          String message = "HelloWorld "+ i;
          channel.basicPublish("", "queue-test", null, message.getBytes());
          System.out.println("Send to mq: " + message);
      }
  } catch (Exception e) {
      e.printStackTrace();
  } finally {
      try {
          if (channel != null) {
              channel.close();
          }
          if (connection != null) {
              connection.close();
          }
      } catch (IOException | TimeoutException e) {
          e.printStackTrace();
      }
  }
  1. consumer code
  ConnectionFactory connectionFactory = new ConnectionFactory();
  connectionFactory.setHost("127.0.0.1");
  connectionFactory.setPort(5682);
  connectionFactory.setVirtualHost("vhost-test");
  try {
      Connection connection = connectionFactory.newConnection();
      Channel channel = connection.createChannel();
      channel.queueDeclare("queue-test", true, false, false, null);
      DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              super.handleDelivery(consumerTag, envelope, properties, body);
              String exchange = envelope.getExchange();
              String routingKey = envelope.getRoutingKey();
              long deliveryTag = envelope.getDeliveryTag();
              String message = new String(body, StandardCharsets.UTF_8);
              System.out.println("receive message: " + message);
          }
      };
      channel.basicConsume("queue-test", true, defaultConsumer);
  } catch (Exception e) {
      e.printStackTrace();
  }
  1. start consumer and producer client.
  2. unload vhost-test.
  3. restart producer client.
  4. In the observation consumer has been unable to consume.

Expected behavior
Unload can still consume normally.

Dominant language
Java
Stars
123
Forks
41
Avg merge
4h 43m
Merged PRs (30d)
4

Contributor guide

No contributing guide indexed for this repository

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

More from streamnative/aop

All issues in streamnative/aop

Similar issues

More Java issues

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.