Add BATCH_BY_KEY Processing Order

README.adoc

@@ -1534,6 +1534,12 @@ ifndef::github_name[]
 toc::[]
 endif::[]
 
+== 0.5.4.0
+
+=== Improvements
+
+* feature: Add batch by key processing order, creating batches of the same key
+
 == 0.5.3.4
 
 === Fixes

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelConsumerOptions.java

@@ -117,7 +117,14 @@ public enum ProcessingOrder {
          * Process messages in key order. Concurrency is at most the number of unique keys in a topic, limited by the
          * max concurrency or uncommitted settings.
          */
-        KEY
+        KEY,
+
+        /**
+         * Together batch size > 1, creates batches of the same key.
+         * Concurrency is at most the number of unique keys in a topic, limited by the max concurrency or uncommitted
+         * settings.
+         */
+        BATCH_BY_KEY
     }
 
     /**

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ShardKey.java

@@ -30,7 +30,7 @@ public static ShardKey of(WorkContainer<?, ?> wc, ProcessingOrder ordering) {
 
     public static ShardKey of(ConsumerRecord<?, ?> rec, ProcessingOrder ordering) {
         return switch (ordering) {
-            case KEY -> ofKey(rec);
+            case KEY, BATCH_BY_KEY -> ofKey(rec);
             case PARTITION, UNORDERED -> ofTopicPartition(rec);
         };
     }

parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/state/WorkManagerTest.java

@@ -641,7 +641,7 @@ void workQueuesEmptyWhenAllWorkComplete() {
     @ParameterizedTest
     @EnumSource
     void resumesFromNextShard(ParallelConsumerOptions.ProcessingOrder order) {
-        Assumptions.assumeFalse(order == KEY); // just want to test ordered vs unordered
+        Assumptions.assumeFalse(order == KEY || order == BATCH_BY_KEY); // just want to test ordered vs unordered
 
         ParallelConsumerOptions<?, ?> build = ParallelConsumerOptions.builder()
                 .ordering(order)