kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject kafka git commit: HOTFIX: call consumer.poll() even when no task is assigned
Date Tue, 27 Oct 2015 20:52:02 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 38a1b6055 -> af42c3789


HOTFIX: call consumer.poll() even when no task is assigned

StreamThread should keep calling consumer.poll() even when no task is assigned. This is necessary
to get a task.

guozhangwang

Author: Yasuhiro Matsuda <yasuhiro@confluent.io>

Reviewers: Guozhang Wang

Closes #373 from ymatsuda/no_task


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/af42c378
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/af42c378
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/af42c378

Branch: refs/heads/trunk
Commit: af42c37899e8fc66590e7f3c4893f8224441f6a8
Parents: 38a1b60
Author: Yasuhiro Matsuda <yasuhiro@confluent.io>
Authored: Tue Oct 27 13:57:19 2015 -0700
Committer: Guozhang Wang <wangguoz@gmail.com>
Committed: Tue Oct 27 13:57:19 2015 -0700

----------------------------------------------------------------------
 .../processor/internals/StreamThread.java       | 25 +++++++++++++-------
 1 file changed, 16 insertions(+), 9 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/af42c378/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index abc5c5d..0bf51d7 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -265,22 +265,29 @@ public class StreamThread extends Thread {
                     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();
+                if (!tasks.isEmpty()) {
+                    // try to process one record from each task
+                    requiresPoll = false;
 
-                    totalNumBuffered += task.process();
-                    requiresPoll = requiresPoll || task.requiresPoll();
+                    for (StreamTask task : tasks.values()) {
+                        long startProcess = time.milliseconds();
 
-                    sensors.processTimeSensor.record(time.milliseconds() - startProcess);
+                        totalNumBuffered += task.process();
+                        requiresPoll = requiresPoll || task.requiresPoll();
+
+                        sensors.processTimeSensor.record(time.milliseconds() - startProcess);
+                    }
+
+                    maybePunctuate();
+                    maybeCommit();
+                } else {
+                    // even when no task is assigned, we must poll to get a task.
+                    requiresPoll = true;
                 }
 
-                maybePunctuate();
                 maybeClean();
-                maybeCommit();
             }
         } catch (Exception e) {
             throw new KafkaException(e);


Mime
View raw message