diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
index 8d69ef14aae77..4f1bdc64215de 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
@@ -680,8 +680,11 @@ private void validateGlobalStoreArguments(final String sourceName,
         if (nodeFactories.containsKey(processorName)) {
             throw new TopologyException("Processor " + processorName + " is already added.");
         }
-        if (stateFactories.containsKey(storeName) || globalStateBuilders.containsKey(storeName)) {
-            throw new TopologyException("StateStore " + storeName + " is already added.");
+        if (stateFactories.containsKey(storeName)) {
+            throw new TopologyException("A different StateStore has already been added with the name " + storeName);
+        }
+        if (globalStateBuilders.containsKey(storeName)) {
+            throw new TopologyException("A different GlobalStateStore has already been added with the name " + storeName);
         }
         if (loggingEnabled) {
             throw new TopologyException("StateStore " + storeName + " for global table must not have logging enabled.");
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
index 8402dfd774f5d..1e87a0e743238 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
@@ -25,6 +25,7 @@
 import org.apache.kafka.streams.errors.TopologyException;
 import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.streams.processor.TopicNameExtractor;
+import org.apache.kafka.streams.state.KeyValueStore;
 import org.apache.kafka.streams.state.StoreBuilder;
 import org.apache.kafka.streams.state.Stores;
 import org.apache.kafka.test.MockKeyValueStoreBuilder;
@@ -62,7 +63,7 @@ public class InternalTopologyBuilderTest {
 
     private final Serde<String> stringSerde = Serdes.String();
     private final InternalTopologyBuilder builder = new InternalTopologyBuilder();
-    private final StoreBuilder<?> storeBuilder = new MockKeyValueStoreBuilder("store", false);
+    private final StoreBuilder<?> storeBuilder = new MockKeyValueStoreBuilder("testStore", false);
 
     @Test
     public void shouldAddSourceWithOffsetReset() {
@@ -372,48 +373,112 @@ public void testAddStateStoreWithSink() {
 
     @Test
     public void shouldNotAllowToAddStoresWithSameName() {
+        final StoreBuilder<KeyValueStore<Object, Object>> otherBuilder =
+            new MockKeyValueStoreBuilder("testStore", false);
+
         builder.addStateStore(storeBuilder);
-        final StoreBuilder otherBuilder = new MockKeyValueStoreBuilder("store", false);
-        try {
-            builder.addStateStore(otherBuilder);
-            fail("Should throw TopologyException with store name conflict");
-        } catch (final TopologyException expected) { /* ok */ }
+
+        final TopologyException exception = assertThrows(
+            TopologyException.class,
+            () -> builder.addStateStore(otherBuilder)
+        );
+
+        assertThat(
+            exception.getMessage(),
+            equalTo("Invalid topology: A different StateStore has already been added with the name testStore")
+        );
     }
 
     @Test
     public void shouldNotAllowToAddStoresWithSameNameWhenFirstStoreIsGlobal() {
-        final StoreBuilder storeBuilder = new MockKeyValueStoreBuilder("store", false).withLoggingDisabled();
+        final StoreBuilder<KeyValueStore<Object, Object>> globalBuilder =
+            new MockKeyValueStoreBuilder("testStore", false).withLoggingDisabled();
+
         builder.addGlobalStore(
-                storeBuilder,
+            globalBuilder,
+            "global-store",
+            null,
+            null,
+            null,
+            "global-topic",
+            "global-processor",
+            new MockProcessorSupplier<>()
+        );
+
+        final TopologyException exception = assertThrows(
+            TopologyException.class,
+            () -> builder.addStateStore(storeBuilder)
+        );
+
+        assertThat(
+            exception.getMessage(),
+            equalTo("Invalid topology: A different GlobalStateStore has already been added with the name testStore")
+        );
+    }
+
+    @Test
+    public void shouldNotAllowToAddStoresWithSameNameWhenSecondStoreIsGlobal() {
+        final StoreBuilder<KeyValueStore<Object, Object>> globalBuilder =
+            new MockKeyValueStoreBuilder("testStore", false).withLoggingDisabled();
+
+        builder.addStateStore(storeBuilder);
+
+        final TopologyException exception = assertThrows(
+            TopologyException.class,
+            () -> builder.addGlobalStore(
+                globalBuilder,
                 "global-store",
                 null,
                 null,
                 null,
                 "global-topic",
                 "global-processor",
-                new MockProcessorSupplier<>());
-        try {
-            builder.addStateStore(storeBuilder);
-            fail("Should throw TopologyException with store name conflict");
-        } catch (final TopologyException expected) { /* ok */ }
+                new MockProcessorSupplier<>()
+            )
+        );
+
+        assertThat(
+            exception.getMessage(),
+            equalTo("Invalid topology: A different StateStore has already been added with the name testStore")
+        );
     }
 
     @Test
-    public void shouldNotAllowToAddStoresWithSameNameWhenSecondStoreIsGlobal() {
-        final StoreBuilder storeBuilder = new MockKeyValueStoreBuilder("store", false).withLoggingDisabled();
-        builder.addStateStore(storeBuilder);
-        try {
-            builder.addGlobalStore(
-                    storeBuilder,
-                    "global-store",
-                    null,
-                    null,
-                    null,
-                    "global-topic",
-                    "global-processor",
-                    new MockProcessorSupplier<>());
-            fail("Should throw TopologyException with store name conflict");
-        } catch (final TopologyException expected) { /* ok */ }
+    public void shouldNotAllowToAddGlobalStoresWithSameName() {
+        final StoreBuilder<KeyValueStore<Object, Object>> firstGlobalBuilder =
+            new MockKeyValueStoreBuilder("testStore", false).withLoggingDisabled();
+        final StoreBuilder<KeyValueStore<Object, Object>> secondGlobalBuilder =
+            new MockKeyValueStoreBuilder("testStore", false).withLoggingDisabled();
+
+        builder.addGlobalStore(
+            firstGlobalBuilder,
+            "global-store",
+            null,
+            null,
+            null,
+            "global-topic",
+            "global-processor",
+            new MockProcessorSupplier<>()
+        );
+
+        final TopologyException exception = assertThrows(
+            TopologyException.class,
+            () -> builder.addGlobalStore(
+                secondGlobalBuilder,
+                "global-store-2",
+                null,
+                null,
+                null,
+                "global-topic",
+                "global-processor-2",
+                new MockProcessorSupplier<>()
+            )
+        );
+
+        assertThat(
+            exception.getMessage(),
+            equalTo("Invalid topology: A different GlobalStateStore has already been added with the name testStore")
+        );
     }
 
     @Test
@@ -682,7 +747,7 @@ public void shouldAssociateStateStoreNameWhenStateStoreSupplierIsInternal() {
         builder.addStateStore(storeBuilder, "processor");
         final Map<String, List<String>> stateStoreNameToSourceTopic = builder.stateStoreNameToSourceTopics();
         assertEquals(1, stateStoreNameToSourceTopic.size());
-        assertEquals(Collections.singletonList("topic"), stateStoreNameToSourceTopic.get("store"));
+        assertEquals(Collections.singletonList("topic"), stateStoreNameToSourceTopic.get("testStore"));
     }
 
     @Test
@@ -692,7 +757,7 @@ public void shouldAssociateStateStoreNameWhenStateStoreSupplierIsExternal() {
         builder.addStateStore(storeBuilder, "processor");
         final Map<String, List<String>> stateStoreNameToSourceTopic = builder.stateStoreNameToSourceTopics();
         assertEquals(1, stateStoreNameToSourceTopic.size());
-        assertEquals(Collections.singletonList("topic"), stateStoreNameToSourceTopic.get("store"));
+        assertEquals(Collections.singletonList("topic"), stateStoreNameToSourceTopic.get("testStore"));
     }
 
     @Test
@@ -704,7 +769,7 @@ public void shouldCorrectlyMapStateStoreToInternalTopics() {
         builder.addStateStore(storeBuilder, "processor");
         final Map<String, List<String>> stateStoreNameToSourceTopic = builder.stateStoreNameToSourceTopics();
         assertEquals(1, stateStoreNameToSourceTopic.size());
-        assertEquals(Collections.singletonList("appId-internal-topic"), stateStoreNameToSourceTopic.get("store"));
+        assertEquals(Collections.singletonList("appId-internal-topic"), stateStoreNameToSourceTopic.get("testStore"));
     }
 
     @Test
@@ -754,11 +819,11 @@ public void shouldAddInternalTopicConfigForNonWindowStores() {
         builder.buildTopology();
         final Map<Integer, InternalTopologyBuilder.TopicsInfo> topicGroups = builder.topicGroups();
         final InternalTopologyBuilder.TopicsInfo topicsInfo = topicGroups.values().iterator().next();
-        final InternalTopicConfig topicConfig = topicsInfo.stateChangelogTopics.get("appId-store-changelog");
+        final InternalTopicConfig topicConfig = topicsInfo.stateChangelogTopics.get("appId-testStore-changelog");
         final Map<String, String> properties = topicConfig.getProperties(Collections.emptyMap(), 10000);
         assertEquals(2, properties.size());
         assertEquals(TopicConfig.CLEANUP_POLICY_COMPACT, properties.get(TopicConfig.CLEANUP_POLICY_CONFIG));
-        assertEquals("appId-store-changelog", topicConfig.name());
+        assertEquals("appId-testStore-changelog", topicConfig.name());
         assertTrue(topicConfig instanceof UnwindowedChangelogTopicConfig);
     }