diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
index 404193a815d..1a2297cd6bf 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
@@ -178,7 +178,7 @@ public class KStreamImpl<K, V> implements KStream<K, V> {
 
         topology.addSource(sourceName, keyDeserializer, valDeserializer, topic);
 
-        return new KStreamImpl<>(topology, sourceName, Collections.<String>emptySet());
+        return new KStreamImpl<>(topology, sourceName, Collections.singleton(sourceName));
     }
 
     @Override