parallel for reading message (apache#159)

Co-authored-by: dockerzhang(张超) <[email protected]>
hangc0276 · Jun 10, 2020 · d15a26d · d15a26d
1 parent f5e2af4
commit d15a26d
Showing 1 changed file with 8 additions and 5 deletions.
diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/MessageFetchContext.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/MessageFetchContext.java
@@ -197,8 +197,10 @@ private void readMessagesInternal(KafkaHeaderAndRequest fetch,
         Map<TopicPartition, CompletableFuture<Entry>> readFutures = readAllCursorOnce(cursors);
         CompletableFuture.allOf(readFutures.values().stream().toArray(CompletableFuture<?>[]::new))
             .whenComplete((ignore, ex) -> {
-                // keep entries since all read completed. currently only read 1 entry each time.
-                readFutures.forEach((kafkaTopic, readEntry) -> {
+                // keep entries since all read completed.
+                readFutures.entrySet().parallelStream().forEach(kafkaTopicReadEntry -> {
+                    TopicPartition kafkaTopic = kafkaTopicReadEntry.getKey();
+                    CompletableFuture<Entry> readEntry = kafkaTopicReadEntry.getValue();
                     try {
                         Entry entry = readEntry.get();
                         List<Entry> entryList = responseValues.computeIfAbsent(kafkaTopic, l -> Lists.newArrayList());
@@ -275,9 +277,10 @@ private void readMessagesInternal(KafkaHeaderAndRequest fetch,
                     }
 
                     AtomicBoolean allPartitionsNoEntry = new AtomicBoolean(true);
-                    responseValues.forEach((kafkaPartition, entries) -> {
-                        final FetchResponse.PartitionData partitionData;
-
+                    responseValues.entrySet().parallelStream().forEach(responseEntrys -> {
+                        final PartitionData partitionData;
+                        TopicPartition kafkaPartition = responseEntrys.getKey();
+                        List<Entry> entries = responseEntrys.getValue();
                         // Add cursor and offset back to TCM when all the read completed.
                         TopicName pulsarTopicName = pulsarTopicName(kafkaPartition, requestHandler.getNamespace());
                         Pair<ManagedCursor, Long> pair = cursors.get(kafkaPartition);