### Question:
StreamUtils { public static <K, V> KStream<K, V> streamFromCollection( StreamsBuilder builder, Properties props, Serde<K> keySerde, Serde<V> valueSerde, Collection<KeyValue<K, V>> values) { return streamFromCollection(builder, props, "temp-" + UUID.randomUUID(), 50, (short) 1, keySerde, valueSerde, values); }  static KStream<K, V> streamFromCollection(
        StreamsBuilder builder,
        Properties props,
        Serde<K> keySerde,
        Serde<V> valueSerde,
        Collection<KeyValue<K, V>> values); static KStream<K, V> streamFromCollection(
        StreamsBuilder builder,
        Properties props,
        String topic,
        int numPartitions,
        short replicationFactor,
        Serde<K> keySerde,
        Serde<V> valueSerde,
        Collection<KeyValue<K, V>> values); static KTable<K, V> tableFromCollection(
        StreamsBuilder builder,
        Properties props,
        Serde<K> keySerde,
        Serde<V> valueSerde,
        Collection<KeyValue<K, V>> values); static KTable<K, V> tableFromCollection(
        StreamsBuilder builder,
        Properties props,
        String topic,
        int numPartitions,
        short replicationFactor,
        Serde<K> keySerde,
        Serde<V> valueSerde,
        Collection<KeyValue<K, V>> values); static List<KeyValue<K, V>> listFromTable(KafkaStreams streams, KTable<K, V> table); static List<KeyValue<K, V>> listFromStore(KafkaStreams streams, String storeName); static NavigableMap<K, V> mapFromTable(KafkaStreams streams, KTable<K, V> table); static NavigableMap<K, V> mapFromStore(KafkaStreams streams, String storeName);  }### Answer:
@Test public void testCollectionToStream() throws Exception { Collection<KeyValue<Integer, Integer>> input = new ArrayList<>(); for (Integer i : LEFT_INPUT) { input.add(new KeyValue<>(i, i)); } StreamsBuilder builder = new StreamsBuilder(); KStream<Integer, Integer> stream = StreamUtils.streamFromCollection( builder, PRODUCER_CONFIG, LEFT_INPUT_TOPIC, 50, (short) 1, Serdes.Integer(), Serdes.Integer(), input); stream.to(OUTPUT_TOPIC); startStreams(builder, Serdes.Integer(), Serdes.Integer()); Thread.sleep(1000); List<KeyValue<Integer, Integer>> records = consumeData( OUTPUT_TOPIC, IntegerDeserializer.class, IntegerDeserializer.class, 26, 10000L); for (KeyValue<Integer, Integer> record : records) { assertEquals(record.key, record.value); } streams.close(); }