title: How to push datastream to kafka topic by retaining the order using string method in Flink Kafka Problem
body: <p>I am trying to create a <code>JSON</code> dataset every <code>500 ms</code> and want to push it to the <code>Kafka</code> topic so that I can set up some windows in the downstream and perform computations. Below is my code:</p>
  <pre><code>package KafkaAsSource
  
  import org.apache.flink.api.common.serialization.SimpleStringSchema
  import org.apache.flink.streaming.api.datastream.DataStream
  import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
  import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.Semantic
  import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaProducer}
  import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper
  
  
  import java.time.format.DateTimeFormatter
  import java.time.LocalDateTime
  import java.util.{Optional, Properties}
  
  object PushingDataToKafka {
  
    def main(args: Array[String]): Unit = {
      val env = StreamExecutionEnvironment.getExecutionEnvironment
      env.setMaxParallelism(256)
      env.enableCheckpointing(5000)
      val stream: DataStream[String] = env.fromElements(createData())
  
      stream.addSink(sendToTopic(stream))
    }
  
    def getProperties(): Properties = {
      val properties = new Properties()
      properties.setProperty(&quot;bootstrap.servers&quot;, &quot;localhost:9092&quot;)
      properties.setProperty(&quot;zookeeper.connect&quot;, &quot;localhost:2181&quot;)
  
      return properties
    }
  
    def createData(): String = {
      val minRange: Int = 0
      val maxRange: Int = 1000
      var jsonData = &quot;&quot;
      for (a &lt;- minRange to maxRange) {
        jsonData = &quot;{\n  \&quot;id\&quot;:\&quot;&quot; + a + &quot;\&quot;,\n  \&quot;Category\&quot;:\&quot;Flink\&quot;,\n  \&quot;eventTime\&quot;:\&quot;&quot; + DateTimeFormatter.ofPattern(&quot;yyyy-MM-dd HH:mm:ss.SSS&quot;).format(LocalDateTime.now) + &quot;\&quot;\n  \n}&quot;
        println(jsonData)
        Thread.sleep(500)
      }
      return jsonData
    }
  
    def sendToTopic(): Properties = {
      val producer = new FlinkKafkaProducer[String](
        &quot;topic&quot;
        ,
        new KeyedSerializationSchemaWrapper[String](new SimpleStringSchema())
        ,
        getProperties(),
        FlinkKafkaProducer.Semantic.EXACTLY_ONCE
      )
      return producer
    }
  }
  </code></pre>
  <p>It gives me below error:</p>
  <pre><code>type mismatch;
   found   : Any
   required: org.apache.flink.streaming.api.functions.sink.SinkFunction[String]
      stream.addSink(sendToTopic())
  </code></pre>
  <p><strong>Modified Code:</strong></p>
  <pre><code>object FlinkTest {
  
    def main(ars: Array[String]): Unit = {
      val env = StreamExecutionEnvironment.getExecutionEnvironment()
      env.setMaxParallelism(256)
      var stream = env.fromElements(&quot;&quot;)
      //env.enableCheckpointing(5000)
      //val stream: DataStream[String] = env.fromElements(&quot;hey mc&quot;, &quot;1&quot;)
  
      val myProducer = new FlinkKafkaProducer[String](
        &quot;maddy&quot;, // target topic
        new KeyedSerializationSchemaWrapper[String](new SimpleStringSchema()), // serialization schema
        getProperties(), // producer config
        FlinkKafkaProducer.Semantic.EXACTLY_ONCE)
      val minRange: Int = 0
      val maxRange: Int = 10
      var jsonData = &quot;&quot;
      for (a &lt;- minRange to maxRange) {
        jsonData = &quot;{\n  \&quot;id\&quot;:\&quot;&quot; + a + &quot;\&quot;,\n  \&quot;Category\&quot;:\&quot;Flink\&quot;,\n  \&quot;eventTime\&quot;:\&quot;&quot; + DateTimeFormatter.ofPattern(&quot;yyyy-MM-dd HH:mm:ss.SSS&quot;).format(LocalDateTime.now) + &quot;\&quot;\n  \n}&quot;
        println(a)
        Thread.sleep(500)
        stream = env.fromElements(jsonData)
        println(jsonData)
        stream.addSink(myProducer)
      }
  
      env.execute(&quot;hey&quot;)
    }
  
    def getProperties(): Properties = {
      val properties = new Properties()
      properties.setProperty(&quot;bootstrap.servers&quot;, &quot;localhost:9092&quot;)
      properties.setProperty(&quot;zookeeper.connect&quot;, &quot;localhost:2181&quot;)
      return properties
    }
    /*
    def createData(): String = {
      val minRange: Int = 0
      val maxRange: Int = 10
      var jsonData = &quot;&quot;
      for (a &lt;- minRange to maxRange) {
        jsonData = &quot;{\n  \&quot;id\&quot;:\&quot;&quot; + a + &quot;\&quot;,\n  \&quot;Category\&quot;:\&quot;Flink\&quot;,\n  \&quot;eventTime\&quot;:\&quot;&quot; + DateTimeFormatter.ofPattern(&quot;yyyy-MM-dd HH:mm:ss.SSS&quot;).format(LocalDateTime.now) + &quot;\&quot;\n  \n}&quot;
        Thread.sleep(500)
      }
      return jsonData
    }
    */
  
  }
  </code></pre>
  <p><code>Modified Code</code> gives me the data in the Kafka topic but it doesn't retain the order. What am I doing wrong here in the loops? Also, had to change the version of Flink to <code>1.12.2</code> from <code>1.13.5</code>.</p>
  <p>I was initially using <code>Flink</code> <code>1.13.5</code>, <code>Connectors</code> and <code>Scala</code> of <code>2.11</code>. What exactly I am missing over here?</p>