How to keep order of events within message groups with amqp reactive-messaging and quarkus?

[crionuke]

Hello! I wish to handle incoming events using reactive-messaging-amqp and quarkus in such way as:

• I need to keep original order of events to handle within the same message group;
• I need concurrent handling for different message groups to achieve enough level or through-put;

But in spite of setting groupId on the side of my producer, my consumer disrupts order.

I have prepared example project to demonstrate that:
https://github.com/crionuke/reactive-amqp-with-groups

Can you explain, is it a bug or I have missed/wrong something?

Producer

return Multi.createFrom().range(0, 10)
        .onItem().transformToUniAndConcatenate(value -> {
            // Use the same group for all outgoing events.
            // It should guarantee one by one handling on the side of consumer
            final var metadata = OutgoingAmqpMetadata.builder()
                    .withGroupId("group")
                    .build();

            final var message = Message.of(Long.valueOf(value))
                    .addMetadata(metadata);

            return outboxEmitter.sendMessage(message);
        })
        .collect().asList()
        .replaceWithVoid();

Consumer
I use transformToUniAndMerge in favor of transformToUniAndConcatenate because I need concurrent handlers.

inboxEvents
        .onItem().transformToUniAndMerge(value -> Uni.createFrom().voidItem()
                .invoke(voidItem -> System.out.println("Incoming value=" + value))
                .onItem().delayIt().by(Duration.ofSeconds(new Random().nextInt(1, 10)))
                .replaceWith(value)
                .invoke(voidItem -> System.out.println("Handled value=" + value)))
        .subscribe().with(value -> System.out.println("Consumed value=" + value));

Output.
You can see my subscriber get a list of all events at once and handle them in arbitrary order.
But in my opinion, the next message in a message group should come to the subscriber only after acknowledging the prev one.

Incoming value=0
Incoming value=1
Incoming value=2
Incoming value=3
Incoming value=4
Handled value=3
Consumed value=3
Handled value=1
Consumed value=1
Handled value=0
Consumed value=0
Handled value=4
Consumed value=4
Handled value=2
Consumed value=2

[ozangunalp]

Also posted on Stackoverflow:

The AMQP clients do not provide any parallelism by message group id out of the box.

You therefore need to manually dispatch event processing to a worker thread pool to process events in parallel. Mutiny APIs allow to do such operations. You can couple that with the KeyedMulti support to group messages by group id (available since Reactive Messaging 4.6.0).

Here is a sample code :

    @Incoming("data")
    @Outgoing("sink")
    Multi<Integer> process(KeyedMulti<String, Integer> group) {
        return group
                .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
                .map(i -> {
                    String id = group.key();
                    try {
                        Thread.sleep(20L * i);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    return i + 1;
                });
    }

    @ApplicationScoped
    static class AmqpKeyValueExtractor implements KeyValueExtractor {

        @Override
        public boolean canExtract(Message<?> first, Type keyType, Type valueType) {
            return first.getMetadata(IncomingAmqpMetadata.class).isPresent();
        }

        @Override
        public Object extractKey(Message<?> message, Type keyType) {
            return message.getMetadata(IncomingAmqpMetadata.class)
                    .map(IncomingAmqpMetadata::getGroupId)
                    .orElse(null);
        }

        @Override
        public Object extractValue(Message<?> message, Type valueType) {
            return message.getPayload();
        }
    }

The KeyValueExtractor is self-explanatory, for each incoming message the code extracts a key and a value to be used in KeyedMulti.

The process method will be called for each new group id, and the method will use runSubscriptionOn to run the processing on a worker thread pool (You cannot use @Blocking annotation in this scenario).

You'll notice that while the message order is preserved inside a group, it'll happen concurrently across all groups.

[crionuke]

@ozangunalp, hello!
Thanks a lot for your response. It looks like one of the interesting solutions to my problem.

For the sake of knowledge transfer, I discovered another approach: using transformToUniAndConcatenate() and multiple subscriptions for @Channel Multi<Message<Long>> that works as expected from the perspective of message groups.