public synchronized <K, V> KStream<K, V> stream(final Collection<String> topics,
                                                    final Consumed<K, V> consumed) {
        Objects.requireNonNull(topics, "topics can't be null");
        Objects.requireNonNull(consumed, "consumed can't be null");
        return internalStreamsBuilder.stream(topics, new ConsumedInternal<>(consumed));
    }
Create a {@link KStream} from the specified topics.
The {@code "auto.offset.reset"} strategy, {@link TimestampExtractor}, key and value deserializers
are defined by the options in {@link Consumed} are used.
<p>
If multiple topics are specified there is no ordering guarantee for records from different topics.
<p>
Note that the specified input topics must be partitioned by key.
If this is not the case it is the user's responsibility to repartition the data before any key based operation
(like aggregation or join) is applied to the returned {@link KStream}.

@param topics the topic names; must contain at least one topic name
@param consumed      the instance of {@link Consumed} used to define optional parameters
@return a {@link KStream} for the specified topics
streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java