AKCORE-253: Optimised batching of requests in ShareConsumeRequestManager

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java

@@ -607,7 +607,7 @@ private ShareFetch<K, V> pollForFetches(final Timer timer) {
         long pollTimeout = Math.min(applicationEventHandler.maximumTimeToWait(), timer.remainingMs());
 
         Map<TopicIdPartition, Acknowledgements> acknowledgementsMap = currentFetch.takeAcknowledgedRecords();
-
+        //log.warn("TAKEACKS : While polling : {}", acknowledgementsMap);
         // If data is available already, return it immediately
         final ShareFetch<K, V> fetch = collect(acknowledgementsMap);
         if (!fetch.isEmpty()) {
@@ -702,6 +702,7 @@ public Map<TopicIdPartition, Optional<KafkaException>> commitSync(final Duration
 
             Timer requestTimer = time.timer(timeout.toMillis());
             Map<TopicIdPartition, Acknowledgements> acknowledgementsMap = acknowledgementsToSend();
+            //log.warn("TAKEACKS : COMMIT-SYNC : {}", acknowledgementsMap);
             if (acknowledgementsMap.isEmpty()) {
                 return Collections.emptyMap();
             } else {
@@ -750,6 +751,7 @@ public void commitAsync() {
             acknowledgeBatchIfImplicitAcknowledgement(false);
 
             Map<TopicIdPartition, Acknowledgements> acknowledgementsMap = acknowledgementsToSend();
+            //rn("TAKEACKS : COMMIT-ASYNC : {}", acknowledgementsMap);
             if (!acknowledgementsMap.isEmpty()) {
                 ShareAcknowledgeAsyncEvent event = new ShareAcknowledgeAsyncEvent(acknowledgementsMap);
                 applicationEventHandler.add(event);

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareInFlightBatch.java

@@ -64,6 +64,7 @@ public int acknowledgeAll(AcknowledgeType type) {
                 recordsAcknowledged++;
             }
         }
+        //System.out.println(recordsAcknowledged + "in SIFBatch : " + acknowledgedRecords);
         return recordsAcknowledged;
     }
 

core/src/test/java/kafka/test/api/ShareConsumerTest.java

@@ -109,7 +109,7 @@ public void createCluster() throws Exception {
                 .setConfigProp("group.share.enable", "true")
                 .setConfigProp("group.share.partition.max.record.locks", "10000")
                 .setConfigProp("group.share.persister.class.name", "org.apache.kafka.server.group.share.DefaultStatePersister")
-                .setConfigProp("group.share.record.lock.duration.ms", "10000")
+                .setConfigProp("group.share.record.lock.duration.ms", "1000")
                 .setConfigProp("offsets.topic.replication.factor", "1")
                 .setConfigProp("share.coordinator.state.topic.min.isr", "1")
                 .setConfigProp("share.coordinator.state.topic.replication.factor", "1")
@@ -1094,6 +1094,7 @@ public void testMultipleConsumersInGroupFailureConcurrentConsumption() {
                 CompletableFuture<Integer> future = new CompletableFuture<>();
                 futuresSuccess.add(future);
                 consumeMessages(totalMessagesConsumed, producerCount * messagesPerProducer, "group1", consumerNumber, 25, true, future);
+                System.out.println(totalMessagesConsumed.get() + " is total msgs consumed");
             });
         }
         producerExecutorService.shutdown();
@@ -1646,8 +1647,8 @@ private CompletableFuture<Integer> produceMessages(int messageCount) {
         Future<?>[] recordFutures = new Future<?>[messageCount];
         int messagesSent = 0;
         try (KafkaProducer<byte[], byte[]> producer = createProducer(new ByteArraySerializer(), new ByteArraySerializer())) {
-            ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, "key".getBytes(), "value".getBytes());
             for (int i = 0; i < messageCount; i++) {
+                ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, ("key" + i).getBytes(), ("value" + i).getBytes());
                 recordFutures[i] = producer.send(record);
             }
             for (int i = 0; i < messageCount; i++) {
@@ -1710,6 +1711,7 @@ private void consumeMessages(AtomicInteger totalMessagesConsumed,
         } catch (Exception e) {
             fail("Consumer " + consumerNumber + " failed with exception: " + e);
         } finally {
+            shareConsumer.commitAsync();
             shareConsumer.close();
             future.complete(messagesConsumed);
         }

core/src/test/resources/log4j.properties

@@ -20,7 +20,9 @@ log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c:%L)%n
 
 log4j.logger.kafka=WARN
 log4j.logger.org.apache.kafka=WARN
-
+log4j.logger.org.apache.kafka.clients.consumer.internals.ShareConsumeRequestManager=WARN
+log4j.logger.org.apache.kafka.clients.consumer.internals.ShareConsumerImpl=WARN
+log4j.logger.org.apache.kafka.clients.consumer.internals.ShareSessionHandler=DEBUG
 
 # zkclient can be verbose, during debugging it is common to adjust it separately
 log4j.logger.org.apache.zookeeper=WARN