Error adding data: The code passed to eventually never returned normally. Attempted 439 times over 10.014200112000001 seconds. Last failure message: assertion failed: stress6 not deleted after timeout.
org.scalatest.concurrent.Eventually$class.tryTryAgain$1(Eventually.scala:420)
 org.scalatest.concurrent.Eventually$class.eventually(Eventually.scala:438)
 org.scalatest.concurrent.Eventually$.eventually(Eventually.scala:478)
 org.scalatest.concurrent.Eventually$class.eventually(Eventually.scala:336)
 org.scalatest.concurrent.Eventually$.eventually(Eventually.scala:478)
 org.apache.spark.sql.kafka010.KafkaTestUtils.verifyTopicDeletion(KafkaTestUtils.scala:307)
 org.apache.spark.sql.kafka010.KafkaTestUtils.deleteTopic(KafkaTestUtils.scala:179)
 org.apache.spark.sql.kafka010.KafkaSourceStressSuite$$anonfun$17$$anonfun$apply$mcV$sp$7$$anonfun$34.apply(KafkaSourceSuite.scala:614)
 org.apache.spark.sql.kafka010.KafkaSourceStressSuite$$anonfun$17$$anonfun$apply$mcV$sp$7$$anonfun$34.apply(KafkaSourceSuite.scala:611)
 org.apache.spark.sql.kafka010.KafkaSourceTest$AddKafkaData$$anonfun$addData$2.apply(KafkaSourceSuite.scala:85)

 Caused by: assertion failed: stress6 not deleted after timeout
 scala.Predef$.assert(Predef.scala:170)
 org.apache.spark.sql.kafka010.KafkaTestUtils$$anonfun$verifyTopicDeletion$1.apply$mcV$sp(KafkaTestUtils.scala:308)
 org.apache.spark.sql.kafka010.KafkaTestUtils$$anonfun$verifyTopicDeletion$1.apply(KafkaTestUtils.scala:308)
 org.apache.spark.sql.kafka010.KafkaTestUtils$$anonfun$verifyTopicDeletion$1.apply(KafkaTestUtils.scala:308)
 org.scalatest.concurrent.Eventually$class.makeAValiantAttempt$1(Eventually.scala:394)
 org.scalatest.concurrent.Eventually$class.tryTryAgain$1(Eventually.scala:408)
 org.scalatest.concurrent.Eventually$class.eventually(Eventually.scala:438)
 org.scalatest.concurrent.Eventually$.eventually(Eventually.scala:478)
 org.scalatest.concurrent.Eventually$class.eventually(Eventually.scala:336)
 org.scalatest.concurrent.Eventually$.eventually(Eventually.scala:478)


== Progress ==
 AssertOnQuery(<condition>, )
 AddKafkaData(topics = Set(stress4, stress2, stress1, stress5, stress3), data = Range(0, 1), message = Add partitiosn)
 CheckAnswer: [1],[2]
 CheckAnswer: [1],[2]
 StopStream
 AddKafkaData(topics = Set(stress4, stress2, stress1, stress5, stress3), data = Range(), message = Add partitiosn)
 StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@dbed7fd,Map())
 AddKafkaData(topics = Set(stress4, stress2, stress1, stress5, stress3), data = Range(2, 3), message = Delete topic stress1)
 CheckAnswer: [1],[2],[3],[4]
 StopStream
 AddKafkaData(topics = Set(stress4, stress2, stress1, stress5, stress3), data = Range(4, 5, 6, 7, 8, 9, 10, 11), message = Add partitiosn)
 StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@a0c5be,Map())
 CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12]
 StopStream
 AddKafkaData(topics = Set(stress4, stress2, stress1, stress5, stress3), data = Range(12, 13, 14, 15, 16, 17, 18, 19, 20), message = )
 StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@5b02a984,Map())
 AddKafkaData(topics = Set(stress4, stress6, stress2, stress1, stress5, stress3), data = Range(21), message = Add topic stress7)
 CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22]
 StopStream
 AddKafkaData(topics = Set(stress4, stress6, stress2, stress1, stress5, stress3), data = Range(22, 23), message = )
 AddKafkaData(topics = Set(stress4, stress6, stress2, stress1, stress5, stress3), data = Range(24, 25), message = )
 StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@4ffced4e,Map())
 CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26]
 StopStream
 AddKafkaData(topics = Set(stress4, stress6, stress2, stress1, stress5, stress3), data = Range(26, 27, 28, 29, 30, 31, 32, 33), message = )
 StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@4eeab3e,Map())
 AddKafkaData(topics = Set(stress4, stress6, stress2, stress1, stress5, stress3), data = Range(), message = Add partitiosn)
 CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30],[31],[32],[33],[34]
 CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30],[31],[32],[33],[34]
 StopStream
 StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@46a8c2b4,Map())
 AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress5, stress3), data = Range(34, 35, 36, 37), message = Add topic stress9)
 CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30],[31],[32],[33],[34],[35],[36],[37],[38]
 StopStream
 AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress5, stress3), data = Range(38, 39, 40, 41, 42, 43, 44, 45), message = Add partitiosn)
 AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress5, stress3), data = Range(46, 47, 48, 49, 50, 51, 52, 53), message = )
 AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress5, stress3), data = Range(54, 55, 56, 57, 58, 59, 60), message = Add partitiosn)
 StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@404ced67,Map())
 CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30],[31],[32],[33],[34],[35],[36],[37],[38],[39],[40],[41],[42],[43],[44],[45],[46],[47],[48],[49],[50],[51],[52],[53],[54],[55],[56],[57],[58],[59],[60],[61]
 StopStream
 AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(61, 62, 63, 64, 65, 66, 67, 68), message = Add topic stress11)
 AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(69, 70, 71, 72, 73, 74, 75, 76, 77), message = )
 AddKafkaData(topics = Set(stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(), message = Add topic stress13)
 StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@2d237460,Map())
 CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30],[31],[32],[33],[34],[35],[36],[37],[38],[39],[40],[41],[42],[43],[44],[45],[46],[47],[48],[49],[50],[51],[52],[53],[54],[55],[56],[57],[58],[59],[60],[61],[62],[63],[64],[65],[66],[67],[68],[69],[70],[71],[72],[73],[74],[75],[76],[77],[78]
 StopStream
 StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@11e71181,Map())
 CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30],[31],[32],[33],[34],[35],[36],[37],[38],[39],[40],[41],[42],[43],[44],[45],[46],[47],[48],[49],[50],[51],[52],[53],[54],[55],[56],[57],[58],[59],[60],[61],[62],[63],[64],[65],[66],[67],[68],[69],[70],[71],[72],[73],[74],[75],[76],[77],[78]
 StopStream
 AddKafkaData(topics = Set(stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(78, 79), message = )
 AddKafkaData(topics = Set(stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(80, 81), message = )
 AddKafkaData(topics = Set(stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(82, 83, 84), message = )
 AddKafkaData(topics = Set(stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(85, 86, 87), message = )
 AddKafkaData(topics = Set(stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(88, 89, 90), message = )
 StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@65f3e805,Map())
 AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(91, 92, 93, 94, 95, 96), message = Add topic stress15)
 AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3), data = Range(97, 98, 99, 100), message = Delete topic stress10)
 AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress8, stress1, stress5, stress3), data = Range(101), message = Delete topic stress2)
 AddKafkaData(topics = Set(stress4, stress6, stress12, stress8, stress1, stress5, stress3), data = Range(102), message = Delete topic stress14)
=> AddKafkaData(topics = Set(stress4, stress12, stress8, stress1, stress5, stress3), data = Range(), message = Delete topic stress6)
 AddKafkaData(topics = Set(stress4, stress12, stress8, stress1, stress5, stress3, stress16), data = Range(103, 104, 105, 106, 107, 108, 109, 110), message = Add topic stress17)
 CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30],[31],[32],[33],[34],[35],[36],[37],[38],[39],[40],[41],[42],[43],[44],[45],[46],[47],[48],[49],[50],[51],[52],[53],[54],[55],[56],[57],[58],[59],[60],[61],[62],[63],[64],[65],[66],[67],[68],[69],[70],[71],[72],[73],[74],[75],[76],[77],[78],[79],[80],[81],[82],[83],[84],[85],[86],[87],[88],[89],[90],[91],[92],[93],[94],[95],[96],[97],[98],[99],[100],[101],[102],[103],[104],[105],[106],[107],[108],[109],[110],[111]

== Stream ==
Output Mode: Append
Stream state: {KafkaSource[SubscribePattern[stress.*]]: {"stress5":{"17":0,"8":0,"11":0,"2":2,"5":0,"14":0,"4":1,"13":0,"7":1,"16":1,"1":2,"10":1,"9":1,"12":1,"3":2,"15":2,"6":1,"0":1},"stress8":{"2":1,"5":0,"4":1,"7":1,"1":1,"3":1,"6":1,"0":1},"stress1":{"23":0,"8":2,"17":0,"11":2,"2":3,"20":0,"5":2,"14":0,"4":2,"13":2,"22":0,"16":0,"7":2,"10":0,"1":4,"19":1,"18":0,"9":1,"3":2,"12":1,"21":0,"15":0,"6":1,"24":0,"0":2},"stress4":{"23":0,"17":1,"8":1,"26":0,"11":1,"20":0,"2":3,"5":0,"14":0,"4":0,"13":0,"22":0,"16":0,"7":0,"25":0,"10":0,"1":2,"19":0,"28":0,"9":0,"27":0,"18":0,"3":0,"12":0,"21":0,"15":0,"6":0,"24":0,"0":1},"stress3":{"8":1,"17":0,"11":0,"20":0,"2":0,"5":0,"14":0,"13":0,"4":0,"16":0,"7":1,"1":5,"10":2,"19":1,"9":2,"18":1,"3":2,"21":0,"12":3,"6":2,"15":2,"0":4},"stress12":{"2":0,"1":0,"0":0}}}
Thread state: alive


== Sink ==
0: 
1: 
2: [1] [2]
3: 
4: [3]
5: [4]
6: [5] [6] [7] [11] [12] [9] [10] [8]
7: [17] [14] [18] [13] [19] [15] [21] [20] [16]
8: [22]
9: [23] [25] [26] [24]
10: [32] [27] [28] [33] [29] [31] [34] [30]
11: 
12: 
13: 
14: [37] [36] [35]
15: [38]
16: [45] [59] [40] [42] [47] [55] [48] [56] [49] [54] [51] [41] [50] [53] [57] [43] [61] [60] [46] [39] [58] [52] [44]
17: [62] [71] [76] [63] [72] [75] [64] [68] [66] [73] [69] [70] [74] [78] [65] [67] [77]
18: [81] [90] [80] [79] [87] [91] [82] [83] [85] [89] [84] [86] [88]
19: [92] [93]
20: [96] [94] [95] [97]
21: 
22: [98]
23: [100] [101] [99]
24: 
25: [102]
26: 
27: [103]
28: 


== Plan ==
== Parsed Logical Plan ==
SerializeFromObject [input[0, int, true] AS value#26]
+- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#25: int
 +- DeserializeToObject newInstance(class scala.Tuple2), obj#24: scala.Tuple2
 +- Project [cast(key#1045 as string) AS key#15, cast(value#1046 as string) AS value#16]
 +- LogicalRDD [key#1045, value#1046, topic#1047, partition#1048, offset#1049L, timestamp#1050L, timestampType#1051]

== Analyzed Logical Plan ==
value: int
SerializeFromObject [input[0, int, true] AS value#26]
+- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#25: int
 +- DeserializeToObject newInstance(class scala.Tuple2), obj#24: scala.Tuple2
 +- Project [cast(key#1045 as string) AS key#15, cast(value#1046 as string) AS value#16]
 +- LogicalRDD [key#1045, value#1046, topic#1047, partition#1048, offset#1049L, timestamp#1050L, timestampType#1051]

== Optimized Logical Plan ==
SerializeFromObject [input[0, int, true] AS value#26]
+- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#25: int
 +- DeserializeToObject newInstance(class scala.Tuple2), obj#24: scala.Tuple2
 +- Project [cast(key#1045 as string) AS key#15, cast(value#1046 as string) AS value#16]
 +- LogicalRDD [key#1045, value#1046, topic#1047, partition#1048, offset#1049L, timestamp#1050L, timestampType#1051]

== Physical Plan ==
*SerializeFromObject [input[0, int, true] AS value#26]
+- *MapElements <function1>, obj#25: int
 +- *DeserializeToObject newInstance(class scala.Tuple2), obj#24: scala.Tuple2
 +- *Project [cast(key#1045 as string) AS key#15, cast(value#1046 as string) AS value#16]
 +- Scan ExistingRDD[key#1045,value#1046,topic#1047,partition#1048,offset#1049L,timestamp#1050L,timestampType#1051]
 

org.scalatest.exceptions.TestFailedException:
Error adding data: The code passed to eventually never returned normally. Attempted 439 times over 10.014200112000001 seconds. Last failure message: assertion failed: stress6 not deleted after timeout.
org.scalatest.concurrent.Eventually$class.tryTryAgain$1(Eventually.scala:420)
org.scalatest.concurrent.Eventually$class.eventually(Eventually.scala:438)
org.scalatest.concurrent.Eventually$.eventually(Eventually.scala:478)
org.scalatest.concurrent.Eventually$class.eventually(Eventually.scala:336)
org.scalatest.concurrent.Eventually$.eventually(Eventually.scala:478)
org.apache.spark.sql.kafka010.KafkaTestUtils.verifyTopicDeletion(KafkaTestUtils.scala:307)
org.apache.spark.sql.kafka010.KafkaTestUtils.deleteTopic(KafkaTestUtils.scala:179)
org.apache.spark.sql.kafka010.KafkaSourceStressSuite$$anonfun$17$$anonfun$apply$mcV$sp$7$$anonfun$34.apply(KafkaSourceSuite.scala:614)
org.apache.spark.sql.kafka010.KafkaSourceStressSuite$$anonfun$17$$anonfun$apply$mcV$sp$7$$anonfun$34.apply(KafkaSourceSuite.scala:611)
org.apache.spark.sql.kafka010.KafkaSourceTest$AddKafkaData$$anonfun$addData$2.apply(KafkaSourceSuite.scala:85)
Caused by: assertion failed: stress6 not deleted after timeout
scala.Predef$.assert(Predef.scala:170)
org.apache.spark.sql.kafka010.KafkaTestUtils$$anonfun$verifyTopicDeletion$1.apply$mcV$sp(KafkaTestUtils.scala:308)
org.apache.spark.sql.kafka010.KafkaTestUtils$$anonfun$verifyTopicDeletion$1.apply(KafkaTestUtils.scala:308)
org.apache.spark.sql.kafka010.KafkaTestUtils$$anonfun$verifyTopicDeletion$1.apply(KafkaTestUtils.scala:308)
org.scalatest.concurrent.Eventually$class.makeAValiantAttempt$1(Eventually.scala:394)
org.scalatest.concurrent.Eventually$class.tryTryAgain$1(Eventually.scala:408)
org.scalatest.concurrent.Eventually$class.eventually(Eventually.scala:438)
org.scalatest.concurrent.Eventually$.eventually(Eventually.scala:478)
org.scalatest.concurrent.Eventually$class.eventually(Eventually.scala:336)
org.scalatest.concurrent.Eventually$.eventually(Eventually.scala:478)
== Progress ==
AssertOnQuery(<condition>, )
AddKafkaData(topics = Set(stress4, stress2, stress1, stress5, stress3), data = Range(0, 1), message = Add partitiosn)
CheckAnswer: [1],[2]
CheckAnswer: [1],[2]
StopStream
AddKafkaData(topics = Set(stress4, stress2, stress1, stress5, stress3), data = Range(), message = Add partitiosn)
StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@dbed7fd,Map())
AddKafkaData(topics = Set(stress4, stress2, stress1, stress5, stress3), data = Range(2, 3), message = Delete topic stress1)
CheckAnswer: [1],[2],[3],[4]
StopStream
AddKafkaData(topics = Set(stress4, stress2, stress1, stress5, stress3), data = Range(4, 5, 6, 7, 8, 9, 10, 11), message = Add partitiosn)
StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@a0c5be,Map())
CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12]
StopStream
AddKafkaData(topics = Set(stress4, stress2, stress1, stress5, stress3), data = Range(12, 13, 14, 15, 16, 17, 18, 19, 20), message = )
StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@5b02a984,Map())
AddKafkaData(topics = Set(stress4, stress6, stress2, stress1, stress5, stress3), data = Range(21), message = Add topic stress7)
CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22]
StopStream
AddKafkaData(topics = Set(stress4, stress6, stress2, stress1, stress5, stress3), data = Range(22, 23), message = )
AddKafkaData(topics = Set(stress4, stress6, stress2, stress1, stress5, stress3), data = Range(24, 25), message = )
StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@4ffced4e,Map())
CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26]
StopStream
AddKafkaData(topics = Set(stress4, stress6, stress2, stress1, stress5, stress3), data = Range(26, 27, 28, 29, 30, 31, 32, 33), message = )
StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@4eeab3e,Map())
AddKafkaData(topics = Set(stress4, stress6, stress2, stress1, stress5, stress3), data = Range(), message = Add partitiosn)
CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30],[31],[32],[33],[34]
CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30],[31],[32],[33],[34]
StopStream
StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@46a8c2b4,Map())
AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress5, stress3), data = Range(34, 35, 36, 37), message = Add topic stress9)
CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30],[31],[32],[33],[34],[35],[36],[37],[38]
StopStream
AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress5, stress3), data = Range(38, 39, 40, 41, 42, 43, 44, 45), message = Add partitiosn)
AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress5, stress3), data = Range(46, 47, 48, 49, 50, 51, 52, 53), message = )
AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress5, stress3), data = Range(54, 55, 56, 57, 58, 59, 60), message = Add partitiosn)
StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@404ced67,Map())
CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30],[31],[32],[33],[34],[35],[36],[37],[38],[39],[40],[41],[42],[43],[44],[45],[46],[47],[48],[49],[50],[51],[52],[53],[54],[55],[56],[57],[58],[59],[60],[61]
StopStream
AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(61, 62, 63, 64, 65, 66, 67, 68), message = Add topic stress11)
AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(69, 70, 71, 72, 73, 74, 75, 76, 77), message = )
AddKafkaData(topics = Set(stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(), message = Add topic stress13)
StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@2d237460,Map())
CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30],[31],[32],[33],[34],[35],[36],[37],[38],[39],[40],[41],[42],[43],[44],[45],[46],[47],[48],[49],[50],[51],[52],[53],[54],[55],[56],[57],[58],[59],[60],[61],[62],[63],[64],[65],[66],[67],[68],[69],[70],[71],[72],[73],[74],[75],[76],[77],[78]
StopStream
StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@11e71181,Map())
CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30],[31],[32],[33],[34],[35],[36],[37],[38],[39],[40],[41],[42],[43],[44],[45],[46],[47],[48],[49],[50],[51],[52],[53],[54],[55],[56],[57],[58],[59],[60],[61],[62],[63],[64],[65],[66],[67],[68],[69],[70],[71],[72],[73],[74],[75],[76],[77],[78]
StopStream
AddKafkaData(topics = Set(stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(78, 79), message = )
AddKafkaData(topics = Set(stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(80, 81), message = )
AddKafkaData(topics = Set(stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(82, 83, 84), message = )
AddKafkaData(topics = Set(stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(85, 86, 87), message = )
AddKafkaData(topics = Set(stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(88, 89, 90), message = )
StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@65f3e805,Map())
AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3, stress10), data = Range(91, 92, 93, 94, 95, 96), message = Add topic stress15)
AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress5, stress3), data = Range(97, 98, 99, 100), message = Delete topic stress10)
AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress8, stress1, stress5, stress3), data = Range(101), message = Delete topic stress2)
AddKafkaData(topics = Set(stress4, stress6, stress12, stress8, stress1, stress5, stress3), data = Range(102), message = Delete topic stress14)
=> AddKafkaData(topics = Set(stress4, stress12, stress8, stress1, stress5, stress3), data = Range(), message = Delete topic stress6)
AddKafkaData(topics = Set(stress4, stress12, stress8, stress1, stress5, stress3, stress16), data = Range(103, 104, 105, 106, 107, 108, 109, 110), message = Add topic stress17)
CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30],[31],[32],[33],[34],[35],[36],[37],[38],[39],[40],[41],[42],[43],[44],[45],[46],[47],[48],[49],[50],[51],[52],[53],[54],[55],[56],[57],[58],[59],[60],[61],[62],[63],[64],[65],[66],[67],[68],[69],[70],[71],[72],[73],[74],[75],[76],[77],[78],[79],[80],[81],[82],[83],[84],[85],[86],[87],[88],[89],[90],[91],[92],[93],[94],[95],[96],[97],[98],[99],[100],[101],[102],[103],[104],[105],[106],[107],[108],[109],[110],[111]
== Stream ==
Output Mode: Append
Stream state: {KafkaSource[SubscribePattern[stress.*]]: {"stress5":{"17":0,"8":0,"11":0,"2":2,"5":0,"14":0,"4":1,"13":0,"7":1,"16":1,"1":2,"10":1,"9":1,"12":1,"3":2,"15":2,"6":1,"0":1},"stress8":{"2":1,"5":0,"4":1,"7":1,"1":1,"3":1,"6":1,"0":1},"stress1":{"23":0,"8":2,"17":0,"11":2,"2":3,"20":0,"5":2,"14":0,"4":2,"13":2,"22":0,"16":0,"7":2,"10":0,"1":4,"19":1,"18":0,"9":1,"3":2,"12":1,"21":0,"15":0,"6":1,"24":0,"0":2},"stress4":{"23":0,"17":1,"8":1,"26":0,"11":1,"20":0,"2":3,"5":0,"14":0,"4":0,"13":0,"22":0,"16":0,"7":0,"25":0,"10":0,"1":2,"19":0,"28":0,"9":0,"27":0,"18":0,"3":0,"12":0,"21":0,"15":0,"6":0,"24":0,"0":1},"stress3":{"8":1,"17":0,"11":0,"20":0,"2":0,"5":0,"14":0,"13":0,"4":0,"16":0,"7":1,"1":5,"10":2,"19":1,"9":2,"18":1,"3":2,"21":0,"12":3,"6":2,"15":2,"0":4},"stress12":{"2":0,"1":0,"0":0}}}
Thread state: alive
== Sink ==
0:
1:
2: [1] [2]
3:
4: [3]
5: [4]
6: [5] [6] [7] [11] [12] [9] [10] [8]
7: [17] [14] [18] [13] [19] [15] [21] [20] [16]
8: [22]
9: [23] [25] [26] [24]
10: [32] [27] [28] [33] [29] [31] [34] [30]
11:
12:
13:
14: [37] [36] [35]
15: [38]
16: [45] [59] [40] [42] [47] [55] [48] [56] [49] [54] [51] [41] [50] [53] [57] [43] [61] [60] [46] [39] [58] [52] [44]
17: [62] [71] [76] [63] [72] [75] [64] [68] [66] [73] [69] [70] [74] [78] [65] [67] [77]
18: [81] [90] [80] [79] [87] [91] [82] [83] [85] [89] [84] [86] [88]
19: [92] [93]
20: [96] [94] [95] [97]
21:
22: [98]
23: [100] [101] [99]
24:
25: [102]
26:
27: [103]
28:
== Plan ==
== Parsed Logical Plan ==
SerializeFromObject [input[0, int, true] AS value#26]
+- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#25: int
+- DeserializeToObject newInstance(class scala.Tuple2), obj#24: scala.Tuple2
+- Project [cast(key#1045 as string) AS key#15, cast(value#1046 as string) AS value#16]
+- LogicalRDD [key#1045, value#1046, topic#1047, partition#1048, offset#1049L, timestamp#1050L, timestampType#1051]
== Analyzed Logical Plan ==
value: int
SerializeFromObject [input[0, int, true] AS value#26]
+- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#25: int
+- DeserializeToObject newInstance(class scala.Tuple2), obj#24: scala.Tuple2
+- Project [cast(key#1045 as string) AS key#15, cast(value#1046 as string) AS value#16]
+- LogicalRDD [key#1045, value#1046, topic#1047, partition#1048, offset#1049L, timestamp#1050L, timestampType#1051]
== Optimized Logical Plan ==
SerializeFromObject [input[0, int, true] AS value#26]
+- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#25: int
+- DeserializeToObject newInstance(class scala.Tuple2), obj#24: scala.Tuple2
+- Project [cast(key#1045 as string) AS key#15, cast(value#1046 as string) AS value#16]
+- LogicalRDD [key#1045, value#1046, topic#1047, partition#1048, offset#1049L, timestamp#1050L, timestampType#1051]
== Physical Plan ==
*SerializeFromObject [input[0, int, true] AS value#26]
+- *MapElements <function1>, obj#25: int
+- *DeserializeToObject newInstance(class scala.Tuple2), obj#24: scala.Tuple2
+- *Project [cast(key#1045 as string) AS key#15, cast(value#1046 as string) AS value#16]
+- Scan ExistingRDD[key#1045,value#1046,topic#1047,partition#1048,offset#1049L,timestamp#1050L,timestampType#1051]
at org.scalatest.Assertions$class.newAssertionFailedException(Assertions.scala:495)
at org.scalatest.FunSuite.newAssertionFailedException(FunSuite.scala:1555)
at org.scalatest.Assertions$class.fail(Assertions.scala:1328)
at org.scalatest.FunSuite.fail(FunSuite.scala:1555)
at org.apache.spark.sql.streaming.StreamTest$class.failTest$1(StreamTest.scala:319)
at org.apache.spark.sql.streaming.StreamTest$$anonfun$testStream$1.apply(StreamTest.scala:488)
at org.apache.spark.sql.streaming.StreamTest$$anonfun$testStream$1.apply(StreamTest.scala:332)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at org.apache.spark.sql.streaming.StreamTest$class.testStream(StreamTest.scala:332)
at org.apache.spark.sql.kafka010.KafkaSourceTest.testStream(KafkaSourceSuite.scala:34)
at org.apache.spark.sql.streaming.StreamTest$class.runStressTest(StreamTest.scala:614)
at org.apache.spark.sql.kafka010.KafkaSourceTest.runStressTest(KafkaSourceSuite.scala:34)
at org.apache.spark.sql.kafka010.KafkaSourceStressSuite$$anonfun$17.apply$mcV$sp(KafkaSourceSuite.scala:590)
at org.apache.spark.sql.kafka010.KafkaSourceStressSuite$$anonfun$17.apply(KafkaSourceSuite.scala:570)
at org.apache.spark.sql.kafka010.KafkaSourceStressSuite$$anonfun$17.apply(KafkaSourceSuite.scala:570)
at org.scalatest.Transformer$$anonfun$apply$1.apply$mcV$sp(Transformer.scala:22)
at org.scalatest.OutcomeOf$class.outcomeOf(OutcomeOf.scala:85)
at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
at org.scalatest.Transformer.apply(Transformer.scala:22)
at org.scalatest.Transformer.apply(Transformer.scala:20)
at org.scalatest.FunSuiteLike$$anon$1.apply(FunSuiteLike.scala:166)
at org.apache.spark.SparkFunSuite.withFixture(SparkFunSuite.scala:68)
at org.scalatest.FunSuiteLike$class.invokeWithFixture$1(FunSuiteLike.scala:163)
at org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:175)
at org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:175)
at org.scalatest.SuperEngine.runTestImpl(Engine.scala:306)
at org.scalatest.FunSuiteLike$class.runTest(FunSuiteLike.scala:175)
at org.apache.spark.sql.kafka010.KafkaSourceTest.org$scalatest$BeforeAndAfterEach$$super$runTest(KafkaSourceSuite.scala:34)
at org.scalatest.BeforeAndAfterEach$class.runTest(BeforeAndAfterEach.scala:255)
at org.apache.spark.sql.kafka010.KafkaSourceTest.runTest(KafkaSourceSuite.scala:34)
at org.scalatest.FunSuiteLike$$anonfun$runTests$1.apply(FunSuiteLike.scala:208)
at org.scalatest.FunSuiteLike$$anonfun$runTests$1.apply(FunSuiteLike.scala:208)
at org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:413)
at org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:401)
at scala.collection.immutable.List.foreach(List.scala:381)
at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:401)
at org.scalatest.SuperEngine.org$scalatest$SuperEngine$$runTestsInBranch(Engine.scala:396)
at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:483)
at org.scalatest.FunSuiteLike$class.runTests(FunSuiteLike.scala:208)
at org.scalatest.FunSuite.runTests(FunSuite.scala:1555)
at org.scalatest.Suite$class.run(Suite.scala:1424)
at org.scalatest.FunSuite.org$scalatest$FunSuiteLike$$super$run(FunSuite.scala:1555)
at org.scalatest.FunSuiteLike$$anonfun$run$1.apply(FunSuiteLike.scala:212)
at org.scalatest.FunSuiteLike$$anonfun$run$1.apply(FunSuiteLike.scala:212)
at org.scalatest.SuperEngine.runImpl(Engine.scala:545)
at org.scalatest.FunSuiteLike$class.run(FunSuiteLike.scala:212)
at org.apache.spark.SparkFunSuite.org$scalatest$BeforeAndAfterAll$$super$run(SparkFunSuite.scala:31)
at org.scalatest.BeforeAndAfterAll$class.liftedTree1$1(BeforeAndAfterAll.scala:257)
at org.scalatest.BeforeAndAfterAll$class.run(BeforeAndAfterAll.scala:256)
at org.apache.spark.SparkFunSuite.run(SparkFunSuite.scala:31)
at org.scalatest.Suite$class.callExecuteOnSuite$1(Suite.scala:1492)
at org.scalatest.Suite$$anonfun$runNestedSuites$1.apply(Suite.scala:1528)
at org.scalatest.Suite$$anonfun$runNestedSuites$1.apply(Suite.scala:1526)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
at org.scalatest.Suite$class.runNestedSuites(Suite.scala:1526)
at org.scalatest.tools.DiscoverySuite.runNestedSuites(DiscoverySuite.scala:29)
at org.scalatest.Suite$class.run(Suite.scala:1421)
at org.scalatest.tools.DiscoverySuite.run(DiscoverySuite.scala:29)
at org.scalatest.tools.SuiteRunner.run(SuiteRunner.scala:55)
at org.scalatest.tools.Runner$$anonfun$doRunRunRunDaDoRunRun$3.apply(Runner.scala:2563)
at org.scalatest.tools.Runner$$anonfun$doRunRunRunDaDoRunRun$3.apply(Runner.scala:2557)
at scala.collection.immutable.List.foreach(List.scala:381)
at org.scalatest.tools.Runner$.doRunRunRunDaDoRunRun(Runner.scala:2557)
at org.scalatest.tools.Runner$$anonfun$runOptionallyWithPassFailReporter$2.apply(Runner.scala:1044)
at org.scalatest.tools.Runner$$anonfun$runOptionallyWithPassFailReporter$2.apply(Runner.scala:1043)
at org.scalatest.tools.Runner$.withClassLoaderAndDispatchReporter(Runner.scala:2722)
at org.scalatest.tools.Runner$.runOptionallyWithPassFailReporter(Runner.scala:1043)
at org.scalatest.tools.Runner$.main(Runner.scala:860)
at org.scalatest.tools.Runner.main(Runner.scala)