cadonna commented on a change in pull request #8902: URL: https://github.com/apache/kafka/pull/8902#discussion_r444724004
########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorStateManager.java ########## @@ -0,0 +1,33 @@ +/* + * 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.processor.internals; + +import java.util.Map; + +public abstract class AbstractProcessorStateManager implements StateManager { Review comment: I do not consider the abstract class as overkill but rather as cleaner. I definitely would like to keep it. Regarding using `changelogTopicPartitionFor()` instead of `changelogFor()` I have the following thoughts: 1. `changelogTopicPartitionFor()` only returns changelogs for registered state stores whereas `changelogFor()` returns the changelogs as specified in the topology builder. That means if a store has logging enabled `changelogFor()` will return a non-null result (global tables and optimized source tables are an exception because they have a changelog topic but the logging is disabled internally) independently whether the store has been registered or not. We need `changelogFor()` at least internally during the registration of state stores. I introduced `changelogTopicPartitionFor()` to have a guard in `logChange()` to only write to changelogs for registered state store. 2. We could use `changelogTopicPartitionFor()` in the initialization of the state store serdes but this would mean we need to move the initialization of the state store serdes after the initialisation of the underlying state stores in `Metered*Store`. That would actually not be an issue, but IMO making a call to a method dependent on the order of calls to other methods is error-prone and if it could be avoided we should avoid it. Since we need `changelogFor()` anyways in `ProcessorStateManager` I would like to keep both. Additionally, initializing the state serde is somehow independent of registering a state store whose main goal is configuring changelogging. 3. I think I can improve on the naming. I will rename `changelogTopicPartitionFor()` to `registeredChangelogPartitionFor()` ---------------------------------------------------------------- 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
