cadonna commented on a change in pull request #9232:
URL: https://github.com/apache/kafka/pull/9232#discussion_r480924297
##########
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) {
+ singleCache = false;
+ } else if (singleCache && cache != valueProviders.cache ||
!singleCache && cache == valueProviders.cache) {
+ throw new IllegalStateException("Caches for store " +
storeName + " of task " + taskId +
+ " are either not all distinct or do not all refer to the
same cache. This is a bug in Kafka Streams. " +
+ "Please open a bug report under
https://issues.apache.org/jira/projects/KAFKA/issues";);
+ }
+ }
+ }
+
+ private void verifyIfSomeAreNull(final String segmentName,
Review comment:
I will rename it to `verifyConsistencyOfValueProvidersAcrossSegments()`.
WDYT?
----------------------------------------------------------------
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:
us...@infra.apache.org
