@SuppressWarnings("unchecked") @Test public void shouldDriveGlobalStore() throws Exception { final StateStoreSupplier storeSupplier = Stores.create("my-store") .withStringKeys().withStringValues().inMemory().disableLogging().build(); final String global = "global"; final String topic = "topic"; final TopologyBuilder topologyBuilder = this.builder .addGlobalStore(storeSupplier, global, STRING_DESERIALIZER, STRING_DESERIALIZER, topic, "processor", define(new StatefulProcessor("my-store"))); driver = new ProcessorTopologyTestDriver(config, topologyBuilder); final KeyValueStore<String, String> globalStore = (KeyValueStore<String, String>) topologyBuilder.globalStateStores().get("my-store"); driver.process(topic, "key1", "value1", STRING_SERIALIZER, STRING_SERIALIZER); driver.process(topic, "key2", "value2", STRING_SERIALIZER, STRING_SERIALIZER); assertEquals("value1", globalStore.get("key1")); assertEquals("value2", globalStore.get("key2")); }
public List<StateStore> globalStateStores() { return globalStateStores; }
ProcessorTopology { public List<StateStore> globalStateStores() { return globalStateStores; } }
ProcessorTopology { public List<StateStore> globalStateStores() { return globalStateStores; } ProcessorTopology(final List<ProcessorNode> processorNodes,
                             final Map<String, SourceNode> sourceByTopics,
                             final Map<String, SinkNode> sinkByTopics,
                             final List<StateStore> stateStores,
                             final Map<String, String> storeToChangelogTopic,
                             final List<StateStore> globalStateStores); }
ProcessorTopology { public List<StateStore> globalStateStores() { return globalStateStores; } ProcessorTopology(final List<ProcessorNode> processorNodes,
                             final Map<String, SourceNode> sourceByTopics,
                             final Map<String, SinkNode> sinkByTopics,
                             final List<StateStore> stateStores,
                             final Map<String, String> storeToChangelogTopic,
                             final List<StateStore> globalStateStores); Set<String> sourceTopics(); SourceNode source(String topic); Set<SourceNode> sources(); Set<String> sinkTopics(); SinkNode sink(String topic); Set<SinkNode> sinks(); List<ProcessorNode> processors(); List<StateStore> stateStores(); Map<String, String> storeToChangelogTopic(); List<StateStore> globalStateStores(); @Override String toString(); String toString(final String indent); }
ProcessorTopology { public List<StateStore> globalStateStores() { return globalStateStores; } ProcessorTopology(final List<ProcessorNode> processorNodes,
                             final Map<String, SourceNode> sourceByTopics,
                             final Map<String, SinkNode> sinkByTopics,
                             final List<StateStore> stateStores,
                             final Map<String, String> storeToChangelogTopic,
                             final List<StateStore> globalStateStores); Set<String> sourceTopics(); SourceNode source(String topic); Set<SourceNode> sources(); Set<String> sinkTopics(); SinkNode sink(String topic); Set<SinkNode> sinks(); List<ProcessorNode> processors(); List<StateStore> stateStores(); Map<String, String> storeToChangelogTopic(); List<StateStore> globalStateStores(); @Override String toString(); String toString(final String indent);  }