@Override
    public void unregister(final Collection<TopicPartition> revokedChangelogs) {
        // 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 the changelog is still in REGISTERED, it means it has not initialized and started
                // restoring yet, and hence we should not try to remove the changelog partition
                if (!changelogMetadata.state().equals(ChangelogState.REGISTERED)) {
                    revokedInitializedChangelogs.add(partition);

                    // if the changelog is not in RESTORING, it means
                    // the corresponding onRestoreStart was not called; in this case
                    // we should not call onRestoreSuspended either
                    if (changelogMetadata.state().equals(ChangelogState.RESTORING)) {
                        final String storeName = changelogMetadata.storeMetadata.store().name();
                        if (changelogMetadata.stateManager.taskType() == Task.TaskType.ACTIVE) {
                            try {
                                stateRestoreListener.onRestoreSuspended(partition, storeName, changelogMetadata.totalRestored);
                            } catch (final Exception e) {
                                throw new StreamsException("State restore listener failed on restore paused", e);
                            }
                        } else if (changelogMetadata.stateManager.taskType() == TaskType.STANDBY) {
                            final StateStoreMetadata storeMetadata = changelogMetadata.stateManager.storeMetadata(partition);
                            // endOffset and storeOffset may be unknown at this point
                            final long storeOffset = storeMetadata.offset() != null ? storeMetadata.offset() : -1;
                            final long endOffset = storeMetadata.endOffset() != null ? storeMetadata.endOffset() : -1;
                            // Unregistering running standby tasks means the task has been promoted to active.
                            final StandbyUpdateListener.SuspendReason suspendReason = 
                                changelogMetadata.stateManager.taskState() == Task.State.RUNNING 
                                    ? StandbyUpdateListener.SuspendReason.PROMOTED
                                    : StandbyUpdateListener.SuspendReason.MIGRATED;
                            try {
                                standbyUpdateListener.onUpdateSuspended(partition, storeName, storeOffset, endOffset, suspendReason);
                            } catch (final Exception e) {
                                throw new StreamsException("Standby updater listener failed on update suspended", e);
                            }
                        }
                    }
                }

                changelogMetadata.clear();
            } else {
                log.debug("Changelog partition {} could not be found," +
                    " it could be already cleaned up during the handling" +
                    " of task corruption and never restore again", partition);
            }
        }

        removeChangelogsFromRestoreConsumer(revokedInitializedChangelogs);

        if (changelogs.isEmpty()) {
            state = ChangelogReaderState.ACTIVE_RESTORING;
        }
    }
restore a changelog with its buffered records if there's any; for active changelogs also check if
it has completed the restoration and can transit to COMPLETED state and trigger restore callbacks

@return number of records restored
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java