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 <[email protected]> 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 <[email protected]> Authored: Wed Jan 25 11:40:17 2017 -0800 Committer: Guozhang Wang <[email protected]> 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); }
