KAFKA-20464: Fix topology corruption when ConnectedStoreProviders share a StoreBuilder (#22102)

[KAFKA-20464](https://issues.apache.org/jira/browse/KAFKA-20464)

`ConnectedStoreProvider` Javadoc documents that multiple providers may
return the same `StoreBuilder` instance to share a store. Currently this
places the processors in separate subtopologies because
`addStateStore()` overwrites the `stateFactories` entry, losing the
previous `connectedProcessorNames`.

The fix is a one-line conditional: skip the put when a compatible
factory already exists. Tests added in `StreamsBuilderTest` and
`InternalTopologyBuilderTest`.

On upgrade, affected topologies will see subtopology regrouping — task
IDs from the previously separate subtopology will no longer exist, and
their state will be restored from changelog topics on first restart.

Reviewers: @UladzislauBlok , Bill Bejeck <bbejeck@apache.org>

streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java

@@ -614,7 +614,10 @@ public class InternalTopologyBuilder {
             throw new TopologyException("A different GlobalStateStore has already been added with the name " + storeName);
         }
 
-        stateFactories.put(storeName, storeFactory);
+        // Reuse existing compatible factory to preserve connectedProcessorNames (KAFKA-20464).
+        if (stateFactory == null || allowOverride) {
+            stateFactories.put(storeName, storeFactory);
+        }
 
         if (processorNames != null) {
             for (final String processorName : processorNames) {

streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java

@@ -46,6 +46,9 @@ import org.apache.kafka.streams.kstream.TableJoined;
 import org.apache.kafka.streams.kstream.TimeWindows;
 import org.apache.kafka.streams.kstream.Windowed;
 import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.api.FixedKeyProcessor;
+import org.apache.kafka.streams.processor.api.FixedKeyProcessorSupplier;
+import org.apache.kafka.streams.processor.api.FixedKeyRecord;
 import org.apache.kafka.streams.processor.api.Processor;
 import org.apache.kafka.streams.processor.api.ProcessorContext;
 import org.apache.kafka.streams.processor.api.ProcessorSupplier;
@@ -1333,6 +1336,78 @@ public class StreamsBuilderTest {
         assertNamesForOperation(topology, "KSTREAM-SOURCE-0000000000", "test-fixed-key-processor");
     }
 
+    @Test
+    public void shouldPlaceProcessorsLinkedByStoreBuilderInSameSubtopology() {
+        final StoreBuilder<KeyValueStore<String, String>> store =
+            Stores.keyValueStoreBuilder(
+                Stores.persistentKeyValueStore("store"),
+                Serdes.String(),
+                Serdes.String());
+
+        final StreamsBuilder localBuilder = new StreamsBuilder();
+        localBuilder.stream("topic1")
+            .processValues(new StoreLinkedFixedKeyProcessorSupplier<>(Set.of(store)));
+        localBuilder.stream("topic2")
+            .processValues(new StoreLinkedFixedKeyProcessorSupplier<>(Set.of(store)));
+
+        assertSingleSubtopologyWithBothProcessorsUsingStore(localBuilder, "store");
+    }
+
+    @Test
+    public void shouldPlaceProcessorsInSameSubtopologyWhenMixingStoreNameAndStoreBuilderLinking() {
+        final StoreBuilder<KeyValueStore<String, String>> store =
+            Stores.keyValueStoreBuilder(
+                Stores.persistentKeyValueStore("store"),
+                Serdes.String(),
+                Serdes.String());
+
+        final StreamsBuilder localBuilder = new StreamsBuilder();
+        localBuilder.addStateStore(store);
+        localBuilder.stream("topic1")
+            .processValues(new StoreLinkedFixedKeyProcessorSupplier<>(Set.of()), store.name());
+        localBuilder.stream("topic2")
+            .processValues(new StoreLinkedFixedKeyProcessorSupplier<>(Set.of(store)));
+
+        assertSingleSubtopologyWithBothProcessorsUsingStore(localBuilder, "store");
+    }
+
+    private static void assertSingleSubtopologyWithBothProcessorsUsingStore(final StreamsBuilder builder,
+                                                                           final String storeName) {
+        final Set<TopologyDescription.Subtopology> subtopologies = builder.build().describe().subtopologies();
+        assertEquals(1, subtopologies.size());
+
+        int processorCount = 0;
+        for (final TopologyDescription.Node node : subtopologies.iterator().next().nodes()) {
+            if (node instanceof TopologyDescription.Processor) {
+                processorCount++;
+                assertEquals(Set.of(storeName), ((TopologyDescription.Processor) node).stores());
+            }
+        }
+        assertEquals(2, processorCount);
+    }
+
+    private static class StoreLinkedFixedKeyProcessorSupplier<K, V> implements FixedKeyProcessorSupplier<K, V, V> {
+        private final Set<StoreBuilder<?>> stores;
+
+        StoreLinkedFixedKeyProcessorSupplier(final Set<StoreBuilder<?>> stores) {
+            this.stores = stores;
+        }
+
+        @Override
+        public Set<StoreBuilder<?>> stores() {
+            return stores;
+        }
+
+        @Override
+        public FixedKeyProcessor<K, V, V> get() {
+            return new FixedKeyProcessor<>() {
+                @Override
+                public void process(final FixedKeyRecord<K, V> record) {
+                }
+            };
+        }
+    }
+
     @Test
     public void shouldUseSpecifiedNameForPrintOperation() {
         builder.stream(STREAM_TOPIC).print(Printed.toSysOut().withName("print-processor"));

streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java

@@ -618,22 +618,6 @@ public class InternalTopologyBuilderTest {
         assertThat(stateStoreNamesUnknownSubtopology, nullValue());
     }
 
-    @Test
-    public void shouldAllowAddingSameStoreBuilderMultipleTimes() {
-        builder.setApplicationId("X");
-        builder.addSource(null, "source-1", null, null, null, "topic-1");
-
-        builder.addStateStore(storeFactory);
-        builder.addProcessor("processor-1", new MockApiProcessorSupplier<>(), "source-1");
-        builder.connectProcessorAndStateStores("processor-1", storeFactory.storeName());
-
-        builder.addStateStore(storeFactory);
-        builder.addProcessor("processor-2", new MockApiProcessorSupplier<>(), "source-1");
-        builder.connectProcessorAndStateStores("processor-2", storeFactory.storeName());
-
-        assertEquals(1, builder.buildTopology().stateStores().size());
-    }
-
     @Test
     public void testTopicGroups() {
         builder.setApplicationId("X");
@@ -711,6 +695,21 @@ public class InternalTopologyBuilderTest {
         assertEquals(expectedTopicGroups, topicGroups);
     }
 
+    @Test
+    public void shouldUniteProcessorsWhenAddStateStoreCalledMultipleTimesWithSameBuilder() {
+        final StoreBuilder<?> sharedStoreBuilder = new MockKeyValueStoreBuilder("shared-store", false);
+
+        builder.addSource(null, "source-1", null, null, null, "topic-1");
+        builder.addProcessor("processor-1", new MockApiProcessorSupplier<>(), "source-1");
+        builder.addStateStore(sharedStoreBuilder, "processor-1");
+
+        builder.addSource(null, "source-2", null, null, null, "topic-2");
+        builder.addProcessor("processor-2", new MockApiProcessorSupplier<>(), "source-2");
+        builder.addStateStore(sharedStoreBuilder, "processor-2");
+
+        assertEquals(1, builder.describe().subtopologies().size());
+    }
+
     @Test
     public void testBuild() {
         builder.addSource(null, "source-1", null, null, null, "topic-1", "topic-1x");