ableegoldman commented on a change in pull request #8248: URL: https://github.com/apache/kafka/pull/8248#discussion_r431513230
########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java ########## @@ -811,17 +812,41 @@ private void prepareChangelogs(final Set<ChangelogMetadata> newPartitionsToResto } } + private RuntimeException invokeOnRestoreEnd(final TopicPartition partition, + final ChangelogMetadata changelogMetadata) { + // only trigger the store's specific listener to make sure we disable bulk loading before transition to standby + final StateStoreMetadata storeMetadata = changelogMetadata.storeMetadata; + final StateRestoreCallback restoreCallback = storeMetadata.restoreCallback(); + final String storeName = storeMetadata.store().name(); + if (restoreCallback instanceof StateRestoreListener) { + try { + ((StateRestoreListener) restoreCallback).onRestoreEnd(partition, storeName, changelogMetadata.totalRestored); + } catch (final RuntimeException e) { + return e; + } + } + return null; + } + @Override - public void remove(final Collection<TopicPartition> revokedChangelogs) { - // Only changelogs that are initialized that been added to the restore consumer's assignment + public void unregister(final Collection<TopicPartition> revokedChangelogs, + final boolean triggerOnRestoreEnd) { + final AtomicReference<RuntimeException> firstException = new AtomicReference<>(null); + + // Only changelogs that are initialized have been added to the restore consumer's assignment final List<TopicPartition> revokedInitializedChangelogs = new ArrayList<>(); for (final TopicPartition partition : revokedChangelogs) { final ChangelogMetadata changelogMetadata = changelogs.remove(partition); if (changelogMetadata != null) { - if (changelogMetadata.state() != ChangelogState.REGISTERED) { + if (triggerOnRestoreEnd && changelogMetadata.state().equals(ChangelogState.RESTORING)) { Review comment: > a new restoration start (for the post-recycled task), without the prior one ever having "ended". This could happen if you have a restoring task that transitions to standby and then back to restoring. But invoking `onRestoreStart` twice in a row in that case is the current behavior, so your listener should presumably already be handling the situation. My impression is that users understand the global restore listener's `onRestoreEnd` to mean that restoration has completed, and invoking it before this would be an unexpected behavior change. I think in an ideal world we would decouple these two callbacks to make the distinction more apparent. ---------------------------------------------------------------- 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