Parallel consumer Key Ordering is not working as expected

[amber13574]

Hi Folks,
I am currently utilizing a parallel consumer to process events concurrently. However, I have encountered an issue when employing KEY ordering, as my project requires this specific ordering. Upon verification, it appears that the expected ordering is not being maintained, as different threads are processing the same key within the same partition. I anticipated that the same key should be processed by the same thread. This is the current implementation.

private void createProcessor(MessageListenerContainer listenerContainer, String consumerGroup, String topic) {
   Consumer<String, String> consumer = cf.createConsumer(consumerGroup, "");
   var options = ParallelConsumerOptions.<String, String>builder()
       .ordering(ProcessingOrder.KEY)
       .consumer(consumer)
       .maxConcurrency(2)
       .commitInterval(Duration.ofMillis(commitInterval))
       .build();
   ParallelStreamProcessor<String, String> processor = ParallelStreamProcessor
       .createEosStreamProcessor(options);
   processor.subscribe(List.of(topic));

   MessageListener listener = (MessageListener) listenerContainer.getContainerProperties().getMessageListener();
   processor.poll(context -> {
     log.info("Receiving message");
     listener.onMessage(context.getSingleConsumerRecord(), null, consumer);

   });

   processors.add(processor);
 }

In this snapshot, we can observe that different threads are processing events with the same key in the same partition. I was expecting thread-1 to process both in order.

Received message on partition 0 with key key-0-0 , offset 68 by thread pc-pool-8-thread-1
Received message on partition 0 with key key-0-0 , offset 69 by thread pc-pool-8-thread-2

Configuration:
Topic with 8 partitions, a single consumer instance (with a concurrency of 2).

[rkolesnev]

It does not dispatch event processing to the same thread in the thread pool per key - but it orders event processing to be sequential per key..
So partition 0 with key key-0-0 , offset 69 will only be executed after partition 0 with key key-0-0 , offset 68 is successfully completed - if you add a delay to user function and log timestamps - that will be visible in logs.
As normally you will have way more keys than processing threads - pinning per key processing to specific threads might delay processing unnecessarily.
For example Key1, Key2, Key3 data - Key1 and Key3 both pinned to thread-1. Key3 cannot be processed until Key1 is done - even if thread-2 is available and free.