root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 3 Packages Projects Releases Wiki Activity

MINOR: small code optimizations in streams 

guozhangwang

Author: Yasuhiro Matsuda <yasuhiro@confluent.io>

Reviewers: Guozhang Wang <wangguoz@gmail.com>

Closes #1176 from ymatsuda/optimize
This commit is contained in:
Yasuhiro Matsuda 2016-04-01 17:14:29 -07:00 committed by Guozhang Wang
parent 75ec67eda8
commit bd5325dd8b
3 changed files with 24 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -49,6 +49,10 @@ public class PartitionGroup {
public TopicPartition partition() { public TopicPartition partition() {
return queue.partition(); return queue.partition();
} }
public RecordQueue queue() {
return queue;
}
} }
// since task is thread-safe, we do not need to synchronize on local variables // since task is thread-safe, we do not need to synchronize on local variables
@ -88,7 +92,7 @@ public class PartitionGroup {
// get the first record from this queue. // get the first record from this queue.
record = queue.poll(); record = queue.poll();
if (queue.size() > 0) { if (!queue.isEmpty()) {
queuesByTime.offer(queue); queuesByTime.offer(queue);
} }
} }

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

@ -179,7 +179,7 @@ public class StreamTask extends AbstractTask implements Punctuator {
// after processing this record, if its partition queue's buffered size has been // after processing this record, if its partition queue's buffered size has been
// decreased to the threshold, we can then resume the consumption on this partition // decreased to the threshold, we can then resume the consumption on this partition
if (partitionGroup.numBuffered(partition) == this.maxBufferedSize) { if (recordInfo.queue().size() == this.maxBufferedSize) {
consumer.resume(singleton(partition)); consumer.resume(singleton(partition));
requiresPoll = true; requiresPoll = true;
} }
@ -320,13 +320,13 @@ public class StreamTask extends AbstractTask implements Punctuator {
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
public <K, V> void forward(K key, V value) { public <K, V> void forward(K key, V value) {
ProcessorNode thisNode = currNode; ProcessorNode thisNode = currNode;
for (ProcessorNode childNode : (List<ProcessorNode<K, V>>) thisNode.children()) { try {
currNode = childNode; for (ProcessorNode childNode : (List<ProcessorNode<K, V>>) thisNode.children()) {
try { currNode = childNode;
childNode.process(key, value); childNode.process(key, value);
} finally {
currNode = thisNode;
} }
} finally {
currNode = thisNode;
} }
} }

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

@ -350,9 +350,12 @@ public class StreamThread extends Thread {
requiresPoll = requiresPoll || task.requiresPoll(); requiresPoll = requiresPoll || task.requiresPoll();
sensors.processTimeSensor.record(time.milliseconds() - startProcess); sensors.processTimeSensor.record(time.milliseconds() - startProcess);
}
maybePunctuate(); maybePunctuate(task);
if (task.commitNeeded())
commitOne(task, time.milliseconds());
}
// if pollTimeMs has passed since the last poll, we poll to respond to a possible rebalance // if pollTimeMs has passed since the last poll, we poll to respond to a possible rebalance
// even when we paused all partitions. // even when we paused all partitions.
@ -424,18 +427,16 @@ public class StreamThread extends Thread {
return true; return true;
} }
private void maybePunctuate() { private void maybePunctuate(StreamTask task) {
for (StreamTask task : activeTasks.values()) { try {
try { long now = time.milliseconds();
long now = time.milliseconds();
if (task.maybePunctuate(now)) if (task.maybePunctuate(now))
sensors.punctuateTimeSensor.record(time.milliseconds() - now); sensors.punctuateTimeSensor.record(time.milliseconds() - now);
} catch (KafkaException e) { } catch (KafkaException e) {
log.error("Failed to punctuate active task #" + task.id() + " in thread [" + this.getName() + "]: ", e); log.error("Failed to punctuate active task #" + task.id() + " in thread [" + this.getName() + "]: ", e);
throw e; throw e;
}
} }
} }
@ -449,16 +450,6 @@ public class StreamThread extends Thread {
lastCommit = now; lastCommit = now;
processStandbyRecords = true; processStandbyRecords = true;
} else {
for (StreamTask task : activeTasks.values()) {
try {
if (task.commitNeeded())
commitOne(task, time.milliseconds());
} catch (KafkaException e) {
log.error("Failed to commit active task #" + task.id() + " in thread [" + this.getName() + "]: ", e);
throw e;
}
}
} }
} }