cadonna commented on a change in pull request #8902:
URL: https://github.com/apache/kafka/pull/8902#discussion_r452165061
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
##########
@@ -73,22 +73,23 @@
private final Set<String> globalNonPersistentStoresTopics = new
HashSet<>();
private final OffsetCheckpoint checkpointFile;
private final Map<TopicPartition, Long> checkpointFileCache;
+ private final Map<String, String> storeToChangelogTopic;
public GlobalStateManagerImpl(final LogContext logContext,
final ProcessorTopology topology,
final Consumer<byte[], byte[]>
globalConsumer,
final StateDirectory stateDirectory,
final StateRestoreListener
stateRestoreListener,
final StreamsConfig config) {
+ storeToChangelogTopic = topology.storeToChangelogTopic();
Review comment:
I would still pass `ProcessorTopology` into the constructor because it
might make the signature of the constructor more stable. I removed the field
for the topology and now we store only `globalStateStores` and
`storeChangelogTopics` in `GlobalStateManagerImpl`.
----------------------------------------------------------------
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:
[email protected]
