Fetch: trigger pending fetches when producing messages.

docs/reference-metrics.md

@@ -72,6 +72,7 @@ The KoP metrics are exposed under "/metrics" at port `8000` along with Pulsar me
 | kop_server_MESSAGE_OUT | Counter | The consumer message out stats. <br> Available labels: *topic*, *partition*, *group*. </br> <ul><li>*topic*: the topic name to consume.</li><li>*partition*: the partition id for the topic to consume</li><li>*group*: the group id for consumer to consumer message from topic-partition</li></ul>|
 | kop_server_ENTRIES_OUT | Counter | The consumer entries out stats. <br> Available labels: *topic*, *partition*, *group*. </br> <ul><li>*topic*: the topic name to consume.</li><li>*partition*: the partition id for the topic to consume</li><li>*group*: the group id for consumer to consumer message from topic-partition</li></ul>|
 | kop_server_CONSUME_MESSAGE_CONVERSIONS | Counter | The consumer message conversions in stats. <br> Available labels: *topic*, *partition*. </br> <ul><li>*topic*: the topic name to consume.</li><li>*partition*: the partition id for the topic to consume</li></ul>|
+| kop_server_WAITING_FETCHES_TRIGGERED | Counter | Number of fetches that have been delayed due to not enough data, and that have been unblocked because some message has been produced|
 
 ### Kop event metrics
 

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

@@ -15,36 +15,67 @@
 
 import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperation;
 import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
+import lombok.extern.slf4j.Slf4j;
 
+@Slf4j
 public class DelayedFetch extends DelayedOperation {
     private final Runnable callback;
     private final AtomicLong bytesReadable;
     private final int minBytes;
+    private final MessageFetchContext messageFetchContext;
+    private final AtomicBoolean restarted = new AtomicBoolean();
+    private final AtomicBoolean someMessageProduced = new AtomicBoolean();
 
-    protected DelayedFetch(long delayMs, AtomicLong bytesReadable, int minBytes, Runnable callback) {
+    protected DelayedFetch(long delayMs, AtomicLong bytesReadable, int minBytes,
+                           MessageFetchContext messageFetchContext) {
         super(delayMs, Optional.empty());
-        this.callback = callback;
         this.bytesReadable = bytesReadable;
         this.minBytes = minBytes;
+        this.messageFetchContext = messageFetchContext;
+        this.callback = messageFetchContext::complete;
     }
 
     @Override
     public void onExpiration() {
+        if (restarted.get()) {
+            return;
+        }
         callback.run();
     }
 
     @Override
     public void onComplete() {
+        if (restarted.get()) {
+            return;
+        }
         callback.run();
     }
 
     @Override
     public boolean tryComplete() {
+        if (someMessageProduced.get()) {
+            // if we are here then we were waiting for the condition
+            // someone wrote some messages to one of the topics
+            // trigger the Fetch from scratch
+            restarted.set(true);
+            messageFetchContext.onDataWrittenToSomePartition();
+            return true;
+        }
         if (bytesReadable.get() < minBytes){
             return false;
         }
         callback.run();
         return true;
     }
+
+    @Override
+    public boolean wakeup() {
+        // In the future we could notify the MessageFetchContext that the
+        // new data is only on this partition and not
+        // on other partitions
+        someMessageProduced.set(true);
+        return true;
+    }
 }

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

@@ -56,6 +56,7 @@
 import io.streamnative.pulsar.handlers.kop.utils.OffsetFinder;
 import io.streamnative.pulsar.handlers.kop.utils.TopicNameUtils;
 import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperation;
+import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperationKey;
 import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperationPurgatory;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
@@ -941,6 +942,11 @@ protected void handleProduceRequest(KafkaHeaderAndRequest produceHar,
                     mergedResponse.putAll(unauthorizedTopicResponsesMap);
                     mergedResponse.putAll(invalidRequestResponses);
                     resultFuture.complete(new ProduceResponse(mergedResponse));
+                    mergedResponse.forEach((_topicPartition, _response) -> {
+                        if (_response.error == Errors.NONE) {
+                            notifyPendingFetches(_topicPartition);
+                        }
+                    });
                 });
             }
         };
@@ -979,6 +985,20 @@ protected void handleProduceRequest(KafkaHeaderAndRequest produceHar,
 
     }
 
+    private void notifyPendingFetches(TopicPartition topicPartition) {
+       ctx.executor().execute(() -> {
+           DelayedOperationKey.TopicPartitionOperationKey key =
+                   new DelayedOperationKey.TopicPartitionOperationKey(topicPartition);
+           int matches = fetchPurgatory.checkAndComplete(key);
+           if (matches > 0) {
+               requestStats.getWaitingFetchesTriggered().add(matches);
+               if (log.isDebugEnabled()) {
+                   log.debug("{} DelayedFetch woke up for {}", matches, topicPartition);
+               }
+           }
+       });
+    }
+
     private void validateRecords(short version, MemoryRecords records) {
         if (version >= 3) {
             Iterator<MutableRecordBatch> iterator = records.batches().iterator();

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

@@ -75,6 +75,7 @@ public interface KopServerStats {
     String PREPARE_METADATA = "PREPARE_METADATA";
     String MESSAGE_READ = "MESSAGE_READ";
     String FETCH_DECODE = "FETCH_DECODE";
+    String WAITING_FETCHES_TRIGGERED = "WAITING_FETCHES_TRIGGERED";
 
     /**
      * Consumer stats.

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

@@ -68,6 +68,7 @@
 import org.apache.kafka.common.requests.IsolationLevel;
 import org.apache.kafka.common.requests.RequestHeader;
 import org.apache.kafka.common.requests.ResponseCallbackWrapper;
+import org.apache.kafka.common.utils.SystemTime;
 import org.apache.pulsar.metadata.api.GetResult;
 
 /**
@@ -83,6 +84,7 @@ protected MessageFetchContext newObject(Handle<MessageFetchContext> handle) {
     };
 
     private final Handle<MessageFetchContext> recyclerHandle;
+    private long startTime;
     private Map<TopicPartition, PartitionData<MemoryRecords>> responseData;
     private ConcurrentLinkedQueue<DecodeResult> decodeResults;
     private KafkaRequestHandler requestHandler;
@@ -95,7 +97,7 @@ protected MessageFetchContext newObject(Handle<MessageFetchContext> handle) {
     private RequestHeader header;
     private volatile CompletableFuture<AbstractResponse> resultFuture;
     private AtomicBoolean hasComplete;
-    private AtomicLong bytesReadable;
+    private AtomicLong bytesRead;
     private DelayedOperationPurgatory<DelayedOperation> fetchPurgatory;
     private String namespacePrefix;
 
@@ -119,8 +121,9 @@ public static MessageFetchContext get(KafkaRequestHandler requestHandler,
         context.header = kafkaHeaderAndRequest.getHeader();
         context.resultFuture = resultFuture;
         context.hasComplete = new AtomicBoolean(false);
-        context.bytesReadable = new AtomicLong(0);
+        context.bytesRead = new AtomicLong(0);
         context.fetchPurgatory = fetchPurgatory;
+        context.startTime = SystemTime.SYSTEM.hiResClockMs();
         return context;
     }
 
@@ -142,6 +145,7 @@ public static MessageFetchContext getForTest(FetchRequest fetchRequest,
         context.header = null;
         context.resultFuture = resultFuture;
         context.hasComplete = new AtomicBoolean(false);
+        context.startTime = SystemTime.SYSTEM.hiResClockMs();
         return context;
     }
 
@@ -163,7 +167,7 @@ private void recycle() {
         header = null;
         resultFuture = null;
         hasComplete = null;
-        bytesReadable = null;
+        bytesRead = null;
         fetchPurgatory = null;
         namespacePrefix = null;
         recyclerHandle.recycle(this);
@@ -195,15 +199,47 @@ private void addErrorPartitionResponse(TopicPartition topicPartition, Errors err
     private void tryComplete() {
         if (resultFuture != null && responseData.size() >= fetchRequest.fetchData().size()
                 && hasComplete.compareAndSet(false, true)) {
-            DelayedFetch delayedFetch = new DelayedFetch(fetchRequest.maxWait(), bytesReadable,
-                    fetchRequest.minBytes(), this::complete);
-            List<Object> delayedFetchKeys =
-                    fetchRequest.fetchData().keySet().stream()
-                            .map(DelayedOperationKey.TopicPartitionOperationKey::new).collect(Collectors.toList());
-            fetchPurgatory.tryCompleteElseWatch(delayedFetch, delayedFetchKeys);
+            boolean errorsOccurred = false;
+            if (responseData
+                    .values()
+                    .stream()
+                    .anyMatch(p->p.error != Errors.NONE)) {
+                // if there is an error no need to wait, the fetch must fail
+                // as soon as possible
+                errorsOccurred = true;
+            }
+            long now = SystemTime.SYSTEM.hiResClockMs();
+            long currentWait = now - this.startTime;
+            long remainingMaxWait = fetchRequest.maxWait() - currentWait;
+            long maxWait = Math.min(remainingMaxWait, fetchRequest.maxWait());
+            if (bytesRead.get() < fetchRequest.minBytes() && !errorsOccurred && maxWait > 0) {
+                // we haven't read enough data, need to wait
+                DelayedFetch delayedFetch = new DelayedFetch(maxWait, bytesRead,
+                        fetchRequest.minBytes(), this);
+                List<Object> delayedFetchKeys =
+                        fetchRequest.fetchData().keySet().stream()
+                                .map(DelayedOperationKey.TopicPartitionOperationKey::new).collect(Collectors.toList());
+                fetchPurgatory.tryCompleteElseWatch(delayedFetch, delayedFetchKeys);
+            } else {
+                this.complete();
+            }
         }
     }
 
+    /**
+     * Restart this Fetch, we were waiting for some data (minBytes)
+     * and someone wrote something on any of the watched partitions.
+     */
+    public void onDataWrittenToSomePartition() {
+        decodeResults.forEach(DecodeResult::recycle);
+        decodeResults.clear();
+        bytesRead.set(0);
+        hasComplete.set(false);
+        responseData.clear();
+        handleFetch();
+    }
+
+
     public void complete() {
         if (resultFuture == null) {
             // the context has been recycled
@@ -329,6 +365,9 @@ private void handlePartitionData(final TopicPartition topicPartition,
         // the future that is returned by getTopicConsumerManager is always completed normally
         topicManager.getTopicConsumerManager(fullTopicName).thenAccept(tcm -> {
             if (tcm == null) {
+                if (log.isDebugEnabled()) {
+                    log.debug("Fetch for {}: failed, topic not owned .", topicPartition);
+                }
                 registerPrepareMetadataFailedEvent(startPrepareMetadataNanos);
                 // remove null future cache
                 KafkaTopicConsumerManagerCache.getInstance().removeAndCloseByTopic(fullTopicName);
@@ -491,7 +530,7 @@ private void handleEntries(final List<Entry> entries,
                     highWatermark, // TODO: should it be changed to the logStartOffset?
                     abortedTransactions,
                     kafkaRecords));
-            bytesReadable.getAndAdd(kafkaRecords.sizeInBytes());
+            bytesRead.getAndAdd(kafkaRecords.sizeInBytes());
             tryComplete();
         });
     }

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

@@ -28,6 +28,7 @@
 import static io.streamnative.pulsar.handlers.kop.KopServerStats.RESPONSE_BLOCKED_LATENCY;
 import static io.streamnative.pulsar.handlers.kop.KopServerStats.RESPONSE_BLOCKED_TIMES;
 import static io.streamnative.pulsar.handlers.kop.KopServerStats.SERVER_SCOPE;
+import static io.streamnative.pulsar.handlers.kop.KopServerStats.WAITING_FETCHES_TRIGGERED;
 
 import io.streamnative.pulsar.handlers.kop.stats.StatsLogger;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -112,6 +113,12 @@ public class RequestStats {
     )
     private final OpStatsLogger fetchDecodeStats;
 
+    @StatsDoc(
+            name = WAITING_FETCHES_TRIGGERED,
+            help = "number of pending fetches that woke up due to some data produced"
+    )
+    private final Counter waitingFetchesTriggered;
+
     public RequestStats(StatsLogger statsLogger) {
         this.statsLogger = statsLogger;
 
@@ -127,6 +134,7 @@ public RequestStats(StatsLogger statsLogger) {
         this.prepareMetadataStats = statsLogger.getOpStatsLogger(PREPARE_METADATA);
         this.messageReadStats = statsLogger.getOpStatsLogger(MESSAGE_READ);
         this.fetchDecodeStats  = statsLogger.getOpStatsLogger(FETCH_DECODE);
+        this.waitingFetchesTriggered = statsLogger.getCounter(WAITING_FETCHES_TRIGGERED);
 
         statsLogger.registerGauge(REQUEST_QUEUE_SIZE, new Gauge<Number>() {
             @Override

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/delayed/DelayedOperation.java

@@ -98,6 +98,13 @@ public boolean isCompleted() {
      */
     public abstract boolean tryComplete();
 
+    /**
+     * Try to wake up the operation.
+     */
+    public boolean wakeup() {
+        return true;
+    }
+
     /**
      * Thread-safe variant of tryComplete() that attempts completion only if the lock can be acquired
      * without blocking.

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/delayed/DelayedOperationPurgatory.java

@@ -226,7 +226,7 @@ public int checkAndComplete(Object key) {
         if (null == watchers) {
             return 0;
         } else {
-            return watchers.tryCompleteWatched();
+            return watchers.tryCompleteWatched(key);
         }
     }
 
@@ -340,7 +340,7 @@ public void watch(T t) {
         }
 
         // traverse the list and try to complete some watched elements
-        public int tryCompleteWatched() {
+        public int tryCompleteWatched(Object key) {
             int completed = 0;
 
             Iterator<T> iter = operations.iterator();
@@ -349,7 +349,7 @@ public int tryCompleteWatched() {
                 if (curr.isCompleted()) {
                     // another thread has completed this operation, just remove it
                     iter.remove();
-                } else if (curr.maybeTryComplete()) {
+                } else if (curr.wakeup() && curr.maybeTryComplete()) {
                     iter.remove();
                     completed += 1;
                 }