title: Structured Streaming: Reading from multiple Kafka topics at once
body: <p>I have a Spark Structured Streaming Application which has to read from 12 Kafka topics (Different Schemas, Avro format) at once, deserialize the data and store in HDFS. When I read from a single topic using my code, it works fine and without errors but on running multiple queries together, I'm getting the following error</p>
  
  <pre><code>java.lang.IllegalStateException: Race while writing batch 0
  </code></pre>
  
  <p>My code is as follows:</p>
  
  <pre><code>def main(args: Array[String]): Unit = {
  
  
    val kafkaProps = Util.loadProperties(kafkaConfigFile).asScala
    val topic_list = ("topic1", "topic2", "topic3", "topic4")
  
    topic_list.foreach(x =&gt; {
  kafkaProps.update("subscribe", x)
  
  val source= Source.fromInputStream(Util.getInputStream("/schema/topics/" + x)).getLines.mkString
  val schemaParser = new Schema.Parser
  val schema = schemaParser.parse(source)
  val sqlTypeSchema = SchemaConverters.toSqlType(schema).dataType.asInstanceOf[StructType]
  
  val kafkaStreamData = spark
    .readStream
    .format("kafka")
    .options(kafkaProps)
    .load()
  
  val udfDeserialize = udf(deserialize(source), DataTypes.createStructType(sqlTypeSchema.fields))
  
  val transformedDeserializedData = kafkaStreamData.select("value").as(Encoders.BINARY)
    .withColumn("rows", udfDeserialize(col("value")))
    .select("rows.*")
  
  val query = transformedDeserializedData
    .writeStream
    .trigger(Trigger.ProcessingTime("5 seconds"))
    .outputMode("append")
    .format("parquet")
    .option("path", "/output/topics/" + x)
    .option("checkpointLocation", checkpointLocation + "//" + x)
    .start()  
  })  
  spark.streams.awaitAnyTermination()  
   }
  </code></pre>