fix can not consume all message if topics has > 1000 partitions (#354)

fixes #353

1, `responseData` is `LinkedHashMap`, which implementation is not synchronized, therefore `responseValues` should not parallel.
2, `getExecutor schedule` may not process so fast for lots of partitions.

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/MessageFetchContext.java

@@ -29,8 +29,6 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
@@ -304,8 +302,7 @@ private void readMessagesInternal(KafkaHeaderAndRequest fetch,
                             fetch.getHeader(), entriesRead.get(), allSize);
                     }
 
-                    AtomicBoolean allPartitionsNoEntry = new AtomicBoolean(true);
-                    responseValues.entrySet().parallelStream().forEach(responseEntries -> {
+                    responseValues.entrySet().forEach(responseEntries -> {
                         final PartitionData partitionData;
                         TopicPartition kafkaPartition = responseEntries.getKey();
                         List<Entry> entries = responseEntries.getValue();
@@ -334,8 +331,6 @@ private void readMessagesInternal(KafkaHeaderAndRequest fetch,
                                 null,
                                 MemoryRecords.EMPTY);
                         } else {
-                            allPartitionsNoEntry.set(false);
-
                             ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) cursors
                                     .get(kafkaPartition).getLeft().getManagedLedger();
                             long highWatermark = MessageIdUtils.getHighWatermark(managedLedger);
@@ -368,29 +363,13 @@ private void readMessagesInternal(KafkaHeaderAndRequest fetch,
                         responseData.put(kafkaPartition, partitionData);
                     });
 
-                    if (allPartitionsNoEntry.get()) {
-                        if (log.isDebugEnabled()) {
-                            log.debug("Request {}: All partitions for request read 0 entry",
-                                fetch.getHeader());
-                        }
-
-                        requestHandler.getPulsarService().getExecutor().schedule(() -> {
-                            resultFuture.complete(
-                                new FetchResponse(Errors.NONE,
+                    resultFuture.complete(
+                            new FetchResponse(
+                                    Errors.NONE,
                                     responseData,
                                     ((Integer) THROTTLE_TIME_MS.defaultValue),
                                     ((FetchRequest) fetch.getRequest()).metadata().sessionId()));
-                            this.recycle();
-                        }, waitTime, TimeUnit.MILLISECONDS);
-                    } else {
-                        resultFuture.complete(
-                            new FetchResponse(
-                                Errors.NONE,
-                                responseData,
-                                ((Integer) THROTTLE_TIME_MS.defaultValue),
-                                ((FetchRequest) fetch.getRequest()).metadata().sessionId()));
-                        this.recycle();
-                    }
+                    this.recycle();
                 } else {
                     if (log.isDebugEnabled()) {
                         log.debug("Request {}: Read time or size not reach, do another round of read before return.",