title: Spark Structured Streaming + Kafka Integration: MicroBatchExecution PartitionOffsets Error
body: <p>I am using Spark Structured Streaming to process the incoming and outgoing data streams from and to Apache Kafka respectively, using the scala code below. </p>
  
  <p>I can successfully read data stream using kafka source, however while trying to write stream to Kafka sink I am getting following error:</p>
  
  <pre><code>ERROR MicroBatchExecution:91 - Query [id = 234750ca-d416-4182-b3cc-4e2c1f922724, runId = 4c4b0931-9876-456f-8d56-752623803332] terminated with error java.lang.IllegalArgumentException: Expected e.g. {"topicA":{"0":23,"1":-1},"topicB":{"0":-2}}, got 1 {"path":"file:///path/to/file.csv","timestamp":1536564701000,"batchId":0}
  at org.apache.spark.sql.kafka010.JsonUtils$.partitionOffsets(JsonUtils.scala:74)
      at org.apache.spark.sql.kafka010.KafkaSourceOffset$.apply(KafkaSourceOffset.scala:64)
      at org.apache.spark.sql.kafka010.KafkaSource$$anon$1.deserialize(KafkaSource.scala:124)
      at org.apache.spark.sql.kafka010.KafkaSource$$anon$1.deserialize(KafkaSource.scala:99)
      at org.apache.spark.sql.execution.streaming.HDFSMetadataLog.get(HDFSMetadataLog.scala:198)
      at org.apache.spark.sql.kafka010.KafkaSource.initialPartitionOffsets$lzycompute(KafkaSource.scala:129)
      at org.apache.spark.sql.kafka010.KafkaSource.initialPartitionOffsets(KafkaSource.scala:97)
      at org.apache.spark.sql.kafka010.KafkaSource.getBatch(KafkaSource.scala:207)
      at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$populateStartOffsets$2.apply(MicroBatchExecution.scala:216)
      at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$populateStartOffsets$2.apply(MicroBatchExecution.scala:213)
      at scala.collection.Iterator$class.foreach(Iterator.scala:893)
      at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
      at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
      at org.apache.spark.sql.execution.streaming.StreamProgress.foreach(StreamProgress.scala:25)
      at org.apache.spark.sql.execution.streaming.MicroBatchExecution.org$apache$spark$sql$execution$streaming$MicroBatchExecution$$populateStartOffsets(MicroBatchExecution.scala:213)
      at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply$mcV$sp(MicroBatchExecution.scala:124)
      at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:121)
      at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:121)
      at org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:271)
      at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:58)
      at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1.apply$mcZ$sp(MicroBatchExecution.scala:121)
      at org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:56)
      at org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:117)
      at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:279)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:189)
  Exception in thread "main" org.apache.spark.sql.streaming.StreamingQueryException: Expected e.g. {"topicA":{"0":23,"1":-1},"topicB":{"0":-2}}, got 1
  {"path":""file:///path/to/file.csv"","timestamp":1536564701000,"batchId":0}
  === Streaming Query ===
  Identifier: [id = 234750ca-d416-4182-b3cc-4e2c1f922724, runId = 851d0cd7-aabe-45c8-8a14-94227f90e174]
  Current Committed Offsets: {KafkaSource[Subscribe[t]]: {"logOffset":2}}
  Current Available Offsets: {KafkaSource[Subscribe[t]]: {"logOffset":3}}
  </code></pre>
  
  <p>Scala code:</p>
  
  <pre><code>object spark_kafka_attempt2 {
  
    def main(args: Array[String]) {
  
      val spark = SparkSession
        .builder
        .appName("spark_kafka_test")
        .getOrCreate()
  
      import spark.implicits._
  
      val input_lines = spark
        .readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094")
        .option("subscribe", "input_stream")
        .option("startingOffsets", "earliest")
        .load()
  
      val inputStreamSchema = new StructType()
        .add("input_id", "long")
        .add("timestamp", "timestamp")
        .add("type", "string")
  
      val lines = input_lines.selectExpr("CAST(value AS STRING)", "CAST(timestamp AS TIMESTAMP)").as[(String, Timestamp)]
        .select(from_json($"value", inputStreamSchema).as("data"), $"timestamp".as("arrival_timestamp"))
        .select("data.*", "arrival_timestamp")
  
  
      val query = lines
        .selectExpr("CAST(input_id AS STRING) AS key", "to_json(struct(*)) AS value")
        .writeStream
        .format("kafka")
        .outputMode("update")
        .option("kafka.bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094")
        .option("topic", "processed_stream")
        .option("checkpointLocation", "/home/local/directory")
        .start()
  
      query.awaitTermination()
    }
  }
  </code></pre>
  
  <p>The code works fine when the output is sent to console, while the error appears while trying to send processed stream to Apache Kafka.</p>
  
  <p>I am using Apache Structured Streaming 2.3.1, Scala 2.11.8 and Apache Kafka 2.0. </p>
  
  <p>Build.sbt file is as follows:</p>
  
  <pre><code>name := "spark_kafka_test"    
  version := "0.1"    
  scalaVersion := "2.11.8"    
  val sparkVersion = "2.3.1"    
  libraryDependencies ++= Seq(
    "org.apache.spark" %% "spark-sql" % sparkVersion,
    "org.apache.spark" %% "spark-sql-kafka-0-10" % sparkVersion
  ) 
  </code></pre>
  
  <p>I am submitting my job to spark as follows:</p>
  
  <pre><code>./spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.1 --class spark_kafka_test --master local[4] /home/salman/Development/spark_kafka_attempt2/target/scala-2.11/spark_kafka_test_2.11-0.1.jar 
  </code></pre>