KAFKA-8980: Refactor state-store-level streams metrics #7584

cadonna · 2019-10-23T12:09:34Z

Refactors metrics according to KIP-444
Introduces StateStoreMetrics as a central provider for state store metrics
Adds metric scope (a.k.a. store type) to the in-memory suppression buffer

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

cadonna · 2019-10-23T12:10:15Z

Call for review: @guozhangwang @vvcephei @bbejeck

cadonna · 2019-10-23T12:13:09Z

...rc/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java

@@ -61,6 +62,7 @@
        new RecordHeaders(new Header[] {new RecordHeader("v", new byte[] {(byte) 1})});
    private static final RecordHeaders V_2_CHANGELOG_HEADERS =
        new RecordHeaders(new Header[] {new RecordHeader("v", new byte[] {(byte) 2})});
+    private static final String METRIC_SCOPE = "in-memory-suppression";


I had to add this metric scope (a.k.a. store type) for the metrics tags. Now the tags would

type=stream-state-metrics, thread-id=[threadId], task-id=[taskId], in-memory-suppression-state-id=[storeName]

What's the tags look like before?

Before it was:

type=stream-buffer-metrics, client-id=[threadId] task-id=[taskId] buffer-id=[storeName]

cadonna · 2019-10-23T12:16:09Z

...s/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java

@@ -50,7 +50,7 @@

    public enum Version {
        LATEST,
-        FROM_100_TO_24
+        FROM_0100_TO_24


Corrected mistake in naming.

cadonna · 2019-10-23T12:18:52Z

streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/StateStoreMetrics.java

+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateToSensor;
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addValueMetricToSensor;
+
+public class StateStoreMetrics {


This class contains all specifications of state-store-level metrics.

cadonna · 2019-10-23T12:20:24Z

...rc/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java

+        bufferSizeSensor = StateStoreMetrics.suppressionBufferSizeSensor(
+            threadId,
+            taskId,
+            METRIC_SCOPE,
+            storeName,
+            streamsMetrics
+        );
+        bufferCountSensor = StateStoreMetrics.suppressionBufferCountSensor(
+            threadId,
+            taskId,
+            METRIC_SCOPE,
+            storeName,
+            streamsMetrics
+        );


Metrics for suppression buffers are fetched in this way now.

Sounds good, I think not relying on context but passing in parameters explicitly is fine (if that's what you mean has changed here).

I wanted just to highlight the most important parts of the PR.

Actually, I would prefer to add the thread ID to the context and use the context for the metrics. I have already tried to do it, but I ran into a ClassCastException because we cast ProcessorContext to InternalProcessorContext (which IMO should be changed in future if possible). Using the context instead of explicit parameters makes the signature of the methods more robust. This is rather minor issue atm.

+1 for adding the thread ID to the context

cadonna · 2019-10-23T12:21:46Z

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java

+        streamsMetrics = (StreamsMetricsImpl) context.metrics();
+
+        putSensor = StateStoreMetrics.putSensor(threadId, taskId, metricsScope, name(), streamsMetrics);
+        putIfAbsentSensor = StateStoreMetrics.putIfAbsentSensor(threadId, taskId, metricsScope, name(), streamsMetrics);
+        putAllSensor = StateStoreMetrics.putAllSensor(threadId, taskId, metricsScope, name(), streamsMetrics);
+        getSensor = StateStoreMetrics.getSensor(threadId, taskId, metricsScope, name(), streamsMetrics);
+        allSensor = StateStoreMetrics.allSensor(threadId, taskId, metricsScope, name(), streamsMetrics);
+        rangeSensor = StateStoreMetrics.rangeSensor(threadId, taskId, metricsScope, name(), streamsMetrics);
+        flushSensor = StateStoreMetrics.flushSensor(threadId, taskId, metricsScope, name(), streamsMetrics);
+        deleteSensor = StateStoreMetrics.deleteSensor(threadId, taskId, metricsScope, name(), streamsMetrics);
+        final Sensor restoreSensor =
+            StateStoreMetrics.restoreSensor(threadId, taskId, metricsScope, name(), streamsMetrics);


Metrics for state stores are fetched in this way now.

guozhangwang · 2019-10-23T18:04:34Z

streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/StateStoreMetrics.java

+                                                       final String storeType,
+                                                       final String storeName,
+                                                       final StreamsMetricsImpl streamsMetrics) {
+        final Sensor sensor = streamsMetrics.storeLevelSensor(


I think in KIP-444 this metric would be replaced with the task-level dropped-records right?

My bad. Will fix it.

I just realized that I need PR #7566 merged before I can fix this, because dropped-records is a task-level sensor.

PR #7566 was merged. Rebased the PR and fixed it.

I could be missing something, but the name still contains expired-window. I thought we were going with dropped-records across all metrics. But this could be a misunderstanding on my part.

I introduced method droppedRecordsSensorOrExpiredWindowRecordDropSensor() in TaskMetrics that is used in the code to get the correct sensor depending on the built-in metrics version. expiredWindowRecordDropSensor() is only called in droppedRecordsSensorOrExpiredWindowRecordDropSensor().

guozhangwang · 2019-10-23T18:05:25Z

...rc/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java

@@ -61,6 +62,7 @@
        new RecordHeaders(new Header[] {new RecordHeader("v", new byte[] {(byte) 1})});
    private static final RecordHeaders V_2_CHANGELOG_HEADERS =
        new RecordHeaders(new Header[] {new RecordHeader("v", new byte[] {(byte) 2})});
+    private static final String METRIC_SCOPE = "in-memory-suppression";


What's the tags look like before?

guozhangwang · 2019-10-23T18:10:15Z

...rc/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java

+        bufferSizeSensor = StateStoreMetrics.suppressionBufferSizeSensor(
+            threadId,
+            taskId,
+            METRIC_SCOPE,
+            storeName,
+            streamsMetrics
+        );
+        bufferCountSensor = StateStoreMetrics.suppressionBufferCountSensor(
+            threadId,
+            taskId,
+            METRIC_SCOPE,
+            storeName,
+            streamsMetrics
+        );


Sounds good, I think not relying on context but passing in parameters explicitly is fine (if that's what you mean has changed here).

guozhangwang · 2019-10-23T18:19:56Z

streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/StateStoreMetrics.java

+        );
+    }
+
+    public static Sensor getSensor(final String threadId,


This is a meta comment: in KIP-439 we also discussed about fixing some metric names which are collapsed and messed with the functionality names, and back then the proposal is:

value-by-key value-by-key-and-time range-by-key range-by-key-range range-by-time-range range-by-key-and-time-range range-by-key-range-and-time-range range-all

Besides the first two, all others are mainly used in IQ only and not in stream processor at all, which means that only the first two are commonly recorded in practice. So I'm thinking maybe we keep the compatibility for the first two while renaming others:

get (in kv-store, it means value-by-key, in window-store, it means value-by-key-time) range-by-key range-by-key-range range-by-time-range range-by-key-and-time-range range-by-key-range-and-time-range range-all

WDYT cc @vvcephei @mjsax who have participated in the discussion of KIP-439 as well.

Hey @guozhangwang . This seems good to me. It shouldn't be too ambiguous either way, and the documentation would have to explain it, anyway.

Ack. I am ok with this suggestions -- will keep it in mind for KIP-439.

Btw: it this a change for KIP-444 what needs to be communicated as follow up to the already accepted KIP?

I think KIP-444 just keep the metrics names on state stores as is -- we will do the renaming on KIP-439 and communicate that in that KIP.

guozhangwang · 2019-10-23T18:22:39Z

streams/src/test/java/org/apache/kafka/streams/integration/MetricsIntegrationTest.java

+            IN_MEMORY_LRUCACHE_TAG_KEY,
+            builtInMetricsVersion
+        );
+        checkRocksDBMetricsByTag(


Thanks for adding the coverage of tag names!

bbejeck

Thanks for the PR @cadonna I've made a pass. Overall looks good, but given the size of the PR, I'd like to make another pass soon.

guozhangwang · 2019-10-24T20:18:02Z

cc @mjsax to take another look at the new state store metrics hierarchy as well since KIP-439 would change it too.

- Refactors metrics according to KIP-444 - Introduces `StateStoreMetrics` as a central provider for state store metrics - Adds metric scope (a.k.a. store type) to the in-memory suppression buffer

guozhangwang · 2019-10-28T20:04:49Z

Made another pass on the rebased PR, lgtm.

cadonna · 2019-10-29T12:59:33Z

Replaced measurements of latency with a common measurement method in StreamsMetricsImpl.

cadonna · 2019-10-29T18:52:11Z

In JDK 11/Scala 2.12, the following tests failed:

kafka.api.SslAdminClientIntegrationTest.testSynchronousAuthorizerAclUpdatesBlockRequestThreads
kafka.api.SslAdminClientIntegrationTest.testCreatePartitions

cadonna · 2019-10-29T18:52:29Z

Retest this, please

bbejeck

LGTM just one minor clarification question.

bbejeck · 2019-10-30T14:33:19Z

...rc/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java

+        bufferSizeSensor = StateStoreMetrics.suppressionBufferSizeSensor(
+            threadId,
+            taskId,
+            METRIC_SCOPE,
+            storeName,
+            streamsMetrics
+        );
+        bufferCountSensor = StateStoreMetrics.suppressionBufferCountSensor(
+            threadId,
+            taskId,
+            METRIC_SCOPE,
+            storeName,
+            streamsMetrics
+        );


+1 for adding the thread ID to the context

bbejeck · 2019-10-30T15:10:34Z

streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/StateStoreMetrics.java

+                                                       final String storeType,
+                                                       final String storeName,
+                                                       final StreamsMetricsImpl streamsMetrics) {
+        final Sensor sensor = streamsMetrics.storeLevelSensor(


I could be missing something, but the name still contains expired-window. I thought we were going with dropped-records across all metrics. But this could be a misunderstanding on my part.

bbejeck · 2019-10-30T16:25:14Z

Merged #7584 into trunk

Fixed a minor conflict in `.gitignore` and fix compiler errors in KafkaUtilities due to `PartitionReplicaAssignment` rename to `ReplicaAssignment`. * apache-github/trunk: (34 commits) HOTFIX: Try to complete Send even if no bytes were written (apache#7622) KAFKA-9080: Revert the check added to validate non-compressed record batch does have continuous incremental offsets KAFKA-8972 (2.4 blocker): TaskManager state should always be updated after rebalance (apache#7620) MINOR: Fix Kafka Streams JavaDocs with regard to new StreamJoined class (apache#7627) MINOR: Fix sensor retrieval in stand0by task's constructor (apache#7632) MINOR: Replace some Java 7 style code with Java 8 style (apache#7623) KAFKA-8868: Generate SubscriptionInfo protocol message (apache#7248) MINOR: Correctly mark offset expiry in GroupMetadataManager's OffsetExpired metric KAFKA-8972 (2.4 blocker): bug fix for restoring task (apache#7617) KAFKA-9093: NullPointerException in KafkaConsumer with group.instance.id (apache#7590) KAFKA-8980: Refactor state-store-level streams metrics (apache#7584) MINOR: Fix documentation for updateCurrentReassignment (apache#7611) MINOR: Preserve backwards-compatibility by renaming the AlterPartitionReassignment metric to PartitionReassignment KAFKA-8972 (2.4 blocker): clear all state for zombie task on TaskMigratedException (apache#7608) KAFKA-9077: Fix reading of metrics of Streams' SimpleBenchmark (apache#7610) KAFKA-8972 (2.4 blocker): correctly release lost partitions during consumer.unsubscribe() (apache#7441) MINOR: improve logging of tasks on shutdown (apache#7597) KAFKA-9048 Pt1: Remove Unnecessary lookup in Fetch Building (apache#7576) MINOR: Fix command examples in kafka-reassign-partitions.sh docs (apache#7583) KAFKA-9102; Increase default zk session timeout and replica max lag [KIP-537] (apache#7596) ...

cadonna commented Oct 23, 2019

View reviewed changes

bbejeck added the streams label Oct 23, 2019

guozhangwang reviewed Oct 23, 2019

View reviewed changes

bbejeck reviewed Oct 23, 2019

View reviewed changes

cadonna added 4 commits October 28, 2019 14:01

KAFKA-8980: Refactor state-store-level streams metrics

ea7d9f3

- Refactors metrics according to KIP-444 - Introduces `StateStoreMetrics` as a central provider for state store metrics - Adds metric scope (a.k.a. store type) to the in-memory suppression buffer

Remove unused import in Rate

b324b39

Remove expired-window-record-drop metrics from latest metrics version

486ed6b

Replace expired-window-record-drop with dropped-records

f54db7d

cadonna force-pushed the AK8980-refactor_store_level_metrics branch from d25c5d4 to f54db7d Compare October 28, 2019 13:54

Replace latency measurement with shared method

430d725

Fix verification in TaskMetricsTest

e5cd569

bbejeck approved these changes Oct 30, 2019

View reviewed changes

bbejeck merged commit fc0f823 into apache:trunk Oct 30, 2019

mjsax added the kip Requires or implements a KIP label Jun 12, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-8980: Refactor state-store-level streams metrics #7584

KAFKA-8980: Refactor state-store-level streams metrics #7584

cadonna commented Oct 23, 2019

cadonna commented Oct 23, 2019

cadonna Oct 23, 2019

guozhangwang Oct 23, 2019

cadonna Oct 24, 2019

cadonna Oct 23, 2019

cadonna Oct 23, 2019

cadonna Oct 23, 2019

guozhangwang Oct 23, 2019

cadonna Oct 24, 2019 •

edited

Loading

bbejeck Oct 30, 2019

cadonna Oct 23, 2019

guozhangwang Oct 23, 2019

cadonna Oct 24, 2019

cadonna Oct 24, 2019 •

edited

Loading

cadonna Oct 28, 2019

bbejeck Oct 30, 2019

cadonna Oct 30, 2019

guozhangwang Oct 23, 2019

guozhangwang Oct 23, 2019

guozhangwang Oct 23, 2019

vvcephei Oct 23, 2019

mjsax Oct 23, 2019

guozhangwang Oct 24, 2019

guozhangwang Oct 23, 2019

bbejeck left a comment

guozhangwang commented Oct 24, 2019

guozhangwang commented Oct 28, 2019

cadonna commented Oct 29, 2019 •

edited

Loading

cadonna commented Oct 29, 2019

cadonna commented Oct 29, 2019

bbejeck left a comment

bbejeck Oct 30, 2019

bbejeck Oct 30, 2019

bbejeck commented Oct 30, 2019

KAFKA-8980: Refactor state-store-level streams metrics #7584

KAFKA-8980: Refactor state-store-level streams metrics #7584

Conversation

cadonna commented Oct 23, 2019

Committer Checklist (excluded from commit message)

cadonna commented Oct 23, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna Oct 24, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna Oct 24, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck left a comment

Choose a reason for hiding this comment

guozhangwang commented Oct 24, 2019

guozhangwang commented Oct 28, 2019

cadonna commented Oct 29, 2019 • edited Loading

cadonna commented Oct 29, 2019

cadonna commented Oct 29, 2019

bbejeck left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck commented Oct 30, 2019

cadonna Oct 24, 2019 •

edited

Loading

cadonna Oct 24, 2019 •

edited

Loading

cadonna commented Oct 29, 2019 •

edited

Loading