diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java index bc2e1501236ba..662ded553addf 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java @@ -229,16 +229,33 @@ public Sensor addLatencyAndThroughputSensor(final String scopeName, final String operationName, final Sensor.RecordingLevel recordingLevel, final String... tags) { + + return addLatencyAndThroughputSensor(null, + scopeName, + entityName, + operationName, + recordingLevel, + tags); + + } + + public Sensor addLatencyAndThroughputSensor(final String taskName, + final String scopeName, + final String entityName, + final String operationName, + final Sensor.RecordingLevel recordingLevel, + final String... tags) { + final Map tagMap = constructTags(scopeName, entityName, tags); final Map allTagMap = constructTags(scopeName, "all", tags); // first add the global operation metrics if not yet, with the global tags only - final Sensor parent = metrics.sensor(sensorName(operationName, null), recordingLevel); + final Sensor parent = metrics.sensor(sensorName(buildUniqueSensorName(operationName, taskName), null), recordingLevel); addLatencyMetrics(scopeName, parent, operationName, allTagMap); addThroughputMetrics(scopeName, parent, operationName, allTagMap); // add the operation metrics with additional tags - final Sensor sensor = metrics.sensor(sensorName(operationName, entityName), recordingLevel, parent); + final Sensor sensor = metrics.sensor(sensorName(buildUniqueSensorName(operationName, taskName), entityName), recordingLevel, parent); addLatencyMetrics(scopeName, sensor, operationName, tagMap); addThroughputMetrics(scopeName, sensor, operationName, tagMap); @@ -247,20 +264,6 @@ public Sensor addLatencyAndThroughputSensor(final String scopeName, return sensor; } - public Sensor addLatencyAndThroughputSensor(final String taskName, - final String scopeName, - final String entityName, - final String operationName, - final Sensor.RecordingLevel recordingLevel, - final String... tags) { - return addLatencyAndThroughputSensor( - scopeName, - entityName, - threadName + "." + taskName + "." + operationName, - recordingLevel, - tags); - } - /** * @throws IllegalArgumentException if tags is not constructed in key-value pairs */ @@ -270,35 +273,44 @@ public Sensor addThroughputSensor(final String scopeName, final String operationName, final Sensor.RecordingLevel recordingLevel, final String... tags) { + + return addThroughputSensor(null, + scopeName, + entityName, + operationName, + recordingLevel, + tags); + + } + + public Sensor addThroughputSensor(final String taskName, + final String scopeName, + final String entityName, + final String operationName, + final Sensor.RecordingLevel recordingLevel, + final String... tags) { + final Map tagMap = constructTags(scopeName, entityName, tags); final Map allTagMap = constructTags(scopeName, "all", tags); // first add the global operation metrics if not yet, with the global tags only - final Sensor parent = metrics.sensor(sensorName(operationName, null), recordingLevel); + final Sensor parent = metrics.sensor(sensorName(buildUniqueSensorName(operationName, taskName), null), recordingLevel); addThroughputMetrics(scopeName, parent, operationName, allTagMap); // add the operation metrics with additional tags - final Sensor sensor = metrics.sensor(sensorName(operationName, entityName), recordingLevel, parent); + final Sensor sensor = metrics.sensor(sensorName(buildUniqueSensorName(operationName, taskName), entityName), recordingLevel, parent); addThroughputMetrics(scopeName, sensor, operationName, tagMap); parentSensors.put(sensor, parent); return sensor; + } - public Sensor addThroughputSensor(final String taskName, - final String scopeName, - final String entityName, - final String operationName, - final Sensor.RecordingLevel recordingLevel, - final String... tags) { - return addThroughputSensor( - scopeName, - entityName, - threadName + "." + taskName + "." + operationName, - recordingLevel, - tags - ); + + private String buildUniqueSensorName(String operationName, String taskName) { + String task = taskName == null ? "" : taskName + "."; + return threadName + "." + task + operationName; } private void addLatencyMetrics(final String scopeName, final Sensor sensor, final String opName, final Map tags) { 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 6cab6e30ce68e..65dd0228b64c9 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 @@ -118,24 +118,27 @@ public void testMetrics() { assertNotNull(metrics.getSensor("name-mock.0_0." + throughputOperation)); for (final String opName : latencyOperations) { - StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-latency-avg", groupName, metricTags); - StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-latency-max", groupName, metricTags); - StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-rate", groupName, metricTags); - StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-total", groupName, metricTags); + StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-latency-avg", groupName, metricTags); + StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-latency-max", groupName, metricTags); + StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-rate", groupName, metricTags); + StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-total", groupName, metricTags); } - assertNotNull(metrics.metrics().get(metrics.metricName("mock.0_0." + throughputOperation + "-rate", groupName, - "The average number of occurrence of " + "mock.0_0." + throughputOperation + " operation per second.", metricTags))); + assertNotNull(metrics.metrics().get(metrics.metricName(throughputOperation + "-rate", groupName, + "The average number of occurrence of " + throughputOperation + " operation per second.", + metricTags))); // test "all" metricTags.put("processor-node-id", "all"); for (final String opName : latencyOperations) { - StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-latency-avg", groupName, metricTags); - StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-latency-max", groupName, metricTags); - StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-rate", groupName, metricTags); - StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-total", groupName, metricTags); + StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-latency-avg", groupName, metricTags); + StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-latency-max", groupName, metricTags); + StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-rate", groupName, metricTags); + StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-total", groupName, metricTags); } - assertNotNull(metrics.metrics().get(metrics.metricName("mock.0_0." + throughputOperation + "-rate", groupName, - "The average number of occurrence of " + "mock.0_0." + throughputOperation + " operation per second.", metricTags))); + assertNotNull(metrics.metrics().get(metrics.metricName(throughputOperation + "-rate", + groupName, + "The average number of occurrence of " + throughputOperation + " operation per second.", + metricTags))); } diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueBytesStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueBytesStoreTest.java index 70bd2ec3854ce..a5e0d79d25a65 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueBytesStoreTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueBytesStoreTest.java @@ -100,7 +100,7 @@ public void shouldWriteBytesToInnerStoreAndRecordPutMetric() { metered.put(key, value); - final KafkaMetric metric = metric("test.0_0.put-rate"); + final KafkaMetric metric = metric("put-rate"); assertTrue(metric.value() > 0); EasyMock.verify(inner); @@ -114,7 +114,7 @@ public void shouldGetBytesFromInnerStoreAndReturnGetMetric() { assertThat(metered.get(key), equalTo(value)); - final KafkaMetric metric = metric("test.0_0.get-rate"); + final KafkaMetric metric = metric("get-rate"); assertTrue(metric.value() > 0); EasyMock.verify(inner); } @@ -127,7 +127,7 @@ public void shouldPutIfAbsentAndRecordPutIfAbsentMetric() { metered.putIfAbsent(key, value); - final KafkaMetric metric = metric("test.0_0.put-if-absent-rate"); + final KafkaMetric metric = metric("put-if-absent-rate"); assertTrue(metric.value() > 0); EasyMock.verify(inner); } @@ -146,7 +146,7 @@ public void shouldPutAllToInnerStoreAndRecordPutAllMetric() { metered.putAll(Collections.singletonList(KeyValue.pair(key, value))); - final KafkaMetric metric = metric("test.0_0.put-all-rate"); + final KafkaMetric metric = metric("put-all-rate"); assertTrue(metric.value() > 0); EasyMock.verify(inner); } @@ -159,7 +159,7 @@ public void shouldDeleteFromInnerStoreAndRecordDeleteMetric() { metered.delete(key); - final KafkaMetric metric = metric("test.0_0.delete-rate"); + final KafkaMetric metric = metric("delete-rate"); assertTrue(metric.value() > 0); EasyMock.verify(inner); } @@ -176,7 +176,7 @@ public void shouldGetRangeFromInnerStoreAndRecordRangeMetric() { assertFalse(iterator.hasNext()); iterator.close(); - final KafkaMetric metric = metric("test.0_0.range-rate"); + final KafkaMetric metric = metric("range-rate"); assertTrue(metric.value() > 0); EasyMock.verify(inner); } @@ -193,7 +193,7 @@ public void shouldGetAllFromInnerStoreAndRecordAllMetric() { assertFalse(iterator.hasNext()); iterator.close(); - final KafkaMetric metric = metric(new MetricName("test.0_0.all-rate", "stream-scope-metrics", "", tags)); + final KafkaMetric metric = metric(new MetricName("all-rate", "stream-scope-metrics", "", tags)); assertTrue(metric.value() > 0); EasyMock.verify(inner); } @@ -206,7 +206,7 @@ public void shouldFlushInnerWhenFlushTimeRecords() { metered.flush(); - final KafkaMetric metric = metric("test.0_0.flush-rate"); + final KafkaMetric metric = metric("flush-rate"); assertTrue(metric.value() > 0); EasyMock.verify(inner); } diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreTest.java index 7ffa77aa5500d..3bd190aa28419 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreTest.java @@ -98,7 +98,7 @@ public void shouldWriteBytesToInnerStoreAndRecordPutMetric() { metered.put(new Windowed<>(key, new SessionWindow(0, 0)), key); - final KafkaMetric metric = metric("test.0_0.put-rate"); + final KafkaMetric metric = metric("put-rate"); assertTrue(((Double) metric.metricValue()) > 0); EasyMock.verify(inner); } @@ -115,7 +115,7 @@ public void shouldFindSessionsFromStoreAndRecordFetchMetric() { assertFalse(iterator.hasNext()); iterator.close(); - final KafkaMetric metric = metric("test.0_0.fetch-rate"); + final KafkaMetric metric = metric("fetch-rate"); assertTrue(metric.value() > 0); EasyMock.verify(inner); } @@ -132,7 +132,7 @@ public void shouldFindSessionRangeFromStoreAndRecordFetchMetric() { assertFalse(iterator.hasNext()); iterator.close(); - final KafkaMetric metric = metric("test.0_0.fetch-rate"); + final KafkaMetric metric = metric("fetch-rate"); assertTrue(metric.value() > 0); EasyMock.verify(inner); } @@ -146,7 +146,7 @@ public void shouldRemoveFromStoreAndRecordRemoveMetric() { metered.remove(new Windowed<>(key, new SessionWindow(0, 0))); - final KafkaMetric metric = metric("test.0_0.remove-rate"); + final KafkaMetric metric = metric("remove-rate"); assertTrue(metric.value() > 0); EasyMock.verify(inner); } @@ -163,7 +163,7 @@ public void shouldFetchForKeyAndRecordFetchMetric() { assertFalse(iterator.hasNext()); iterator.close(); - final KafkaMetric metric = metric("test.0_0.fetch-rate"); + final KafkaMetric metric = metric("fetch-rate"); assertTrue(metric.value() > 0); EasyMock.verify(inner); } @@ -180,7 +180,7 @@ public void shouldFetchRangeFromStoreAndRecordFetchMetric() { assertFalse(iterator.hasNext()); iterator.close(); - final KafkaMetric metric = metric("test.0_0.fetch-rate"); + final KafkaMetric metric = metric("fetch-rate"); assertTrue(metric.value() > 0); EasyMock.verify(inner); } @@ -188,7 +188,7 @@ public void shouldFetchRangeFromStoreAndRecordFetchMetric() { @Test public void shouldRecordRestoreTimeOnInit() { init(); - final KafkaMetric metric = metric("test.0_0.restore-rate"); + final KafkaMetric metric = metric("restore-rate"); assertTrue(metric.value() > 0); } diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredWindowStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredWindowStoreTest.java index 70677770904eb..19bd523a8d54e 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredWindowStoreTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredWindowStoreTest.java @@ -89,8 +89,8 @@ public void shouldRecordRestoreLatencyOnInit() { EasyMock.replay(innerStoreMock); store.init(context, store); final Map metrics = context.metrics().metrics(); - assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.restore-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue()); - assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.restore-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue()); + assertEquals(1.0, getMetricByNameFilterByTags(metrics, "restore-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue()); + assertEquals(1.0, getMetricByNameFilterByTags(metrics, "restore-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue()); } @Test @@ -103,8 +103,8 @@ public void shouldRecordPutLatency() { store.init(context, store); store.put("a", "a"); final Map metrics = context.metrics().metrics(); - assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.put-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue()); - assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.put-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue()); + assertEquals(1.0, getMetricByNameFilterByTags(metrics, "put-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue()); + assertEquals(1.0, getMetricByNameFilterByTags(metrics, "put-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue()); EasyMock.verify(innerStoreMock); } @@ -116,8 +116,8 @@ public void shouldRecordFetchLatency() { store.init(context, store); store.fetch("a", 1, 1).close(); // recorded on close; final Map metrics = context.metrics().metrics(); - assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.fetch-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue()); - assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.fetch-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue()); + assertEquals(1.0, getMetricByNameFilterByTags(metrics, "fetch-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue()); + assertEquals(1.0, getMetricByNameFilterByTags(metrics, "fetch-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue()); EasyMock.verify(innerStoreMock); } @@ -129,8 +129,8 @@ public void shouldRecordFetchRangeLatency() { store.init(context, store); store.fetch("a", "b", 1, 1).close(); // recorded on close; final Map metrics = context.metrics().metrics(); - assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.fetch-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue()); - assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.fetch-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue()); + assertEquals(1.0, getMetricByNameFilterByTags(metrics, "fetch-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue()); + assertEquals(1.0, getMetricByNameFilterByTags(metrics, "fetch-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue()); EasyMock.verify(innerStoreMock); } @@ -144,8 +144,8 @@ public void shouldRecordFlushLatency() { store.init(context, store); store.flush(); final Map metrics = context.metrics().metrics(); - assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.flush-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue()); - assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.flush-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue()); + assertEquals(1.0, getMetricByNameFilterByTags(metrics, "flush-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue()); + assertEquals(1.0, getMetricByNameFilterByTags(metrics, "flush-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue()); EasyMock.verify(innerStoreMock); }