title: How to send a KeyValue list to Kafka?
body: <p>I am trying to send a List[KeyValue] to the topic, in a Kafka Streams app. But the stream expects a single KeyValue. How I can send the KeyValues to the stream, instead of the whole list?</p>
  
  <pre><code>class MainTopology {
  
    val createTopology = (sourceTopic: String, sinkTopic: String) =&gt; {
      val builder = new StreamsBuilder()
      builder.stream(sourceTopic)
        .map[String, Option[JsonMessage]]((key, value) =&gt; toJsonEvent(key, value))
        .map[String, String]((key, value) =&gt; toFormattedEvents(key, value))
        .to(sinkTopic)
      builder.build()
    }
  
    private val toJsonEvent = (key: String, value: String) =&gt; {
      println(value)
      val jsonEventsAsCaseClasses = jsonToClass(value)
      new KeyValue(key, jsonEventsAsCaseClasses)
    }
  
    private val toFormattedEvents = (key: String, value: Option[JsonMessage]) =&gt; {
      val jsonEvents: List[String] = formatEvents(value)
      jsonEvents.map(event =&gt; new KeyValue(key,event))
    }
  
  }
  </code></pre>
  
  <p>The second map is not compiling due to this.</p>
  
  <p><code>Expression of type List[KeyValue[String, String]] doesn't conform to expected type KeyValue[_ &lt;: String, _ &lt;: String]</code></p>
  
  <p>I update my code, but now it is throwing another error:</p>
  
  <pre><code>    val builder = new StreamsBuilder()
      builder.stream(sourceTopic)
        .map[String, Option[JsonMessage]]((key, value) =&gt; toJsonEvent(key, value))
        .flatMap(
        (key, value) =&gt; toFormattedEvents(key, value)
      )
        .to(sinkTopic)
      builder.build()
    }
  
    private val toJsonEvent = (key: String, value: String) =&gt; {
      println(value)
      val jsonEventsAsCaseClasses = jsonToClass(value)
      new KeyValue(key, jsonEventsAsCaseClasses)
    }
  
    private val toFormattedEvents = (key: String, value: Option[JsonMessage]) =&gt; {
      val jsonEvents: List[String] = formatEvents(value)
      jsonEvents.map(event =&gt; new KeyValue(key,event)).toIterable.asJava
    }
  </code></pre>
  
  <pre><code>Error:(15, 8) inferred type arguments [?1,?0] do not conform to method flatMap's type parameter bounds [KR,VR]
        .flatMap(
  Error:(16, 20) type mismatch;
   found   : org.apache.kafka.streams.kstream.KeyValueMapper[String,Option[org.minsait.streams.model.JsonMessage],Iterable[org.apache.kafka.streams.KeyValue[String,String]]]
   required: org.apache.kafka.streams.kstream.KeyValueMapper[_ &gt;: String, _ &gt;: Option[org.minsait.streams.model.JsonMessage], _ &lt;: Iterable[_ &lt;: org.apache.kafka.streams.KeyValue[_ &lt;: KR, _ &lt;: VR]]]
        (key, value) =&gt; toFormattedEvents(key, value)
  </code></pre>