[
https://issues.apache.org/jira/browse/KAFKA-20464?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
chomingi reassigned KAFKA-20464:
--------------------------------
Assignee: chomingi
> Linking stores using ConnectedStoreProvider causes corrupt topology
> -------------------------------------------------------------------
>
> Key: KAFKA-20464
> URL: https://issues.apache.org/jira/browse/KAFKA-20464
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Reporter: Stef Noten
> Assignee: chomingi
> Priority: Minor
>
> When linking more than 1 processor to a given store by passing the same
> StoreBuilder instance, the result is a corrupt topology. However, this is
> explicitly supported according the javadoc of ConnectedStoreProvider.
> Both processors should end up in the same subtopology, so that their tasks of
> the same partition run on the same StreamThread. However, they may end up in
> different subtopologies and threads, causing corrupt reads/updates.
> h2. Example
> {code:java}
> private static final StoreBuilder<KeyValueStore<String, String>> STORE =
> Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("store"),
> Serdes.String(), Serdes.String());
> @Test // SUCCEEDS
> void processorsLinkedByStoreNameEndUpInSameSubtopology() {
> var builder = new StreamsBuilder();
> builder.addStateStore(STORE);
> builder.stream("topic1").processValues(new
> ProcessorSupplierLinkedToStores<>(Set.of()), STORE.name());
> builder.stream("topic2").processValues(new
> ProcessorSupplierLinkedToStores<>(Set.of()), STORE.name());
> var topologyDescription = builder.build().describe();
> System.out.println(topologyDescription);
> assertEquals(topologyDescription.subtopologies().size(), 1);
> }
> @Test // FAILS!
> void processorsLinkedByStoreBuildersEndUpInSameSubtopology() {
> var builder = new StreamsBuilder();
> builder.stream("topic1").processValues(new
> ProcessorSupplierLinkedToStores<>(Set.of(STORE)));
> builder.stream("topic2").processValues(new
> ProcessorSupplierLinkedToStores<>(Set.of(STORE)));
> var topologyDescription = builder.build().describe();
> System.out.println(topologyDescription);
> assertEquals(topologyDescription.subtopologies().size(), 1);
> }
> private static class ProcessorSupplierLinkedToStores<K, V> implements
> FixedKeyProcessorSupplier<K, V, V> {
> private final Set<StoreBuilder<?>> stores;
> public ProcessorSupplierLinkedToStores(Set<StoreBuilder<?>> stores) {
> this.stores = stores;
> }
> @Override
> public Set<StoreBuilder<?>> stores() {
> return stores;
> }
> @Override
> public FixedKeyProcessor<K, V, V> get() {
> return new FixedKeyProcessor<K, V, V>() {
> @Override
> public void process(FixedKeyRecord<K, V> record) {
> }
> };
> }
> }{code}
> h3. Topology when linking by name (correct)
> {code:java}
> Topologies:
> Sub-topology: 0
> Source: KSTREAM-SOURCE-0000000000 (topics: [topic1])
> --> KSTREAM-PROCESSVALUES-0000000001
> Source: KSTREAM-SOURCE-0000000002 (topics: [topic2])
> --> KSTREAM-PROCESSVALUES-0000000003
> Processor: KSTREAM-PROCESSVALUES-0000000001 (stores: [store])
> --> none
> <-- KSTREAM-SOURCE-0000000000
> Processor: KSTREAM-PROCESSVALUES-0000000003 (stores: [store])
> --> none
> <-- KSTREAM-SOURCE-0000000002
> {code}
> h3. Topology when linking with ConnectedStoreProvider.stores (corrupt)
> {code:java}
> Topologies:
> Sub-topology: 0
> Source: KSTREAM-SOURCE-0000000000 (topics: [topic1])
> --> KSTREAM-PROCESSVALUES-0000000001
> Processor: KSTREAM-PROCESSVALUES-0000000001 (stores: [store])
> --> none
> <-- KSTREAM-SOURCE-0000000000
> Sub-topology: 1
> Source: KSTREAM-SOURCE-0000000002 (topics: [topic2])
> --> KSTREAM-PROCESSVALUES-0000000003
> Processor: KSTREAM-PROCESSVALUES-0000000003 (stores: [store])
> --> none
> <-- KSTREAM-SOURCE-0000000002{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
