You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "chomingi (Jira)" <ji...@apache.org> on 2026/07/20 04:59:00 UTC
[jira] [Assigned] (KAFKA-20464) Linking stores using ConnectedStoreProvider causes corrupt topology
[ 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)