You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gu...@apache.org on 2017/01/25 19:40:21 UTC
kafka git commit: HOTFIX: KIP-104 follow-up: remove duplicated
per-processor-node throughput metrics, and also rename some other sensors
Repository: kafka
Updated Branches:
refs/heads/trunk 2ca3e59bb -> 0971e62e5
HOTFIX: KIP-104 follow-up: remove duplicated per-processor-node throughput metrics, and also rename some other sensors
This is the HOTFIX PR for any issues detected with KIP-104 until code freeze. Note: do not merge until close to code freeze.
The name changes reflect feedback received while writing the documentation.
Author: Eno Thereska <en...@gmail.com>
Reviewers: Dan Norwood, Guozhang Wang
Closes #2398 from enothereska/hotfix-streams-metrics
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/0971e62e
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/0971e62e
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/0971e62e
Branch: refs/heads/trunk
Commit: 0971e62e50b94a74aae170fb49d1e532e39b2136
Parents: 2ca3e59
Author: Eno Thereska <en...@gmail.com>
Authored: Wed Jan 25 11:40:17 2017 -0800
Committer: Guozhang Wang <wa...@gmail.com>
Committed: Wed Jan 25 11:40:17 2017 -0800
----------------------------------------------------------------------
.../apache/kafka/streams/StreamsMetrics.java | 16 ++++--
.../processor/internals/ProcessorNode.java | 17 +++---
.../streams/processor/internals/SourceNode.java | 2 +-
.../streams/processor/internals/StreamTask.java | 2 +-
.../processor/internals/StreamThread.java | 58 ++++++++++----------
.../processor/internals/StreamsMetricsImpl.java | 8 +--
.../state/internals/MeteredKeyValueStore.java | 18 +++---
.../internals/MeteredSegmentedBytesStore.java | 12 ++--
.../processor/internals/ProcessorNodeTest.java | 10 ++--
.../processor/internals/StreamTaskTest.java | 6 +-
.../processor/internals/StreamThreadTest.java | 42 +++++++-------
.../internals/StreamsMetricsImplTest.java | 4 +-
.../MeteredSegmentedBytesStoreTest.java | 2 +-
13 files changed, 99 insertions(+), 98 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/0971e62e/streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java b/streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java
index afd3497..7b68137 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java
@@ -37,7 +37,7 @@ public interface StreamsMetrics {
Map<MetricName, ? extends Metric> metrics();
/**
- * Add a latency sensor and default associated metrics. Metrics include both latency ones
+ * Add a latency and throughput sensor and default associated metrics. Metrics include both latency ones
* (average and max latency) and throughput ones (operations/time unit).
*
* @param scopeName Name of the scope, could be the type of the state store, etc.
@@ -47,10 +47,13 @@ public interface StreamsMetrics {
* @param tags Additional tags of the sensor.
* @return The added sensor.
*/
- Sensor addLatencySensor(String scopeName, String entityName, String operationName, Sensor.RecordingLevel recordingLevel, String... tags);
+ Sensor addLatencyAndThroughputSensor(String scopeName, String entityName, String operationName, Sensor.RecordingLevel recordingLevel, String... tags);
/**
- * Record the given latency value of the sensor.
+ * Record the given latency value of the sensor. If the passed sensor includes throughput metrics,
+ * e.g., when created by the {@link #addLatencyAndThroughputSensor(String, String, String, Sensor.RecordingLevel, String...)}
+ * method, then the throughput metrics will also be recorded from this event.
+ *
* @param sensor sensor whose latency we are recording.
* @param startNs start of measurement time in nanoseconds.
* @param endNs end of measurement time in nanoseconds.
@@ -59,7 +62,8 @@ public interface StreamsMetrics {
/**
* Add a throughput sensor and default associated metrics. Metrics include throughput ones
- * (operations/time unit).
+ * (operations/time unit). This sensor is a strict subset of the sensor created by
+ * {@link #addLatencyAndThroughputSensor(String, String, String, Sensor.RecordingLevel, String...)}.
*
* @param scopeName Name of the scope, could be the type of the state store, etc.
* @param entityName Name of the entity, could be the name of the state store instance, etc.
@@ -81,7 +85,7 @@ public interface StreamsMetrics {
/**
* Generic method to create a sensor.
* Note that for most cases it is advisable to use {@link #addThroughputSensor(String, String, String, Sensor.RecordingLevel, String...)}
- * or {@link #addLatencySensor(String, String, String, Sensor.RecordingLevel, String...)} to ensure
+ * or {@link #addLatencyAndThroughputSensor(String, String, String, Sensor.RecordingLevel, String...)} to ensure
* metric name well-formedness and conformity with the rest of the streams code base. However,
* if the above two methods are not sufficient, this method can also be used.
* @param name Name of the sensor.
@@ -92,7 +96,7 @@ public interface StreamsMetrics {
/**
* Generic method to create a sensor with parent sensors.
* Note that for most cases it is advisable to use {@link #addThroughputSensor(String, String, String, Sensor.RecordingLevel, String...)}
- * or {@link #addLatencySensor(String, String, String, Sensor.RecordingLevel, String...)} to ensure
+ * or {@link #addLatencyAndThroughputSensor(String, String, String, Sensor.RecordingLevel, String...)} to ensure
* metric name well-formedness and conformity with the rest of the streams code base. However,
* if the above two methods are not sufficient, this method can also be used.
* @param name Name of the sensor.
http://git-wip-us.apache.org/repos/asf/kafka/blob/0971e62e/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
index b66e3df..fa4cc2b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
@@ -132,9 +132,6 @@ public class ProcessorNode<K, V> {
this.value = value;
this.nodeMetrics.metrics.measureLatencyNs(time, processDelegate, nodeMetrics.nodeProcessTimeSensor);
-
- // record throughput
- this.nodeMetrics.metrics.recordThroughput(nodeMetrics.nodeThroughputSensor, 1);
}
public void punctuate(long timestamp) {
@@ -174,7 +171,7 @@ public class ProcessorNode<K, V> {
final Sensor nodeProcessTimeSensor;
final Sensor nodePunctuateTimeSensor;
- final Sensor nodeThroughputSensor;
+ final Sensor sourceNodeForwardSensor;
final Sensor nodeCreationSensor;
final Sensor nodeDestructionSensor;
@@ -189,18 +186,18 @@ public class ProcessorNode<K, V> {
this.metricTags.put(tagKey, tagValue);
// these are all latency metrics
- this.nodeProcessTimeSensor = metrics.addLatencySensor(scope, sensorNamePrefix + "." + name, "process", Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
- this.nodePunctuateTimeSensor = metrics.addLatencySensor(scope, sensorNamePrefix + "." + name, "punctuate", Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
- this.nodeCreationSensor = metrics.addLatencySensor(scope, sensorNamePrefix + "." + name, "create", Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
- this.nodeDestructionSensor = metrics.addLatencySensor(scope, sensorNamePrefix + "." + name, "destroy", Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
- this.nodeThroughputSensor = metrics.addThroughputSensor(scope, sensorNamePrefix + "." + name, "process-throughput", Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
+ this.nodeProcessTimeSensor = metrics.addLatencyAndThroughputSensor(scope, sensorNamePrefix + "." + name, "process", Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
+ this.nodePunctuateTimeSensor = metrics.addLatencyAndThroughputSensor(scope, sensorNamePrefix + "." + name, "punctuate", Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
+ this.nodeCreationSensor = metrics.addLatencyAndThroughputSensor(scope, sensorNamePrefix + "." + name, "create", Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
+ this.nodeDestructionSensor = metrics.addLatencyAndThroughputSensor(scope, sensorNamePrefix + "." + name, "destroy", Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
+ this.sourceNodeForwardSensor = metrics.addThroughputSensor(scope, sensorNamePrefix + "." + name, "forward", Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
}
public void removeAllSensors() {
metrics.removeSensor(nodeProcessTimeSensor);
metrics.removeSensor(nodePunctuateTimeSensor);
- metrics.removeSensor(nodeThroughputSensor);
+ metrics.removeSensor(sourceNodeForwardSensor);
metrics.removeSensor(nodeCreationSensor);
metrics.removeSensor(nodeDestructionSensor);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/0971e62e/streams/src/main/java/org/apache/kafka/streams/processor/internals/SourceNode.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/SourceNode.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/SourceNode.java
index 3406606..d44929c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/SourceNode.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/SourceNode.java
@@ -68,7 +68,7 @@ public class SourceNode<K, V> extends ProcessorNode<K, V> {
@Override
public void process(final K key, final V value) {
context.forward(key, value);
- nodeMetrics.nodeThroughputSensor.record();
+ nodeMetrics.sourceNodeForwardSensor.record();
}
/**
http://git-wip-us.apache.org/repos/asf/kafka/blob/0971e62e/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
index b714221..7375fb5 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
@@ -408,7 +408,7 @@ public class StreamTask extends AbstractTask implements Punctuator {
public TaskMetrics(StreamsMetrics metrics) {
String name = id.toString();
this.metrics = (StreamsMetricsImpl) metrics;
- this.taskCommitTimeSensor = metrics.addLatencySensor("task", name, "commit", Sensor.RecordingLevel.DEBUG, "streams-task-id", name);
+ this.taskCommitTimeSensor = metrics.addLatencyAndThroughputSensor("task", name, "commit", Sensor.RecordingLevel.DEBUG, "streams-task-id", name);
}
public void removeAllSensors() {
http://git-wip-us.apache.org/repos/asf/kafka/blob/0971e62e/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 b445a51..0128142 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
@@ -827,7 +827,7 @@ public class StreamThread extends Thread {
protected StreamTask createStreamTask(TaskId id, Collection<TopicPartition> partitions) {
log.info("{} Creating active task {} with assigned partitions [{}]", logPrefix, id, partitions);
- streamsMetrics.taskCreationSensor.record();
+ streamsMetrics.taskCreatedSensor.record();
final ProcessorTopology topology = builder.build(id.topicGroupId);
final RecordCollector recordCollector = new RecordCollectorImpl(producer, id.toString());
@@ -940,7 +940,7 @@ public class StreamThread extends Thread {
StandbyTask createStandbyTask(TaskId id, Collection<TopicPartition> partitions) {
log.info("{} Creating new standby task {} with assigned partitions [{}]", logPrefix, id, partitions);
- streamsMetrics.taskCreationSensor.record();
+ streamsMetrics.taskCreatedSensor.record();
ProcessorTopology topology = builder.build(id.topicGroupId);
@@ -1044,7 +1044,7 @@ public class StreamThread extends Thread {
public void apply(final AbstractTask task) {
log.info("{} Closing a task {}", StreamThread.this.logPrefix, task.id());
task.close();
- streamsMetrics.taskDestructionSensor.record();
+ streamsMetrics.tasksClosedSensor.record();
}
}, "close");
}
@@ -1055,7 +1055,7 @@ public class StreamThread extends Thread {
public void apply(final AbstractTask task) {
log.info("{} Closing a task's topology {}", StreamThread.this.logPrefix, task.id());
task.closeTopology();
- streamsMetrics.taskDestructionSensor.record();
+ streamsMetrics.tasksClosedSensor.record();
}
}, "close");
}
@@ -1111,40 +1111,40 @@ public class StreamThread extends Thread {
final Sensor pollTimeSensor;
final Sensor processTimeSensor;
final Sensor punctuateTimeSensor;
- final Sensor taskCreationSensor;
- final Sensor taskDestructionSensor;
+ final Sensor taskCreatedSensor;
+ final Sensor tasksClosedSensor;
final Sensor skippedRecordsSensor;
public StreamsMetricsThreadImpl(Metrics metrics, String groupName, String prefix, Map<String, String> tags) {
super(metrics, groupName, tags);
- this.commitTimeSensor = metrics.sensor(prefix + ".commit-time", Sensor.RecordingLevel.INFO);
- this.commitTimeSensor.add(metrics.metricName("commit-time-avg", this.groupName, "The average commit time in ms", this.tags), new Avg());
- this.commitTimeSensor.add(metrics.metricName("commit-time-max", this.groupName, "The maximum commit time in ms", this.tags), new Max());
- this.commitTimeSensor.add(metrics.metricName("commit-calls-rate", this.groupName, "The average per-second number of commit calls", this.tags), new Rate(new Count()));
+ this.commitTimeSensor = metrics.sensor(prefix + ".commit-latency", Sensor.RecordingLevel.INFO);
+ this.commitTimeSensor.add(metrics.metricName("commit-latency-avg", this.groupName, "The average commit time in ms", this.tags), new Avg());
+ this.commitTimeSensor.add(metrics.metricName("commit-latency-max", this.groupName, "The maximum commit time in ms", this.tags), new Max());
+ this.commitTimeSensor.add(metrics.metricName("commit-rate", this.groupName, "The average per-second number of commit calls", this.tags), new Rate(new Count()));
- this.pollTimeSensor = metrics.sensor(prefix + ".poll-time", Sensor.RecordingLevel.INFO);
- this.pollTimeSensor.add(metrics.metricName("poll-time-avg", this.groupName, "The average poll time in ms", this.tags), new Avg());
- this.pollTimeSensor.add(metrics.metricName("poll-time-max", this.groupName, "The maximum poll time in ms", this.tags), new Max());
- this.pollTimeSensor.add(metrics.metricName("poll-calls-rate", this.groupName, "The average per-second number of record-poll calls", this.tags), new Rate(new Count()));
+ this.pollTimeSensor = metrics.sensor(prefix + ".poll-latency", Sensor.RecordingLevel.INFO);
+ this.pollTimeSensor.add(metrics.metricName("poll-latency-avg", this.groupName, "The average poll time in ms", this.tags), new Avg());
+ this.pollTimeSensor.add(metrics.metricName("poll-latency-max", this.groupName, "The maximum poll time in ms", this.tags), new Max());
+ this.pollTimeSensor.add(metrics.metricName("poll-rate", this.groupName, "The average per-second number of record-poll calls", this.tags), new Rate(new Count()));
- this.processTimeSensor = metrics.sensor(prefix + ".process-time", Sensor.RecordingLevel.INFO);
- this.processTimeSensor.add(metrics.metricName("process-time-avg", this.groupName, "The average process time in ms", this.tags), new Avg());
- this.processTimeSensor.add(metrics.metricName("process-time-max", this.groupName, "The maximum process time in ms", this.tags), new Max());
- this.processTimeSensor.add(metrics.metricName("process-calls-rate", this.groupName, "The average per-second number of process calls", this.tags), new Rate(new Count()));
+ this.processTimeSensor = metrics.sensor(prefix + ".process-latency", Sensor.RecordingLevel.INFO);
+ this.processTimeSensor.add(metrics.metricName("process-latency-avg", this.groupName, "The average process time in ms", this.tags), new Avg());
+ this.processTimeSensor.add(metrics.metricName("process-latency-max", this.groupName, "The maximum process time in ms", this.tags), new Max());
+ this.processTimeSensor.add(metrics.metricName("process-rate", this.groupName, "The average per-second number of process calls", this.tags), new Rate(new Count()));
- this.punctuateTimeSensor = metrics.sensor(prefix + ".punctuate-time", Sensor.RecordingLevel.INFO);
- this.punctuateTimeSensor.add(metrics.metricName("punctuate-time-avg", this.groupName, "The average punctuate time in ms", this.tags), new Avg());
- this.punctuateTimeSensor.add(metrics.metricName("punctuate-time-max", this.groupName, "The maximum punctuate time in ms", this.tags), new Max());
- this.punctuateTimeSensor.add(metrics.metricName("punctuate-calls-rate", this.groupName, "The average per-second number of punctuate calls", this.tags), new Rate(new Count()));
+ this.punctuateTimeSensor = metrics.sensor(prefix + ".punctuate-latency", Sensor.RecordingLevel.INFO);
+ this.punctuateTimeSensor.add(metrics.metricName("punctuate-latency-avg", this.groupName, "The average punctuate time in ms", this.tags), new Avg());
+ this.punctuateTimeSensor.add(metrics.metricName("punctuate-latency-max", this.groupName, "The maximum punctuate time in ms", this.tags), new Max());
+ this.punctuateTimeSensor.add(metrics.metricName("punctuate-rate", this.groupName, "The average per-second number of punctuate calls", this.tags), new Rate(new Count()));
- this.taskCreationSensor = metrics.sensor(prefix + ".task-creation", Sensor.RecordingLevel.INFO);
- this.taskCreationSensor.add(metrics.metricName("task-creation-rate", this.groupName, "The average per-second number of newly created tasks", this.tags), new Rate(new Count()));
+ this.taskCreatedSensor = metrics.sensor(prefix + ".task-created", Sensor.RecordingLevel.INFO);
+ this.taskCreatedSensor.add(metrics.metricName("task-created-rate", this.groupName, "The average per-second number of newly created tasks", this.tags), new Rate(new Count()));
- this.taskDestructionSensor = metrics.sensor(prefix + ".task-destruction", Sensor.RecordingLevel.INFO);
- this.taskDestructionSensor.add(metrics.metricName("task-destruction-rate", this.groupName, "The average per-second number of destructed tasks", this.tags), new Rate(new Count()));
+ this.tasksClosedSensor = metrics.sensor(prefix + ".task-closed", Sensor.RecordingLevel.INFO);
+ this.tasksClosedSensor.add(metrics.metricName("task-closed-rate", this.groupName, "The average per-second number of closed tasks", this.tags), new Rate(new Count()));
this.skippedRecordsSensor = metrics.sensor(prefix + ".skipped-records");
- this.skippedRecordsSensor.add(metrics.metricName("skipped-records-count", this.groupName, "The average per-second number of skipped records.", this.tags), new Rate(new Count()));
+ this.skippedRecordsSensor.add(metrics.metricName("skipped-records-rate", this.groupName, "The average per-second number of skipped records.", this.tags), new Rate(new Count()));
}
@@ -1159,8 +1159,8 @@ public class StreamThread extends Thread {
removeSensor(pollTimeSensor);
removeSensor(processTimeSensor);
removeSensor(punctuateTimeSensor);
- removeSensor(taskCreationSensor);
- removeSensor(taskDestructionSensor);
+ removeSensor(taskCreatedSensor);
+ removeSensor(tasksClosedSensor);
removeSensor(skippedRecordsSensor);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/0971e62e/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetricsImpl.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetricsImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetricsImpl.java
index e29476b..99ab03c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetricsImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetricsImpl.java
@@ -113,7 +113,7 @@ public class StreamsMetricsImpl implements StreamsMetrics {
* @throws IllegalArgumentException if tags is not constructed in key-value pairs
*/
@Override
- public Sensor addLatencySensor(String scopeName, String entityName, String operationName, Sensor.RecordingLevel recordingLevel, String... tags) {
+ public Sensor addLatencyAndThroughputSensor(String scopeName, String entityName, String operationName, Sensor.RecordingLevel recordingLevel, String... tags) {
Map<String, String> tagMap = tagMap(tags);
// first add the global operation metrics if not yet, with the global tags only
@@ -150,15 +150,15 @@ public class StreamsMetricsImpl implements StreamsMetrics {
}
private void addLatencyMetrics(String scopeName, Sensor sensor, String entityName, String opName, Map<String, String> tags) {
- maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-avg-latency", groupNameFromScope(scopeName),
+ maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-latency-avg", groupNameFromScope(scopeName),
"The average latency of " + entityName + " " + opName + " operation.", tags), new Avg());
- maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-max-latency", groupNameFromScope(scopeName),
+ maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-latency-max", groupNameFromScope(scopeName),
"The max latency of " + entityName + " " + opName + " operation.", tags), new Max());
addThroughputMetrics(scopeName, sensor, entityName, opName, tags);
}
private void addThroughputMetrics(String scopeName, Sensor sensor, String entityName, String opName, Map<String, String> tags) {
- maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-qps", groupNameFromScope(scopeName),
+ maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-rate", groupNameFromScope(scopeName),
"The average number of occurrence of " + entityName + " " + opName + " operation per second.", tags), new Rate(new Count()));
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/0971e62e/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
index 7dc2d33..dc6b7a5 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
@@ -117,15 +117,15 @@ public class MeteredKeyValueStore<K, V> extends WrappedStateStore.AbstractWrappe
this.context = context;
this.root = root;
this.metrics = (StreamsMetricsImpl) context.metrics();
- this.putTime = this.metrics.addLatencySensor(metricScope, name, "put", Sensor.RecordingLevel.DEBUG);
- this.putIfAbsentTime = this.metrics.addLatencySensor(metricScope, name, "put-if-absent", Sensor.RecordingLevel.DEBUG);
- this.getTime = this.metrics.addLatencySensor(metricScope, name, "get", Sensor.RecordingLevel.DEBUG);
- this.deleteTime = this.metrics.addLatencySensor(metricScope, name, "delete", Sensor.RecordingLevel.DEBUG);
- this.putAllTime = this.metrics.addLatencySensor(metricScope, name, "put-all", Sensor.RecordingLevel.DEBUG);
- this.allTime = this.metrics.addLatencySensor(metricScope, name, "all", Sensor.RecordingLevel.DEBUG);
- this.rangeTime = this.metrics.addLatencySensor(metricScope, name, "range", Sensor.RecordingLevel.DEBUG);
- this.flushTime = this.metrics.addLatencySensor(metricScope, name, "flush", Sensor.RecordingLevel.DEBUG);
- this.restoreTime = this.metrics.addLatencySensor(metricScope, name, "restore", Sensor.RecordingLevel.DEBUG);
+ this.putTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name, "put", Sensor.RecordingLevel.DEBUG);
+ this.putIfAbsentTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name, "put-if-absent", Sensor.RecordingLevel.DEBUG);
+ this.getTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name, "get", Sensor.RecordingLevel.DEBUG);
+ this.deleteTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name, "delete", Sensor.RecordingLevel.DEBUG);
+ this.putAllTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name, "put-all", Sensor.RecordingLevel.DEBUG);
+ this.allTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name, "all", Sensor.RecordingLevel.DEBUG);
+ this.rangeTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name, "range", Sensor.RecordingLevel.DEBUG);
+ this.flushTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name, "flush", Sensor.RecordingLevel.DEBUG);
+ this.restoreTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name, "restore", Sensor.RecordingLevel.DEBUG);
// register and possibly restore the state from the logs
metrics.measureLatencyNs(time, initDelegate, this.restoreTime);
http://git-wip-us.apache.org/repos/asf/kafka/blob/0971e62e/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSegmentedBytesStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSegmentedBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSegmentedBytesStore.java
index 4eb3936..4d3771e 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSegmentedBytesStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSegmentedBytesStore.java
@@ -53,13 +53,13 @@ class MeteredSegmentedBytesStore extends WrappedStateStore.AbstractWrappedStateS
public void init(ProcessorContext context, StateStore root) {
final String name = name();
this.metrics = context.metrics();
- this.putTime = this.metrics.addLatencySensor(metricScope, name, "put", Sensor.RecordingLevel.DEBUG);
- this.fetchTime = this.metrics.addLatencySensor(metricScope, name, "fetch", Sensor.RecordingLevel.DEBUG);
- this.flushTime = this.metrics.addLatencySensor(metricScope, name, "flush", Sensor.RecordingLevel.DEBUG);
- this.getTime = this.metrics.addLatencySensor(metricScope, name, "get", Sensor.RecordingLevel.DEBUG);
- this.removeTime = this.metrics.addLatencySensor(metricScope, name, "remove", Sensor.RecordingLevel.DEBUG);
+ this.putTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name, "put", Sensor.RecordingLevel.DEBUG);
+ this.fetchTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name, "fetch", Sensor.RecordingLevel.DEBUG);
+ this.flushTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name, "flush", Sensor.RecordingLevel.DEBUG);
+ this.getTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name, "get", Sensor.RecordingLevel.DEBUG);
+ this.removeTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name, "remove", Sensor.RecordingLevel.DEBUG);
- final Sensor restoreTime = this.metrics.addLatencySensor(metricScope, name, "restore", Sensor.RecordingLevel.DEBUG);
+ final Sensor restoreTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name, "restore", Sensor.RecordingLevel.DEBUG);
// register and possibly restore the state from the logs
final long startNs = time.nanoseconds();
try {
http://git-wip-us.apache.org/repos/asf/kafka/blob/0971e62e/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
index b1d5f08..f2c7e53 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
@@ -103,7 +103,7 @@ public class ProcessorNodeTest {
String name = "task." + context.taskId() + "." + node.name();
String[] entities = {"all", name};
String[] latencyOperations = {"process", "punctuate", "create", "destroy"};
- String throughputOperation = "process-throughput";
+ String throughputOperation = "forward";
String groupName = "stream-processor-node-metrics";
Map<String, String> tags = Collections.singletonMap("processor-node-id", node.name());
@@ -115,14 +115,14 @@ public class ProcessorNodeTest {
for (String entity : entities) {
for (String operation : latencyOperations) {
- assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-avg-latency", groupName,
+ assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-latency-avg", groupName,
"The average latency in milliseconds of " + entity + " " + operation + " operation.", tags)));
- assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-max-latency", groupName,
+ assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-latency-max", groupName,
"The max latency in milliseconds of " + entity + " " + operation + " operation.", tags)));
- assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-qps", groupName,
+ assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-rate", groupName,
"The average number of occurrence of " + entity + " " + operation + " operation per second.", tags)));
}
- assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + throughputOperation + "-qps", groupName,
+ assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + throughputOperation + "-rate", groupName,
"The average number of occurrence of " + entity + " " + throughputOperation + " operation per second.", tags)));
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/0971e62e/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
index 4e95911..0479b9d 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
@@ -196,11 +196,11 @@ public class StreamTaskTest {
assertNotNull(metrics.getSensor(name + "-" + operation));
for (String entity : entities) {
- assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-avg-latency", groupName,
+ assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-latency-avg", groupName,
"The average latency in milliseconds of " + entity + " " + operation + " operation.", tags)));
- assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-max-latency", groupName,
+ assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-latency-max", groupName,
"The max latency in milliseconds of " + entity + " " + operation + " operation.", tags)));
- assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-qps", groupName,
+ assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-rate", groupName,
"The average number of occurrence of " + entity + " " + operation + " operation per second.", tags)));
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/0971e62e/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
index 20d428b..0e98f56 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
@@ -438,29 +438,29 @@ public class StreamThreadTest {
String defaultPrefix = "thread." + thread.threadClientId();
Map<String, String> defaultTags = Collections.singletonMap("client-id", thread.threadClientId());
- assertNotNull(metrics.getSensor(defaultPrefix + ".commit-time"));
- assertNotNull(metrics.getSensor(defaultPrefix + ".poll-time"));
- assertNotNull(metrics.getSensor(defaultPrefix + ".process-time"));
- assertNotNull(metrics.getSensor(defaultPrefix + ".punctuate-time"));
- assertNotNull(metrics.getSensor(defaultPrefix + ".task-creation"));
- assertNotNull(metrics.getSensor(defaultPrefix + ".task-destruction"));
+ assertNotNull(metrics.getSensor(defaultPrefix + ".commit-latency"));
+ assertNotNull(metrics.getSensor(defaultPrefix + ".poll-latency"));
+ assertNotNull(metrics.getSensor(defaultPrefix + ".process-latency"));
+ assertNotNull(metrics.getSensor(defaultPrefix + ".punctuate-latency"));
+ assertNotNull(metrics.getSensor(defaultPrefix + ".task-created"));
+ assertNotNull(metrics.getSensor(defaultPrefix + ".task-closed"));
assertNotNull(metrics.getSensor(defaultPrefix + ".skipped-records"));
- assertNotNull(metrics.metrics().get(metrics.metricName("commit-time-avg", defaultGroupName, "The average commit time in ms", defaultTags)));
- assertNotNull(metrics.metrics().get(metrics.metricName("commit-time-max", defaultGroupName, "The maximum commit time in ms", defaultTags)));
- assertNotNull(metrics.metrics().get(metrics.metricName("commit-calls-rate", defaultGroupName, "The average per-second number of commit calls", defaultTags)));
- assertNotNull(metrics.metrics().get(metrics.metricName("poll-time-avg", defaultGroupName, "The average poll time in ms", defaultTags)));
- assertNotNull(metrics.metrics().get(metrics.metricName("poll-time-max", defaultGroupName, "The maximum poll time in ms", defaultTags)));
- assertNotNull(metrics.metrics().get(metrics.metricName("poll-calls-rate", defaultGroupName, "The average per-second number of record-poll calls", defaultTags)));
- assertNotNull(metrics.metrics().get(metrics.metricName("process-time-avg", defaultGroupName, "The average process time in ms", defaultTags)));
- assertNotNull(metrics.metrics().get(metrics.metricName("process-time-max", defaultGroupName, "The maximum process time in ms", defaultTags)));
- assertNotNull(metrics.metrics().get(metrics.metricName("process-calls-rate", defaultGroupName, "The average per-second number of process calls", defaultTags)));
- assertNotNull(metrics.metrics().get(metrics.metricName("punctuate-time-avg", defaultGroupName, "The average punctuate time in ms", defaultTags)));
- assertNotNull(metrics.metrics().get(metrics.metricName("punctuate-time-max", defaultGroupName, "The maximum punctuate time in ms", defaultTags)));
- assertNotNull(metrics.metrics().get(metrics.metricName("punctuate-calls-rate", defaultGroupName, "The average per-second number of punctuate calls", defaultTags)));
- assertNotNull(metrics.metrics().get(metrics.metricName("task-creation-rate", defaultGroupName, "The average per-second number of newly created tasks", defaultTags)));
- assertNotNull(metrics.metrics().get(metrics.metricName("task-destruction-rate", defaultGroupName, "The average per-second number of destructed tasks", defaultTags)));
- assertNotNull(metrics.metrics().get(metrics.metricName("skipped-records-count", defaultGroupName, "The average per-second number of skipped records.", defaultTags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName("commit-latency-avg", defaultGroupName, "The average commit time in ms", defaultTags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName("commit-latency-max", defaultGroupName, "The maximum commit time in ms", defaultTags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName("commit-rate", defaultGroupName, "The average per-second number of commit calls", defaultTags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName("poll-latency-avg", defaultGroupName, "The average poll time in ms", defaultTags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName("poll-latency-max", defaultGroupName, "The maximum poll time in ms", defaultTags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName("poll-rate", defaultGroupName, "The average per-second number of record-poll calls", defaultTags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName("process-latency-avg", defaultGroupName, "The average process time in ms", defaultTags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName("process-latency-max", defaultGroupName, "The maximum process time in ms", defaultTags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName("process-rate", defaultGroupName, "The average per-second number of process calls", defaultTags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName("punctuate-latency-avg", defaultGroupName, "The average punctuate time in ms", defaultTags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName("punctuate-latency-max", defaultGroupName, "The maximum punctuate time in ms", defaultTags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName("punctuate-rate", defaultGroupName, "The average per-second number of punctuate calls", defaultTags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName("task-created-rate", defaultGroupName, "The average per-second number of newly created tasks", defaultTags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName("task-closed-rate", defaultGroupName, "The average per-second number of closed tasks", defaultTags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName("skipped-records-rate", defaultGroupName, "The average per-second number of skipped records.", defaultTags)));
}
@Test
http://git-wip-us.apache.org/repos/asf/kafka/blob/0971e62e/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsMetricsImplTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsMetricsImplTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsMetricsImplTest.java
index b7c0c2c..c6bc250 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsMetricsImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsMetricsImplTest.java
@@ -61,7 +61,7 @@ public class StreamsMetricsImplTest {
Sensor sensor1a = streamsMetrics.addSensor(sensorName, Sensor.RecordingLevel.DEBUG, sensor1);
streamsMetrics.removeSensor(sensor1a);
- Sensor sensor2 = streamsMetrics.addLatencySensor(scope, entity, operation, Sensor.RecordingLevel.DEBUG);
+ Sensor sensor2 = streamsMetrics.addLatencyAndThroughputSensor(scope, entity, operation, Sensor.RecordingLevel.DEBUG);
streamsMetrics.removeSensor(sensor2);
Sensor sensor3 = streamsMetrics.addThroughputSensor(scope, entity, operation, Sensor.RecordingLevel.DEBUG);
@@ -77,7 +77,7 @@ public class StreamsMetricsImplTest {
Map<String, String> tags = new HashMap<>();
StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(new Metrics(), groupName, tags);
- Sensor sensor1 = streamsMetrics.addLatencySensor(scope, entity, operation, Sensor.RecordingLevel.DEBUG);
+ Sensor sensor1 = streamsMetrics.addLatencyAndThroughputSensor(scope, entity, operation, Sensor.RecordingLevel.DEBUG);
Map<MetricName, ? extends Metric> metrics = streamsMetrics.metrics();
// 6 metrics plus a common metric that keeps track of total registered metrics in Metrics() constructor
http://git-wip-us.apache.org/repos/asf/kafka/blob/0971e62e/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSegmentedBytesStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSegmentedBytesStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSegmentedBytesStoreTest.java
index 9160a73..205e990 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSegmentedBytesStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSegmentedBytesStoreTest.java
@@ -57,7 +57,7 @@ public class MeteredSegmentedBytesStoreTest {
}
@Override
- public Sensor addLatencySensor(String scopeName, String entityName, String operationName, Sensor.RecordingLevel recordLevel, String... tags) {
+ public Sensor addLatencyAndThroughputSensor(String scopeName, String entityName, String operationName, Sensor.RecordingLevel recordLevel, String... tags) {
return metrics.sensor(operationName);
}