You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2019/08/03 21:03:29 UTC
[kafka] branch trunk updated: MINOR: Avoid dividing by zero (#7143)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 320f7b0 MINOR: Avoid dividing by zero (#7143)
320f7b0 is described below
commit 320f7b0d7afd75c6d76ee33770461e93bef43689
Author: Matthias J. Sax <ma...@confluent.io>
AuthorDate: Sat Aug 3 14:03:15 2019 -0700
MINOR: Avoid dividing by zero (#7143)
Reviews: A. Sophie Blee-Goldman <so...@confluent.io>, Bill Bejeck <bi...@confluent.io>, Bruno Cadonna <br...@confluent.io>, Boyang Chen <bo...@confluent.io>, Guozhang Wang <gu...@confluent.io>
---
.../apache/kafka/streams/processor/internals/StreamThread.java | 9 ++++-----
1 file changed, 4 insertions(+), 5 deletions(-)
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 1633b30..d3efa9e 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
@@ -1042,7 +1042,7 @@ public class StreamThread extends Thread {
* or if the task producer got fenced (EOS)
*/
boolean maybeCommit() {
- int committed = 0;
+ final int committed;
if (now - lastCommitMs > commitTimeMs) {
if (log.isTraceEnabled()) {
@@ -1050,7 +1050,7 @@ public class StreamThread extends Thread {
taskManager.activeTaskIds(), taskManager.standbyTaskIds(), now - lastCommitMs, commitTimeMs);
}
- committed += taskManager.commitAll();
+ committed = taskManager.commitAll();
if (committed > 0) {
final long intervalCommitLatency = advanceNowAndComputeLatency();
commitSensor.record(intervalCommitLatency / (double) committed, now);
@@ -1067,11 +1067,10 @@ public class StreamThread extends Thread {
lastCommitMs = now;
processStandbyRecords = true;
} else {
- final int commitPerRequested = taskManager.maybeCommitActiveTasksPerUserRequested();
- if (commitPerRequested > 0) {
+ committed = taskManager.maybeCommitActiveTasksPerUserRequested();
+ if (committed > 0) {
final long requestCommitLatency = advanceNowAndComputeLatency();
commitSensor.record(requestCommitLatency / (double) committed, now);
- committed += commitPerRequested;
}
}