KAFKA-2654: optimize unnecessary poll(0) away

streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java

@@ -154,6 +154,11 @@ public int numBuffered(TopicPartition partition) {
         return recordQueue.size();
     }
 
+    public int topQueueSize() {
+        RecordQueue recordQueue = queuesByTime.peek();
+        return (recordQueue == null) ? 0 : recordQueue.size();
+    }
+
     public int numBuffered() {
         return totalBuffered;
     }

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

@@ -64,6 +64,8 @@ public class StreamTask implements Punctuator {
     private StampedRecord currRecord = null;
     private ProcessorNode currNode = null;
 
+    private boolean requiresPoll = true;
+
     /**
      * Create {@link StreamTask} with its assigned partitions
      *
@@ -173,8 +175,12 @@ public int process() {
             StampedRecord record = partitionGroup.nextRecord(recordInfo);
 
             // if there is no record to process, return immediately
-            if (record == null)
+            if (record == null) {
+                requiresPoll = true;
                 return 0;
+            }
+
+            requiresPoll = false;
 
             try {
                 // process the record by passing to the source node of the topology
@@ -196,6 +202,11 @@ public int process() {
                 // decreased to the threshold, we can then resume the consumption on this partition
                 if (partitionGroup.numBuffered(partition) == this.maxBufferedSize) {
                     consumer.resume(partition);
+                    requiresPoll = true;
+                }
+
+                if (partitionGroup.topQueueSize() <= this.maxBufferedSize) {
+                    requiresPoll = true;
                 }
             } finally {
                 this.currRecord = null;
@@ -206,6 +217,10 @@ public int process() {
         }
     }
 
+    public boolean requiresPoll() {
+        return requiresPoll;
+    }
+
     /**
      * Possibly trigger registered punctuation functions if
      * current time has reached the defined stamp

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -231,33 +231,38 @@ private void shutdown() {
     private void runLoop() {
         try {
             int totalNumBuffered = 0;
+            boolean requiresPoll = true;
 
             consumer.subscribe(new ArrayList<>(builder.sourceTopics()), rebalanceListener);
 
             while (stillRunning()) {
-                long startPoll = time.milliseconds();
-
                 // try to fetch some records if necessary
-                ConsumerRecords<byte[], byte[]> records = consumer.poll(totalNumBuffered == 0 ? this.pollTimeMs : 0);
+                if (requiresPoll) {
+                    long startPoll = time.milliseconds();
+
+                    ConsumerRecords<byte[], byte[]> records = consumer.poll(totalNumBuffered == 0 ? this.pollTimeMs : 0);
 
-                if (!records.isEmpty()) {
-                    for (StreamTask task : tasks.values()) {
-                        for (TopicPartition partition : task.partitions()) {
-                            task.addRecords(partition, records.records(partition));
+                    if (!records.isEmpty()) {
+                        for (StreamTask task : tasks.values()) {
+                            for (TopicPartition partition : task.partitions()) {
+                                task.addRecords(partition, records.records(partition));
+                            }
                         }
                     }
-                }
 
-                long endPoll = time.milliseconds();
-                sensors.pollTimeSensor.record(endPoll - startPoll);
+                    long endPoll = time.milliseconds();
+                    sensors.pollTimeSensor.record(endPoll - startPoll);
+                }
 
                 // try to process one record from each task
                 totalNumBuffered = 0;
+                requiresPoll = false;
 
                 for (StreamTask task : tasks.values()) {
                     long startProcess = time.milliseconds();
 
                     totalNumBuffered += task.process();
+                    requiresPoll = requiresPoll || task.requiresPoll();
 
                     sensors.processTimeSensor.record(time.milliseconds() - startProcess);
                 }