This is an automated email from the ASF dual-hosted git repository. mjsax pushed a commit to branch 3.6 in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.6 by this push: new 4f1fdfe9864 KAFKA-13973: Fix inflated block cache metrics (#14317) 4f1fdfe9864 is described below commit 4f1fdfe98648f0c7bd41d1897a419aafec2b9fe0 Author: Nick Telford <nick.telf...@gmail.com> AuthorDate: Mon Sep 18 10:09:10 2023 +0100 KAFKA-13973: Fix inflated block cache metrics (#14317) All block cache metrics are being multiplied by the total number of column families. In a `RocksDBTimestampedStore`, we have 2 column families (the default, and the timestamped values), which causes all block cache metrics in these stores to become doubled. The cause is that our metrics recorder uses `getAggregatedLongProperty` to fetch block cache metrics. `getAggregatedLongProperty` queries the property on each column family in the database, and sums the results. Since we always configure all column families to share the same block cache, that causes the same block cache to be queried multiple times for its metrics, with the results added togehter, effectively multiplying the real value by the total number of column families. To fix this, we should simply use `getLongProperty`, which queries a single column family (the default one). Since all column families share the same block cache, querying just one of them will give us the correct metrics for that shared block cache. Note: the same block cache is shared among all column families of a store irrespective of whether the user has configured a shared block cache across multiple stores. Reviewers: Matthias J. Sax <matth...@confluent.io>, Bruno Cadonna <cado...@apache.org> --- .../state/internals/RocksDBTimestampedStore.java | 2 +- .../internals/metrics/RocksDBMetricsRecorder.java | 4 +- .../metrics/RocksDBBlockCacheMetricsTest.java | 136 +++++++++++++++++++++ .../metrics/RocksDBMetricsRecorderGaugesTest.java | 19 ++- 4 files changed, 155 insertions(+), 6 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java index dd070597945..d2876c63f5b 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java @@ -50,7 +50,7 @@ import static org.apache.kafka.streams.state.TimestampedBytesStore.convertToTime public class RocksDBTimestampedStore extends RocksDBStore implements TimestampedBytesStore { private static final Logger log = LoggerFactory.getLogger(RocksDBTimestampedStore.class); - RocksDBTimestampedStore(final String name, + public RocksDBTimestampedStore(final String name, final String metricsScope) { super(name, metricsScope); } diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorder.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorder.java index df68f2e8024..7680b060802 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorder.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorder.java @@ -373,14 +373,14 @@ public class RocksDBMetricsRecorder { // values of RocksDB properties are of type unsigned long in C++, i.e., in Java we need to use // BigInteger and construct the object from the byte representation of the value result = new BigInteger(1, longToBytes( - valueProvider.db.getAggregatedLongProperty(ROCKSDB_PROPERTIES_PREFIX + propertyName) + valueProvider.db.getLongProperty(ROCKSDB_PROPERTIES_PREFIX + propertyName) )); break; } else { // values of RocksDB properties are of type unsigned long in C++, i.e., in Java we need to use // BigInteger and construct the object from the byte representation of the value result = result.add(new BigInteger(1, longToBytes( - valueProvider.db.getAggregatedLongProperty(ROCKSDB_PROPERTIES_PREFIX + propertyName) + valueProvider.db.getLongProperty(ROCKSDB_PROPERTIES_PREFIX + propertyName) ))); } } catch (final RocksDBException e) { diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBBlockCacheMetricsTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBBlockCacheMetricsTest.java new file mode 100644 index 00000000000..ecff033e037 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBBlockCacheMetricsTest.java @@ -0,0 +1,136 @@ +/* + * 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.metrics; + +import org.apache.kafka.common.MetricName; +import org.apache.kafka.common.metrics.KafkaMetric; +import org.apache.kafka.common.utils.Utils; +import org.apache.kafka.streams.processor.StateStoreContext; +import org.apache.kafka.streams.processor.TaskId; +import org.apache.kafka.streams.processor.internals.ProcessorContextUtils; +import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; +import org.apache.kafka.streams.state.internals.BlockBasedTableConfigWithAccessibleCache; +import org.apache.kafka.streams.state.internals.RocksDBStore; +import org.apache.kafka.streams.state.internals.RocksDBTimestampedStore; +import org.apache.kafka.test.MockInternalProcessorContext; +import org.apache.kafka.test.TestUtils; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.io.File; +import java.io.IOException; +import java.math.BigInteger; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Properties; +import java.util.stream.Stream; + +import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.STATE_STORE_LEVEL_GROUP; +import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.STORE_ID_TAG; +import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.TASK_ID_TAG; +import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.THREAD_ID_TAG; +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class RocksDBBlockCacheMetricsTest { + + private static final String STORE_NAME = "test"; + private static final String METRICS_SCOPE = "test-scope"; + + private static TaskId taskId = new TaskId(0, 0); + + public static Stream<Arguments> stores() { + final File stateDir = TestUtils.tempDirectory("state"); + return Stream.of( + Arguments.of(new RocksDBStore(STORE_NAME, METRICS_SCOPE), new MockInternalProcessorContext(new Properties(), taskId, stateDir)), + Arguments.of(new RocksDBTimestampedStore(STORE_NAME, METRICS_SCOPE), new MockInternalProcessorContext(new Properties(), taskId, stateDir)) + ); + } + + static void withStore(final RocksDBStore store, final StateStoreContext context, final Runnable function) { + store.init(context, store); + try { + function.run(); + } finally { + store.close(); + try { + Utils.delete(context.stateDir()); + } catch (final IOException e) { + // ignore + } + } + } + + @ParameterizedTest + @MethodSource("stores") + public void shouldRecordCorrectBlockCacheCapacity(final RocksDBStore store, final StateStoreContext ctx) { + withStore(store, ctx, () -> + assertMetric(ctx, STATE_STORE_LEVEL_GROUP, RocksDBMetrics.CAPACITY_OF_BLOCK_CACHE, BigInteger.valueOf(50 * 1024 * 1024L))); + } + + @ParameterizedTest + @MethodSource("stores") + public void shouldRecordCorrectBlockCacheUsage(final RocksDBStore store, final StateStoreContext ctx) { + withStore(store, ctx, () -> { + final BlockBasedTableConfigWithAccessibleCache tableFormatConfig = (BlockBasedTableConfigWithAccessibleCache) store.getOptions().tableFormatConfig(); + final long usage = tableFormatConfig.blockCache().getUsage(); + assertMetric(ctx, STATE_STORE_LEVEL_GROUP, RocksDBMetrics.USAGE_OF_BLOCK_CACHE, BigInteger.valueOf(usage)); + }); + } + + @ParameterizedTest + @MethodSource("stores") + public void shouldRecordCorrectBlockCachePinnedUsage(final RocksDBStore store, final StateStoreContext ctx) { + withStore(store, ctx, () -> { + final BlockBasedTableConfigWithAccessibleCache tableFormatConfig = (BlockBasedTableConfigWithAccessibleCache) store.getOptions().tableFormatConfig(); + final long usage = tableFormatConfig.blockCache().getPinnedUsage(); + assertMetric(ctx, STATE_STORE_LEVEL_GROUP, RocksDBMetrics.PINNED_USAGE_OF_BLOCK_CACHE, BigInteger.valueOf(usage)); + }); + } + + public <T> void assertMetric(final StateStoreContext context, final String group, final String metricName, final T expected) { + final StreamsMetricsImpl metrics = ProcessorContextUtils.getMetricsImpl(context); + final MetricName name = metrics.metricsRegistry().metricName( + metricName, + group, + "Ignored", + storeLevelTagMap(taskId.toString(), METRICS_SCOPE, STORE_NAME) + ); + final KafkaMetric metric = (KafkaMetric) metrics.metrics().get(name); + assertEquals(expected, metric.metricValue(), String.format("Value for metric '%s-%s' was incorrect", group, metricName)); + } + + public Map<String, String> threadLevelTagMap(final String threadId) { + final Map<String, String> tagMap = new LinkedHashMap<>(); + tagMap.put(THREAD_ID_TAG, threadId); + return tagMap; + } + + public Map<String, String> taskLevelTagMap(final String threadId, final String taskId) { + final Map<String, String> tagMap = threadLevelTagMap(threadId); + tagMap.put(TASK_ID_TAG, taskId); + return tagMap; + } + + public Map<String, String> storeLevelTagMap(final String taskName, + final String storeType, + final String storeName) { + final Map<String, String> tagMap = taskLevelTagMap(Thread.currentThread().getName(), taskName); + tagMap.put(storeType + "-" + STORE_ID_TAG, storeName); + return tagMap; + } +} diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderGaugesTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderGaugesTest.java index 9c21d1dba9e..a66116e733e 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderGaugesTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderGaugesTest.java @@ -218,7 +218,20 @@ public class RocksDBMetricsRecorderGaugesTest { } private void runAndVerifyBlockCacheMetricsWithMultipleCaches(final String propertyName) throws Exception { - runAndVerifySumOfProperties(propertyName); + final StreamsMetricsImpl streamsMetrics = + new StreamsMetricsImpl(new Metrics(), "test-client", StreamsConfig.METRICS_LATEST, new MockTime()); + final RocksDBMetricsRecorder recorder = new RocksDBMetricsRecorder(METRICS_SCOPE, STORE_NAME); + + recorder.init(streamsMetrics, TASK_ID); + recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1); + recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2); + + final long recordedValue1 = 5L; + final long recordedValue2 = 3L; + when(dbToAdd1.getLongProperty(ROCKSDB_PROPERTIES_PREFIX + propertyName)).thenReturn(recordedValue1); + when(dbToAdd2.getLongProperty(ROCKSDB_PROPERTIES_PREFIX + propertyName)).thenReturn(recordedValue2); + + verifyMetrics(streamsMetrics, propertyName, recordedValue1 + recordedValue2); } private void runAndVerifyBlockCacheMetricsWithSingleCache(final String propertyName) throws Exception { @@ -231,8 +244,8 @@ public class RocksDBMetricsRecorderGaugesTest { recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd1, statisticsToAdd2); final long recordedValue = 5L; - when(dbToAdd1.getAggregatedLongProperty(ROCKSDB_PROPERTIES_PREFIX + propertyName)).thenReturn(recordedValue); - when(dbToAdd2.getAggregatedLongProperty(ROCKSDB_PROPERTIES_PREFIX + propertyName)).thenReturn(recordedValue); + when(dbToAdd1.getLongProperty(ROCKSDB_PROPERTIES_PREFIX + propertyName)).thenReturn(recordedValue); + when(dbToAdd2.getLongProperty(ROCKSDB_PROPERTIES_PREFIX + propertyName)).thenReturn(recordedValue); verifyMetrics(streamsMetrics, propertyName, recordedValue); }