1

Using this document as a reference: https://blog.rabbitmq.com/posts/2021/07/rabbitmq-streams-first-application I have created a stream in RabbitMQ and added 1 million messages to it.

try (Environment environment = Environment.builder().uri("rabbitmq-stream://guest:guest@localhost:5552")
                .build()) {

            environment.streamCreator().stream("tenth-application-stream").create();
            Producer producer = environment.producerBuilder().stream("tenth-application-stream") 
                    .build();

        int messageCount = 1_000_000;

        CountDownLatch confirmLatch = new CountDownLatch(messageCount);
        IntStream.range(0, messageCount).forEach(i -> {
            Message message = producer.messageBuilder().properties().creationTime(System.currentTimeMillis())
                    .messageId(i).messageBuilder().properties().groupId(String.valueOf(i)).messageBuilder()
                    .addData(String.valueOf(i).getBytes(StandardCharsets.UTF_8)).build();
            producer.send(message, confirmationStatus -> confirmLatch.countDown());
        });
        try {
            boolean done = confirmLatch.await(10, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    } catch (StreamException e) {
        System.out.println(e);
    }

I'm trying to read back all of the messages in the stream by doing the following:

    Consumer consumer = environment.consumerBuilder().stream("tenth-application-stream")
            .offset(OffsetSpecification.first()).messageHandler((context, message) -> {
                System.out.println(message.getBody().toString());

            }).build();

However, this will only print up to message 499. How do i see all of the messages in the stream?

5
  • Have you verified that there are 1M messages in the stream Q? Did you set the Utils.waitAtMost function to allow sufficient time to read all messages? Mar 24, 2022 at 20:35
  • You should use a CountDownLatch in the listener as well, to avoid exiting the program before all the expected messages made it to the application. The complete code of the consumer program would be useful. Mar 25, 2022 at 9:25
  • @ArnaudCogoluègnes That solved the problem! If you make it an answer, i can mark it as the answer. Thanks for your help.
    – BigBug
    Mar 25, 2022 at 15:01
  • 1
    @BigBug Good to hear the CountDownLatch trick solved your problem. To be fair @chad-knutson had the first hunch, so I let them make an answer if they want. Mar 28, 2022 at 6:52
  • Sounds good! @ChadKnutson If you'd like to add an answer i can mark it.
    – BigBug
    Mar 31, 2022 at 15:37

1 Answer 1

1

Since your client has read 499 messages, we know that some messages were published by the producer. Have you verified that there were in fact more than 499 messages in the queue? That can be done using the RabbitMQ web manager. A nice feature of steam queues is that messages can remain in the queue after consumers have read them.

If the rest of the messages were published, then it is likely that the consumer closed its connection before all messages were consumed. The example from the blog (repo here) uses the Utils.waitAtMost method to delay closing the connection:

Utils.waitAtMost(60, () -> messageConsumed.get() >= 1_000_000);

This waits until either 60 seconds has elapsed or 1M messages have been consumed. Alternatively, you could use the CountDownLatch method to keep the connection open until the goal is reached.

Neither of these options seem ideal for a production use case, but they work well for a proof of concept.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.