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/11 19:35:09 UTC
[1/3] kafka git commit: KAFKA-3715: Add granular metrics to Kafka
Streams and add hierarhical logging levels to Metrics
Repository: kafka
Updated Branches:
refs/heads/trunk 3d60f1e64 -> e43cf2240
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
index 69474b8..c3df49d 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
@@ -5,9 +5,9 @@
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -16,14 +16,28 @@
*/
package org.apache.kafka.streams.state;
+import java.io.File;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.Set;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.MetricsReporter;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsMetrics;
@@ -32,6 +46,7 @@ import org.apache.kafka.streams.processor.StateRestoreCallback;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.StreamPartitioner;
import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.processor.internals.ProcessorNode;
import org.apache.kafka.streams.processor.internals.RecordCollector;
import org.apache.kafka.streams.processor.internals.RecordCollectorImpl;
@@ -40,16 +55,6 @@ import org.apache.kafka.test.MockProcessorContext;
import org.apache.kafka.test.MockTimestampExtractor;
import org.apache.kafka.test.TestUtils;
-import java.io.File;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Properties;
-import java.util.Set;
-
/**
* A component that provides a {@link #context() ProcessingContext} that can be supplied to a {@link KeyValueStore} so that
* all entries written to the Kafka topic by the store during {@link KeyValueStore#flush()} are captured for testing purposes.
@@ -171,8 +176,8 @@ public class KeyValueStoreTestDriver<K, V> {
Serializer<V> valueSerializer,
Deserializer<V> valueDeserializer) {
StateSerdes<K, V> serdes = new StateSerdes<K, V>("unexpected",
- Serdes.serdeFrom(keySerializer, keyDeserializer),
- Serdes.serdeFrom(valueSerializer, valueDeserializer));
+ Serdes.serdeFrom(keySerializer, keyDeserializer),
+ Serdes.serdeFrom(valueSerializer, valueDeserializer));
return new KeyValueStoreTestDriver<K, V>(serdes);
}
@@ -181,18 +186,13 @@ public class KeyValueStoreTestDriver<K, V> {
private final List<KeyValue<K, V>> restorableEntries = new LinkedList<>();
private final MockProcessorContext context;
private final Map<String, StateStore> storeMap = new HashMap<>();
- private static final long DEFAULT_CACHE_SIZE_BYTES = 1 * 1024 * 1024L;
- private final ThreadCache cache = new ThreadCache(DEFAULT_CACHE_SIZE_BYTES);
- private final StreamsMetrics metrics = new StreamsMetrics() {
- @Override
- public Sensor addLatencySensor(String scopeName, String entityName, String operationName, String... tags) {
- return null;
- }
+ private MockTime time = new MockTime();
+ private MetricConfig config = new MetricConfig();
+ private Metrics metrics = new Metrics(config, Arrays.asList((MetricsReporter) new JmxReporter()), time, true);
- @Override
- public void recordLatency(Sensor sensor, long startNs, long endNs) {
- }
- };
+ private static final long DEFAULT_CACHE_SIZE_BYTES = 1 * 1024 * 1024L;
+ private final ThreadCache cache = new ThreadCache("testCache", DEFAULT_CACHE_SIZE_BYTES, new MockStreamsMetrics(new Metrics()));
+ private final StreamsMetrics streamsMetrics = new MockStreamsMetrics(metrics);
private final RecordCollector recordCollector;
private File stateDir = null;
@@ -211,6 +211,7 @@ public class KeyValueStoreTestDriver<K, V> {
recordFlushed(key, value);
}
+
@Override
public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer,
StreamPartitioner<? super K1, ? super V1> partitioner) {
@@ -254,7 +255,7 @@ public class KeyValueStoreTestDriver<K, V> {
@Override
public StreamsMetrics metrics() {
- return metrics;
+ return streamsMetrics;
}
@Override
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreTest.java
index 60eed96..4eadfa3 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreTest.java
@@ -17,12 +17,14 @@
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.streams.kstream.internals.CacheFlushListener;
import org.apache.kafka.streams.kstream.internals.Change;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import org.apache.kafka.streams.processor.internals.RecordCollector;
import org.apache.kafka.streams.state.KeyValueIterator;
@@ -59,7 +61,7 @@ public class CachingKeyValueStoreTest {
cacheFlushListener = new CacheFlushListenerStub<>();
store = new CachingKeyValueStore<>(underlyingStore, Serdes.String(), Serdes.String());
store.setFlushListener(cacheFlushListener);
- cache = new ThreadCache(maxCacheSizeBytes);
+ cache = new ThreadCache("testCache", maxCacheSizeBytes, new MockStreamsMetrics(new Metrics()));
final MockProcessorContext context = new MockProcessorContext(null, null, null, null, (RecordCollector) null, cache);
topic = "topic";
context.setRecordContext(new ProcessorRecordContext(10, 0, 0, topic));
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
index d453316..a4e8df3 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
@@ -24,6 +25,7 @@ import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.kstream.internals.SessionKeySerde;
import org.apache.kafka.streams.kstream.internals.TimeWindow;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import org.apache.kafka.streams.processor.internals.RecordCollector;
import org.apache.kafka.streams.state.KeyValueIterator;
@@ -60,7 +62,7 @@ public class CachingSessionStoreTest {
cachingStore = new CachingSessionStore<>(underlying,
Serdes.String(),
Serdes.Long());
- cache = new ThreadCache(MAX_CACHE_SIZE_BYTES);
+ cache = new ThreadCache("testCache", MAX_CACHE_SIZE_BYTES, new MockStreamsMetrics(new Metrics()));
final MockProcessorContext context = new MockProcessorContext(null, TestUtils.tempDirectory(), null, null, (RecordCollector) null, cache);
context.setRecordContext(new ProcessorRecordContext(DEFAULT_TIMESTAMP, 0, 0, "topic"));
cachingStore.init(context, cachingStore);
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingWindowStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingWindowStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingWindowStoreTest.java
index 427798d..37fc9a0 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingWindowStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingWindowStoreTest.java
@@ -17,12 +17,14 @@
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.kstream.internals.TimeWindow;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import org.apache.kafka.streams.processor.internals.RecordCollector;
import org.apache.kafka.streams.state.KeyValueIterator;
@@ -63,7 +65,7 @@ public class CachingWindowStoreTest {
Serdes.String(),
WINDOW_SIZE);
cachingStore.setFlushListener(cacheListener);
- cache = new ThreadCache(MAX_CACHE_SIZE_BYTES);
+ cache = new ThreadCache("testCache", MAX_CACHE_SIZE_BYTES, new MockStreamsMetrics(new Metrics()));
topic = "topic";
final MockProcessorContext context = new MockProcessorContext(null, TestUtils.tempDirectory(), null, null, (RecordCollector) null, cache);
context.setRecordContext(new ProcessorRecordContext(DEFAULT_TIMESTAMP, 0, 0, topic));
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingSegmentedBytesStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingSegmentedBytesStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingSegmentedBytesStoreTest.java
index d4f9e47..621feb3 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingSegmentedBytesStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingSegmentedBytesStoreTest.java
@@ -18,9 +18,11 @@
package org.apache.kafka.streams.state.internals;
import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.test.MockProcessorContext;
import org.apache.kafka.test.NoOpRecordCollector;
import org.apache.kafka.test.SegmentedBytesStoreStub;
@@ -55,7 +57,7 @@ public class ChangeLoggingSegmentedBytesStoreTest {
Serdes.String(),
Serdes.Long(),
collector,
- new ThreadCache(0));
+ new ThreadCache("testCache", 0, new MockStreamsMetrics(new Metrics())));
context.setTime(0);
store.init(context, store);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheKeyValueStoreIteratorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheKeyValueStoreIteratorTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheKeyValueStoreIteratorTest.java
index 14abaa0..df2fbca 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheKeyValueStoreIteratorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheKeyValueStoreIteratorTest.java
@@ -17,8 +17,10 @@
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.StateSerdes;
@@ -38,7 +40,7 @@ public class MergedSortedCacheKeyValueStoreIteratorTest {
@Before
public void setUp() throws Exception {
store = new InMemoryKeyValueStore<>(namespace);
- cache = new ThreadCache(10000L);
+ cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
}
@Test
@@ -142,7 +144,7 @@ public class MergedSortedCacheKeyValueStoreIteratorTest {
@Test
public void shouldPeekNextKey() throws Exception {
final KeyValueStore<Bytes, byte[]> kv = new InMemoryKeyValueStore<>("one");
- final ThreadCache cache = new ThreadCache(1000000L);
+ final ThreadCache cache = new ThreadCache("testCache", 1000000L, new MockStreamsMetrics(new Metrics()));
byte[][] bytes = {{0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}, {10}};
final String namespace = "one";
for (int i = 0; i < bytes.length - 1; i += 2) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWindowStoreIteratorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWindowStoreIteratorTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWindowStoreIteratorTest.java
index c33f174..b04f248 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWindowStoreIteratorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWindowStoreIteratorTest.java
@@ -17,9 +17,11 @@
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.StateSerdes;
import org.apache.kafka.test.KeyValueIteratorStub;
@@ -36,7 +38,7 @@ public class MergedSortedCacheWindowStoreIteratorTest {
@Test
public void shouldIterateOverValueFromBothIterators() throws Exception {
final List<KeyValue<Bytes, byte[]>> storeValues = new ArrayList<>();
- final ThreadCache cache = new ThreadCache(1000000L);
+ final ThreadCache cache = new ThreadCache("testCache", 1000000L, new MockStreamsMetrics(new Metrics()));
final String namespace = "one";
final StateSerdes<String, String> stateSerdes = new StateSerdes<>("foo", Serdes.String(), Serdes.String());
final List<KeyValue<Long, byte[]>> expectedKvPairs = new ArrayList<>();
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/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 1587f13..6306512 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
@@ -17,6 +17,8 @@
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.serialization.Serdes;
@@ -30,7 +32,9 @@ import org.apache.kafka.test.TestUtils;
import org.junit.Before;
import org.junit.Test;
+import java.util.Collections;
import java.util.HashSet;
+import java.util.Map;
import java.util.Set;
import static org.junit.Assert.assertTrue;
@@ -39,6 +43,7 @@ public class MeteredSegmentedBytesStoreTest {
private final SegmentedBytesStoreStub bytesStore = new SegmentedBytesStoreStub();
private final MeteredSegmentedBytesStore store = new MeteredSegmentedBytesStore(bytesStore, "scope", new MockTime());
private final Set<String> latencyRecorded = new HashSet<>();
+ private final Set<String> throughputRecorded = new HashSet<>();
@SuppressWarnings("unchecked")
@Before
@@ -47,7 +52,12 @@ public class MeteredSegmentedBytesStoreTest {
final StreamsMetrics streamsMetrics = new StreamsMetrics() {
@Override
- public Sensor addLatencySensor(final String scopeName, final String entityName, final String operationName, final String... tags) {
+ public Map<MetricName, ? extends Metric> metrics() {
+ return Collections.unmodifiableMap(metrics.metrics());
+ }
+
+ @Override
+ public Sensor addLatencySensor(String scopeName, String entityName, String operationName, Sensor.RecordingLevel recordLevel, String... tags) {
return metrics.sensor(operationName);
}
@@ -55,6 +65,32 @@ public class MeteredSegmentedBytesStoreTest {
public void recordLatency(final Sensor sensor, final long startNs, final long endNs) {
latencyRecorded.add(sensor.name());
}
+
+ @Override
+ public Sensor addThroughputSensor(String scopeName, String entityName, String operationName, Sensor.RecordingLevel recordLevel, String... tags) {
+ return metrics.sensor(operationName);
+ }
+
+ @Override
+ public void recordThroughput(Sensor sensor, long value) {
+ throughputRecorded.add(sensor.name());
+ }
+
+ @Override
+ public void removeSensor(Sensor sensor) {
+ metrics.removeSensor(sensor.name());
+ }
+
+ @Override
+ public Sensor addSensor(String name, Sensor.RecordingLevel recordLevel) {
+ return metrics.sensor(name);
+ }
+
+ @Override
+ public Sensor addSensor(String name, Sensor.RecordingLevel recordLevel, Sensor... parents) {
+ return metrics.sensor(name);
+ }
+
};
final MockProcessorContext context = new MockProcessorContext(null,
@@ -62,7 +98,7 @@ public class MeteredSegmentedBytesStoreTest {
Serdes.String(),
Serdes.Long(),
new NoOpRecordCollector(),
- new ThreadCache(0)) {
+ new ThreadCache("testCache", 0, streamsMetrics)) {
@Override
public StreamsMetrics metrics() {
return streamsMetrics;
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/state/internals/NamedCacheTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/NamedCacheTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/NamedCacheTest.java
index 99deb50..8efa024 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/NamedCacheTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/NamedCacheTest.java
@@ -17,8 +17,10 @@
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.junit.Before;
import org.junit.Test;
@@ -26,7 +28,9 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
@@ -38,10 +42,12 @@ import static org.junit.Assert.assertSame;
public class NamedCacheTest {
private NamedCache cache;
+ private MockStreamsMetrics streamMetrics;
@Before
public void setUp() throws Exception {
- cache = new NamedCache("name");
+ streamMetrics = new MockStreamsMetrics(new Metrics());
+ cache = new NamedCache("name", streamMetrics);
}
@Test
@@ -68,6 +74,30 @@ public class NamedCacheTest {
}
@Test
+ public void testMetrics() throws Exception {
+ final String scope = "record-cache";
+ final String entityName = cache.name();
+ final String opName = "hitRatio";
+ final String tagKey = "record-cache-id";
+ final String tagValue = cache.name();
+ final String groupName = "stream-" + scope + "-metrics";
+ final Map<String, String> metricTags = new LinkedHashMap<>();
+ metricTags.put(tagKey, tagValue);
+
+ assertNotNull(streamMetrics.registry().getSensor(entityName + "-" + opName));
+ assertNotNull(streamMetrics.registry().metrics().get(streamMetrics.registry().metricName(entityName +
+ "-" + opName + "-avg", groupName, "The current count of " + entityName + " " + opName +
+ " operation.", metricTags)));
+ assertNotNull(streamMetrics.registry().metrics().get(streamMetrics.registry().metricName(entityName +
+ "-" + opName + "-min", groupName, "The current count of " + entityName + " " + opName +
+ " operation.", metricTags)));
+ assertNotNull(streamMetrics.registry().metrics().get(streamMetrics.registry().metricName(entityName +
+ "-" + opName + "-max", groupName, "The current count of " + entityName + " " + opName +
+ " operation.", metricTags)));
+
+ }
+
+ @Test
public void shouldKeepTrackOfSize() throws Exception {
final LRUCacheEntry value = new LRUCacheEntry(new byte[]{0});
cache.put(Bytes.wrap(new byte[]{0}), value);
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSegmentedBytesStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSegmentedBytesStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSegmentedBytesStoreTest.java
index d4c81c3..9ff2762 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSegmentedBytesStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSegmentedBytesStoreTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.Utils;
@@ -24,6 +25,7 @@ import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.kstream.internals.SessionKeySerde;
import org.apache.kafka.streams.kstream.internals.TimeWindow;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.test.MockProcessorContext;
import org.apache.kafka.test.NoOpRecordCollector;
@@ -65,7 +67,7 @@ public class RocksDBSegmentedBytesStoreTest {
Serdes.String(),
Serdes.Long(),
new NoOpRecordCollector(),
- new ThreadCache(0));
+ new ThreadCache("testCache", 0, new MockStreamsMetrics(new Metrics())));
bytesStore.init(context, bytesStore);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreTest.java
index a664e3b..ab4f5da 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreTest.java
@@ -17,10 +17,12 @@
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.kstream.internals.TimeWindow;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.SessionStore;
import org.apache.kafka.test.MockProcessorContext;
@@ -56,7 +58,7 @@ public class RocksDBSessionStoreTest {
Serdes.String(),
Serdes.Long(),
new NoOpRecordCollector(),
- new ThreadCache(0));
+ new ThreadCache("testCache", 0, new MockStreamsMetrics(new Metrics())));
sessionStore.init(context, sessionStore);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java
index fc30740..a522592 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java
@@ -20,6 +20,7 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.Serializer;
@@ -27,6 +28,7 @@ import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import org.apache.kafka.streams.processor.internals.RecordCollector;
import org.apache.kafka.streams.processor.internals.RecordCollectorImpl;
@@ -35,6 +37,7 @@ import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.streams.state.WindowStoreIterator;
import org.apache.kafka.test.MockProcessorContext;
import org.apache.kafka.test.TestUtils;
+import org.junit.Before;
import org.junit.Test;
import java.io.File;
@@ -67,8 +70,17 @@ public class RocksDBWindowStoreTest {
private final Serde<String> stringSerde = Serdes.String();
private final StateSerdes<Integer, String> serdes = new StateSerdes<>("", intSerde, stringSerde);
private static final long DEFAULT_CACHE_SIZE_BYTES = 1 * 1024 * 1024L;
+ private ThreadCache cache;
private final Segments segments = new Segments(windowName, retentionPeriod, numSegments);
+ @Before
+ public void setUp() {
+ cache = new ThreadCache("testCache", DEFAULT_CACHE_SIZE_BYTES, new MockStreamsMetrics(new Metrics()));
+ }
+
+
+
+
@SuppressWarnings("unchecked")
protected <K, V> WindowStore<K, V> createWindowStore(ProcessorContext context, final boolean enableCaching, final boolean retainDuplicates) {
final RocksDBWindowStoreSupplier supplier = new RocksDBWindowStoreSupplier<>(windowName, retentionPeriod, numSegments, retainDuplicates, intSerde, stringSerde, windowSize, true, Collections.<String, String>emptyMap(), enableCaching);
@@ -90,7 +102,7 @@ public class RocksDBWindowStoreTest {
MockProcessorContext context = new MockProcessorContext(
null, baseDir,
byteArraySerde, byteArraySerde,
- recordCollector, new ThreadCache(DEFAULT_CACHE_SIZE_BYTES));
+ recordCollector, cache);
final WindowStore<Integer, String> windowStore = createWindowStore(context, false, true);
long currentTime = 0;
@@ -141,7 +153,7 @@ public class RocksDBWindowStoreTest {
MockProcessorContext context = new MockProcessorContext(
null, baseDir,
byteArraySerde, byteArraySerde,
- recordCollector, new ThreadCache(DEFAULT_CACHE_SIZE_BYTES));
+ recordCollector, cache);
WindowStore<Integer, String> store = createWindowStore(context, false, true);
try {
@@ -215,7 +227,7 @@ public class RocksDBWindowStoreTest {
MockProcessorContext context = new MockProcessorContext(
null, baseDir,
byteArraySerde, byteArraySerde,
- recordCollector, new ThreadCache(DEFAULT_CACHE_SIZE_BYTES));
+ recordCollector, cache);
WindowStore<Integer, String> store = createWindowStore(context, false, true);
try {
@@ -304,7 +316,7 @@ public class RocksDBWindowStoreTest {
MockProcessorContext context = new MockProcessorContext(
null, baseDir,
byteArraySerde, byteArraySerde,
- recordCollector, new ThreadCache(DEFAULT_CACHE_SIZE_BYTES));
+ recordCollector, cache);
WindowStore<Integer, String> store = createWindowStore(context, false, true);
try {
@@ -391,7 +403,7 @@ public class RocksDBWindowStoreTest {
MockProcessorContext context = new MockProcessorContext(
null, baseDir,
byteArraySerde, byteArraySerde,
- recordCollector, new ThreadCache(DEFAULT_CACHE_SIZE_BYTES));
+ recordCollector, cache);
WindowStore<Integer, String> store = createWindowStore(context, false, true);
try {
@@ -447,7 +459,7 @@ public class RocksDBWindowStoreTest {
MockProcessorContext context = new MockProcessorContext(
null, baseDir,
byteArraySerde, byteArraySerde,
- recordCollector, new ThreadCache(DEFAULT_CACHE_SIZE_BYTES));
+ recordCollector, cache);
WindowStore<Integer, String> store = createWindowStore(context, true, false);
assertTrue(store instanceof CachedStateStore);
@@ -475,7 +487,7 @@ public class RocksDBWindowStoreTest {
MockProcessorContext context = new MockProcessorContext(
null, baseDir,
byteArraySerde, byteArraySerde,
- recordCollector, new ThreadCache(DEFAULT_CACHE_SIZE_BYTES));
+ recordCollector, cache);
WindowStore<Integer, String> store = createWindowStore(context, false, true);
RocksDBWindowStore<Integer, String> inner =
@@ -605,7 +617,7 @@ public class RocksDBWindowStoreTest {
MockProcessorContext context = new MockProcessorContext(
null, baseDir,
byteArraySerde, byteArraySerde,
- recordCollector, new ThreadCache(DEFAULT_CACHE_SIZE_BYTES));
+ recordCollector, cache);
WindowStore<Integer, String> store = createWindowStore(context, false, true);
try {
@@ -654,7 +666,7 @@ public class RocksDBWindowStoreTest {
MockProcessorContext context = new MockProcessorContext(
null, baseDir,
byteArraySerde, byteArraySerde,
- recordCollector, new ThreadCache(DEFAULT_CACHE_SIZE_BYTES));
+ recordCollector, cache);
WindowStore<Integer, String> store = createWindowStore(context, false, true);
@@ -702,7 +714,7 @@ public class RocksDBWindowStoreTest {
MockProcessorContext context = new MockProcessorContext(
null, baseDir,
byteArraySerde, byteArraySerde,
- recordCollector, new ThreadCache(DEFAULT_CACHE_SIZE_BYTES));
+ recordCollector, cache);
WindowStore<Integer, String> store = createWindowStore(context, false, true);
try {
@@ -802,7 +814,7 @@ public class RocksDBWindowStoreTest {
MockProcessorContext context = new MockProcessorContext(
null, baseDir,
byteArraySerde, byteArraySerde,
- recordCollector, new ThreadCache(DEFAULT_CACHE_SIZE_BYTES));
+ recordCollector, cache);
File storeDir = new File(baseDir, windowName);
@@ -862,7 +874,7 @@ public class RocksDBWindowStoreTest {
MockProcessorContext context = new MockProcessorContext(
null, baseDir,
byteArraySerde, byteArraySerde,
- recordCollector, new ThreadCache(DEFAULT_CACHE_SIZE_BYTES));
+ recordCollector, cache);
final WindowStore<Integer, String> windowStore = createWindowStore(context, false, true);
context.setRecordContext(createRecordContext(0));
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/state/internals/SegmentIteratorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/SegmentIteratorTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/SegmentIteratorTest.java
index 3869480..3d2da31 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/SegmentIteratorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/SegmentIteratorTest.java
@@ -17,9 +17,11 @@
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.test.MockProcessorContext;
import org.apache.kafka.test.NoOpRecordCollector;
@@ -53,7 +55,7 @@ public class SegmentIteratorTest {
Serdes.String(),
Serdes.String(),
new NoOpRecordCollector(),
- new ThreadCache(0));
+ new ThreadCache("testCache", 0, new MockStreamsMetrics(new Metrics())));
segmentOne.openDB(context);
segmentTwo.openDB(context);
segmentOne.put(Bytes.wrap("a".getBytes()), "1".getBytes());
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/state/internals/SegmentsTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/SegmentsTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/SegmentsTest.java
index 9c2f688..47207ec 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/SegmentsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/SegmentsTest.java
@@ -17,7 +17,9 @@
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.test.MockProcessorContext;
import org.apache.kafka.test.NoOpRecordCollector;
import org.apache.kafka.test.TestUtils;
@@ -46,7 +48,7 @@ public class SegmentsTest {
Serdes.String(),
Serdes.Long(),
new NoOpRecordCollector(),
- new ThreadCache(0));
+ new ThreadCache("testCache", 0, new MockStreamsMetrics(new Metrics())));
segments = new Segments("test", 4 * 60 * 1000, NUM_SEGMENTS);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java
index fe53ec0..a6f7f81 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java
@@ -18,11 +18,10 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.StreamsConfig;
-import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.TopologyBuilder;
@@ -31,6 +30,7 @@ import org.apache.kafka.streams.processor.internals.StateDirectory;
import org.apache.kafka.streams.processor.internals.StreamTask;
import org.apache.kafka.streams.processor.internals.StreamThread;
import org.apache.kafka.streams.processor.internals.StreamsMetadataState;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.apache.kafka.streams.state.ReadOnlyWindowStore;
@@ -187,7 +187,7 @@ public class StreamThreadStateStoreProviderTest {
.singletonList(new TopicPartition("topic", taskId.partition)), topology,
clientSupplier.consumer,
clientSupplier.restoreConsumer,
- streamsConfig, new TheStreamMetrics(), stateDirectory, null, new NoOpRecordCollector()) {
+ streamsConfig, new MockStreamsMetrics(new Metrics()), stateDirectory, null, new MockTime(), new NoOpRecordCollector()) {
@Override
protected void initializeOffsetLimits() {
@@ -221,21 +221,4 @@ public class StreamThreadStateStoreProviderTest {
clientSupplier.restoreConsumer
.updateEndOffsets(offsets);
}
-
- private static class TheStreamMetrics implements StreamsMetrics {
-
- @Override
- public Sensor addLatencySensor(final String scopeName,
- final String entityName,
- final String operationName,
- final String... tags) {
- return null;
- }
-
- @Override
- public void recordLatency(final Sensor sensor, final long startNs, final long endNs) {
-
- }
- }
-
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/state/internals/ThreadCacheTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ThreadCacheTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ThreadCacheTest.java
index 8aaf9e4..97a1f8b 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/ThreadCacheTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/ThreadCacheTest.java
@@ -16,8 +16,11 @@
*/
package org.apache.kafka.streams.state.internals;
+
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.junit.Test;
import java.io.IOException;
@@ -45,8 +48,9 @@ public class ThreadCacheTest {
new KeyValue<>("K5", "V5"));
final KeyValue<String, String> kv = toInsert.get(0);
final String name = "name";
- ThreadCache cache = new ThreadCache(
- toInsert.size() * memoryCacheEntrySize(kv.key.getBytes(), kv.value.getBytes(), ""));
+ ThreadCache cache = new ThreadCache("testCache",
+ toInsert.size() * memoryCacheEntrySize(kv.key.getBytes(), kv.value.getBytes(), ""),
+ new MockStreamsMetrics(new Metrics()));
for (int i = 0; i < toInsert.size(); i++) {
byte[] key = toInsert.get(i).key.getBytes();
@@ -77,7 +81,7 @@ public class ThreadCacheTest {
System.gc();
long prevRuntimeMemory = runtime.totalMemory() - runtime.freeMemory();
- ThreadCache cache = new ThreadCache(desiredCacheSize);
+ ThreadCache cache = new ThreadCache("testCache", desiredCacheSize, new MockStreamsMetrics(new Metrics()));
long size = cache.sizeBytes();
assertEquals(size, 0);
for (int i = 0; i < numElements; i++) {
@@ -151,8 +155,9 @@ public class ThreadCacheTest {
new KeyValue<>("K5", "V5"));
final KeyValue<String, String> kv = toInsert.get(0);
final String namespace = "kafka";
- ThreadCache cache = new ThreadCache(
- memoryCacheEntrySize(kv.key.getBytes(), kv.value.getBytes(), ""));
+ ThreadCache cache = new ThreadCache("testCache",
+ memoryCacheEntrySize(kv.key.getBytes(), kv.value.getBytes(), ""),
+ new MockStreamsMetrics(new Metrics()));
cache.addDirtyEntryFlushListener(namespace, new ThreadCache.DirtyEntryFlushListener() {
@Override
public void apply(final List<ThreadCache.DirtyEntry> dirty) {
@@ -180,7 +185,7 @@ public class ThreadCacheTest {
@Test
public void shouldDelete() throws Exception {
- final ThreadCache cache = new ThreadCache(10000L);
+ final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
final byte[] key = new byte[]{0};
cache.put("name", key, dirtyEntry(key));
@@ -191,7 +196,7 @@ public class ThreadCacheTest {
@Test
public void shouldNotFlushAfterDelete() throws Exception {
final byte[] key = new byte[]{0};
- final ThreadCache cache = new ThreadCache(10000L);
+ final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
final List<ThreadCache.DirtyEntry> received = new ArrayList<>();
final String namespace = "namespace";
cache.addDirtyEntryFlushListener(namespace, new ThreadCache.DirtyEntryFlushListener() {
@@ -211,7 +216,7 @@ public class ThreadCacheTest {
@Test
public void shouldNotBlowUpOnNonExistentKeyWhenDeleting() throws Exception {
- final ThreadCache cache = new ThreadCache(10000L);
+ final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
final byte[] key = new byte[]{0};
cache.put("name", key, dirtyEntry(key));
@@ -220,13 +225,13 @@ public class ThreadCacheTest {
@Test
public void shouldNotBlowUpOnNonExistentNamespaceWhenDeleting() throws Exception {
- final ThreadCache cache = new ThreadCache(10000L);
+ final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
assertNull(cache.delete("name", new byte[]{1}));
}
@Test
public void shouldNotClashWithOverlappingNames() throws Exception {
- final ThreadCache cache = new ThreadCache(10000L);
+ final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
final byte[] nameByte = new byte[]{0};
final byte[] name1Byte = new byte[]{1};
cache.put("name", nameByte, dirtyEntry(nameByte));
@@ -238,7 +243,7 @@ public class ThreadCacheTest {
@Test
public void shouldPeekNextKey() throws Exception {
- final ThreadCache cache = new ThreadCache(10000L);
+ final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
final byte[] theByte = {0};
final String namespace = "streams";
cache.put(namespace, theByte, dirtyEntry(theByte));
@@ -249,7 +254,7 @@ public class ThreadCacheTest {
@Test
public void shouldGetSameKeyAsPeekNext() throws Exception {
- final ThreadCache cache = new ThreadCache(10000L);
+ final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
final byte[] theByte = {0};
final String namespace = "streams";
cache.put(namespace, theByte, dirtyEntry(theByte));
@@ -259,21 +264,21 @@ public class ThreadCacheTest {
@Test(expected = NoSuchElementException.class)
public void shouldThrowIfNoPeekNextKey() throws Exception {
- final ThreadCache cache = new ThreadCache(10000L);
+ final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
final ThreadCache.MemoryLRUCacheBytesIterator iterator = cache.range("", new byte[]{0}, new byte[]{1});
iterator.peekNextKey();
}
@Test
public void shouldReturnFalseIfNoNextKey() throws Exception {
- final ThreadCache cache = new ThreadCache(10000L);
+ final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
final ThreadCache.MemoryLRUCacheBytesIterator iterator = cache.range("", new byte[]{0}, new byte[]{1});
assertFalse(iterator.hasNext());
}
@Test
public void shouldPeekAndIterateOverRange() throws Exception {
- final ThreadCache cache = new ThreadCache(10000L);
+ final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
final byte[][] bytes = {{0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}, {10}};
final String namespace = "streams";
for (final byte[] aByte : bytes) {
@@ -295,7 +300,7 @@ public class ThreadCacheTest {
public void shouldSkipEntriesWhereValueHasBeenEvictedFromCache() throws Exception {
final String namespace = "streams";
final int entrySize = memoryCacheEntrySize(new byte[1], new byte[1], "");
- final ThreadCache cache = new ThreadCache(entrySize * 5);
+ final ThreadCache cache = new ThreadCache("testCache", entrySize * 5, new MockStreamsMetrics(new Metrics()));
cache.addDirtyEntryFlushListener(namespace, new ThreadCache.DirtyEntryFlushListener() {
@Override
public void apply(final List<ThreadCache.DirtyEntry> dirty) {
@@ -317,7 +322,7 @@ public class ThreadCacheTest {
@Test
public void shouldFlushDirtyEntriesForNamespace() throws Exception {
- final ThreadCache cache = new ThreadCache(100000);
+ final ThreadCache cache = new ThreadCache("testCache", 100000, new MockStreamsMetrics(new Metrics()));
final List<byte[]> received = new ArrayList<>();
cache.addDirtyEntryFlushListener("1", new ThreadCache.DirtyEntryFlushListener() {
@Override
@@ -339,7 +344,7 @@ public class ThreadCacheTest {
@Test
public void shouldNotFlushCleanEntriesForNamespace() throws Exception {
- final ThreadCache cache = new ThreadCache(100000);
+ final ThreadCache cache = new ThreadCache("testCache", 100000, new MockStreamsMetrics(new Metrics()));
final List<byte[]> received = new ArrayList<>();
cache.addDirtyEntryFlushListener("1", new ThreadCache.DirtyEntryFlushListener() {
@Override
@@ -379,13 +384,13 @@ public class ThreadCacheTest {
@Test
public void shouldEvictImmediatelyIfCacheSizeIsVerySmall() throws Exception {
- final ThreadCache cache = new ThreadCache(1);
+ final ThreadCache cache = new ThreadCache("testCache", 1, new MockStreamsMetrics(new Metrics()));
shouldEvictImmediatelyIfCacheSizeIsZeroOrVerySmall(cache);
}
@Test
public void shouldEvictImmediatelyIfCacheSizeIsZero() throws Exception {
- final ThreadCache cache = new ThreadCache(0);
+ final ThreadCache cache = new ThreadCache("testCache", 0, new MockStreamsMetrics(new Metrics()));
shouldEvictImmediatelyIfCacheSizeIsZeroOrVerySmall(cache);
}
@@ -393,7 +398,7 @@ public class ThreadCacheTest {
public void shouldEvictAfterPutAll() throws Exception {
final List<ThreadCache.DirtyEntry> received = new ArrayList<>();
final String namespace = "namespace";
- final ThreadCache cache = new ThreadCache(1);
+ final ThreadCache cache = new ThreadCache("testCache", 1, new MockStreamsMetrics(new Metrics()));
cache.addDirtyEntryFlushListener(namespace, new ThreadCache.DirtyEntryFlushListener() {
@Override
public void apply(final List<ThreadCache.DirtyEntry> dirty) {
@@ -409,7 +414,7 @@ public class ThreadCacheTest {
@Test
public void shouldPutAll() throws Exception {
- final ThreadCache cache = new ThreadCache(100000);
+ final ThreadCache cache = new ThreadCache("testCache", 100000, new MockStreamsMetrics(new Metrics()));
cache.putAll("name", Arrays.asList(KeyValue.pair(new byte[]{0}, dirtyEntry(new byte[]{5})),
KeyValue.pair(new byte[]{1}, dirtyEntry(new byte[]{6}))));
@@ -420,7 +425,7 @@ public class ThreadCacheTest {
@Test
public void shouldNotForwardCleanEntryOnEviction() throws Exception {
- final ThreadCache cache = new ThreadCache(0);
+ final ThreadCache cache = new ThreadCache("testCache", 0, new MockStreamsMetrics(new Metrics()));
final List<ThreadCache.DirtyEntry> received = new ArrayList<>();
cache.addDirtyEntryFlushListener("name", new ThreadCache.DirtyEntryFlushListener() {
@Override
@@ -433,7 +438,7 @@ public class ThreadCacheTest {
}
@Test
public void shouldPutIfAbsent() throws Exception {
- final ThreadCache cache = new ThreadCache(100000);
+ final ThreadCache cache = new ThreadCache("testCache", 100000, new MockStreamsMetrics(new Metrics()));
final byte[] key = {10};
final byte[] value = {30};
assertNull(cache.putIfAbsent("n", key, dirtyEntry(value)));
@@ -445,7 +450,7 @@ public class ThreadCacheTest {
public void shouldEvictAfterPutIfAbsent() throws Exception {
final List<ThreadCache.DirtyEntry> received = new ArrayList<>();
final String namespace = "namespace";
- final ThreadCache cache = new ThreadCache(1);
+ final ThreadCache cache = new ThreadCache("testCache", 1, new MockStreamsMetrics(new Metrics()));
cache.addDirtyEntryFlushListener(namespace, new ThreadCache.DirtyEntryFlushListener() {
@Override
public void apply(final List<ThreadCache.DirtyEntry> dirty) {
@@ -463,7 +468,7 @@ public class ThreadCacheTest {
@Test
public void shouldNotLoopForEverWhenEvictingAndCurrentCacheIsEmpty() throws Exception {
final int maxCacheSizeInBytes = 100;
- final ThreadCache threadCache = new ThreadCache(maxCacheSizeInBytes);
+ final ThreadCache threadCache = new ThreadCache("testCache", maxCacheSizeInBytes, new MockStreamsMetrics(new Metrics()));
// trigger a put into another cache on eviction from "name"
threadCache.addDirtyEntryFlushListener("name", new ThreadCache.DirtyEntryFlushListener() {
@Override
@@ -496,7 +501,7 @@ public class ThreadCacheTest {
@Test
public void shouldCleanupNamedCacheOnClose() throws Exception {
- final ThreadCache cache = new ThreadCache(100000);
+ final ThreadCache cache = new ThreadCache("testCache", 100000, new MockStreamsMetrics(new Metrics()));
cache.put("one", new byte[]{1}, cleanEntry(new byte[] {1}));
cache.put("two", new byte[]{1}, cleanEntry(new byte[] {1}));
assertEquals(cache.size(), 2);
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/test/KStreamTestDriver.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/test/KStreamTestDriver.java b/streams/src/test/java/org/apache/kafka/test/KStreamTestDriver.java
index c12c612..e471300 100644
--- a/streams/src/test/java/org/apache/kafka/test/KStreamTestDriver.java
+++ b/streams/src/test/java/org/apache/kafka/test/KStreamTestDriver.java
@@ -18,11 +18,13 @@
package org.apache.kafka.test;
import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.StreamPartitioner;
@@ -75,7 +77,7 @@ public class KStreamTestDriver {
builder.setApplicationId("TestDriver");
this.topology = builder.build(null);
this.stateDir = stateDir;
- this.cache = new ThreadCache(cacheSize);
+ this.cache = new ThreadCache("testCache", cacheSize, new MockStreamsMetrics(new Metrics()));
this.context = new MockProcessorContext(this, stateDir, keySerde, valSerde, new MockRecordCollector(), cache);
this.context.setRecordContext(new ProcessorRecordContext(0, 0, 0, "topic"));
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java b/streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java
index f058e30..9ec2dfd 100644
--- a/streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java
+++ b/streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java
@@ -17,12 +17,24 @@
package org.apache.kafka.test;
-import org.apache.kafka.common.metrics.Sensor;
+import java.io.File;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.MetricsReporter;
import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.processor.internals.ProcessorNode;
import org.apache.kafka.streams.processor.internals.RecordCollector;
import org.apache.kafka.streams.processor.internals.RecordContext;
@@ -32,12 +44,7 @@ import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.state.StateSerdes;
import org.apache.kafka.streams.state.internals.ThreadCache;
-import java.io.File;
-import java.util.Collections;
-import java.util.HashMap;
import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
public class MockProcessorContext implements InternalProcessorContext, RecordCollector.Supplier {
@@ -46,8 +53,13 @@ public class MockProcessorContext implements InternalProcessorContext, RecordCol
private final Serde<?> valSerde;
private final RecordCollector.Supplier recordCollectorSupplier;
private final File stateDir;
+ private final MockTime time = new MockTime();
+ private MetricConfig config = new MetricConfig();
+ private final Metrics metrics;
+ private final StreamsMetrics streamsMetrics;
private final ThreadCache cache;
private Map<String, StateStore> storeMap = new LinkedHashMap<>();
+
private Map<String, StateRestoreCallback> restoreFuncs = new HashMap<>();
long timestamp = -1L;
@@ -82,7 +94,9 @@ public class MockProcessorContext implements InternalProcessorContext, RecordCol
this.keySerde = keySerde;
this.valSerde = valSerde;
this.recordCollectorSupplier = collectorSupplier;
+ this.metrics = new Metrics(config, Arrays.asList((MetricsReporter) new JmxReporter()), time, true);
this.cache = cache;
+ this.streamsMetrics = new MockStreamsMetrics(metrics);
}
@Override
@@ -102,6 +116,10 @@ public class MockProcessorContext implements InternalProcessorContext, RecordCol
this.timestamp = timestamp;
}
+ public Metrics baseMetrics() {
+ return metrics;
+ }
+
@Override
public TaskId taskId() {
return new TaskId(0, 0);
@@ -137,15 +155,7 @@ public class MockProcessorContext implements InternalProcessorContext, RecordCol
@Override
public StreamsMetrics metrics() {
- return new StreamsMetrics() {
- @Override
- public Sensor addLatencySensor(String scopeName, String entityName, String operationName, String... tags) {
- return null;
- }
- @Override
- public void recordLatency(Sensor sensor, long startNs, long endNs) {
- }
- };
+ return streamsMetrics;
}
@Override
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java b/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
index 7dad408..3cf0624 100644
--- a/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
+++ b/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
@@ -5,9 +5,9 @@
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -16,6 +16,15 @@
*/
package org.apache.kafka.test;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
@@ -23,11 +32,12 @@ import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.processor.StateStore;
@@ -41,19 +51,10 @@ import org.apache.kafka.streams.processor.internals.ProcessorTopology;
import org.apache.kafka.streams.processor.internals.RecordCollectorImpl;
import org.apache.kafka.streams.processor.internals.StateDirectory;
import org.apache.kafka.streams.processor.internals.StreamTask;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.internals.ThreadCache;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Queue;
-import java.util.concurrent.atomic.AtomicLong;
-
/**
* This class makes it easier to write tests to verify the behavior of topologies created with a {@link TopologyBuilder}.
* You can test simple topologies that have a single processor, or very complex topologies that have multiple sources, processors,
@@ -171,25 +172,19 @@ public class ProcessorTopologyTestDriver {
offsetsByTopicPartition.put(tp, new AtomicLong());
}
consumer.assign(offsetsByTopicPartition.keySet());
-
+ StreamsMetrics streamsMetrics = new MockStreamsMetrics(new Metrics());
task = new StreamTask(id,
- applicationId,
- partitionsByTopic.values(),
- topology,
- consumer,
- restoreStateConsumer,
- config,
- new StreamsMetrics() {
- @Override
- public Sensor addLatencySensor(String scopeName, String entityName, String operationName, String... tags) {
- return null;
- }
-
- @Override
- public void recordLatency(Sensor sensor, long startNs, long endNs) {
- // do nothing
- }
- }, new StateDirectory(applicationId, TestUtils.tempDirectory().getPath()), new ThreadCache(1024 * 1024), new RecordCollectorImpl(producer, "id"));
+ applicationId,
+ partitionsByTopic.values(),
+ topology,
+ consumer,
+ restoreStateConsumer,
+ config,
+ streamsMetrics,
+ new StateDirectory(applicationId, TestUtils.tempDirectory().getPath()),
+ new ThreadCache("testCache", 1024 * 1024, streamsMetrics),
+ new MockTime(),
+ new RecordCollectorImpl(producer, "id"));
}
/**
@@ -345,7 +340,7 @@ public class ProcessorTopologyTestDriver {
// consumer.subscribe(new TopicPartition(topicName, 1));
// Set up the partition that matches the ID (which is what ProcessorStateManager expects) ...
List<PartitionInfo> partitionInfos = new ArrayList<>();
- partitionInfos.add(new PartitionInfo(topicName , id.partition, null, null, null));
+ partitionInfos.add(new PartitionInfo(topicName, id.partition, null, null, null));
consumer.updatePartitions(topicName, partitionInfos);
consumer.updateEndOffsets(Collections.singletonMap(new TopicPartition(topicName, id.partition), 0L));
}
[2/3] kafka git commit: KAFKA-3715: Add granular metrics to Kafka
Streams and add hierarhical logging levels to Metrics
Posted by gu...@apache.org.
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/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
new file mode 100644
index 0000000..e29476b
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetricsImpl.java
@@ -0,0 +1,210 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.streams.processor.internals;
+
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.MeasurableStat;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Count;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Rate;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.streams.StreamsMetrics;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+public class StreamsMetricsImpl implements StreamsMetrics {
+ private static final Logger log = LoggerFactory.getLogger(StreamsMetricsImpl.class);
+
+ final Metrics metrics;
+ final String groupName;
+ final Map<String, String> tags;
+ final Map<Sensor, Sensor> parentSensors;
+
+ public StreamsMetricsImpl(Metrics metrics, String groupName, Map<String, String> tags) {
+ Objects.requireNonNull(metrics, "Metrics cannot be null");
+
+ this.metrics = metrics;
+ this.groupName = groupName;
+ this.tags = tags;
+ this.parentSensors = new HashMap<>();
+ }
+
+ public Metrics registry() {
+ return metrics;
+ }
+
+ @Override
+ public Sensor addSensor(String name, Sensor.RecordingLevel recordingLevel) {
+ return metrics.sensor(name, recordingLevel);
+ }
+
+ @Override
+ public Sensor addSensor(String name, Sensor.RecordingLevel recordingLevel, Sensor... parents) {
+ return metrics.sensor(name, recordingLevel, parents);
+ }
+
+ @Override
+ public Map<MetricName, ? extends Metric> metrics() {
+ return Collections.unmodifiableMap(this.metrics.metrics());
+ }
+
+ @Override
+ public void recordLatency(Sensor sensor, long startNs, long endNs) {
+ sensor.record(endNs - startNs);
+ }
+
+ @Override
+ public void recordThroughput(Sensor sensor, long value) {
+ sensor.record(value);
+ }
+
+
+ private String groupNameFromScope(String scopeName) {
+ return "stream-" + scopeName + "-metrics";
+ }
+
+ private String sensorName(String operationName, String entityName) {
+ if (entityName == null) {
+ return operationName;
+ } else {
+ return entityName + "-" + operationName;
+ }
+ }
+
+ private Map<String, String> tagMap(String... tags) {
+ // extract the additional tags if there are any
+ Map<String, String> tagMap = new HashMap<>(this.tags);
+ if ((tags.length % 2) != 0)
+ throw new IllegalArgumentException("Tags needs to be specified in key-value pairs");
+
+ for (int i = 0; i < tags.length; i += 2)
+ tagMap.put(tags[i], tags[i + 1]);
+
+ return tagMap;
+ }
+
+
+
+ /**
+ * @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) {
+ Map<String, String> tagMap = tagMap(tags);
+
+ // first add the global operation metrics if not yet, with the global tags only
+ Sensor parent = metrics.sensor(sensorName(operationName, null), recordingLevel);
+ addLatencyMetrics(scopeName, parent, "all", operationName, tagMap);
+
+ // add the operation metrics with additional tags
+ Sensor sensor = metrics.sensor(sensorName(operationName, entityName), recordingLevel, parent);
+ addLatencyMetrics(scopeName, sensor, entityName, operationName, tagMap);
+
+ parentSensors.put(sensor, parent);
+
+ return sensor;
+ }
+
+ /**
+ * @throws IllegalArgumentException if tags is not constructed in key-value pairs
+ */
+ @Override
+ public Sensor addThroughputSensor(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
+ Sensor parent = metrics.sensor(sensorName(operationName, null), recordingLevel);
+ addThroughputMetrics(scopeName, parent, "all", operationName, tagMap);
+
+ // add the operation metrics with additional tags
+ Sensor sensor = metrics.sensor(sensorName(operationName, entityName), recordingLevel, parent);
+ addThroughputMetrics(scopeName, sensor, entityName, operationName, tagMap);
+
+ parentSensors.put(sensor, parent);
+
+ return sensor;
+ }
+
+ 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),
+ "The average latency of " + entityName + " " + opName + " operation.", tags), new Avg());
+ maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-max-latency", 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),
+ "The average number of occurrence of " + entityName + " " + opName + " operation per second.", tags), new Rate(new Count()));
+ }
+
+ private void maybeAddMetric(Sensor sensor, MetricName name, MeasurableStat stat) {
+ if (!metrics.metrics().containsKey(name)) {
+ sensor.add(name, stat);
+ } else {
+ log.debug("Trying to add metric twice " + name);
+ }
+ }
+
+ /**
+ * Helper function. Measure the latency of an action. This is equivalent to
+ * startTs = time.nanoseconds()
+ * action.run()
+ * endTs = time.nanoseconds()
+ * sensor.record(endTs - startTs)
+ * @param time Time object.
+ * @param action Action to run.
+ * @param sensor Sensor to record value.
+ */
+ public void measureLatencyNs(final Time time, final Runnable action, final Sensor sensor) {
+ long startNs = -1;
+ if (sensor.shouldRecord()) {
+ startNs = time.nanoseconds();
+ }
+ action.run();
+ if (startNs != -1) {
+ recordLatency(sensor, startNs, time.nanoseconds());
+ }
+ }
+
+ /**
+ * Deletes a sensor and its parents, if any
+ */
+ @Override
+ public void removeSensor(Sensor sensor) {
+ Sensor parent = null;
+ Objects.requireNonNull(sensor, "Sensor is null");
+
+ metrics.removeSensor(sensor.name());
+ parent = parentSensors.get(sensor);
+ if (parent != null) {
+ metrics.removeSensor(parent.name());
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/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 fb8b3b1..926e5d4 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
@@ -20,9 +20,9 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.KeyValue;
-import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.internals.StreamsMetricsImpl;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
@@ -50,7 +50,56 @@ public class MeteredKeyValueStore<K, V> implements KeyValueStore<K, V> {
private Sensor rangeTime;
private Sensor flushTime;
private Sensor restoreTime;
- private StreamsMetrics metrics;
+ private StreamsMetricsImpl metrics;
+
+
+ private K key;
+ private V value;
+ private Runnable getDelegate = new Runnable() {
+ @Override
+ public void run() {
+ value = inner.get(key);
+ }
+ };
+ private Runnable putDelegate = new Runnable() {
+ @Override
+ public void run() {
+ inner.put(key, value);
+ }
+ };
+ private Runnable putIfAbsentDelegate = new Runnable() {
+ @Override
+ public void run() {
+ value = inner.putIfAbsent(key, value);
+ }
+ };
+ private List<KeyValue<K, V>> entries;
+ private Runnable putAllDelegate = new Runnable() {
+ @Override
+ public void run() {
+ inner.putAll(entries);
+ }
+ };
+ private Runnable deleteDelegate = new Runnable() {
+ @Override
+ public void run() {
+ value = inner.delete(key);
+ }
+ };
+ private Runnable flushDelegate = new Runnable() {
+ @Override
+ public void run() {
+ inner.flush();
+ }
+ };
+ private ProcessorContext context;
+ private StateStore root;
+ private Runnable initDelegate = new Runnable() {
+ @Override
+ public void run() {
+ inner.init(context, root);
+ }
+ };
// always wrap the store with the metered store
public MeteredKeyValueStore(final KeyValueStore<K, V> inner, String metricScope, Time time) {
@@ -67,24 +116,21 @@ public class MeteredKeyValueStore<K, V> implements KeyValueStore<K, V> {
@Override
public void init(ProcessorContext context, StateStore root) {
final String name = name();
- this.metrics = context.metrics();
- this.putTime = this.metrics.addLatencySensor(metricScope, name, "put");
- this.putIfAbsentTime = this.metrics.addLatencySensor(metricScope, name, "put-if-absent");
- this.getTime = this.metrics.addLatencySensor(metricScope, name, "get");
- this.deleteTime = this.metrics.addLatencySensor(metricScope, name, "delete");
- this.putAllTime = this.metrics.addLatencySensor(metricScope, name, "put-all");
- this.allTime = this.metrics.addLatencySensor(metricScope, name, "all");
- this.rangeTime = this.metrics.addLatencySensor(metricScope, name, "range");
- this.flushTime = this.metrics.addLatencySensor(metricScope, name, "flush");
- this.restoreTime = this.metrics.addLatencySensor(metricScope, name, "restore");
+ 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);
// register and possibly restore the state from the logs
- long startNs = time.nanoseconds();
- try {
- inner.init(context, root);
- } finally {
- this.metrics.recordLatency(this.restoreTime, startNs, time.nanoseconds());
- }
+ metrics.measureLatencyNs(time, initDelegate, this.restoreTime);
}
@Override
@@ -99,52 +145,37 @@ public class MeteredKeyValueStore<K, V> implements KeyValueStore<K, V> {
@Override
public V get(K key) {
- long startNs = time.nanoseconds();
- try {
- return this.inner.get(key);
- } finally {
- this.metrics.recordLatency(this.getTime, startNs, time.nanoseconds());
- }
+ this.key = key;
+ metrics.measureLatencyNs(time, getDelegate, this.getTime);
+ return value;
}
@Override
public void put(K key, V value) {
- long startNs = time.nanoseconds();
- try {
- this.inner.put(key, value);
- } finally {
- this.metrics.recordLatency(this.putTime, startNs, time.nanoseconds());
- }
+ this.key = key;
+ this.value = value;
+ metrics.measureLatencyNs(time, putDelegate, this.putTime);
}
@Override
public V putIfAbsent(K key, V value) {
- long startNs = time.nanoseconds();
- try {
- return this.inner.putIfAbsent(key, value);
- } finally {
- this.metrics.recordLatency(this.putIfAbsentTime, startNs, time.nanoseconds());
- }
+ this.key = key;
+ this.value = value;
+ metrics.measureLatencyNs(time, putIfAbsentDelegate, this.putIfAbsentTime);
+ return this.value;
}
@Override
public void putAll(List<KeyValue<K, V>> entries) {
- long startNs = time.nanoseconds();
- try {
- this.inner.putAll(entries);
- } finally {
- this.metrics.recordLatency(this.putAllTime, startNs, time.nanoseconds());
- }
+ this.entries = entries;
+ metrics.measureLatencyNs(time, putAllDelegate, this.putAllTime);
}
@Override
public V delete(K key) {
- long startNs = time.nanoseconds();
- try {
- return this.inner.delete(key);
- } finally {
- this.metrics.recordLatency(this.deleteTime, startNs, time.nanoseconds());
- }
+ this.key = key;
+ metrics.measureLatencyNs(time, deleteDelegate, this.deleteTime);
+ return value;
}
@Override
@@ -169,12 +200,7 @@ public class MeteredKeyValueStore<K, V> implements KeyValueStore<K, V> {
@Override
public void flush() {
- long startNs = time.nanoseconds();
- try {
- this.inner.flush();
- } finally {
- this.metrics.recordLatency(this.flushTime, startNs, time.nanoseconds());
- }
+ metrics.measureLatencyNs(time, flushDelegate, this.flushTime);
}
private class MeteredKeyValueIterator<K1, V1> implements KeyValueIterator<K1, V1> {
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/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 5181c08..e0ed03e 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
@@ -55,13 +55,13 @@ class MeteredSegmentedBytesStore implements SegmentedBytesStore {
public void init(ProcessorContext context, StateStore root) {
final String name = name();
this.metrics = context.metrics();
- this.putTime = this.metrics.addLatencySensor(metricScope, name, "put");
- this.fetchTime = this.metrics.addLatencySensor(metricScope, name, "fetch");
- this.flushTime = this.metrics.addLatencySensor(metricScope, name, "flush");
- this.getTime = this.metrics.addLatencySensor(metricScope, name, "get");
- this.removeTime = this.metrics.addLatencySensor(metricScope, name, "remove");
+ 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);
- final Sensor restoreTime = this.metrics.addLatencySensor(metricScope, name, "restore");
+ final Sensor restoreTime = this.metrics.addLatencySensor(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/e43cf224/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredWindowStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredWindowStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredWindowStore.java
index 6533460..c725c1a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredWindowStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredWindowStore.java
@@ -19,10 +19,10 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.internals.StreamsMetricsImpl;
import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.streams.state.WindowStoreIterator;
@@ -36,7 +36,38 @@ public class MeteredWindowStore<K, V> implements WindowStore<K, V> {
private Sensor fetchTime;
private Sensor flushTime;
private Sensor restoreTime;
- private StreamsMetrics metrics;
+ private StreamsMetricsImpl metrics;
+
+ private ProcessorContext context;
+ private StateStore root;
+ private Runnable initDelegate = new Runnable() {
+ @Override
+ public void run() {
+ inner.init(context, root);
+ }
+ };
+
+ private K key;
+ private V value;
+ private long timestamp;
+ private Runnable putDelegate = new Runnable() {
+ @Override
+ public void run() {
+ inner.put(key, value);
+ }
+ };
+ private Runnable putTsDelegate = new Runnable() {
+ @Override
+ public void run() {
+ inner.put(key, value, timestamp);
+ }
+ };
+ private Runnable flushDelegate = new Runnable() {
+ @Override
+ public void run() {
+ inner.flush();
+ }
+ };
// always wrap the store with the metered store
public MeteredWindowStore(final WindowStore<K, V> inner, String metricScope, Time time) {
@@ -53,19 +84,16 @@ public class MeteredWindowStore<K, V> implements WindowStore<K, V> {
@Override
public void init(ProcessorContext context, StateStore root) {
final String name = name();
- this.metrics = context.metrics();
- this.putTime = this.metrics.addLatencySensor(metricScope, name, "put");
- this.fetchTime = this.metrics.addLatencySensor(metricScope, name, "fetch");
- this.flushTime = this.metrics.addLatencySensor(metricScope, name, "flush");
- this.restoreTime = this.metrics.addLatencySensor(metricScope, name, "restore");
+ this.context = context;
+ this.root = root;
+ this.metrics = (StreamsMetricsImpl) 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.restoreTime = this.metrics.addLatencySensor(metricScope, name, "restore", Sensor.RecordingLevel.DEBUG);
// register and possibly restore the state from the logs
- long startNs = time.nanoseconds();
- try {
- inner.init(context, root);
- } finally {
- this.metrics.recordLatency(this.restoreTime, startNs, time.nanoseconds());
- }
+ metrics.measureLatencyNs(time, initDelegate, this.restoreTime);
}
@Override
@@ -85,22 +113,17 @@ public class MeteredWindowStore<K, V> implements WindowStore<K, V> {
@Override
public void put(K key, V value) {
- long startNs = time.nanoseconds();
- try {
- this.inner.put(key, value);
- } finally {
- this.metrics.recordLatency(this.putTime, startNs, time.nanoseconds());
- }
+ this.key = key;
+ this.value = value;
+ metrics.measureLatencyNs(time, putDelegate, this.putTime);
}
@Override
public void put(K key, V value, long timestamp) {
- long startNs = time.nanoseconds();
- try {
- this.inner.put(key, value, timestamp);
- } finally {
- this.metrics.recordLatency(this.putTime, startNs, time.nanoseconds());
- }
+ this.key = key;
+ this.value = value;
+ this.timestamp = timestamp;
+ metrics.measureLatencyNs(time, putTsDelegate, this.putTime);
}
@Override
@@ -110,12 +133,7 @@ public class MeteredWindowStore<K, V> implements WindowStore<K, V> {
@Override
public void flush() {
- long startNs = time.nanoseconds();
- try {
- this.inner.flush();
- } finally {
- this.metrics.recordLatency(this.flushTime, startNs, time.nanoseconds());
- }
+ metrics.measureLatencyNs(time, flushDelegate, this.flushTime);
}
private class MeteredWindowStoreIterator<E> implements WindowStoreIterator<E> {
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/main/java/org/apache/kafka/streams/state/internals/NamedCache.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/NamedCache.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/NamedCache.java
index 70f8676..849caa7 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/NamedCache.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/NamedCache.java
@@ -17,15 +17,22 @@
package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Min;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsMetrics;
+import org.apache.kafka.streams.processor.internals.StreamsMetricsImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
@@ -39,8 +46,7 @@ class NamedCache {
private LRUNode tail;
private LRUNode head;
private long currentSizeBytes;
- private ThreadCacheMetrics metrics;
-
+ private NamedCacheMetrics namedCacheMetrics;
// JMX stats
private Sensor hitRatio = null;
@@ -51,15 +57,13 @@ class NamedCache {
private long numOverwrites = 0;
private long numFlushes = 0;
- NamedCache(final String name) {
- this(name, null);
- }
-
- NamedCache(final String name, final ThreadCacheMetrics metrics) {
+ NamedCache(final String name, final StreamsMetrics metrics) {
this.name = name;
- this.metrics = metrics != null ? metrics : new ThreadCache.NullThreadCacheMetrics();
+ this.namedCacheMetrics = new NamedCacheMetrics(metrics);
+ }
- this.hitRatio = this.metrics.addCacheSensor(name, "hitRatio");
+ synchronized final String name() {
+ return name;
}
synchronized long hits() {
@@ -181,7 +185,7 @@ class NamedCache {
return null;
} else {
numReadHits++;
- metrics.recordCacheSensor(hitRatio, (double) numReadHits / (double) (numReadHits + numReadMisses));
+ namedCacheMetrics.hitRatioSensor.record((double) numReadHits / (double) (numReadHits + numReadMisses));
}
return node;
}
@@ -308,6 +312,7 @@ class NamedCache {
currentSizeBytes = 0;
dirtyKeys.clear();
cache.clear();
+ namedCacheMetrics.removeAllSensors();
}
/**
@@ -353,4 +358,38 @@ class NamedCache {
}
}
+ class NamedCacheMetrics {
+ final StreamsMetricsImpl metrics;
+ final String groupName;
+ final Map<String, String> metricTags;
+ final Sensor hitRatioSensor;
+
+
+ public NamedCacheMetrics(StreamsMetrics metrics) {
+ final String scope = "record-cache";
+ final String entityName = name;
+ final String opName = "hitRatio";
+ final String tagKey = "record-cache-id";
+ final String tagValue = name;
+ this.groupName = "stream-" + scope + "-metrics";
+ this.metrics = (StreamsMetricsImpl) metrics;
+ this.metricTags = new LinkedHashMap<>();
+ this.metricTags.put(tagKey, tagValue);
+
+
+ hitRatioSensor = this.metrics.registry().sensor(entityName + "-" + opName, Sensor.RecordingLevel.DEBUG);
+
+ hitRatioSensor.add(this.metrics.registry().metricName(entityName + "-" + opName + "-avg", groupName,
+ "The current count of " + entityName + " " + opName + " operation.", metricTags), new Avg());
+ hitRatioSensor.add(this.metrics.registry().metricName(entityName + "-" + opName + "-min", groupName,
+ "The current count of " + entityName + " " + opName + " operation.", metricTags), new Min());
+ hitRatioSensor.add(this.metrics.registry().metricName(entityName + "-" + opName + "-max", groupName,
+ "The current count of " + entityName + " " + opName + " operation.", metricTags), new Max());
+
+ }
+
+ public void removeAllSensors() {
+ metrics.removeSensor(hitRatioSensor);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java
index a66c961..6f33a63 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java
@@ -16,9 +16,9 @@
*/
package org.apache.kafka.streams.state.internals;
-import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.processor.internals.RecordContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -41,7 +41,7 @@ public class ThreadCache {
private final String name;
private final long maxCacheSizeBytes;
private final Map<String, NamedCache> caches = new HashMap<>();
- private final ThreadCacheMetrics metrics;
+ private final StreamsMetrics metrics;
// internal stats
private long numPuts = 0;
@@ -55,14 +55,10 @@ public class ThreadCache {
void apply(final List<DirtyEntry> dirty);
}
- public ThreadCache(long maxCacheSizeBytes) {
- this(null, maxCacheSizeBytes, null);
- }
-
- public ThreadCache(final String name, long maxCacheSizeBytes, final ThreadCacheMetrics metrics) {
+ public ThreadCache(final String name, long maxCacheSizeBytes, final StreamsMetrics metrics) {
this.name = name;
this.maxCacheSizeBytes = maxCacheSizeBytes;
- this.metrics = metrics != null ? metrics : new NullThreadCacheMetrics();
+ this.metrics = metrics;
}
public long puts() {
@@ -152,7 +148,7 @@ public class ThreadCache {
public MemoryLRUCacheBytesIterator range(final String namespace, final byte[] from, final byte[] to) {
final NamedCache cache = getCache(namespace);
if (cache == null) {
- return new MemoryLRUCacheBytesIterator(Collections.<Bytes>emptyIterator(), new NamedCache(namespace));
+ return new MemoryLRUCacheBytesIterator(Collections.<Bytes>emptyIterator(), new NamedCache(namespace, this.metrics));
}
return new MemoryLRUCacheBytesIterator(cache.keyRange(cacheKey(from), cacheKey(to)), cache);
}
@@ -160,7 +156,7 @@ public class ThreadCache {
public MemoryLRUCacheBytesIterator all(final String namespace) {
final NamedCache cache = getCache(namespace);
if (cache == null) {
- return new MemoryLRUCacheBytesIterator(Collections.<Bytes>emptyIterator(), new NamedCache(namespace));
+ return new MemoryLRUCacheBytesIterator(Collections.<Bytes>emptyIterator(), new NamedCache(namespace, this.metrics));
}
return new MemoryLRUCacheBytesIterator(cache.allKeys(), cache);
}
@@ -327,17 +323,4 @@ public class ThreadCache {
}
}
- public static class NullThreadCacheMetrics implements ThreadCacheMetrics {
- @Override
- public Sensor addCacheSensor(String entityName, String operationName, String... tags) {
- return null;
- }
-
- @Override
- public void recordCacheSensor(Sensor sensor, double value) {
- // do nothing
- }
-
- }
-
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCacheMetrics.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCacheMetrics.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCacheMetrics.java
deleted file mode 100644
index cad2697..0000000
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCacheMetrics.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.kafka.streams.state.internals;
-
-import org.apache.kafka.common.metrics.Sensor;
-
-/**
- * The Kafka Streams metrics interface for adding metric sensors and collecting metric values.
- */
-public interface ThreadCacheMetrics {
-
- /**
- * Add the hit ratio sensor.
- * @param entityName Name of the entity, could be the name of the cache instance, etc.
- * @param operationName Name of the operation, could be "hit ratio".
- * @param tags Additional tags of the sensor.
- * @return The added sensor.
- */
- Sensor addCacheSensor(String entityName, String operationName, String... tags);
-
- /**
- * Record the given value of the sensor.
- */
- void recordCacheSensor(Sensor sensor, double value);
-}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java b/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
index 5804407..50efe5b 100644
--- a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
@@ -17,6 +17,10 @@
package org.apache.kafka.streams;
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
@@ -38,6 +42,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -142,6 +147,41 @@ public class KafkaStreamsTest {
}
}
+ @Test
+ public void testNumberDefaultMetrics() {
+ final KafkaStreams streams = createKafkaStreams();
+ final Map<MetricName, ? extends Metric> metrics = streams.metrics();
+ // all 15 default StreamThread metrics + 1 metric that keeps track of number of metrics
+ assertEquals(metrics.size(), 16);
+ }
+
+ @Test(expected = ConfigException.class)
+ public void testIllegalMetricsConfig() {
+ final Properties props = new Properties();
+ props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "appId");
+ props.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+ props.setProperty(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG, "illegalConfig");
+ final KStreamBuilder builder = new KStreamBuilder();
+ final KafkaStreams streams = new KafkaStreams(builder, props);
+
+ }
+
+ @Test
+ public void testLegalMetricsConfig() {
+ final Properties props = new Properties();
+ props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "appId");
+ props.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+ props.setProperty(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG, Sensor.RecordingLevel.INFO.toString());
+ final KStreamBuilder builder1 = new KStreamBuilder();
+ final KafkaStreams streams1 = new KafkaStreams(builder1, props);
+ streams1.close();
+
+ props.setProperty(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG, Sensor.RecordingLevel.DEBUG.toString());
+ final KStreamBuilder builder2 = new KStreamBuilder();
+ final KafkaStreams streams2 = new KafkaStreams(builder2, props);
+
+ }
+
@Test(expected = IllegalStateException.class)
public void shouldNotGetAllTasksWhenNotRunning() throws Exception {
final KafkaStreams streams = createKafkaStreams();
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
index ba95522..0ecaf3a 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.streams.kstream.internals;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.Aggregator;
@@ -26,6 +27,7 @@ import org.apache.kafka.streams.kstream.Merger;
import org.apache.kafka.streams.kstream.SessionWindows;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.processor.Processor;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.internals.RocksDBSessionStoreSupplier;
import org.apache.kafka.streams.state.SessionStore;
@@ -89,7 +91,8 @@ public class KStreamSessionWindowAggregateProcessorTest {
@Before
public void initializeStore() {
final File stateDir = TestUtils.tempDirectory();
- context = new MockProcessorContext(new KStreamTestDriver(new KStreamBuilder(), stateDir), stateDir, Serdes.String(), Serdes.String(), new NoOpRecordCollector(), new ThreadCache(100000)) {
+ context = new MockProcessorContext(new KStreamTestDriver(new KStreamBuilder(), stateDir), stateDir,
+ Serdes.String(), Serdes.String(), new NoOpRecordCollector(), new ThreadCache("testCache", 100000, new MockStreamsMetrics(new Metrics()))) {
@Override
public <K, V> void forward(final K key, final V value) {
results.add(KeyValue.pair(key, value));
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java b/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java
index 264be3b..7ba6161 100644
--- a/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java
+++ b/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java
@@ -129,6 +129,8 @@ public class SimpleBenchmark {
benchmark.processStreamWithSink(SOURCE_TOPIC);
// simple stream performance source->store
benchmark.processStreamWithStateStore(SOURCE_TOPIC);
+ // simple stream performance source->cache->store
+ benchmark.processStreamWithCachedStateStore(SOURCE_TOPIC);
// simple streams performance KSTREAM-KTABLE join
benchmark.kStreamKTableJoin(JOIN_TOPIC_1_PREFIX + "kStreamKTable", JOIN_TOPIC_2_PREFIX + "kStreamKTable");
// simple streams performance KSTREAM-KSTREAM join
@@ -302,11 +304,8 @@ public class SimpleBenchmark {
}
}
- public void processStreamWithStateStore(String topic) {
- CountDownLatch latch = new CountDownLatch(1);
-
- final KafkaStreams streams = createKafkaStreamsWithStateStore(topic, stateDir, kafka, zookeeper, latch);
-
+ private void internalProcessStreamWithStore(final KafkaStreams streams, final CountDownLatch latch,
+ final String message) {
Thread thread = new Thread() {
public void run() {
streams.start();
@@ -326,7 +325,7 @@ public class SimpleBenchmark {
long endTime = System.currentTimeMillis();
- System.out.println("Streams Performance [MB/sec read+store]: " + megaBytePerSec(endTime - startTime));
+ System.out.println(message + megaBytePerSec(endTime - startTime));
streams.close();
try {
@@ -335,6 +334,21 @@ public class SimpleBenchmark {
// ignore
}
}
+ public void processStreamWithStateStore(String topic) {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ final KafkaStreams streams = createKafkaStreamsWithStateStore(topic, stateDir, kafka, zookeeper, latch, false);
+ internalProcessStreamWithStore(streams, latch, "Streams Performance [MB/sec read+store]: ");
+
+ }
+
+ public void processStreamWithCachedStateStore(String topic) {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ final KafkaStreams streams = createKafkaStreamsWithStateStore(topic, stateDir, kafka, zookeeper, latch, true);
+
+ internalProcessStreamWithStore(streams, latch, "Streams Performance [MB/sec read+cache+store]: ");
+ }
/**
* Produce values to a topic
@@ -552,9 +566,10 @@ public class SimpleBenchmark {
}
private KafkaStreams createKafkaStreamsWithStateStore(String topic, File stateDir, String kafka, String zookeeper,
- final CountDownLatch latch) {
+ final CountDownLatch latch,
+ boolean enableCaching) {
Properties props = new Properties();
- props.put(StreamsConfig.APPLICATION_ID_CONFIG, "simple-benchmark-streams-with-store");
+ props.put(StreamsConfig.APPLICATION_ID_CONFIG, "simple-benchmark-streams-with-store" + enableCaching);
props.put(StreamsConfig.STATE_DIR_CONFIG, stateDir.toString());
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
@@ -562,8 +577,11 @@ public class SimpleBenchmark {
KStreamBuilder builder = new KStreamBuilder();
- builder.addStateStore(Stores.create("store").withIntegerKeys().withByteArrayValues().persistent().build());
-
+ if (enableCaching) {
+ builder.addStateStore(Stores.create("store").withIntegerKeys().withByteArrayValues().persistent().enableCaching().build());
+ } else {
+ builder.addStateStore(Stores.create("store").withIntegerKeys().withByteArrayValues().persistent().build());
+ }
KStream<Integer, byte[]> source = builder.stream(INTEGER_SERDE, BYTE_SERDE, topic);
source.process(new ProcessorSupplier<Integer, byte[]>() {
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractTaskTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractTaskTest.java
index b1c19ba..cf161f8 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractTaskTest.java
@@ -25,6 +25,7 @@ import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.AuthorizationException;
import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.streams.errors.ProcessorStateException;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.TaskId;
@@ -72,7 +73,7 @@ public class AbstractTaskTest {
consumer,
false,
new StateDirectory("app", TestUtils.tempDirectory().getPath()),
- new ThreadCache(0)) {
+ new ThreadCache("testCache", 0, new MockStreamsMetrics(new Metrics()))) {
@Override
public void commit() {
// do nothing
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/processor/internals/MockStreamsMetrics.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/MockStreamsMetrics.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/MockStreamsMetrics.java
new file mode 100644
index 0000000..f6cbbf2
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/MockStreamsMetrics.java
@@ -0,0 +1,28 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.internals;
+
+import java.util.Collections;
+import org.apache.kafka.common.metrics.Metrics;
+
+public class MockStreamsMetrics extends StreamsMetricsImpl {
+
+ public MockStreamsMetrics(Metrics metrics) {
+ super(metrics, "mock-stream-metrics",
+ Collections.<String, String>emptyMap());
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/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 3aafc00..b1d5f08 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
@@ -17,12 +17,19 @@
package org.apache.kafka.streams.processor.internals;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.state.StateSerdes;
+import org.apache.kafka.test.MockProcessorContext;
import org.junit.Test;
import java.util.Collections;
+import java.util.Map;
+
+import static org.junit.Assert.assertNotNull;
public class ProcessorNodeTest {
@@ -62,4 +69,62 @@ public class ProcessorNodeTest {
}
}
+ private static class NoOpProcessor implements Processor {
+ @Override
+ public void init(final ProcessorContext context) {
+
+ }
+
+ @Override
+ public void process(final Object key, final Object value) {
+
+ }
+
+ @Override
+ public void punctuate(final long timestamp) {
+
+ }
+
+ @Override
+ public void close() {
+
+ }
+ }
+
+ @Test
+ public void testMetrics() {
+ final StateSerdes anyStateSerde = StateSerdes.withBuiltinTypes("anyName", Bytes.class, Bytes.class);
+
+ final MockProcessorContext context = new MockProcessorContext(anyStateSerde, new RecordCollectorImpl(null, null));
+ final ProcessorNode node = new ProcessorNode("name", new NoOpProcessor(), Collections.emptySet());
+ node.init(context);
+
+ Metrics metrics = context.baseMetrics();
+ String name = "task." + context.taskId() + "." + node.name();
+ String[] entities = {"all", name};
+ String[] latencyOperations = {"process", "punctuate", "create", "destroy"};
+ String throughputOperation = "process-throughput";
+ String groupName = "stream-processor-node-metrics";
+ Map<String, String> tags = Collections.singletonMap("processor-node-id", node.name());
+
+ for (String operation : latencyOperations) {
+ assertNotNull(metrics.getSensor(operation));
+ assertNotNull(metrics.getSensor(name + "-" + operation));
+ }
+ assertNotNull(metrics.getSensor(throughputOperation));
+
+ for (String entity : entities) {
+ for (String operation : latencyOperations) {
+ assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-avg-latency", groupName,
+ "The average latency in milliseconds of " + entity + " " + operation + " operation.", tags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-max-latency", groupName,
+ "The max latency in milliseconds of " + entity + " " + operation + " operation.", tags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-qps", groupName,
+ "The average number of occurrence of " + entity + " " + operation + " operation per second.", tags)));
+ }
+ assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + throughputOperation + "-qps", groupName,
+ "The average number of occurrence of " + entity + " " + throughputOperation + " operation per second.", tags)));
+ }
+ }
+
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
index 10a86fe..c5fecfa 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
@@ -24,13 +24,12 @@ import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.StreamsConfig;
-import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.TaskId;
@@ -321,17 +320,8 @@ public class StandbyTaskTest {
builder.stream("topic").groupByKey().count("my-store");
final ProcessorTopology topology = builder.build(0);
StreamsConfig config = createConfig(baseDir);
- new StandbyTask(taskId, applicationId, partitions, topology, consumer, restoreStateConsumer, config, new StreamsMetrics() {
- @Override
- public Sensor addLatencySensor(final String scopeName, final String entityName, final String operationName, final String... tags) {
- return null;
- }
-
- @Override
- public void recordLatency(final Sensor sensor, final long startNs, final long endNs) {
-
- }
- }, stateDirectory);
+ new StandbyTask(taskId, applicationId, partitions, topology, consumer, restoreStateConsumer, config,
+ new MockStreamsMetrics(new Metrics()), stateDirectory);
}
private List<ConsumerRecord<byte[], byte[]>> records(ConsumerRecord<byte[], byte[]>... recs) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/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 3d50007..adabde7 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
@@ -23,14 +23,17 @@ import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.processor.AbstractProcessor;
import org.apache.kafka.streams.processor.ProcessorContext;
@@ -58,6 +61,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -131,7 +135,8 @@ public class StreamTaskTest {
public void testProcessOrder() throws Exception {
StreamsConfig config = createConfig(baseDir);
recordCollector = new RecordCollectorImpl(producer, "taskId");
- StreamTask task = new StreamTask(new TaskId(0, 0), "applicationId", partitions, topology, consumer, restoreStateConsumer, config, null, stateDirectory, null, recordCollector);
+ StreamTask task = new StreamTask(new TaskId(0, 0), "applicationId", partitions, topology, consumer,
+ restoreStateConsumer, config, new MockStreamsMetrics(new Metrics()), stateDirectory, null, new MockTime(), recordCollector);
task.addRecords(partition1, records(
new ConsumerRecord<>(partition1.topic(), partition1.partition(), 10, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue),
@@ -171,14 +176,41 @@ public class StreamTaskTest {
task.close();
+ }
+ @Test
+ public void testMetrics() throws Exception {
+ StreamsConfig config = createConfig(baseDir);
+ recordCollector = new RecordCollectorImpl(producer, "taskId");
+ Metrics metrics = new Metrics();
+ StreamTask task = new StreamTask(new TaskId(0, 0), "applicationId", partitions, topology, consumer,
+ restoreStateConsumer, config, new MockStreamsMetrics(metrics), stateDirectory, null, new MockTime(), recordCollector);
+ String name = task.id().toString();
+ String[] entities = {"all", name};
+ String operation = "commit";
+
+ String groupName = "stream-task-metrics";
+ Map<String, String> tags = Collections.singletonMap("streams-task-id", name);
+
+ assertNotNull(metrics.getSensor(operation));
+ assertNotNull(metrics.getSensor(name + "-" + operation));
+
+ for (String entity : entities) {
+ assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-avg-latency", groupName,
+ "The average latency in milliseconds of " + entity + " " + operation + " operation.", tags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-max-latency", groupName,
+ "The max latency in milliseconds of " + entity + " " + operation + " operation.", tags)));
+ assertNotNull(metrics.metrics().get(metrics.metricName(entity + "-" + operation + "-qps", groupName,
+ "The average number of occurrence of " + entity + " " + operation + " operation per second.", tags)));
+ }
}
@SuppressWarnings("unchecked")
@Test
public void testPauseResume() throws Exception {
StreamsConfig config = createConfig(baseDir);
- StreamTask task = new StreamTask(new TaskId(1, 1), "applicationId", partitions, topology, consumer, restoreStateConsumer, config, null, stateDirectory, null, recordCollector);
+ StreamTask task = new StreamTask(new TaskId(1, 1), "applicationId", partitions, topology, consumer,
+ restoreStateConsumer, config, new MockStreamsMetrics(new Metrics()), stateDirectory, null, new MockTime(), recordCollector);
task.addRecords(partition1, records(
new ConsumerRecord<>(partition1.topic(), partition1.partition(), 10, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue),
@@ -237,7 +269,8 @@ public class StreamTaskTest {
@Test
public void testMaybePunctuate() throws Exception {
StreamsConfig config = createConfig(baseDir);
- StreamTask task = new StreamTask(new TaskId(0, 0), "applicationId", partitions, topology, consumer, restoreStateConsumer, config, null, stateDirectory, null, recordCollector);
+ StreamTask task = new StreamTask(new TaskId(0, 0), "applicationId", partitions, topology, consumer,
+ restoreStateConsumer, config, new MockStreamsMetrics(new Metrics()), stateDirectory, null, new MockTime(), recordCollector);
task.addRecords(partition1, records(
new ConsumerRecord<>(partition1.topic(), partition1.partition(), 20, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue),
@@ -310,13 +343,17 @@ public class StreamTaskTest {
final List<ProcessorNode> processorNodes = Collections.<ProcessorNode>singletonList(processorNode);
final Map<String, SourceNode> sourceNodes
= Collections.<String, SourceNode>singletonMap(topic1[0], processorNode);
+ final StreamsMetrics streamsMetrics = new MockStreamsMetrics(new Metrics());
final ProcessorTopology topology = new ProcessorTopology(processorNodes,
sourceNodes,
Collections.<String, SinkNode>emptyMap(),
Collections.<StateStore>emptyList(),
Collections.<String, String>emptyMap(),
Collections.<StateStore, ProcessorNode>emptyMap());
- final StreamTask streamTask = new StreamTask(new TaskId(0, 0), "applicationId", partitions, topology, consumer, restoreStateConsumer, config, null, stateDirectory, new ThreadCache(0), recordCollector);
+
+ final StreamTask streamTask = new StreamTask(new TaskId(0, 0), "applicationId", partitions,
+ topology, consumer, restoreStateConsumer, config, streamsMetrics, stateDirectory, new ThreadCache("testCache", 0, streamsMetrics), new MockTime(), recordCollector);
+
final int offset = 20;
streamTask.addRecords(partition1, Collections.singletonList(
new ConsumerRecord<>(partition1.topic(), partition1.partition(), offset, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue)));
@@ -364,7 +401,10 @@ public class StreamTaskTest {
Collections.<StateStore>emptyList(),
Collections.<String, String>emptyMap(),
Collections.<StateStore, ProcessorNode>emptyMap());
- final StreamTask streamTask = new StreamTask(new TaskId(0, 0), "applicationId", partitions, topology, consumer, restoreStateConsumer, config, null, stateDirectory, new ThreadCache(0), recordCollector);
+
+ final StreamsMetrics streamsMetrics = new MockStreamsMetrics(new Metrics());
+ final StreamTask streamTask = new StreamTask(new TaskId(0, 0), "applicationId", partitions,
+ topology, consumer, restoreStateConsumer, config, streamsMetrics, stateDirectory, new ThreadCache("testCache", 0, streamsMetrics), new MockTime(), recordCollector);
try {
streamTask.punctuate(punctuator, 1);
@@ -391,7 +431,9 @@ public class StreamTaskTest {
flushed.set(true);
}
};
- final StreamTask streamTask = new StreamTask(new TaskId(0, 0), "appId", partitions, topology, consumer, restoreStateConsumer, createConfig(baseDir), null, stateDirectory, new ThreadCache(0), recordCollector);
+ final StreamsMetrics streamsMetrics = new MockStreamsMetrics(new Metrics());
+ final StreamTask streamTask = new StreamTask(new TaskId(0, 0), "appId", partitions, topology, consumer,
+ restoreStateConsumer, createConfig(baseDir), streamsMetrics, stateDirectory, new ThreadCache("testCache", 0, streamsMetrics), new MockTime(), recordCollector);
streamTask.flushState();
assertTrue(flushed.get());
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/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 d787ed5..454b4c2 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
@@ -5,9 +5,9 @@
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -17,6 +17,19 @@
package org.apache.kafka.streams.processor.internals;
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.UUID;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.MockConsumer;
@@ -27,7 +40,6 @@ import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
@@ -42,39 +54,34 @@ import org.apache.kafka.test.MockInternalTopicManager;
import org.apache.kafka.test.MockProcessorSupplier;
import org.apache.kafka.test.MockStateStoreSupplier;
import org.apache.kafka.test.MockTimestampExtractor;
+import org.junit.Before;
import org.apache.kafka.test.TestUtils;
import org.junit.Assert;
import org.junit.Test;
-import java.io.File;
-import java.nio.ByteBuffer;
-import java.nio.file.Files;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Set;
-import java.util.UUID;
import java.util.regex.Pattern;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import static org.junit.Assert.assertThat;
+
public class StreamThreadTest {
private final String clientId = "clientId";
private final String applicationId = "stream-thread-test";
- private final UUID processId = UUID.randomUUID();
+ private UUID processId = UUID.randomUUID();
+
+ @Before
+ public void setUp() throws Exception {
+ processId = UUID.randomUUID();
+ }
private TopicPartition t1p1 = new TopicPartition("topic1", 1);
private TopicPartition t1p2 = new TopicPartition("topic1", 2);
@@ -84,22 +91,22 @@ public class StreamThreadTest {
private TopicPartition t3p2 = new TopicPartition("topic3", 2);
private List<PartitionInfo> infos = Arrays.asList(
- new PartitionInfo("topic1", 0, Node.noNode(), new Node[0], new Node[0]),
- new PartitionInfo("topic1", 1, Node.noNode(), new Node[0], new Node[0]),
- new PartitionInfo("topic1", 2, Node.noNode(), new Node[0], new Node[0]),
- new PartitionInfo("topic2", 0, Node.noNode(), new Node[0], new Node[0]),
- new PartitionInfo("topic2", 1, Node.noNode(), new Node[0], new Node[0]),
- new PartitionInfo("topic2", 2, Node.noNode(), new Node[0], new Node[0]),
- new PartitionInfo("topic3", 0, Node.noNode(), new Node[0], new Node[0]),
- new PartitionInfo("topic3", 1, Node.noNode(), new Node[0], new Node[0]),
- new PartitionInfo("topic3", 2, Node.noNode(), new Node[0], new Node[0])
+ new PartitionInfo("topic1", 0, Node.noNode(), new Node[0], new Node[0]),
+ new PartitionInfo("topic1", 1, Node.noNode(), new Node[0], new Node[0]),
+ new PartitionInfo("topic1", 2, Node.noNode(), new Node[0], new Node[0]),
+ new PartitionInfo("topic2", 0, Node.noNode(), new Node[0], new Node[0]),
+ new PartitionInfo("topic2", 1, Node.noNode(), new Node[0], new Node[0]),
+ new PartitionInfo("topic2", 2, Node.noNode(), new Node[0], new Node[0]),
+ new PartitionInfo("topic3", 0, Node.noNode(), new Node[0], new Node[0]),
+ new PartitionInfo("topic3", 1, Node.noNode(), new Node[0], new Node[0]),
+ new PartitionInfo("topic3", 2, Node.noNode(), new Node[0], new Node[0])
);
private Cluster metadata = new Cluster("cluster", Arrays.asList(Node.noNode()), infos, Collections.<String>emptySet(),
Collections.<String>emptySet());
private final PartitionAssignor.Subscription subscription =
- new PartitionAssignor.Subscription(Arrays.asList("topic1", "topic2", "topic3"), subscriptionUserData());
+ new PartitionAssignor.Subscription(Arrays.asList("topic1", "topic2", "topic3"), subscriptionUserData());
private ByteBuffer subscriptionUserData() {
UUID uuid = UUID.randomUUID();
@@ -149,18 +156,10 @@ public class StreamThreadTest {
Producer<byte[], byte[]> producer,
Consumer<byte[], byte[]> restoreConsumer,
StreamsConfig config,
+ StreamsMetrics metrics,
StateDirectory stateDirectory) {
- super(id, applicationId, partitions, topology, consumer, restoreConsumer, config, new StreamsMetrics() {
- @Override
- public Sensor addLatencySensor(final String scopeName, final String entityName, final String operationName, final String... tags) {
- return null;
- }
-
- @Override
- public void recordLatency(final Sensor sensor, final long startNs, final long endNs) {
-
- }
- }, stateDirectory, null, new RecordCollectorImpl(producer, id.toString()));
+ super(id, applicationId, partitions, topology, consumer, restoreConsumer, config, metrics,
+ stateDirectory, null, new MockTime(), new RecordCollectorImpl(producer, id.toString()));
}
@Override
@@ -212,8 +211,10 @@ public class StreamThreadTest {
@Override
protected StreamTask createStreamTask(TaskId id, Collection<TopicPartition> partitionsForTask) {
+
ProcessorTopology topology = builder.build(id.topicGroupId);
- return new TestStreamTask(id, applicationId, partitionsForTask, topology, consumer, producer, restoreConsumer, config, stateDirectory);
+ return new TestStreamTask(id, applicationId, partitionsForTask, topology, consumer,
+ producer, restoreConsumer, config, new MockStreamsMetrics(new Metrics()), stateDirectory);
}
};
@@ -424,6 +425,44 @@ public class StreamThreadTest {
}
@Test
+ public void testMetrics() throws Exception {
+ TopologyBuilder builder = new TopologyBuilder().setApplicationId("MetricsApp");
+ StreamsConfig config = new StreamsConfig(configProps());
+ MockClientSupplier clientSupplier = new MockClientSupplier();
+
+ Metrics metrics = new Metrics();
+ StreamThread thread = new StreamThread(builder, config, clientSupplier, applicationId,
+ clientId, processId, metrics, new MockTime(), new StreamsMetadataState(builder));
+ String defaultGroupName = "stream-metrics";
+ 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 + ".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)));
+ }
+
+ @Test
public void testMaybeClean() throws Exception {
File baseDir = Files.createTempDirectory("test").toFile();
try {
@@ -460,7 +499,8 @@ public class StreamThreadTest {
@Override
protected StreamTask createStreamTask(TaskId id, Collection<TopicPartition> partitionsForTask) {
ProcessorTopology topology = builder.build(id.topicGroupId);
- return new TestStreamTask(id, applicationId, partitionsForTask, topology, consumer, producer, restoreConsumer, config, stateDirectory);
+ return new TestStreamTask(id, applicationId, partitionsForTask, topology, consumer,
+ producer, restoreConsumer, config, new MockStreamsMetrics(new Metrics()), stateDirectory);
}
};
@@ -587,7 +627,8 @@ public class StreamThreadTest {
@Override
protected StreamTask createStreamTask(TaskId id, Collection<TopicPartition> partitionsForTask) {
ProcessorTopology topology = builder.build(id.topicGroupId);
- return new TestStreamTask(id, applicationId, partitionsForTask, topology, consumer, producer, restoreConsumer, config, stateDirectory);
+ return new TestStreamTask(id, applicationId, partitionsForTask, topology, consumer,
+ producer, restoreConsumer, config, new MockStreamsMetrics(new Metrics()), stateDirectory);
}
};
@@ -810,7 +851,8 @@ public class StreamThreadTest {
@Override
protected StreamTask createStreamTask(final TaskId id, final Collection<TopicPartition> partitions) {
final ProcessorTopology topology = builder.build(id.topicGroupId);
- final TestStreamTask task = new TestStreamTask(id, "appId", partitions, topology, consumer, producer, restoreConsumer, config, stateDirectory);
+ final TestStreamTask task = new TestStreamTask(id, "appId", partitions, topology, consumer,
+ producer, restoreConsumer, config, new MockStreamsMetrics(new Metrics()), stateDirectory);
createdTasks.put(partitions, task);
return task;
}
@@ -864,6 +906,7 @@ public class StreamThreadTest {
clientSupplier.producer,
clientSupplier.restoreConsumer,
config,
+ new MockStreamsMetrics(new Metrics()),
new StateDirectory(applicationId, config.getString(StreamsConfig.STATE_DIR_CONFIG))) {
@Override
public void close() {
@@ -914,6 +957,7 @@ public class StreamThreadTest {
clientSupplier.producer,
clientSupplier.restoreConsumer,
config,
+ new MockStreamsMetrics(new Metrics()),
new StateDirectory(applicationId, config.getString(StreamsConfig.STATE_DIR_CONFIG))) {
@Override
public void flushState() {
@@ -964,6 +1008,7 @@ public class StreamThreadTest {
clientSupplier.producer,
clientSupplier.restoreConsumer,
config,
+ new MockStreamsMetrics(new Metrics()),
new StateDirectory(applicationId, config.getString(StreamsConfig.STATE_DIR_CONFIG))) {
@Override
public void closeTopology() {
@@ -1013,6 +1058,7 @@ public class StreamThreadTest {
clientSupplier.producer,
clientSupplier.restoreConsumer,
config,
+ new MockStreamsMetrics(new Metrics()),
new StateDirectory(applicationId, config.getString(StreamsConfig.STATE_DIR_CONFIG))) {
@Override
public void flushState() {
@@ -1058,7 +1104,7 @@ public class StreamThreadTest {
partitionAssignor.setInternalTopicManager(internalTopicManager);
Map<String, PartitionAssignor.Assignment> assignments =
- partitionAssignor.assign(metadata, Collections.singletonMap("client", subscription));
+ partitionAssignor.assign(metadata, Collections.singletonMap("client", subscription));
partitionAssignor.onAssignment(assignments.get("client"));
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/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
new file mode 100644
index 0000000..b7c0c2c
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsMetricsImplTest.java
@@ -0,0 +1,110 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.internals;
+
+
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+public class StreamsMetricsImplTest {
+
+ @Test(expected = NullPointerException.class)
+ public void testNullMetrics() throws Exception {
+ String groupName = "doesNotMatter";
+ Map<String, String> tags = new HashMap<>();
+ StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(null, groupName, tags);
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void testRemoveNullSensor() {
+ String groupName = "doesNotMatter";
+ Map<String, String> tags = new HashMap<>();
+ StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(new Metrics(), groupName, tags);
+ streamsMetrics.removeSensor(null);
+ }
+
+ @Test
+ public void testRemoveSensor() {
+ String groupName = "doesNotMatter";
+ String sensorName = "sensor1";
+ String scope = "scope";
+ String entity = "entity";
+ String operation = "put";
+ Map<String, String> tags = new HashMap<>();
+ StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(new Metrics(), groupName, tags);
+
+ Sensor sensor1 = streamsMetrics.addSensor(sensorName, Sensor.RecordingLevel.DEBUG);
+ streamsMetrics.removeSensor(sensor1);
+
+ Sensor sensor1a = streamsMetrics.addSensor(sensorName, Sensor.RecordingLevel.DEBUG, sensor1);
+ streamsMetrics.removeSensor(sensor1a);
+
+ Sensor sensor2 = streamsMetrics.addLatencySensor(scope, entity, operation, Sensor.RecordingLevel.DEBUG);
+ streamsMetrics.removeSensor(sensor2);
+
+ Sensor sensor3 = streamsMetrics.addThroughputSensor(scope, entity, operation, Sensor.RecordingLevel.DEBUG);
+ streamsMetrics.removeSensor(sensor3);
+ }
+
+ @Test
+ public void testLatencyMetrics() {
+ String groupName = "doesNotMatter";
+ String scope = "scope";
+ String entity = "entity";
+ String operation = "put";
+ Map<String, String> tags = new HashMap<>();
+ StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(new Metrics(), groupName, tags);
+
+ Sensor sensor1 = streamsMetrics.addLatencySensor(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
+ assertEquals(metrics.size(), 7);
+
+ streamsMetrics.removeSensor(sensor1);
+ metrics = streamsMetrics.metrics();
+ assertEquals(metrics.size(), 1);
+ }
+
+ @Test
+ public void testThroughputMetrics() {
+ String groupName = "doesNotMatter";
+ String scope = "scope";
+ String entity = "entity";
+ String operation = "put";
+ Map<String, String> tags = new HashMap<>();
+ StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(new Metrics(), groupName, tags);
+
+ Sensor sensor1 = streamsMetrics.addThroughputSensor(scope, entity, operation, Sensor.RecordingLevel.DEBUG);
+
+ Map<MetricName, ? extends Metric> metrics = streamsMetrics.metrics();
+ // 2 metrics plus a common metric that keeps track of total registered metrics in Metrics() constructor
+ assertEquals(metrics.size(), 3);
+
+ streamsMetrics.removeSensor(sensor1);
+ metrics = streamsMetrics.metrics();
+ assertEquals(metrics.size(), 1);
+ }
+}
[3/3] kafka git commit: KAFKA-3715: Add granular metrics to Kafka
Streams and add hierarhical logging levels to Metrics
Posted by gu...@apache.org.
KAFKA-3715: Add granular metrics to Kafka Streams and add hierarhical logging levels to Metrics
Kafka Streams: add granular metrics per node, also expose ability to \u2026
\u2026register non latency metrics in StreamsMetrics
from https://github.com/apache/kafka/pull/1362#issuecomment-218326690-------
We can consider adding metrics for process / punctuate / commit rate at the granularity of each processor node in addition to the global rate mentioned above. This is very helpful in debugging.
We can consider adding rate / total cumulated metrics for context.forward indicating how many records were forwarded downstream from this processor node as well. This is helpful in debugging.
We can consider adding metrics for each stream partition's timestamp.
This is helpful in debugging.
## Besides the latency metrics, we can also add throughput latency in terms of source records consumed.
More discussions here https://issues.apache.org/jira/browse/KAFKA-3715
Author: Eno Thereska <en...@confluent.io>
Author: aartigupta <aa...@gmail.com>
Author: aartiguptaa <aa...@gmail.com>
Author: aartigupta <ag...@agupta-m01.vmware.com>
Reviewers: Greg Fodor, Ismael Juma, Damian Guy, Guozhang Wang
Closes #1446 from aartigupta/trunk
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/e43cf224
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/e43cf224
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/e43cf224
Branch: refs/heads/trunk
Commit: e43cf22401a0aa8328749365f7157dde20e634b2
Parents: 3d60f1e
Author: Eno Thereska <en...@confluent.io>
Authored: Wed Jan 11 11:34:58 2017 -0800
Committer: Guozhang Wang <wa...@gmail.com>
Committed: Wed Jan 11 11:34:58 2017 -0800
----------------------------------------------------------------------
.../kafka/clients/CommonClientConfigs.java | 3 +
.../kafka/clients/consumer/ConsumerConfig.java | 12 ++
.../kafka/common/metrics/MetricConfig.java | 13 ++
.../apache/kafka/common/metrics/Metrics.java | 80 +++++--
.../org/apache/kafka/common/metrics/Sensor.java | 102 +++++++--
.../apache/kafka/common/metrics/SensorTest.java | 58 +++++
.../main/scala/kafka/server/KafkaConfig.scala | 7 +-
.../unit/kafka/server/KafkaConfigTest.scala | 1 +
.../org/apache/kafka/streams/KafkaStreams.java | 13 ++
.../org/apache/kafka/streams/StreamsConfig.java | 11 +
.../apache/kafka/streams/StreamsMetrics.java | 72 ++++++-
.../processor/internals/ProcessorNode.java | 111 +++++++++-
.../streams/processor/internals/SinkNode.java | 7 +-
.../streams/processor/internals/SourceNode.java | 7 +-
.../processor/internals/StandbyContextImpl.java | 4 +-
.../streams/processor/internals/StreamTask.java | 58 +++--
.../processor/internals/StreamThread.java | 200 +++++++-----------
.../processor/internals/StreamsMetricsImpl.java | 210 +++++++++++++++++++
.../state/internals/MeteredKeyValueStore.java | 134 +++++++-----
.../internals/MeteredSegmentedBytesStore.java | 12 +-
.../state/internals/MeteredWindowStore.java | 80 ++++---
.../streams/state/internals/NamedCache.java | 59 +++++-
.../streams/state/internals/ThreadCache.java | 29 +--
.../state/internals/ThreadCacheMetrics.java | 40 ----
.../apache/kafka/streams/KafkaStreamsTest.java | 40 ++++
...reamSessionWindowAggregateProcessorTest.java | 5 +-
.../kafka/streams/perf/SimpleBenchmark.java | 38 +++-
.../processor/internals/AbstractTaskTest.java | 3 +-
.../processor/internals/MockStreamsMetrics.java | 28 +++
.../processor/internals/ProcessorNodeTest.java | 65 ++++++
.../processor/internals/StandbyTaskTest.java | 16 +-
.../processor/internals/StreamTaskTest.java | 54 ++++-
.../processor/internals/StreamThreadTest.java | 136 ++++++++----
.../internals/StreamsMetricsImplTest.java | 110 ++++++++++
.../streams/state/KeyValueStoreTestDriver.java | 57 ++---
.../internals/CachingKeyValueStoreTest.java | 4 +-
.../internals/CachingSessionStoreTest.java | 4 +-
.../state/internals/CachingWindowStoreTest.java | 4 +-
.../ChangeLoggingSegmentedBytesStoreTest.java | 4 +-
...gedSortedCacheKeyValueStoreIteratorTest.java | 6 +-
...ergedSortedCacheWindowStoreIteratorTest.java | 4 +-
.../MeteredSegmentedBytesStoreTest.java | 40 +++-
.../streams/state/internals/NamedCacheTest.java | 32 ++-
.../RocksDBSegmentedBytesStoreTest.java | 4 +-
.../internals/RocksDBSessionStoreTest.java | 4 +-
.../state/internals/RocksDBWindowStoreTest.java | 36 ++--
.../state/internals/SegmentIteratorTest.java | 4 +-
.../streams/state/internals/SegmentsTest.java | 4 +-
.../StreamThreadStateStoreProviderTest.java | 23 +-
.../state/internals/ThreadCacheTest.java | 59 +++---
.../apache/kafka/test/KStreamTestDriver.java | 4 +-
.../apache/kafka/test/MockProcessorContext.java | 40 ++--
.../kafka/test/ProcessorTopologyTestDriver.java | 61 +++---
53 files changed, 1627 insertions(+), 585 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java
index 3327815..944b09d 100644
--- a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java
+++ b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java
@@ -63,6 +63,9 @@ public class CommonClientConfigs {
public static final String METRICS_NUM_SAMPLES_CONFIG = "metrics.num.samples";
public static final String METRICS_NUM_SAMPLES_DOC = "The number of samples maintained to compute metrics.";
+ public static final String METRICS_RECORDING_LEVEL_CONFIG = "metrics.recording.level";
+ public static final String METRICS_RECORDING_LEVEL_DOC = "The highest recording level for metrics.";
+
public static final String METRIC_REPORTER_CLASSES_CONFIG = "metric.reporters";
public static final String METRIC_REPORTER_CLASSES_DOC = "A list of classes to use as metrics reporters. Implementing the <code>MetricReporter</code> interface allows plugging in classes that will be notified of new metric creation. The JmxReporter is always included to register JMX statistics.";
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
index 13cc4c4..ed809a9 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
@@ -17,6 +17,7 @@ import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigDef.Importance;
import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.serialization.Deserializer;
import java.util.Collections;
@@ -175,6 +176,11 @@ public class ConsumerConfig extends AbstractConfig {
public static final String METRICS_NUM_SAMPLES_CONFIG = CommonClientConfigs.METRICS_NUM_SAMPLES_CONFIG;
/**
+ * <code>metrics.log.level</code>
+ */
+ public static final String METRICS_RECORDING_LEVEL_CONFIG = CommonClientConfigs.METRICS_RECORDING_LEVEL_CONFIG;
+
+ /**
* <code>metric.reporters</code>
*/
public static final String METRIC_REPORTER_CLASSES_CONFIG = CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG;
@@ -327,6 +333,12 @@ public class ConsumerConfig extends AbstractConfig {
atLeast(1),
Importance.LOW,
CommonClientConfigs.METRICS_NUM_SAMPLES_DOC)
+ .define(METRICS_RECORDING_LEVEL_CONFIG,
+ Type.STRING,
+ Sensor.RecordingLevel.INFO.toString(),
+ in(Sensor.RecordingLevel.INFO.toString(), Sensor.RecordingLevel.DEBUG.toString()),
+ Importance.LOW,
+ CommonClientConfigs.METRICS_RECORDING_LEVEL_DOC)
.define(METRIC_REPORTER_CLASSES_CONFIG,
Type.LIST,
"",
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/clients/src/main/java/org/apache/kafka/common/metrics/MetricConfig.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/MetricConfig.java b/clients/src/main/java/org/apache/kafka/common/metrics/MetricConfig.java
index a423334..0766cc3 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/MetricConfig.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/MetricConfig.java
@@ -30,6 +30,7 @@ public class MetricConfig {
private long eventWindow;
private long timeWindowMs;
private Map<String, String> tags;
+ private Sensor.RecordingLevel recordingLevel;
public MetricConfig() {
super();
@@ -38,6 +39,7 @@ public class MetricConfig {
this.eventWindow = Long.MAX_VALUE;
this.timeWindowMs = TimeUnit.MILLISECONDS.convert(30, TimeUnit.SECONDS);
this.tags = new LinkedHashMap<>();
+ this.recordingLevel = Sensor.RecordingLevel.INFO;
}
public Quota quota() {
@@ -86,4 +88,15 @@ public class MetricConfig {
this.samples = samples;
return this;
}
+
+ public Sensor.RecordingLevel recordLevel() {
+ return this.recordingLevel;
+ }
+
+ public MetricConfig recordLevel(Sensor.RecordingLevel recordingLevel) {
+ this.recordingLevel = recordingLevel;
+ return this;
+ }
+
+
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java b/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
index 8beb107..7b303fa 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
@@ -12,6 +12,12 @@
*/
package org.apache.kafka.common.metrics;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.Closeable;
import java.util.ArrayList;
import java.util.Collections;
@@ -25,12 +31,6 @@ import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
-import org.apache.kafka.common.MetricName;
-import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.common.utils.Utils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
/**
* A registry of sensors and metrics.
* <p>
@@ -236,35 +236,74 @@ public class Metrics implements Closeable {
}
/**
- * Get or create a sensor with the given unique name and no parent sensors.
+ * Get or create a sensor with the given unique name and no parent sensors. This uses
+ * a default recording level of INFO.
* @param name The sensor name
* @return The sensor
*/
public Sensor sensor(String name) {
- return sensor(name, null, (Sensor[]) null);
+ return this.sensor(name, Sensor.RecordingLevel.INFO);
}
/**
+ * Get or create a sensor with the given unique name and no parent sensors and with a given
+ * recording level.
+ * @param name The sensor name.
+ * @param recordingLevel The recording level.
+ * @return The sensor
+ */
+ public Sensor sensor(String name, Sensor.RecordingLevel recordingLevel) {
+ return sensor(name, null, recordingLevel, (Sensor[]) null);
+ }
+
+
+ /**
* Get or create a sensor with the given unique name and zero or more parent sensors. All parent sensors will
- * receive every value recorded with this sensor.
+ * receive every value recorded with this sensor. This uses a default recording level of INFO.
* @param name The name of the sensor
* @param parents The parent sensors
* @return The sensor that is created
*/
public Sensor sensor(String name, Sensor... parents) {
- return sensor(name, null, parents);
+ return this.sensor(name, Sensor.RecordingLevel.INFO, parents);
}
/**
* Get or create a sensor with the given unique name and zero or more parent sensors. All parent sensors will
* receive every value recorded with this sensor.
+ * @param name The name of the sensor.
+ * @param parents The parent sensors.
+ * @param recordingLevel The recording level.
+ * @return The sensor that is created
+ */
+ public Sensor sensor(String name, Sensor.RecordingLevel recordingLevel, Sensor... parents) {
+ return sensor(name, null, recordingLevel, parents);
+ }
+
+ /**
+ * Get or create a sensor with the given unique name and zero or more parent sensors. All parent sensors will
+ * receive every value recorded with this sensor. This uses a default recording level of INFO.
* @param name The name of the sensor
* @param config A default configuration to use for this sensor for metrics that don't have their own config
* @param parents The parent sensors
* @return The sensor that is created
*/
public synchronized Sensor sensor(String name, MetricConfig config, Sensor... parents) {
- return sensor(name, config, Long.MAX_VALUE, parents);
+ return this.sensor(name, config, Sensor.RecordingLevel.INFO, parents);
+ }
+
+
+ /**
+ * Get or create a sensor with the given unique name and zero or more parent sensors. All parent sensors will
+ * receive every value recorded with this sensor.
+ * @param name The name of the sensor
+ * @param config A default configuration to use for this sensor for metrics that don't have their own config
+ * @param recordingLevel The recording level.
+ * @param parents The parent sensors
+ * @return The sensor that is created
+ */
+ public synchronized Sensor sensor(String name, MetricConfig config, Sensor.RecordingLevel recordingLevel, Sensor... parents) {
+ return sensor(name, config, Long.MAX_VALUE, recordingLevel, parents);
}
/**
@@ -275,12 +314,13 @@ public class Metrics implements Closeable {
* @param inactiveSensorExpirationTimeSeconds If no value if recorded on the Sensor for this duration of time,
* it is eligible for removal
* @param parents The parent sensors
+ * @param recordingLevel The recording level.
* @return The sensor that is created
*/
- public synchronized Sensor sensor(String name, MetricConfig config, long inactiveSensorExpirationTimeSeconds, Sensor... parents) {
+ public synchronized Sensor sensor(String name, MetricConfig config, long inactiveSensorExpirationTimeSeconds, Sensor.RecordingLevel recordingLevel, Sensor... parents) {
Sensor s = getSensor(name);
if (s == null) {
- s = new Sensor(this, name, parents, config == null ? this.config : config, time, inactiveSensorExpirationTimeSeconds);
+ s = new Sensor(this, name, parents, config == null ? this.config : config, time, inactiveSensorExpirationTimeSeconds, recordingLevel);
this.sensors.put(name, s);
if (parents != null) {
for (Sensor parent : parents) {
@@ -298,6 +338,20 @@ public class Metrics implements Closeable {
}
/**
+ * Get or create a sensor with the given unique name and zero or more parent sensors. All parent sensors will
+ * receive every value recorded with this sensor. This uses a default recording level of INFO.
+ * @param name The name of the sensor
+ * @param config A default configuration to use for this sensor for metrics that don't have their own config
+ * @param inactiveSensorExpirationTimeSeconds If no value if recorded on the Sensor for this duration of time,
+ * it is eligible for removal
+ * @param parents The parent sensors
+ * @return The sensor that is created
+ */
+ public synchronized Sensor sensor(String name, MetricConfig config, long inactiveSensorExpirationTimeSeconds, Sensor... parents) {
+ return this.sensor(name, config, inactiveSensorExpirationTimeSeconds, Sensor.RecordingLevel.INFO, parents);
+ }
+
+ /**
* Remove a sensor (if it exists), associated metrics and its children.
*
* @param name The name of the sensor to be removed
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java b/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
index 4f630f9..4a9b488 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
@@ -12,18 +12,19 @@
*/
package org.apache.kafka.common.metrics;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.CompoundStat.NamedMeasurable;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
+import java.util.Locale;
import java.util.Set;
import java.util.concurrent.TimeUnit;
-import org.apache.kafka.common.MetricName;
-import org.apache.kafka.common.metrics.CompoundStat.NamedMeasurable;
-import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.common.utils.Utils;
-
/**
* A sensor applies a continuous sequence of numerical values to a set of associated metrics. For example a sensor on
* message size would record a sequence of message sizes using the {@link #record(double)} api and would maintain a set
@@ -41,7 +42,63 @@ public final class Sensor {
private volatile long lastRecordTime;
private final long inactiveSensorExpirationTimeMs;
- Sensor(Metrics registry, String name, Sensor[] parents, MetricConfig config, Time time, long inactiveSensorExpirationTimeSeconds) {
+ public enum RecordingLevel {
+ INFO(0, "INFO"), DEBUG(1, "DEBUG");
+
+ private static final RecordingLevel[] ID_TO_TYPE;
+ private static final int MIN_RECORDING_LEVEL_KEY = 0;
+ public static final int MAX_RECORDING_LEVEL_KEY;
+
+ static {
+ int maxRL = -1;
+ for (RecordingLevel level : RecordingLevel.values()) {
+ maxRL = Math.max(maxRL, level.id);
+ }
+ RecordingLevel[] idToName = new RecordingLevel[maxRL + 1];
+ for (RecordingLevel level : RecordingLevel.values()) {
+ idToName[level.id] = level;
+ }
+ ID_TO_TYPE = idToName;
+ MAX_RECORDING_LEVEL_KEY = maxRL;
+ }
+
+ /** an english description of the api--this is for debugging and can change */
+ public final String name;
+
+ /** the permanent and immutable id of an API--this can't change ever */
+ public final short id;
+
+ RecordingLevel(int id, String name) {
+ this.id = (short) id;
+ this.name = name;
+ }
+
+ public static RecordingLevel forId(int id) {
+ if (id < MIN_RECORDING_LEVEL_KEY || id > MAX_RECORDING_LEVEL_KEY)
+ throw new IllegalArgumentException(String.format("Unexpected RecordLevel id `%s`, it should be between `%s` " +
+ "and `%s` (inclusive)", id, MIN_RECORDING_LEVEL_KEY, MAX_RECORDING_LEVEL_KEY));
+ return ID_TO_TYPE[id];
+ }
+
+ /** Case insensitive lookup by protocol name */
+ public static RecordingLevel forName(String name) {
+ return RecordingLevel.valueOf(name.toUpperCase(Locale.ROOT));
+ }
+
+ public boolean shouldRecord(final int configId) {
+ if (configId == DEBUG.id) {
+ return true;
+ } else {
+ return configId == this.id;
+ }
+ }
+
+ }
+
+ private final RecordingLevel recordingLevel;
+
+ Sensor(Metrics registry, String name, Sensor[] parents, MetricConfig config, Time time,
+ long inactiveSensorExpirationTimeSeconds, RecordingLevel recordingLevel) {
super();
this.registry = registry;
this.name = Utils.notNull(name);
@@ -52,6 +109,7 @@ public final class Sensor {
this.time = time;
this.inactiveSensorExpirationTimeMs = TimeUnit.MILLISECONDS.convert(inactiveSensorExpirationTimeSeconds, TimeUnit.SECONDS);
this.lastRecordTime = time.milliseconds();
+ this.recordingLevel = recordingLevel;
checkForest(new HashSet<Sensor>());
}
@@ -74,17 +132,27 @@ public final class Sensor {
* Record an occurrence, this is just short-hand for {@link #record(double) record(1.0)}
*/
public void record() {
- record(1.0);
+ if (shouldRecord()) {
+ record(1.0);
+ }
}
/**
+ * @return true if the sensor's record level indicates that the metric will be recorded, false otherwise
+ */
+ public boolean shouldRecord() {
+ return this.recordingLevel.shouldRecord(config.recordLevel().id);
+ }
+ /**
* Record a value with this sensor
* @param value The value to record
* @throws QuotaViolationException if recording this value moves a metric beyond its configured maximum or minimum
* bound
*/
public void record(double value) {
- record(value, time.milliseconds());
+ if (shouldRecord()) {
+ record(value, time.milliseconds());
+ }
}
/**
@@ -96,15 +164,17 @@ public final class Sensor {
* bound
*/
public void record(double value, long timeMs) {
- this.lastRecordTime = timeMs;
- synchronized (this) {
- // increment all the stats
- for (int i = 0; i < this.stats.size(); i++)
- this.stats.get(i).record(config, value, timeMs);
- checkQuotas(timeMs);
+ if (shouldRecord()) {
+ this.lastRecordTime = timeMs;
+ synchronized (this) {
+ // increment all the stats
+ for (int i = 0; i < this.stats.size(); i++)
+ this.stats.get(i).record(config, value, timeMs);
+ checkQuotas(timeMs);
+ }
+ for (int i = 0; i < parents.length; i++)
+ parents[i].record(value, timeMs);
}
- for (int i = 0; i < parents.length; i++)
- parents[i].record(value, timeMs);
}
/**
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java b/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java
new file mode 100644
index 0000000..9aadc82
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java
@@ -0,0 +1,58 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE
+ * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file
+ * to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the
+ * License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package org.apache.kafka.common.metrics;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.kafka.common.utils.SystemTime;
+import org.junit.Test;
+
+public class SensorTest {
+ @Test
+ public void testRecordLevelEnum() {
+ Sensor.RecordingLevel configLevel = Sensor.RecordingLevel.INFO;
+ assertTrue(Sensor.RecordingLevel.INFO.shouldRecord(configLevel.id));
+ assertFalse(Sensor.RecordingLevel.DEBUG.shouldRecord(configLevel.id));
+
+ configLevel = Sensor.RecordingLevel.DEBUG;
+ assertTrue(Sensor.RecordingLevel.INFO.shouldRecord(configLevel.id));
+ assertTrue(Sensor.RecordingLevel.DEBUG.shouldRecord(configLevel.id));
+
+ assertEquals(Sensor.RecordingLevel.valueOf(Sensor.RecordingLevel.DEBUG.toString()),
+ Sensor.RecordingLevel.DEBUG);
+ assertEquals(Sensor.RecordingLevel.valueOf(Sensor.RecordingLevel.INFO.toString()),
+ Sensor.RecordingLevel.INFO);
+ }
+
+ @Test
+ public void testShouldRecord() {
+ MetricConfig debugConfig = new MetricConfig().recordLevel(Sensor.RecordingLevel.DEBUG);
+ MetricConfig infoConfig = new MetricConfig().recordLevel(Sensor.RecordingLevel.INFO);
+
+ Sensor infoSensor = new Sensor(null, "infoSensor", null, debugConfig, new SystemTime(),
+ 0, Sensor.RecordingLevel.INFO);
+ assertTrue(infoSensor.shouldRecord());
+ infoSensor = new Sensor(null, "infoSensor", null, debugConfig, new SystemTime(),
+ 0, Sensor.RecordingLevel.DEBUG);
+ assertTrue(infoSensor.shouldRecord());
+
+ Sensor debugSensor = new Sensor(null, "debugSensor", null, infoConfig, new SystemTime(),
+ 0, Sensor.RecordingLevel.INFO);
+ assertTrue(debugSensor.shouldRecord());
+ debugSensor = new Sensor(null, "debugSensor", null, infoConfig, new SystemTime(),
+ 0, Sensor.RecordingLevel.DEBUG);
+ assertFalse(debugSensor.shouldRecord());
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/core/src/main/scala/kafka/server/KafkaConfig.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index cf9daa2..73fee6c 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -18,6 +18,7 @@
package kafka.server
import java.util.Properties
+
import kafka.api.{ApiVersion, KAFKA_0_10_0_IV1}
import kafka.cluster.EndPoint
import kafka.consumer.ConsumerConfig
@@ -28,7 +29,7 @@ import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.common.config.ConfigDef.ValidList
import org.apache.kafka.common.config.SaslConfigs
import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, SslConfigs}
-import org.apache.kafka.common.metrics.MetricsReporter
+import org.apache.kafka.common.metrics.{MetricsReporter, Sensor}
import org.apache.kafka.common.protocol.SecurityProtocol
import org.apache.kafka.common.record.TimestampType
@@ -163,6 +164,7 @@ object Defaults {
val MetricNumSamples = 2
val MetricSampleWindowMs = 30000
val MetricReporterClasses = ""
+ val MetricRecordingLevel = Sensor.RecordingLevel.INFO.toString()
/** ********* SSL configuration ***********/
val PrincipalBuilderClass = SslConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS
@@ -326,6 +328,7 @@ object KafkaConfig {
val MetricSampleWindowMsProp = CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_CONFIG
val MetricNumSamplesProp: String = CommonClientConfigs.METRICS_NUM_SAMPLES_CONFIG
val MetricReporterClassesProp: String = CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG
+ val MetricRecordingLevelProp: String = CommonClientConfigs.METRICS_RECORDING_LEVEL_CONFIG
/** ********* SSL Configuration ****************/
val PrincipalBuilderClassProp = SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG
@@ -547,6 +550,7 @@ object KafkaConfig {
val MetricSampleWindowMsDoc = CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_DOC
val MetricNumSamplesDoc = CommonClientConfigs.METRICS_NUM_SAMPLES_DOC
val MetricReporterClassesDoc = CommonClientConfigs.METRIC_REPORTER_CLASSES_DOC
+ val MetricRecordingLevelDoc = CommonClientConfigs.METRICS_RECORDING_LEVEL_DOC
/** ********* SSL Configuration ****************/
val PrincipalBuilderClassDoc = SslConfigs.PRINCIPAL_BUILDER_CLASS_DOC
@@ -716,6 +720,7 @@ object KafkaConfig {
.define(MetricNumSamplesProp, INT, Defaults.MetricNumSamples, atLeast(1), LOW, MetricNumSamplesDoc)
.define(MetricSampleWindowMsProp, LONG, Defaults.MetricSampleWindowMs, atLeast(1), LOW, MetricSampleWindowMsDoc)
.define(MetricReporterClassesProp, LIST, Defaults.MetricReporterClasses, LOW, MetricReporterClassesDoc)
+ .define(MetricRecordingLevelProp, STRING, Defaults.MetricRecordingLevel, LOW, MetricRecordingLevelDoc)
/** ********* Quota configuration ***********/
.define(ProducerQuotaBytesPerSecondDefaultProp, LONG, Defaults.ProducerQuotaBytesPerSecondDefault, atLeast(1), HIGH, ProducerQuotaBytesPerSecondDefaultDoc)
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
index 93d0413..d5e2ce3 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -532,6 +532,7 @@ class KafkaConfigTest {
case KafkaConfig.MetricNumSamplesProp => assertPropertyInvalid(getBaseProperties, name, "not_a_number", "-1", "0")
case KafkaConfig.MetricSampleWindowMsProp => assertPropertyInvalid(getBaseProperties, name, "not_a_number", "-1", "0")
case KafkaConfig.MetricReporterClassesProp => // ignore string
+ case KafkaConfig.MetricRecordingLevelProp => // ignore string
case KafkaConfig.RackProp => // ignore string
//SSL Configs
case KafkaConfig.PrincipalBuilderClassProp =>
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
index cf49896..fc47672 100644
--- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
+++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
@@ -17,11 +17,14 @@
package org.apache.kafka.streams;
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.metrics.JmxReporter;
import org.apache.kafka.common.metrics.MetricConfig;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.MetricsReporter;
+import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.processor.StateStore;
@@ -42,6 +45,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -205,6 +209,14 @@ public class KafkaStreams {
return state;
}
+ /**
+ * Get read-only handle on global metrics registry
+ * @return Map of all metrics.
+ */
+ public Map<MetricName, ? extends Metric> metrics() {
+ return Collections.unmodifiableMap(this.metrics.metrics());
+ }
+
private class StreamStateListener implements StreamThread.StateListener {
@Override
public synchronized void onChange(final StreamThread thread, final StreamThread.State newState, final StreamThread.State oldState) {
@@ -272,6 +284,7 @@ public class KafkaStreams {
reporters.add(new JmxReporter(JMX_PREFIX));
final MetricConfig metricConfig = new MetricConfig().samples(config.getInt(StreamsConfig.METRICS_NUM_SAMPLES_CONFIG))
+ .recordLevel(Sensor.RecordingLevel.forName(config.getString(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG)))
.timeWindow(config.getLong(StreamsConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG),
TimeUnit.MILLISECONDS);
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
index 6b20e98..956ae0b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -25,6 +25,7 @@ import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigDef.Importance;
import org.apache.kafka.common.config.ConfigDef.Type;
import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.errors.StreamsException;
@@ -39,6 +40,7 @@ import java.util.Map;
import java.util.Set;
import static org.apache.kafka.common.config.ConfigDef.Range.atLeast;
+import static org.apache.kafka.common.config.ConfigDef.ValidString.in;
/**
* Configuration for Kafka Streams. Documentation for these configurations can be found in the <a
@@ -120,6 +122,9 @@ public class StreamsConfig extends AbstractConfig {
/** <code>metrics.num.samples</code> */
public static final String METRICS_NUM_SAMPLES_CONFIG = CommonClientConfigs.METRICS_NUM_SAMPLES_CONFIG;
+ /** <code>metrics.record.level</code> */
+ public static final String METRICS_RECORDING_LEVEL_CONFIG = CommonClientConfigs.METRICS_RECORDING_LEVEL_CONFIG;
+
/** <code>metric.reporters</code> */
public static final String METRIC_REPORTER_CLASSES_CONFIG = CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG;
@@ -262,6 +267,12 @@ public class StreamsConfig extends AbstractConfig {
atLeast(1),
Importance.LOW,
CommonClientConfigs.METRICS_NUM_SAMPLES_DOC)
+ .define(METRICS_RECORDING_LEVEL_CONFIG,
+ Type.STRING,
+ Sensor.RecordingLevel.INFO.toString(),
+ in(Sensor.RecordingLevel.INFO.toString(), Sensor.RecordingLevel.DEBUG.toString()),
+ Importance.LOW,
+ CommonClientConfigs.METRICS_RECORDING_LEVEL_DOC)
.define(APPLICATION_SERVER_CONFIG,
Type.STRING,
"",
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/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 c0870c6..afd3497 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java
@@ -17,26 +17,94 @@
package org.apache.kafka.streams;
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.metrics.Sensor;
+import java.util.Map;
+
/**
* The Kafka Streams metrics interface for adding metric sensors and collecting metric values.
*/
+@InterfaceStability.Unstable
public interface StreamsMetrics {
/**
- * Add the latency sensor.
+ * Get read-only handle on global metrics registry
+ * @return Map of all metrics.
+ */
+ Map<MetricName, ? extends Metric> metrics();
+
+ /**
+ * Add a latency 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.
* @param entityName Name of the entity, could be the name of the state store instance, etc.
+ * @param recordingLevel The recording level (e.g., INFO or DEBUG) for this sensor.
* @param operationName Name of the operation, could be get / put / delete / etc.
* @param tags Additional tags of the sensor.
* @return The added sensor.
*/
- Sensor addLatencySensor(String scopeName, String entityName, String operationName, String... tags);
+ Sensor addLatencySensor(String scopeName, String entityName, String operationName, Sensor.RecordingLevel recordingLevel, String... tags);
/**
* Record the given latency value of the sensor.
+ * @param sensor sensor whose latency we are recording.
+ * @param startNs start of measurement time in nanoseconds.
+ * @param endNs end of measurement time in nanoseconds.
*/
void recordLatency(Sensor sensor, long startNs, long endNs);
+
+ /**
+ * Add a throughput sensor and default associated metrics. Metrics include throughput ones
+ * (operations/time unit).
+ *
+ * @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.
+ * @param recordingLevel The recording level (e.g., INFO or DEBUG) for this sensor.
+ * @param operationName Name of the operation, could be get / put / delete / etc.
+ * @param tags Additional tags of the sensor.
+ * @return The added sensor.
+ */
+ Sensor addThroughputSensor(String scopeName, String entityName, String operationName, Sensor.RecordingLevel recordingLevel, String... tags);
+
+ /**
+ * Records the throughput value of a sensor.
+ * @param sensor addSensor whose throughput we are recording.
+ * @param value throughput value.
+ */
+ void recordThroughput(Sensor sensor, long value);
+
+
+ /**
+ * 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
+ * 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.
+ * @param recordingLevel The recording level (e.g., INFO or DEBUG) for this sensor.
+ */
+ Sensor addSensor(String name, Sensor.RecordingLevel recordingLevel);
+
+ /**
+ * 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
+ * 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.
+ * @param recordingLevel The recording level (e.g., INFO or DEBUG) for this sensor.
+ */
+ Sensor addSensor(String name, Sensor.RecordingLevel recordingLevel, Sensor... parents);
+
+ /**
+ * Remove a sensor.
+ * @param sensor Sensor to be removed.
+ */
+ void removeSensor(Sensor sensor);
}
+
+
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/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 f7f8e8d..f165ebf 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
@@ -5,9 +5,9 @@
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -17,12 +17,18 @@
package org.apache.kafka.streams.processor.internals;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.utils.SystemTime;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import java.util.ArrayList;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.Set;
public class ProcessorNode<K, V> {
@@ -31,6 +37,42 @@ public class ProcessorNode<K, V> {
private final String name;
private final Processor<K, V> processor;
+ protected NodeMetrics nodeMetrics;
+ private Time time;
+
+ private K key;
+ private V value;
+ private Runnable processDelegate = new Runnable() {
+ @Override
+ public void run() {
+ processor.process(key, value);
+ }
+ };
+ private ProcessorContext context;
+ private Runnable initDelegate = new Runnable() {
+ @Override
+ public void run() {
+ if (processor != null) {
+ processor.init(context);
+ }
+ }
+ };
+ private Runnable closeDelegate = new Runnable() {
+ @Override
+ public void run() {
+ if (processor != null) {
+ processor.close();
+ }
+ }
+ };
+
+ private long timestamp;
+ private Runnable punctuateDelegate = new Runnable() {
+ @Override
+ public void run() {
+ processor().punctuate(timestamp);
+ }
+ };
public final Set<String> stateStores;
@@ -38,11 +80,13 @@ public class ProcessorNode<K, V> {
this(name, null, null);
}
+
public ProcessorNode(String name, Processor<K, V> processor, Set<String> stateStores) {
this.name = name;
this.processor = processor;
this.children = new ArrayList<>();
this.stateStores = stateStores;
+ this.time = new SystemTime();
}
@@ -62,9 +106,12 @@ public class ProcessorNode<K, V> {
children.add(child);
}
+
public void init(ProcessorContext context) {
+ this.context = context;
try {
- processor.init(context);
+ nodeMetrics = new NodeMetrics(context.metrics(), name, "task." + context.taskId());
+ nodeMetrics.metrics.measureLatencyNs(time, initDelegate, nodeMetrics.nodeCreationSensor);
} catch (Exception e) {
throw new StreamsException(String.format("failed to initialize processor %s", name), e);
}
@@ -72,14 +119,27 @@ public class ProcessorNode<K, V> {
public void close() {
try {
- processor.close();
+ nodeMetrics.metrics.measureLatencyNs(time, closeDelegate, nodeMetrics.nodeDestructionSensor);
+ nodeMetrics.removeAllSensors();
} catch (Exception e) {
throw new StreamsException(String.format("failed to close processor %s", name), e);
}
}
+
public void process(final K key, final V value) {
- processor.process(key, value);
+ this.key = key;
+ 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) {
+ this.timestamp = timestamp;
+ this.nodeMetrics.metrics.measureLatencyNs(time, punctuateDelegate, nodeMetrics.nodePunctuateTimeSensor);
}
/**
@@ -106,4 +166,43 @@ public class ProcessorNode<K, V> {
}
return sb.toString();
}
+
+ protected class NodeMetrics {
+ final StreamsMetricsImpl metrics;
+ final String metricGrpName;
+ final Map<String, String> metricTags;
+
+ final Sensor nodeProcessTimeSensor;
+ final Sensor nodePunctuateTimeSensor;
+ final Sensor nodeThroughputSensor;
+ final Sensor nodeCreationSensor;
+ final Sensor nodeDestructionSensor;
+
+
+ public NodeMetrics(StreamsMetrics metrics, String name, String sensorNamePrefix) {
+ final String scope = "processor-node";
+ final String tagKey = "processor-node-id";
+ final String tagValue = name;
+ this.metrics = (StreamsMetricsImpl) metrics;
+ this.metricGrpName = "stream-processor-node-metrics";
+ this.metricTags = new LinkedHashMap<>();
+ 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);
+
+ }
+
+ public void removeAllSensors() {
+ metrics.removeSensor(nodeProcessTimeSensor);
+ metrics.removeSensor(nodePunctuateTimeSensor);
+ metrics.removeSensor(nodeThroughputSensor);
+ metrics.removeSensor(nodeCreationSensor);
+ metrics.removeSensor(nodeDestructionSensor);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
index 65bad39..0ebfda7 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
@@ -53,6 +53,7 @@ public class SinkNode<K, V> extends ProcessorNode<K, V> {
@SuppressWarnings("unchecked")
@Override
public void init(ProcessorContext context) {
+ super.init(context);
this.context = context;
// if serializers are null, get the default ones from the context
@@ -63,7 +64,6 @@ public class SinkNode<K, V> extends ProcessorNode<K, V> {
if (this.valSerializer instanceof ChangedSerializer &&
((ChangedSerializer) this.valSerializer).inner() == null)
((ChangedSerializer) this.valSerializer).setInner(context.valueSerde().serializer());
-
}
@@ -91,11 +91,6 @@ public class SinkNode<K, V> extends ProcessorNode<K, V> {
}
}
- @Override
- public void close() {
- // do nothing
- }
-
/**
* @return a string representation of this node, useful for debugging.
*/
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/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 ad9c213..771f504 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
@@ -46,6 +46,7 @@ public class SourceNode<K, V> extends ProcessorNode<K, V> {
@SuppressWarnings("unchecked")
@Override
public void init(ProcessorContext context) {
+ super.init(context);
this.context = context;
// if deserializers are null, get the default ones from the context
@@ -64,11 +65,7 @@ public class SourceNode<K, V> extends ProcessorNode<K, V> {
@Override
public void process(final K key, final V value) {
context.forward(key, value);
- }
-
- @Override
- public void close() {
- // do nothing
+ nodeMetrics.nodeThroughputSensor.record();
}
/**
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyContextImpl.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyContextImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyContextImpl.java
index b9367ab..86bdc19 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyContextImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyContextImpl.java
@@ -69,7 +69,7 @@ public class StandbyContextImpl implements InternalProcessorContext, RecordColle
private final StreamsConfig config;
private final Serde<?> keySerde;
private final Serde<?> valSerde;
- private final ThreadCache zeroSizedCache = new ThreadCache(0);
+ private final ThreadCache zeroSizedCache;
private boolean initialized;
@@ -86,7 +86,7 @@ public class StandbyContextImpl implements InternalProcessorContext, RecordColle
this.config = config;
this.keySerde = config.keySerde();
this.valSerde = config.valueSerde();
-
+ this.zeroSizedCache = new ThreadCache("zeroCache", 0, this.metrics);
this.initialized = false;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/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 9369c01..07ae761 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
@@ -22,6 +22,8 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.errors.StreamsException;
@@ -59,8 +61,24 @@ public class StreamTask extends AbstractTask implements Punctuator {
private boolean commitRequested = false;
private boolean commitOffsetNeeded = false;
-
private boolean requiresPoll = true;
+ private final Time time;
+ private final TaskMetrics metrics;
+ private Runnable commitDelegate = new Runnable() {
+ @Override
+ public void run() {
+ log.debug("{} Committing its state", logPrefix);
+ // 1) flush local state
+ stateMgr.flush(processorContext);
+
+ log.trace("{} Start flushing its producer's sent records upon committing its state", logPrefix);
+ // 2) flush produced records in the downstream and change logs of local states
+ recordCollector.flush();
+
+ // 3) commit consumed offsets if it is dirty already
+ commitOffsets();
+ }
+ };
/**
* Create {@link StreamTask} with its assigned partitions
@@ -85,10 +103,12 @@ public class StreamTask extends AbstractTask implements Punctuator {
StreamsMetrics metrics,
StateDirectory stateDirectory,
ThreadCache cache,
+ Time time,
final RecordCollector recordCollector) {
super(id, applicationId, partitions, topology, consumer, restoreConsumer, false, stateDirectory, cache);
this.punctuationQueue = new PunctuationQueue();
this.maxBufferedSize = config.getInt(StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ this.metrics = new TaskMetrics(metrics);
// create queues for each assigned partition and associate them
// to corresponding source nodes in the processor topology
@@ -113,7 +133,7 @@ public class StreamTask extends AbstractTask implements Punctuator {
// initialize the topology with its own context
this.processorContext = new ProcessorContextImpl(id, this, config, this.recordCollector, stateMgr, metrics, cache);
-
+ this.time = time;
// initialize the state stores
log.info("{} Initializing state stores", logPrefix);
initializeStateStores();
@@ -242,7 +262,7 @@ public class StreamTask extends AbstractTask implements Punctuator {
log.trace("{} Punctuating processor {} with timestamp {}", logPrefix, node.name(), timestamp);
try {
- node.processor().punctuate(timestamp);
+ node.punctuate(timestamp);
} catch (KafkaException ke) {
throw new StreamsException(String.format("Exception caught in punctuate. taskId=%s processor=%s", id, node.name()), ke);
} finally {
@@ -250,21 +270,11 @@ public class StreamTask extends AbstractTask implements Punctuator {
}
}
-
/**
* Commit the current task state
*/
public void commit() {
- log.debug("{} Committing its state", logPrefix);
- // 1) flush local state
- stateMgr.flush(processorContext);
-
- log.trace("{} Start flushing its producer's sent records upon committing its state", logPrefix);
- // 2) flush produced records in the downstream and change logs of local states
- recordCollector.flush();
-
- // 3) commit consumed offsets if it is dirty already
- commitOffsets();
+ metrics.metrics.measureLatencyNs(time, commitDelegate, metrics.taskCommitTimeSensor);
}
/**
@@ -321,7 +331,7 @@ public class StreamTask extends AbstractTask implements Punctuator {
for (ProcessorNode node : this.topology.processors()) {
processorContext.setCurrentNode(node);
try {
- node.init(this.processorContext);
+ node.init(processorContext);
} finally {
processorContext.setCurrentNode(null);
}
@@ -361,6 +371,7 @@ public class StreamTask extends AbstractTask implements Punctuator {
this.partitionGroup.close();
closeTopology();
+ metrics.removeAllSensors();
}
@Override
@@ -385,9 +396,26 @@ public class StreamTask extends AbstractTask implements Punctuator {
return super.toString();
}
+ protected class TaskMetrics {
+ final StreamsMetricsImpl metrics;
+ final Sensor taskCommitTimeSensor;
+
+
+ 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);
+ }
+
+ public void removeAllSensors() {
+ metrics.removeSensor(taskCommitTimeSensor);
+ }
+ }
+
@Override
public void flushState() {
super.flushState();
recordCollector.flush();
}
+
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/e43cf224/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 65e857f..90194c6 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
@@ -24,20 +24,16 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.metrics.MeasurableStat;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.metrics.stats.Avg;
import org.apache.kafka.common.metrics.stats.Count;
import org.apache.kafka.common.metrics.stats.Max;
-import org.apache.kafka.common.metrics.stats.Min;
import org.apache.kafka.common.metrics.stats.Rate;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.KafkaClientSupplier;
import org.apache.kafka.streams.StreamsConfig;
-import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.errors.LockException;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TaskIdFormatException;
@@ -46,7 +42,6 @@ import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.TopologyBuilder;
import org.apache.kafka.streams.state.HostInfo;
import org.apache.kafka.streams.state.internals.ThreadCache;
-import org.apache.kafka.streams.state.internals.ThreadCacheMetrics;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -112,7 +107,7 @@ public class StreamThread extends Thread {
private final Set<Integer> validTransitions = new HashSet<>();
- State(final Integer...validTransitions) {
+ State(final Integer... validTransitions) {
this.validTransitions.addAll(Arrays.asList(validTransitions));
}
@@ -124,6 +119,7 @@ public class StreamThread extends Thread {
return validTransitions.contains(newState.ordinal());
}
}
+
private volatile State state = State.NOT_RUNNING;
private StateListener stateListener = null;
@@ -142,10 +138,8 @@ public class StreamThread extends Thread {
}
/**
- * Set the {@link StateListener} to be notified when state changes.
- * Note this API is internal to Kafka Streams and is not intended to be used by an
- * external application.
- * @param listener
+ * Set the {@link StateListener} to be notified when state changes. Note this API is internal to
+ * Kafka Streams and is not intended to be used by an external application.
*/
public void setStateListener(final StateListener listener) {
this.stateListener = listener;
@@ -203,7 +197,7 @@ public class StreamThread extends Thread {
private final long pollTimeMs;
private final long cleanTimeMs;
private final long commitTimeMs;
- private final StreamsMetricsImpl sensors;
+ private final StreamsMetricsThreadImpl streamsMetrics;
final StateDirectory stateDirectory;
private StreamPartitionAssignor partitionAssignor = null;
@@ -276,6 +270,10 @@ public class StreamThread extends Thread {
return state == State.RUNNING;
}
+ public String threadClientId() {
+ return threadClientId;
+ }
+
public StreamThread(TopologyBuilder builder,
StreamsConfig config,
KafkaClientSupplier clientSupplier,
@@ -297,13 +295,14 @@ public class StreamThread extends Thread {
this.partitionGrouper = config.getConfiguredInstance(StreamsConfig.PARTITION_GROUPER_CLASS_CONFIG, PartitionGrouper.class);
this.streamsMetadataState = streamsMetadataState;
threadClientId = clientId + "-" + threadName;
- this.sensors = new StreamsMetricsImpl(metrics);
+ this.streamsMetrics = new StreamsMetricsThreadImpl(metrics, "stream-metrics", "thread." + threadClientId,
+ Collections.singletonMap("client-id", threadClientId));
if (config.getLong(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG) < 0) {
log.warn("Negative cache size passed in thread [{}]. Reverting to cache size of 0 bytes.", threadName);
}
long cacheSizeBytes = Math.max(0, config.getLong(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG) /
config.getInt(StreamsConfig.NUM_STREAM_THREADS_CONFIG));
- this.cache = new ThreadCache(threadClientId, cacheSizeBytes, this.sensors);
+ this.cache = new ThreadCache(threadClientId, cacheSizeBytes, this.streamsMetrics);
this.logPrefix = String.format("stream-thread [%s]", threadName);
@@ -348,8 +347,9 @@ public class StreamThread extends Thread {
/**
* Execute the stream processors
+ *
* @throws KafkaException for any Kafka-related exceptions
- * @throws Exception for any other non-Kafka exceptions
+ * @throws Exception for any other non-Kafka exceptions
*/
@Override
public void run() {
@@ -415,6 +415,7 @@ public class StreamThread extends Thread {
log.info("{} Stream thread shutdown complete", logPrefix);
setState(State.NOT_RUNNING);
+ streamsMetrics.removeAllSensors();
}
private RuntimeException unAssignChangeLogPartitions() {
@@ -452,8 +453,13 @@ public class StreamThread extends Thread {
/**
+<<<<<<< HEAD
+ * Similar to shutdownTasksAndState, however does not close the task managers, in the hope that
+ * soon the tasks will be assigned again
+=======
* Similar to shutdownTasksAndState, however does not close the task managers,
* in the hope that soon the tasks will be assigned again
+>>>>>>> apache-kafka/trunk
*/
private void suspendTasksAndState() {
log.debug("{} suspendTasksAndState: suspending all active tasks [{}] and standby tasks [{}]", logPrefix,
@@ -536,8 +542,8 @@ public class StreamThread extends Thread {
}
/**
- * Compute the latency based on the current marked timestamp,
- * and update the marked timestamp with the current system timestamp.
+ * Compute the latency based on the current marked timestamp, and update the marked timestamp
+ * with the current system timestamp.
*
* @return latency
*/
@@ -579,7 +585,7 @@ public class StreamThread extends Thread {
StreamTask task = activeTasksByPartition.get(partition);
numAddedRecords += task.addRecords(partition, records.records(partition));
}
- sensors.skippedRecordsSensor.record(records.count() - numAddedRecords, timerStartedMs);
+ streamsMetrics.skippedRecordsSensor.record(records.count() - numAddedRecords, timerStartedMs);
polledRecords = true;
} else {
polledRecords = false;
@@ -587,7 +593,7 @@ public class StreamThread extends Thread {
// only record poll latency is long poll is required
if (longPoll) {
- sensors.pollTimeSensor.record(computeLatency());
+ streamsMetrics.pollTimeSensor.record(computeLatency());
}
}
@@ -603,7 +609,7 @@ public class StreamThread extends Thread {
requiresPoll = requiresPoll || task.requiresPoll();
- sensors.processTimeSensor.record(computeLatency());
+ streamsMetrics.processTimeSensor.record(computeLatency());
maybePunctuate(task);
@@ -680,7 +686,7 @@ public class StreamThread extends Thread {
// check whether we should punctuate based on the task's partition group timestamp;
// which are essentially based on record timestamp.
if (task.maybePunctuate())
- sensors.punctuateTimeSensor.record(computeLatency());
+ streamsMetrics.punctuateTimeSensor.record(computeLatency());
} catch (KafkaException e) {
log.error("{} Failed to punctuate active task {}: ", logPrefix, task.id(), e);
@@ -745,7 +751,7 @@ public class StreamThread extends Thread {
throw e;
}
- sensors.commitTimeSensor.record(computeLatency());
+ streamsMetrics.commitTimeSensor.record(computeLatency());
}
/**
@@ -788,11 +794,11 @@ public class StreamThread extends Thread {
protected StreamTask createStreamTask(TaskId id, Collection<TopicPartition> partitions) {
log.info("{} Creating active task {} with assigned partitions [{}]", logPrefix, id, partitions);
- sensors.taskCreationSensor.record();
+ streamsMetrics.taskCreationSensor.record();
final ProcessorTopology topology = builder.build(id.topicGroupId);
final RecordCollector recordCollector = new RecordCollectorImpl(producer, id.toString());
- return new StreamTask(id, applicationId, partitions, topology, consumer, restoreConsumer, config, sensors, stateDirectory, cache, recordCollector);
+ return new StreamTask(id, applicationId, partitions, topology, consumer, restoreConsumer, config, streamsMetrics, stateDirectory, cache, time, recordCollector);
}
private StreamTask findMatchingSuspendedTask(final TaskId taskId, final Set<TopicPartition> partitions) {
@@ -901,12 +907,12 @@ public class StreamThread extends Thread {
StandbyTask createStandbyTask(TaskId id, Collection<TopicPartition> partitions) {
log.info("{} Creating new standby task {} with assigned partitions [{}]", logPrefix, id, partitions);
- sensors.taskCreationSensor.record();
+ streamsMetrics.taskCreationSensor.record();
ProcessorTopology topology = builder.build(id.topicGroupId);
if (!topology.stateStores().isEmpty()) {
- return new StandbyTask(id, applicationId, partitions, topology, consumer, restoreConsumer, config, sensors, stateDirectory);
+ return new StandbyTask(id, applicationId, partitions, topology, consumer, restoreConsumer, config, streamsMetrics, stateDirectory);
} else {
return null;
}
@@ -1005,7 +1011,7 @@ public class StreamThread extends Thread {
public void apply(final AbstractTask task) {
log.info("{} Closing a task {}", StreamThread.this.logPrefix, task.id());
task.close();
- sensors.taskDestructionSensor.record();
+ streamsMetrics.taskDestructionSensor.record();
}
}, "close");
}
@@ -1016,7 +1022,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();
- sensors.taskDestructionSensor.record();
+ streamsMetrics.taskDestructionSensor.record();
}
}, "close");
}
@@ -1063,12 +1069,11 @@ public class StreamThread extends Thread {
return sb.toString();
}
- private class StreamsMetricsImpl implements StreamsMetrics, ThreadCacheMetrics {
- final Metrics metrics;
- final String metricGrpName;
- final String sensorNamePrefix;
- final Map<String, String> metricTags;
-
+ /**
+ * This class extends {@link #StreamsMetricsImpl(Metrics, String, String, Map)} and
+ * overrides one of its functions for efficiency
+ */
+ private class StreamsMetricsThreadImpl extends StreamsMetricsImpl {
final Sensor commitTimeSensor;
final Sensor pollTimeSensor;
final Sensor processTimeSensor;
@@ -1077,117 +1082,54 @@ public class StreamThread extends Thread {
final Sensor taskDestructionSensor;
final Sensor skippedRecordsSensor;
- public StreamsMetricsImpl(Metrics metrics) {
- this.metrics = metrics;
- this.metricGrpName = "stream-metrics";
- this.sensorNamePrefix = "thread." + threadClientId;
- this.metricTags = Collections.singletonMap("client-id", threadClientId);
+ 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(sensorNamePrefix + ".commit-time");
- this.commitTimeSensor.add(metrics.metricName("commit-time-avg", metricGrpName, "The average commit time in ms", metricTags), new Avg());
- this.commitTimeSensor.add(metrics.metricName("commit-time-max", metricGrpName, "The maximum commit time in ms", metricTags), new Max());
- this.commitTimeSensor.add(metrics.metricName("commit-calls-rate", metricGrpName, "The average per-second number of commit calls", metricTags), 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(sensorNamePrefix + ".poll-time");
- this.pollTimeSensor.add(metrics.metricName("poll-time-avg", metricGrpName, "The average poll time in ms", metricTags), new Avg());
- this.pollTimeSensor.add(metrics.metricName("poll-time-max", metricGrpName, "The maximum poll time in ms", metricTags), new Max());
- this.pollTimeSensor.add(metrics.metricName("poll-calls-rate", metricGrpName, "The average per-second number of record-poll calls", metricTags), 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(sensorNamePrefix + ".process-time");
- this.processTimeSensor.add(metrics.metricName("process-time-avg-ms", metricGrpName, "The average process time in ms", metricTags), new Avg());
- this.processTimeSensor.add(metrics.metricName("process-time-max-ms", metricGrpName, "The maximum process time in ms", metricTags), new Max());
- this.processTimeSensor.add(metrics.metricName("process-calls-rate", metricGrpName, "The average per-second number of process calls", metricTags), 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(sensorNamePrefix + ".punctuate-time");
- this.punctuateTimeSensor.add(metrics.metricName("punctuate-time-avg", metricGrpName, "The average punctuate time in ms", metricTags), new Avg());
- this.punctuateTimeSensor.add(metrics.metricName("punctuate-time-max", metricGrpName, "The maximum punctuate time in ms", metricTags), new Max());
- this.punctuateTimeSensor.add(metrics.metricName("punctuate-calls-rate", metricGrpName, "The average per-second number of punctuate calls", metricTags), 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.taskCreationSensor = metrics.sensor(sensorNamePrefix + ".task-creation");
- this.taskCreationSensor.add(metrics.metricName("task-creation-rate", metricGrpName, "The average per-second number of newly created tasks", metricTags), 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.taskDestructionSensor = metrics.sensor(sensorNamePrefix + ".task-destruction");
- this.taskDestructionSensor.add(metrics.metricName("task-destruction-rate", metricGrpName, "The average per-second number of destructed tasks", metricTags), 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 = metrics.sensor(sensorNamePrefix + ".skipped-records");
- this.skippedRecordsSensor.add(metrics.metricName("skipped-records-count", metricGrpName, "The average per-second number of skipped records.", metricTags), new Rate(new Count()));
}
+
@Override
public void recordLatency(Sensor sensor, long startNs, long endNs) {
sensor.record(endNs - startNs, timerStartedMs);
}
- @Override
- public void recordCacheSensor(Sensor sensor, double count) {
- sensor.record(count);
- }
-
- /**
- * @throws IllegalArgumentException if tags is not constructed in key-value pairs
- */
- @Override
- public Sensor addLatencySensor(String scopeName, String entityName, String operationName, String... tags) {
- // extract the additional tags if there are any
- Map<String, String> tagMap = new HashMap<>(this.metricTags);
- if ((tags.length % 2) != 0)
- throw new IllegalArgumentException("Tags needs to be specified in key-value pairs");
-
- for (int i = 0; i < tags.length; i += 2)
- tagMap.put(tags[i], tags[i + 1]);
-
- String metricGroupName = "stream-" + scopeName + "-metrics";
-
- // first add the global operation metrics if not yet, with the global tags only
- Sensor parent = metrics.sensor(sensorNamePrefix + "." + scopeName + "-" + operationName);
- addLatencyMetrics(metricGroupName, parent, "all", operationName, this.metricTags);
-
- // add the store operation metrics with additional tags
- Sensor sensor = metrics.sensor(sensorNamePrefix + "." + scopeName + "-" + entityName + "-" + operationName, parent);
- addLatencyMetrics(metricGroupName, sensor, entityName, operationName, tagMap);
-
- return sensor;
- }
-
- @Override
- public Sensor addCacheSensor(String entityName, String operationName, String... tags) {
- // extract the additional tags if there are any
- Map<String, String> tagMap = new HashMap<>(this.metricTags);
- if ((tags.length % 2) != 0)
- throw new IllegalArgumentException("Tags needs to be specified in key-value pairs");
-
- for (int i = 0; i < tags.length; i += 2)
- tagMap.put(tags[i], tags[i + 1]);
-
- String metricGroupName = "stream-thread-cache-metrics";
-
- Sensor sensor = metrics.sensor(sensorNamePrefix + "-" + entityName + "-" + operationName);
- addCacheMetrics(metricGroupName, sensor, entityName, operationName, tagMap);
- return sensor;
-
- }
-
- private void addCacheMetrics(String metricGrpName, Sensor sensor, String entityName, String opName, Map<String, String> tags) {
- maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-avg", metricGrpName,
- "The current count of " + entityName + " " + opName + " operation.", tags), new Avg());
- maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-min", metricGrpName,
- "The current count of " + entityName + " " + opName + " operation.", tags), new Min());
- maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-max", metricGrpName,
- "The current count of " + entityName + " " + opName + " operation.", tags), new Max());
- }
-
- private void addLatencyMetrics(String metricGrpName, Sensor sensor, String entityName, String opName, Map<String, String> tags) {
- maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-avg-latency-ms", metricGrpName,
- "The average latency in milliseconds of " + entityName + " " + opName + " operation.", tags), new Avg());
- maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-max-latency-ms", metricGrpName,
- "The max latency in milliseconds of " + entityName + " " + opName + " operation.", tags), new Max());
- maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-qps", metricGrpName,
- "The average number of occurrence of " + entityName + " " + opName + " operation per second.", tags), new Rate(new Count()));
- }
+ public void removeAllSensors() {
+ removeSensor(commitTimeSensor);
+ removeSensor(pollTimeSensor);
+ removeSensor(processTimeSensor);
+ removeSensor(punctuateTimeSensor);
+ removeSensor(taskCreationSensor);
+ removeSensor(taskDestructionSensor);
+ removeSensor(skippedRecordsSensor);
- private void maybeAddMetric(Sensor sensor, MetricName name, MeasurableStat stat) {
- if (!metrics.metrics().containsKey(name))
- sensor.add(name, stat);
}
}