ProcessorStateManager { public StateStore getStore(String name) { return stores.get(name); } ProcessorStateManager(String applicationId, int defaultPartition, Collection<TopicPartition> sources, File baseDir, Consumer<byte[], byte[]> restoreConsumer, boolean isStandby); static String storeChangelogTopic(String applicationId, String storeName); static FileLock lockStateDirectory(File stateDir); File baseDir(); void register(StateStore store, boolean loggingEnabled, StateRestoreCallback stateRestoreCallback); Map<TopicPartition, Long> checkpointedOffsets(); List<ConsumerRecord<byte[], byte[]>> updateStandbyStates(TopicPartition storePartition, List<ConsumerRecord<byte[], byte[]>> records); void putOffsetLimit(TopicPartition partition, long limit); StateStore getStore(String name); void flush(); void close(Map<TopicPartition, Long> ackedOffsets); static final String STATE_CHANGELOG_TOPIC_SUFFIX; static final String CHECKPOINT_FILE_NAME; static final String LOCK_FILE_NAME; }
@Test public void testGetStore() throws IOException { File baseDir = Files.createTempDirectory(stateDir).toFile(); try { MockRestoreConsumer restoreConsumer = new MockRestoreConsumer(); restoreConsumer.updatePartitions(nonPersistentStoreTopicName, Utils.mkList( new PartitionInfo(nonPersistentStoreTopicName, 1, Node.noNode(), new Node[0], new Node[0]) )); MockStateStoreSupplier.MockStateStore mockStateStore = new MockStateStoreSupplier.MockStateStore(nonPersistentStoreName, false); ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, 1, noPartitions, baseDir, restoreConsumer, false); try { stateMgr.register(mockStateStore, true, mockStateStore.stateRestoreCallback); assertNull(stateMgr.getStore("noSuchStore")); assertEquals(mockStateStore, stateMgr.getStore(nonPersistentStoreName)); } finally { stateMgr.close(Collections.<TopicPartition, Long>emptyMap()); } } finally { Utils.delete(baseDir); } }