@Test public void testGetStore() throws IOException { final MockStateStoreSupplier.MockStateStore mockStateStore = new MockStateStoreSupplier.MockStateStore(nonPersistentStoreName, false); final ProcessorStateManager stateMgr = new ProcessorStateManager( new TaskId(0, 1), noPartitions, false, stateDirectory, Collections.<String, String>emptyMap(), changelogReader, false); try { stateMgr.register(mockStateStore, true, mockStateStore.stateRestoreCallback); assertNull(stateMgr.getStore("noSuchStore")); assertEquals(mockStateStore, stateMgr.getStore(nonPersistentStoreName)); } finally { stateMgr.close(Collections.<TopicPartition, Long>emptyMap()); } }
@Override public StateStore getStore(final String name) { return stores.get(name); }
ProcessorStateManager implements StateManager { @Override public StateStore getStore(final String name) { return stores.get(name); } }
ProcessorStateManager implements StateManager { @Override public StateStore getStore(final String name) { return stores.get(name); } ProcessorStateManager(final TaskId taskId,
                                 final Collection<TopicPartition> sources,
                                 final boolean isStandby,
                                 final StateDirectory stateDirectory,
                                 final Map<String, String> storeToChangelogTopic,
                                 final ChangelogReader changelogReader,
                                 final boolean eosEnabled); }
ProcessorStateManager implements StateManager { @Override public StateStore getStore(final String name) { return stores.get(name); } ProcessorStateManager(final TaskId taskId,
                                 final Collection<TopicPartition> sources,
                                 final boolean isStandby,
                                 final StateDirectory stateDirectory,
                                 final Map<String, String> storeToChangelogTopic,
                                 final ChangelogReader changelogReader,
                                 final boolean eosEnabled); static String storeChangelogTopic(final String applicationId, final String storeName); @Override File baseDir(); @Override void register(final StateStore store,
                         final boolean loggingEnabled,
                         final StateRestoreCallback stateRestoreCallback); @Override Map<TopicPartition, Long> checkpointed(); @Override StateStore getStore(final String name); @Override void flush(); @Override void close(final Map<TopicPartition, Long> ackedOffsets); @Override void checkpoint(final Map<TopicPartition, Long> ackedOffsets); @Override StateStore getGlobalStore(final String name); }
ProcessorStateManager implements StateManager { @Override public StateStore getStore(final String name) { return stores.get(name); } ProcessorStateManager(final TaskId taskId,
                                 final Collection<TopicPartition> sources,
                                 final boolean isStandby,
                                 final StateDirectory stateDirectory,
                                 final Map<String, String> storeToChangelogTopic,
                                 final ChangelogReader changelogReader,
                                 final boolean eosEnabled); static String storeChangelogTopic(final String applicationId, final String storeName); @Override File baseDir(); @Override void register(final StateStore store,
                         final boolean loggingEnabled,
                         final StateRestoreCallback stateRestoreCallback); @Override Map<TopicPartition, Long> checkpointed(); @Override StateStore getStore(final String name); @Override void flush(); @Override void close(final Map<TopicPartition, Long> ackedOffsets); @Override void checkpoint(final Map<TopicPartition, Long> ackedOffsets); @Override StateStore getGlobalStore(final String name); static final String STATE_CHANGELOG_TOPIC_SUFFIX; }