[SPARK-22956][SS] Bug fix for 2 streams union failover scenario

[xuanyuanking]

What changes were proposed in this pull request?

This problem reported by @yanlin-Lynn @ivoson and @LiangchangZ. Thanks!

When we union 2 streams from kafka or other sources, while one of them have no continues data coming and in the same time task restart, this will cause an IllegalStateException. This mainly cause because the code in MicroBatchExecution , while one stream has no continues data, its comittedOffset same with availableOffset during populateStartOffsets, and currentPartitionOffsets not properly handled in KafkaSource. Also, maybe we should also consider this scenario in other Source.

How was this patch tested?

Add a UT in KafkaSourceSuite.scala

[SparkQA]

Test build #85675 has finished for PR 20150 at commit aa3d7b7.

• This patch fails to build.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #85678 has finished for PR 20150 at commit fa64187.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[xuanyuanking]

cc @zsxwing

[xuanyuanking]

cc @gatorsmile @cloud-fan

[zsxwing]

@xuanyuanking could you post the full stack trace about this issue?

[xuanyuanking]

Hi Shixiong, thanks a lot for your reply.
The full stack below can reproduce by running the added UT based on original code base.

Assert on query failed: : Query [id = 3421db21-652e-47af-9d54-2b74a222abed, runId = cd8d7c94-1286-44a5-b000-a8d870aef6fa] terminated with exception: Partition topic-0-0's offset was changed from 10 to 5, some data may have been missed. 
Some data may have been lost because they are not available in Kafka any more; either the
 data was aged out by Kafka or the topic may have been deleted before all the data in the
 topic was processed. If you don't want your streaming query to fail on such cases, set the
 source option "failOnDataLoss" to "false".
    
org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:295)
	org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:189)

	Caused by: 	Partition topic-0-0's offset was changed from 10 to 5, some data may have been missed. 
Some data may have been lost because they are not available in Kafka any more; either the
 data was aged out by Kafka or the topic may have been deleted before all the data in the
 topic was processed. If you don't want your streaming query to fail on such cases, set the
 source option "failOnDataLoss" to "false".
    
	org.apache.spark.sql.kafka010.KafkaSource.org$apache$spark$sql$kafka010$KafkaSource$$reportDataLoss(KafkaSource.scala:332)
		org.apache.spark.sql.kafka010.KafkaSource$$anonfun$8.apply(KafkaSource.scala:291)
		org.apache.spark.sql.kafka010.KafkaSource$$anonfun$8.apply(KafkaSource.scala:289)
		scala.collection.TraversableLike$$anonfun$filterImpl$1.apply(TraversableLike.scala:248)
		scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
		scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
		scala.collection.TraversableLike$class.filterImpl(TraversableLike.scala:247)
		scala.collection.TraversableLike$class.filter(TraversableLike.scala:259)
		scala.collection.AbstractTraversable.filter(Traversable.scala:104)
		org.apache.spark.sql.kafka010.KafkaSource.getBatch(KafkaSource.scala:289)

[zsxwing]

Thanks for fixing this. Looks good to me. Just some nits.

[zsxwing]

nit: could you add an assert(sliceStart <= sliceEnd, s"sliceStart: $sliceStart sliceEnd: $sliceEnd") above batches.slice(sliceStart, sliceEnd) to make sure getBatch will not be called with wrong offsets.

[xuanyuanking]

DONE

[zsxwing]

nit: test("SPARK-22956: currentPartitionOffsets should be set when no new data comes in")

[xuanyuanking]

DONE

[zsxwing]

You can clean these codes a bit using the following snippet:

    testStream(kafka)(
      StartStream(ProcessingTime(100), clock),
      waitUntilBatchProcessed,
      // 5 from smaller topic, 5 from bigger one
      CheckLastBatch((0 to 4) ++ (100 to 104): _*),
      AdvanceManualClock(100),
      waitUntilBatchProcessed,
      // 5 from smaller topic, 5 from bigger one
      CheckLastBatch((5 to 9) ++ (105 to 109): _*),
      AdvanceManualClock(100),
      waitUntilBatchProcessed,
      // smaller topic empty, 5 from bigger one
      CheckLastBatch(110 to 114: _*),
      StopStream,
      StartStream(ProcessingTime(100), clock),
      waitUntilBatchProcessed,
      // smallest now empty, 5 from bigger one
      CheckLastBatch(115 to 119: _*),
      AdvanceManualClock(100),
      waitUntilBatchProcessed,
      // smallest now empty, 5 from bigger one
      CheckLastBatch(120 to 124: _*)
    )

[xuanyuanking]

Cool, this made the code more cleaner.

[SparkQA]

Test build #86128 has finished for PR 20150 at commit bf8af29.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[zsxwing]

Thanks! Merging to master and 2.3.

[xuanyuanking]

Thanks for your review! Shixiong