@Override
    public Set<String> initialize() {
        try {
            checkpointFileCache.putAll(checkpointFile.read());
        } catch (final IOException e) {
            throw new StreamsException("Failed to read checkpoints for global state globalStores", e);
        }

        final Set<String> changelogTopics = new HashSet<>();
        for (final StateStore stateStore : topology.globalStateStores()) {
            final String sourceTopic = storeToChangelogTopic.get(stateStore.name());
            changelogTopics.add(sourceTopic);
            stateStore.init((StateStoreContext) globalProcessorContext, stateStore);
        }

        // make sure each topic-partition from checkpointFileCache is associated with a global state store
        checkpointFileCache.keySet().forEach(tp -> {
            if (!changelogTopics.contains(tp.topic())) {
                log.error(
                    "Encountered a topic-partition in the global checkpoint file not associated with any global" +
                        " state store, topic-partition: {}, checkpoint file: {}. If this topic-partition is no longer valid," +
                        " an application reset and state store directory cleanup will be required.",
                    tp.topic(),
                    checkpointFile
                );
                throw new StreamsException("Encountered a topic-partition not associated with any global state store");
            }
        });
        return Collections.unmodifiableSet(globalStoreNames);
    }
@Test
    public void shouldThrowStreamsExceptionForOldTopicPartitions() throws IOException {
        final HashMap<TopicPartition, Long> expectedOffsets = new HashMap<>();
        expectedOffsets.put(t1, 1L);
        expectedOffsets.put(t2, 1L);
        expectedOffsets.put(t3, 1L);
        expectedOffsets.put(t4, 1L);

        // add an old topic (a topic not associated with any global state store)
        final HashMap<TopicPartition, Long> startOffsets = new HashMap<>(expectedOffsets);
        final TopicPartition tOld = new TopicPartition("oldTopic", 1);
        startOffsets.put(tOld, 1L);

        // start with a checkpoint file will all topic-partitions: expected and old (not
        // associated with any global state store).
        final OffsetCheckpoint checkpoint = new OffsetCheckpoint(checkpointFile);
        checkpoint.write(startOffsets);

        // initialize will throw exception
        final StreamsException e = assertThrows(StreamsException.class, () -> stateManager.initialize());
        assertThat(e.getMessage(), equalTo("Encountered a topic-partition not associated with any global state store"));
    }