@Test
public void testClose() throws IOException {

    File baseDir = Files.createTempDirectory(stateDir).toFile();
    File checkpointFile = new File(baseDir, ProcessorStateManager.CHECKPOINT_FILE_NAME);
    try {
        // write an empty checkpoint file
        OffsetCheckpoint oldCheckpoint = new OffsetCheckpoint(checkpointFile);
        oldCheckpoint.write(Collections.<TopicPartition, Long>emptyMap());
        MockRestoreConsumer restoreConsumer = new MockRestoreConsumer();
        restoreConsumer.updatePartitions(persistentStoreTopicName, Utils.mkList(
                new PartitionInfo(persistentStoreTopicName, 1, Node.noNode(), new Node[0], new Node[0])
        ));
        restoreConsumer.updatePartitions(nonPersistentStoreTopicName, Utils.mkList(
                new PartitionInfo(nonPersistentStoreTopicName, 1, Node.noNode(), new Node[0], new Node[0])
        ));
        // set up ack'ed offsets
        HashMap<TopicPartition, Long> ackedOffsets = new HashMap<>();
        ackedOffsets.put(new TopicPartition(persistentStoreTopicName, 1), 123L);
        ackedOffsets.put(new TopicPartition(nonPersistentStoreTopicName, 1), 456L);
        ackedOffsets.put(new TopicPartition(ProcessorStateManager.storeChangelogTopic(applicationId, "otherTopic"), 1), 789L);
        MockStateStoreSupplier.MockStateStore persistentStore = new MockStateStoreSupplier.MockStateStore(persistentStoreName, true);
        MockStateStoreSupplier.MockStateStore nonPersistentStore = new MockStateStoreSupplier.MockStateStore(nonPersistentStoreName, false);
        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, 1, noPartitions, baseDir, restoreConsumer, false);
        try {
            // make sure the checkpoint file is deleted
            assertFalse(checkpointFile.exists());
            restoreConsumer.reset();
            stateMgr.register(persistentStore, true, persistentStore.stateRestoreCallback);
            restoreConsumer.reset();
            stateMgr.register(nonPersistentStore, true, nonPersistentStore.stateRestoreCallback);
        } finally {
            // close the state manager with the ack'ed offsets
            stateMgr.close(ackedOffsets);
        }
        // make sure all stores are closed, and the checkpoint file is written.
        assertTrue(persistentStore.flushed);
        assertTrue(persistentStore.closed);
        assertTrue(nonPersistentStore.flushed);
        assertTrue(nonPersistentStore.closed);
        assertTrue(checkpointFile.exists());
        // the checkpoint file should contain an offset from the persistent store only.
        OffsetCheckpoint newCheckpoint = new OffsetCheckpoint(checkpointFile);
        Map<TopicPartition, Long> checkpointedOffsets = newCheckpoint.read();
        assertEquals(1, checkpointedOffsets.size());
        assertEquals(new Long(123L + 1L), checkpointedOffsets.get(new TopicPartition(persistentStoreTopicName, 1)));
    } finally {
        Utils.delete(baseDir);
    }
}