ezhou413 commented on code in PR #22267:
URL: https://github.com/apache/kafka/pull/22267#discussion_r3229838941
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -182,7 +182,21 @@ private void registerMetrics() {
);
if (!persistent()) {
StateStoreMetrics.addNumKeysGauge(taskId.toString(), metricsScope,
name(), streamsMetrics,
- (config, now) -> wrapped().approximateNumEntries());
+ (config, now) -> {
+ final InMemoryKeyValueStore inMemoryStore =
findInMemoryKeyValueStore(wrapped());
+ return inMemoryStore != null ?
inMemoryStore.approximateNumEntries() : -1L;
+ }
+ );
+ }
+ }
+
+ private static InMemoryKeyValueStore findInMemoryKeyValueStore(final
StateStore store) {
+ if (store instanceof InMemoryKeyValueStore) {
+ return (InMemoryKeyValueStore) store;
+ } else if (store instanceof WrappedStateStore) {
+ return findInMemoryKeyValueStore(((WrappedStateStore<?, ?, ?>)
store).wrapped());
+ } else {
+ return null;
Review Comment:
I don't think null is ever a valid value but it seems we can fall into this
case if a custom state store is used. I guess the question is do we want to
register this metric when a custom state store is used?
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
