@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;
        }
    }
@Test
    public void shouldNotThrowOnUnknownRevokedPartition() {
        try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StoreChangelogReader.class)) {
            appender.setClassLogger(StoreChangelogReader.class, Level.DEBUG);
            changelogReader.unregister(Collections.singletonList(new TopicPartition("unknown", 0)));

            assertThat(
                appender.getMessages(),
                hasItem("test-reader Changelog partition unknown-0 could not be found," +
                    " it could be already cleaned up during the handling of task corruption and never restore again")
            );
        }
    }