cadonna commented on a change in pull request #8902: URL: https://github.com/apache/kafka/pull/8902#discussion_r445457706
########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java ########## @@ -578,4 +577,10 @@ private StateStoreMetadata findStore(final TopicPartition changelogPartition) { return found.isEmpty() ? null : found.get(0); } + + @Override + public TopicPartition changelogTopicPartitionFor(final String storeName) { + final StateStoreMetadata storeMetadata = stores.get(storeName); + return storeMetadata == null ? null : storeMetadata.changelogPartition; Review comment: I would say it indicates a bug, if the state store should be changelogged. My point is that retrieving a changelog partition for a state store does not necessarily mean that the state store needs to be registered. However, since `registeredChangelogPartitionFor()` is currently only called for changelogging, I will add two `IllegalStateException`s to this method: one for non registered stores and one for stores with logging disabled. ---------------------------------------------------------------- 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