boolean changelogAsSource(final TopicPartition partition) {
        return sourcePartitions.contains(partition);
    }
@Test
    public void shouldReportChangelogAsSource() {
        final ProcessorStateManager stateMgr = new ProcessorStateManager(
            taskId,
            Task.TaskType.STANDBY,
            false,
            logContext,
            stateDirectory,
            changelogReader,
            mkMap(
                mkEntry(persistentStoreName, persistentStoreTopicName),
                mkEntry(persistentStoreTwoName, persistentStoreTwoTopicName),
                mkEntry(nonPersistentStoreName, nonPersistentStoreTopicName)
            ),
            mkSet(persistentStorePartition, nonPersistentStorePartition),
            false);

        assertTrue(stateMgr.changelogAsSource(persistentStorePartition));
        assertTrue(stateMgr.changelogAsSource(nonPersistentStorePartition));
        assertFalse(stateMgr.changelogAsSource(persistentStoreTwoPartition));
    }