@SuppressWarnings("unchecked")
@Test
public void testUpdate() throws Exception {

    File baseDir = Files.createTempDirectory("test").toFile();
    try {
        StreamsConfig config = createConfig(baseDir);
        StandbyTask task = new StandbyTask(taskId, applicationId, topicPartitions, topology, consumer, restoreStateConsumer, config, null);
        restoreStateConsumer.assign(new ArrayList<>(task.changeLogPartitions()));
        for (ConsumerRecord<Integer, Integer> record : Arrays.asList(
                new ConsumerRecord<>(partition2.topic(), partition2.partition(), 10, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, 1, 100),
                new ConsumerRecord<>(partition2.topic(), partition2.partition(), 20, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, 2, 100),
                new ConsumerRecord<>(partition2.topic(), partition2.partition(), 30, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, 3, 100))) {
            restoreStateConsumer.bufferRecord(record);
        }
        for (Map.Entry<TopicPartition, Long> entry : task.checkpointedOffsets().entrySet()) {
            TopicPartition partition = entry.getKey();
            long offset = entry.getValue();
            if (offset >= 0) {
                restoreStateConsumer.seek(partition, offset);
            } else {
                restoreStateConsumer.seekToBeginning(singleton(partition));
            }
        }
        task.update(partition2, restoreStateConsumer.poll(100).records(partition2));
        StandbyContextImpl context = (StandbyContextImpl) task.context();
        MockStateStoreSupplier.MockStateStore store1 =
                (MockStateStoreSupplier.MockStateStore) context.getStateMgr().getStore(storeName1);
        MockStateStoreSupplier.MockStateStore store2 =
                (MockStateStoreSupplier.MockStateStore) context.getStateMgr().getStore(storeName2);
        assertEquals(Collections.emptyList(), store1.keys);
        assertEquals(Utils.mkList(1, 2, 3), store2.keys);
        task.close();
        File taskDir = new File(StreamThread.makeStateDir(applicationId, baseDir.getCanonicalPath()), taskId.toString());
        OffsetCheckpoint checkpoint = new OffsetCheckpoint(new File(taskDir, ProcessorStateManager.CHECKPOINT_FILE_NAME));
        Map<TopicPartition, Long> offsets = checkpoint.read();
        assertEquals(1, offsets.size());
        assertEquals(new Long(30L + 1L), offsets.get(partition2));
    } finally {
        Utils.delete(baseDir);
    }
}