[GitHub] [kafka] cadonna commented on a change in pull request #9232: KAFKA-9924: Add remaining property-based RocksDB metrics as described in KIP-607

[GitBox <gi...@apache.org>]

cadonna commented on a change in pull request #9232:
URL: https://github.com/apache/kafka/pull/9232#discussion_r480914782



##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorder.java
##########
@@ -150,14 +176,55 @@ private void verifyStatistics(final String segmentName, final Statistics statist
                 statistics != null &&
                 storeToValueProviders.values().stream().anyMatch(valueProviders -> valueProviders.statistics == null))) {
 
-            throw new IllegalStateException("Statistics for store \"" + segmentName + "\" of task " + taskId +
-                " is" + (statistics == null ? " " : " not ") + "null although the statistics of another store in this " +
+            throw new IllegalStateException("Statistics for segment " + segmentName + " of task " + taskId +
+                " is" + (statistics == null ? " " : " not ") + "null although the statistics of another segment in this " +
                 "metrics recorder is" + (statistics != null ? " " : " not ") + "null. " +
                 "This is a bug in Kafka Streams. " +
                 "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues");
         }
     }
 
+    private void verifyDbAndCacheAndStatistics(final String segmentName,
+                                               final RocksDB db,
+                                               final Cache cache,
+                                               final Statistics statistics) {
+        for (final DbAndCacheAndStatistics valueProviders : storeToValueProviders.values()) {
+            verifyIfSomeAreNull(segmentName, statistics, valueProviders.statistics, "statistics");
+            verifyIfSomeAreNull(segmentName, cache, valueProviders.cache, "cache");
+            if (db == valueProviders.db) {
+                throw new IllegalStateException("DB instance for store " + segmentName + " of task " + taskId +
+                    " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
+                    "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues");
+            }
+            if (storeToValueProviders.size() == 1 && cache != valueProviders.cache) {

Review comment:
       If RocksDB uses the memory bounded configuration, all RocksDB instances use the same cache, i.e., the reference to the same cache. 
   At this point in the code the value providers for the given segment with the cache `cache` has not been added yet to the value providers used in this recorder. To understand if different caches are used for different segments (i.e., `singleCache = false`), it is not enough to just check if only one single cache has been already added, we also need to check if the already added cache (i.e., `valueProviders.cache`) is different from the cache to add (`cache`).       




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org