kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject kafka git commit: MINOR: StreamThread performance optimization
Date Wed, 16 Dec 2015 07:53:27 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 4ad165c07 -> 841d2d1a2


MINOR: StreamThread performance optimization

guozhangwang

Author: Yasuhiro Matsuda <yasuhiro@confluent.io>

Reviewers: Guozhang Wang

Closes #680 from ymatsuda/perf


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

Branch: refs/heads/trunk
Commit: 841d2d1a26af94ec95c480dbf2453f9c7d28c2f7
Parents: 4ad165c
Author: Yasuhiro Matsuda <yasuhiro@confluent.io>
Authored: Tue Dec 15 23:53:23 2015 -0800
Committer: Guozhang Wang <wangguoz@gmail.com>
Committed: Tue Dec 15 23:53:23 2015 -0800

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


http://git-wip-us.apache.org/repos/asf/kafka/blob/841d2d1a/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 37d24bb..578357a 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
@@ -84,6 +84,8 @@ public class StreamThread extends Thread {
     private final AtomicBoolean running;
     private final Map<TaskId, StreamTask> activeTasks;
     private final Map<TaskId, StandbyTask> standbyTasks;
+    private final Map<TopicPartition, StreamTask> activeTasksByPartition;
+    private final Map<TopicPartition, StandbyTask> standbyTasksByPartition;
     private final Set<TaskId> prevTasks;
     private final Time time;
     private final File stateDir;
@@ -157,6 +159,8 @@ public class StreamThread extends Thread {
         // initialize the task list
         this.activeTasks = new HashMap<>();
         this.standbyTasks = new HashMap<>();
+        this.activeTasksByPartition = new HashMap<>();
+        this.standbyTasksByPartition = new HashMap<>();
         this.prevTasks = new HashSet<>();
 
         // standby ktables
@@ -261,7 +265,7 @@ public class StreamThread extends Thread {
             removeStreamTasks();
             removeStandbyTasks();
         } catch (Throwable e) {
-            // already logged in removePartition()
+            // already logged in removeStreamTasks() and removeStandbyTasks()
         }
 
         log.info("Stream thread shutdown complete [" + this.getName() + "]");
@@ -279,15 +283,16 @@ public class StreamThread extends Thread {
             while (stillRunning()) {
                 // try to fetch some records if necessary
                 if (requiresPoll) {
+                    requiresPoll = false;
+
                     long startPoll = time.milliseconds();
 
                     ConsumerRecords<byte[], byte[]> records = consumer.poll(totalNumBuffered
== 0 ? this.pollTimeMs : 0);
 
                     if (!records.isEmpty()) {
-                        for (StreamTask task : activeTasks.values()) {
-                            for (TopicPartition partition : task.partitions()) {
-                                task.addRecords(partition, records.records(partition));
-                            }
+                        for (TopicPartition partition : records.partitions()) {
+                            StreamTask task = activeTasksByPartition.get(partition);
+                            task.addRecords(partition, records.records(partition));
                         }
                     }
 
@@ -299,8 +304,6 @@ public class StreamThread extends Thread {
 
                 if (!activeTasks.isEmpty()) {
                     // try to process one record from each task
-                    requiresPoll = false;
-
                     for (StreamTask task : activeTasks.values()) {
                         long startProcess = time.milliseconds();
 
@@ -329,16 +332,15 @@ public class StreamThread extends Thread {
         if (!standbyTasks.isEmpty()) {
             if (processStandbyRecords) {
                 if (!standbyRecords.isEmpty()) {
-                    for (StandbyTask task : standbyTasks.values()) {
-                        for (TopicPartition partition : task.changeLogPartitions()) {
-                            List<ConsumerRecord<byte[], byte[]>> remaining =
standbyRecords.remove(partition);
+                    for (TopicPartition partition : standbyRecords.keySet()) {
+                        StandbyTask task = standbyTasksByPartition.get(partition);
+                        List<ConsumerRecord<byte[], byte[]>> remaining = standbyRecords.remove(partition);
+                        if (remaining != null) {
+                            remaining = task.update(partition, remaining);
                             if (remaining != null) {
-                                remaining = task.update(partition, remaining);
-                                if (remaining != null) {
-                                    standbyRecords.put(partition, remaining);
-                                } else {
-                                    restoreConsumer.resume(partition);
-                                }
+                                standbyRecords.put(partition, remaining);
+                            } else {
+                                restoreConsumer.resume(partition);
                             }
                         }
                     }
@@ -349,13 +351,12 @@ public class StreamThread extends Thread {
             ConsumerRecords<byte[], byte[]> records = restoreConsumer.poll(0);
 
             if (!records.isEmpty()) {
-                for (StandbyTask task : standbyTasks.values()) {
-                    for (TopicPartition partition : task.changeLogPartitions()) {
-                        List<ConsumerRecord<byte[], byte[]>> remaining = task.update(partition,
records.records(partition));
-                        if (remaining != null) {
-                            restoreConsumer.pause(partition);
-                            standbyRecords.put(partition, remaining);
-                        }
+                for (TopicPartition partition : records.partitions()) {
+                    StandbyTask task = standbyTasksByPartition.get(partition);
+                    List<ConsumerRecord<byte[], byte[]>> remaining = task.update(partition,
records.records(partition));
+                    if (remaining != null) {
+                        restoreConsumer.pause(partition);
+                        standbyRecords.put(partition, remaining);
                     }
                 }
             }
@@ -547,9 +548,16 @@ public class StreamThread extends Thread {
         }
 
         // create the active tasks
-        for (TaskId taskId : partitionsForTask.keySet()) {
+        for (Map.Entry<TaskId, Set<TopicPartition>> entry : partitionsForTask.entrySet())
{
+            TaskId taskId = entry.getKey();
+            Set<TopicPartition> partitions = entry.getValue();
+
             try {
-                activeTasks.put(taskId, createStreamTask(taskId, partitionsForTask.get(taskId)));
+                StreamTask task = createStreamTask(taskId, partitions);
+                activeTasks.put(taskId, task);
+
+                for (TopicPartition partition : partitions)
+                    activeTasksByPartition.put(partition, task);
             } catch (Exception e) {
                 log.error("Failed to create an active task #" + taskId + " in thread [" +
this.getName() + "]: ", e);
                 throw e;
@@ -567,6 +575,7 @@ public class StreamThread extends Thread {
         prevTasks.addAll(activeTasks.keySet());
 
         activeTasks.clear();
+        activeTasksByPartition.clear();
     }
 
     private void closeOne(AbstractTask task) {
@@ -605,6 +614,9 @@ public class StreamThread extends Thread {
             StandbyTask task = createStandbyTask(taskId, partitions);
             if (task != null) {
                 standbyTasks.put(taskId, task);
+                for (TopicPartition partition : partitions) {
+                    standbyTasksByPartition.put(partition, task);
+                }
                 // collect checked pointed offsets to position the restore consumer
                 // this include all partitions from which we restore states
                 checkpointedOffsets.putAll(task.checkpointedOffsets());
@@ -633,6 +645,7 @@ public class StreamThread extends Thread {
         restoreConsumer.assign(Collections.<TopicPartition>emptyList());
 
         standbyTasks.clear();
+        standbyTasksByPartition.clear();
     }
 
     private void ensureCopartitioning(Collection<Set<String>> copartitionGroups)
{


Mime
View raw message