&#010;Error adding data: Timeout after waiting for 10000 ms.&#010;org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:78)&#010; org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:30)&#010; org.apache.spark.sql.kafka010.KafkaTestUtils$$anonfun$2.apply(KafkaTestUtils.scala:269)&#010; org.apache.spark.sql.kafka010.KafkaTestUtils$$anonfun$2.apply(KafkaTestUtils.scala:263)&#010; scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)&#010; scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)&#010; scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)&#010; scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)&#010; scala.collection.TraversableLike$class.map(TraversableLike.scala:234)&#010; scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:186)&#010;&#010;&#010;== Progress ==&#010; AssertOnQuery(<condition>, )&#010; AddKafkaData(topics = Set(stress4, stress2, stress1, stress5, stress3), data = Range(0, 1, 2), message = )&#010; AddKafkaData(topics = Set(stress1, stress2, stress3, stress4), data = Range(3, 4), message = Delete topic stress5)&#010; AddKafkaData(topics = Set(stress1, stress2, stress3, stress4), data = Range(), message = )&#010; AddKafkaData(topics = Set(stress4, stress6, stress2, stress1, stress3), data = Range(5, 6), message = Add topic stress7)&#010; AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress3), data = Range(), message = Add topic stress9)&#010; CheckAnswer: [1],[2],[3],[4],[5],[6],[7]&#010; AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress3, stress10), data = Range(7, 8, 9, 10), message = Add topic stress11)&#010; AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress3, stress10), data = Range(11), message = Add partition)&#010; AddKafkaData(topics = Set(stress4, stress6, stress12, stress2, stress8, stress1, stress3, stress10), data = Range(), message = Add topic stress13)&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress3, stress10), data = Range(12, 13, 14, 15), message = Add topic stress15)&#010; CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16]&#010; StopStream&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress3, stress10), data = Range(16), message = )&#010; StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@3edd6b40,Map(),null)&#010; CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17]&#010; StopStream&#010; StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@338de11c,Map(),null)&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress3, stress10), data = Range(17, 18, 19, 20), message = Add partition)&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress3, stress10, stress16), data = Range(21, 22, 23, 24, 25, 26), message = Add topic stress17)&#010; 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]&#010; StopStream&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress3, stress10, stress16), data = Range(27, 28), message = )&#010; StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@4e09a3a9,Map(),null)&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress3, stress10, stress16), data = Range(29, 30, 31, 32, 33, 34, 35, 36), message = )&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress3, stress10, stress16), data = Range(37, 38, 39), message = Delete topic stress8)&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress18, stress3, stress10, stress16), data = Range(40, 41, 42), message = Add topic stress19)&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress18, stress3, stress10, stress16), data = Range(43), message = )&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress18, stress3, stress10, stress16), data = Range(44, 45, 46, 47, 48), message = )&#010;=> AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress18, stress3, stress10, stress16), data = Range(49, 50, 51, 52, 53, 54), message = )&#010; 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]&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress3, stress10, stress16), data = Range(55, 56, 57), message = Delete topic stress18)&#010; 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]&#010; StopStream&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress3, stress10, stress16), data = Range(58, 59, 60, 61, 62, 63, 64, 65), message = Add partition)&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress3, stress10, stress16), data = Range(66, 67, 68, 69, 70, 71, 72), message = )&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress3, stress10, stress16), data = Range(73, 74, 75, 76), message = )&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress3, stress10, stress16), data = Range(77), message = )&#010; StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@6108efc2,Map(),null)&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress3, stress10, stress16), data = Range(78, 79, 80, 81, 82, 83, 84), message = )&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress3, stress10, stress16), data = Range(), message = Add topic stress21)&#010; 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]&#010; StopStream&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress3, stress10, stress16), data = Range(85, 86, 87, 88, 89), message = Add partition)&#010; StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@510e9e24,Map(),null)&#010; 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]&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress3, stress10, stress16), data = Range(90, 91, 92, 93, 94, 95), message = )&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress22, stress3, stress10, stress16), data = Range(96, 97, 98, 99, 100, 101, 102, 103), message = Add topic stress23)&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress22, stress3, stress10, stress16), data = Range(104, 105, 106, 107, 108, 109, 110, 111, 112), message = Add partition)&#010; 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],[112],[113]&#010; StopStream&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress22, stress3, stress10, stress16), data = Range(113, 114, 115, 116, 117, 118, 119), message = Add partition)&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress22, stress3, stress10, stress16), data = Range(), message = )&#010; StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@667437b3,Map(),null)&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress22, stress3, stress10, stress16), data = Range(120, 121, 122, 123, 124, 125), message = )&#010; AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress22, stress3, stress10, stress16), data = Range(126, 127, 128, 129, 130, 131, 132), message = )&#010; AddKafkaData(topics = Set(stress14, stress24, stress4, stress6, stress12, stress2, stress20, stress1, stress22, stress3, stress10, stress16), data = Range(133), message = Add topic stress25)&#010; 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],[112],[113],[114],[115],[116],[117],[118],[119],[120],[121],[122],[123],[124],[125],[126],[127],[128],[129],[130],[131],[132],[133],[134]&#010;&#010;== Stream ==&#010;Output Mode: Append&#010;Stream state: {KafkaV2[SubscribePattern[stress.*]]: {"stress16":{"2":1,"5":0,"4":0,"1":1,"3":1,"0":0},"stress10":{"2":0,"5":0,"4":0,"1":0,"3":0,"0":0},"stress1":{"2":1,"5":0,"4":1,"1":1,"3":2,"6":0,"0":7},"stress4":{"2":1,"5":1,"4":1,"1":1,"3":0,"6":1,"0":5},"stress3":{"8":2,"2":2,"5":1,"4":0,"7":1,"10":2,"1":0,"9":0,"3":1,"6":1,"0":8},"stress6":{"8":1,"11":0,"2":3,"5":0,"4":1,"7":0,"10":0,"1":1,"9":0,"3":0,"6":0,"0":1},"stress18":{"2":0,"1":0,"3":0,"0":0},"stress12":{"2":0,"1":1,"0":0},"stress2":{"8":0,"2":0,"5":0,"4":0,"7":0,"1":4,"3":0,"6":0,"0":4},"stress14":{"2":1,"5":2,"4":1,"1":2,"3":2,"0":2}}}&#010;Thread state: alive&#010;Thread stack trace: org.apache.kafka.common.requests.MetadataResponse.cluster(MetadataResponse.java:385)&#010;org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.handleCompletedMetadataResponse(NetworkClient.java:1001)&#010;org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:817)&#010;org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:544)&#010;org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:265)&#010;org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:236)&#010;org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:227)&#010;org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:161)&#010;org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.ensureFreshMetadata(ConsumerNetworkClient.java:175)&#010;org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:336)&#010;org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1214)&#010;org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1183)&#010;org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1123)&#010;org.apache.spark.sql.kafka010.KafkaOffsetReader$$anonfun$fetchLatestOffsets$1$$anonfun$apply$9.apply(KafkaOffsetReader.scala:199)&#010;org.apache.spark.sql.kafka010.KafkaOffsetReader$$anonfun$fetchLatestOffsets$1$$anonfun$apply$9.apply(KafkaOffsetReader.scala:197)&#010;org.apache.spark.sql.kafka010.KafkaOffsetReader$$anonfun$org$apache$spark$sql$kafka010$KafkaOffsetReader$$withRetriesWithoutInterrupt$1.apply$mcV$sp(KafkaOffsetReader.scala:288)&#010;org.apache.spark.sql.kafka010.KafkaOffsetReader$$anonfun$org$apache$spark$sql$kafka010$KafkaOffsetReader$$withRetriesWithoutInterrupt$1.apply(KafkaOffsetReader.scala:287)&#010;org.apache.spark.sql.kafka010.KafkaOffsetReader$$anonfun$org$apache$spark$sql$kafka010$KafkaOffsetReader$$withRetriesWithoutInterrupt$1.apply(KafkaOffsetReader.scala:287)&#010;org.apache.spark.util.UninterruptibleThread.runUninterruptibly(UninterruptibleThread.scala:77)&#010;org.apache.spark.sql.kafka010.KafkaOffsetReader.org$apache$spark$sql$kafka010$KafkaOffsetReader$$withRetriesWithoutInterrupt(KafkaOffsetReader.scala:286)&#010;org.apache.spark.sql.kafka010.KafkaOffsetReader$$anonfun$fetchLatestOffsets$1.apply(KafkaOffsetReader.scala:197)&#010;org.apache.spark.sql.kafka010.KafkaOffsetReader$$anonfun$fetchLatestOffsets$1.apply(KafkaOffsetReader.scala:197)&#010;org.apache.spark.sql.kafka010.KafkaOffsetReader.runUninterruptibly(KafkaOffsetReader.scala:255)&#010;org.apache.spark.sql.kafka010.KafkaOffsetReader.fetchLatestOffsets(KafkaOffsetReader.scala:196)&#010;org.apache.spark.sql.kafka010.KafkaMicroBatchReadSupport.latestOffset(KafkaMicroBatchReadSupport.scala:87)&#010;org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch$1$$anonfun$5$$anonfun$apply$9.apply(MicroBatchExecution.scala:351)&#010;org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch$1$$anonfun$5$$anonfun$apply$9.apply(MicroBatchExecution.scala:347)&#010;org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:323)&#010;org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:59)&#010;org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch$1$$anonfun$5.apply(MicroBatchExecution.scala:347)&#010;org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch$1$$anonfun$5.apply(MicroBatchExecution.scala:339)&#010;scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)&#010;scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)&#010;scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)&#010;scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)&#010;scala.collection.TraversableLike$class.map(TraversableLike.scala:234)&#010;scala.collection.AbstractTraversable.map(Traversable.scala:104)&#010;org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch$1.apply$mcZ$sp(MicroBatchExecution.scala:339)&#010;org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch$1.apply(MicroBatchExecution.scala:335)&#010;org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch$1.apply(MicroBatchExecution.scala:335)&#010;org.apache.spark.sql.execution.streaming.MicroBatchExecution.withProgressLocked(MicroBatchExecution.scala:557)&#010;org.apache.spark.sql.execution.streaming.MicroBatchExecution.org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch(MicroBatchExecution.scala:335)&#010;org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply$mcV$sp(MicroBatchExecution.scala:181)&#010;org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:164)&#010;org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:164)&#010;org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:323)&#010;org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:59)&#010;org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1.apply$mcZ$sp(MicroBatchExecution.scala:164)&#010;org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:56)&#010;org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:158)&#010;org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:281)&#010;org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:191)&#010;&#010;&#010;== Sink ==&#010;0: &#010;1: [1] [2]&#010;2: [3]&#010;3: [4] [5]&#010;4: [6] [7]&#010;5: &#010;6: &#010;7: [8]&#010;8: [11] [10] [9]&#010;9: &#010;10: &#010;11: &#010;12: [12]&#010;13: &#010;14: [14] [16] [15] [13]&#010;15: &#010;16: [17]&#010;17: &#010;18: &#010;19: [21] [19] [18] [20]&#010;20: &#010;21: [23] [22] [26] [24] [25]&#010;22: [27]&#010;23: &#010;24: [28] [29]&#010;25: [34] [33] [30] [31] [32]&#010;26: [37] [35] [36]&#010;27: [39] [40] [38]&#010;28: [43] [41] [42]&#010;29: &#010;30: [44]&#010;31: [46] [45]&#010;32: [48] [49] [47]&#010;&#010;&#010;== Plan ==&#010;== Parsed Logical Plan ==&#010;SerializeFromObject [input[0, int, false] AS value#11115]&#010;+- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#11114: int&#010; +- DeserializeToObject newInstance(class scala.Tuple2), obj#11113: scala.Tuple2&#010; +- Project [cast(key#11089 as string) AS key#11103, cast(value#11090 as string) AS value#11104]&#010; +- Project [key#11480 AS key#11089, value#11481 AS value#11090, topic#11482 AS topic#11091, partition#11483 AS partition#11092, offset#11484L AS offset#11093L, timestamp#11485 AS timestamp#11094, timestampType#11486 AS timestampType#11095]&#010; +- Streaming RelationV2 kafka[key#11480, value#11481, topic#11482, partition#11483, offset#11484L, timestamp#11485, timestampType#11486] (Options: [kafka.metadata.max.age.ms=1,failOnDataLoss=false,kafka.bootstrap.servers=127.0.0.1:40631,kafka.d...)&#010;&#010;== Analyzed Logical Plan ==&#010;value: int&#010;SerializeFromObject [input[0, int, false] AS value#11115]&#010;+- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#11114: int&#010; +- DeserializeToObject newInstance(class scala.Tuple2), obj#11113: scala.Tuple2&#010; +- Project [cast(key#11089 as string) AS key#11103, cast(value#11090 as string) AS value#11104]&#010; +- Project [key#11480 AS key#11089, value#11481 AS value#11090, topic#11482 AS topic#11091, partition#11483 AS partition#11092, offset#11484L AS offset#11093L, timestamp#11485 AS timestamp#11094, timestampType#11486 AS timestampType#11095]&#010; +- Streaming RelationV2 kafka[key#11480, value#11481, topic#11482, partition#11483, offset#11484L, timestamp#11485, timestampType#11486] (Options: [kafka.metadata.max.age.ms=1,failOnDataLoss=false,kafka.bootstrap.servers=127.0.0.1:40631,kafka.d...)&#010;&#010;== Optimized Logical Plan ==&#010;SerializeFromObject [input[0, int, false] AS value#11115]&#010;+- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#11114: int&#010; +- DeserializeToObject newInstance(class scala.Tuple2), obj#11113: scala.Tuple2&#010; +- Project [cast(key#11480 as string) AS key#11103, cast(value#11481 as string) AS value#11104]&#010; +- Streaming RelationV2 kafka[key#11480, value#11481, topic#11482, partition#11483, offset#11484L, timestamp#11485, timestampType#11486] (Options: [kafka.metadata.max.age.ms=1,failOnDataLoss=false,kafka.bootstrap.servers=127.0.0.1:40631,kafka.d...)&#010;&#010;== Physical Plan ==&#010;*(1) SerializeFromObject [input[0, int, false] AS value#11115]&#010;+- *(1) MapElements <function1>, obj#11114: int&#010; +- *(1) DeserializeToObject newInstance(class scala.Tuple2), obj#11113: scala.Tuple2&#010; +- *(1) Project [cast(key#11480 as string) AS key#11103, cast(value#11481 as string) AS value#11104]&#010; +- *(1) Project [key#11480, value#11481, topic#11482, partition#11483, offset#11484L, timestamp#11485, timestampType#11486]&#010; +- *(1) ScanV2 kafka[key#11480, value#11481, topic#11482, partition#11483, offset#11484L, timestamp#11485, timestampType#11486] (Options: [kafka.metadata.max.age.ms=1,failOnDataLoss=false,kafka.bootstrap.servers=127.0.0.1:40631,kafka.d...)&#010;&#010; &#010;

      org.scalatest.exceptions.TestFailedException: 
Error adding data: Timeout after waiting for 10000 ms.
org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:78)
	org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:30)
	org.apache.spark.sql.kafka010.KafkaTestUtils$$anonfun$2.apply(KafkaTestUtils.scala:269)
	org.apache.spark.sql.kafka010.KafkaTestUtils$$anonfun$2.apply(KafkaTestUtils.scala:263)
	scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
	scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
	scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
	scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:186)


== Progress ==
   AssertOnQuery(<condition>, )
   AddKafkaData(topics = Set(stress4, stress2, stress1, stress5, stress3), data = Range(0, 1, 2), message = )
   AddKafkaData(topics = Set(stress1, stress2, stress3, stress4), data = Range(3, 4), message = Delete topic stress5)
   AddKafkaData(topics = Set(stress1, stress2, stress3, stress4), data = Range(), message = )
   AddKafkaData(topics = Set(stress4, stress6, stress2, stress1, stress3), data = Range(5, 6), message = Add topic stress7)
   AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress3), data = Range(), message = Add topic stress9)
   CheckAnswer: [1],[2],[3],[4],[5],[6],[7]
   AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress3, stress10), data = Range(7, 8, 9, 10), message = Add topic stress11)
   AddKafkaData(topics = Set(stress4, stress6, stress2, stress8, stress1, stress3, stress10), data = Range(11), message = Add partition)
   AddKafkaData(topics = Set(stress4, stress6, stress12, stress2, stress8, stress1, stress3, stress10), data = Range(), message = Add topic stress13)
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress3, stress10), data = Range(12, 13, 14, 15), message = Add topic stress15)
   CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16]
   StopStream
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress3, stress10), data = Range(16), message = )
   StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@3edd6b40,Map(),null)
   CheckAnswer: [1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17]
   StopStream
   StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@338de11c,Map(),null)
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress3, stress10), data = Range(17, 18, 19, 20), message = Add partition)
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress3, stress10, stress16), data = Range(21, 22, 23, 24, 25, 26), 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]
   StopStream
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress3, stress10, stress16), data = Range(27, 28), message = )
   StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@4e09a3a9,Map(),null)
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress8, stress1, stress3, stress10, stress16), data = Range(29, 30, 31, 32, 33, 34, 35, 36), message = )
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress3, stress10, stress16), data = Range(37, 38, 39), message = Delete topic stress8)
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress18, stress3, stress10, stress16), data = Range(40, 41, 42), message = Add topic stress19)
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress18, stress3, stress10, stress16), data = Range(43), message = )
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress18, stress3, stress10, stress16), data = Range(44, 45, 46, 47, 48), message = )
=> AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress18, stress3, stress10, stress16), data = Range(49, 50, 51, 52, 53, 54), message = )
   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]
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress3, stress10, stress16), data = Range(55, 56, 57), message = Delete topic stress18)
   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]
   StopStream
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress3, stress10, stress16), data = Range(58, 59, 60, 61, 62, 63, 64, 65), message = Add partition)
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress3, stress10, stress16), data = Range(66, 67, 68, 69, 70, 71, 72), message = )
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress3, stress10, stress16), data = Range(73, 74, 75, 76), message = )
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress3, stress10, stress16), data = Range(77), message = )
   StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@6108efc2,Map(),null)
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress1, stress3, stress10, stress16), data = Range(78, 79, 80, 81, 82, 83, 84), message = )
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress3, stress10, stress16), data = Range(), message = Add topic stress21)
   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]
   StopStream
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress3, stress10, stress16), data = Range(85, 86, 87, 88, 89), message = Add partition)
   StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@510e9e24,Map(),null)
   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]
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress3, stress10, stress16), data = Range(90, 91, 92, 93, 94, 95), message = )
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress22, stress3, stress10, stress16), data = Range(96, 97, 98, 99, 100, 101, 102, 103), message = Add topic stress23)
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress22, stress3, stress10, stress16), data = Range(104, 105, 106, 107, 108, 109, 110, 111, 112), message = Add partition)
   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],[112],[113]
   StopStream
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress22, stress3, stress10, stress16), data = Range(113, 114, 115, 116, 117, 118, 119), message = Add partition)
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress22, stress3, stress10, stress16), data = Range(), message = )
   StartStream(ProcessingTime(0),org.apache.spark.util.SystemClock@667437b3,Map(),null)
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress22, stress3, stress10, stress16), data = Range(120, 121, 122, 123, 124, 125), message = )
   AddKafkaData(topics = Set(stress14, stress4, stress6, stress12, stress2, stress20, stress1, stress22, stress3, stress10, stress16), data = Range(126, 127, 128, 129, 130, 131, 132), message = )
   AddKafkaData(topics = Set(stress14, stress24, stress4, stress6, stress12, stress2, stress20, stress1, stress22, stress3, stress10, stress16), data = Range(133), message = Add topic stress25)
   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],[112],[113],[114],[115],[116],[117],[118],[119],[120],[121],[122],[123],[124],[125],[126],[127],[128],[129],[130],[131],[132],[133],[134]

== Stream ==
Output Mode: Append
Stream state: {KafkaV2[SubscribePattern[stress.*]]: {"stress16":{"2":1,"5":0,"4":0,"1":1,"3":1,"0":0},"stress10":{"2":0,"5":0,"4":0,"1":0,"3":0,"0":0},"stress1":{"2":1,"5":0,"4":1,"1":1,"3":2,"6":0,"0":7},"stress4":{"2":1,"5":1,"4":1,"1":1,"3":0,"6":1,"0":5},"stress3":{"8":2,"2":2,"5":1,"4":0,"7":1,"10":2,"1":0,"9":0,"3":1,"6":1,"0":8},"stress6":{"8":1,"11":0,"2":3,"5":0,"4":1,"7":0,"10":0,"1":1,"9":0,"3":0,"6":0,"0":1},"stress18":{"2":0,"1":0,"3":0,"0":0},"stress12":{"2":0,"1":1,"0":0},"stress2":{"8":0,"2":0,"5":0,"4":0,"7":0,"1":4,"3":0,"6":0,"0":4},"stress14":{"2":1,"5":2,"4":1,"1":2,"3":2,"0":2}}}
Thread state: alive
Thread stack trace: org.apache.kafka.common.requests.MetadataResponse.cluster(MetadataResponse.java:385)
org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.handleCompletedMetadataResponse(NetworkClient.java:1001)
org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:817)
org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:544)
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:265)
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:236)
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:227)
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:161)
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.ensureFreshMetadata(ConsumerNetworkClient.java:175)
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:336)
org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1214)
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1183)
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1123)
org.apache.spark.sql.kafka010.KafkaOffsetReader$$anonfun$fetchLatestOffsets$1$$anonfun$apply$9.apply(KafkaOffsetReader.scala:199)
org.apache.spark.sql.kafka010.KafkaOffsetReader$$anonfun$fetchLatestOffsets$1$$anonfun$apply$9.apply(KafkaOffsetReader.scala:197)
org.apache.spark.sql.kafka010.KafkaOffsetReader$$anonfun$org$apache$spark$sql$kafka010$KafkaOffsetReader$$withRetriesWithoutInterrupt$1.apply$mcV$sp(KafkaOffsetReader.scala:288)
org.apache.spark.sql.kafka010.KafkaOffsetReader$$anonfun$org$apache$spark$sql$kafka010$KafkaOffsetReader$$withRetriesWithoutInterrupt$1.apply(KafkaOffsetReader.scala:287)
org.apache.spark.sql.kafka010.KafkaOffsetReader$$anonfun$org$apache$spark$sql$kafka010$KafkaOffsetReader$$withRetriesWithoutInterrupt$1.apply(KafkaOffsetReader.scala:287)
org.apache.spark.util.UninterruptibleThread.runUninterruptibly(UninterruptibleThread.scala:77)
org.apache.spark.sql.kafka010.KafkaOffsetReader.org$apache$spark$sql$kafka010$KafkaOffsetReader$$withRetriesWithoutInterrupt(KafkaOffsetReader.scala:286)
org.apache.spark.sql.kafka010.KafkaOffsetReader$$anonfun$fetchLatestOffsets$1.apply(KafkaOffsetReader.scala:197)
org.apache.spark.sql.kafka010.KafkaOffsetReader$$anonfun$fetchLatestOffsets$1.apply(KafkaOffsetReader.scala:197)
org.apache.spark.sql.kafka010.KafkaOffsetReader.runUninterruptibly(KafkaOffsetReader.scala:255)
org.apache.spark.sql.kafka010.KafkaOffsetReader.fetchLatestOffsets(KafkaOffsetReader.scala:196)
org.apache.spark.sql.kafka010.KafkaMicroBatchReadSupport.latestOffset(KafkaMicroBatchReadSupport.scala:87)
org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch$1$$anonfun$5$$anonfun$apply$9.apply(MicroBatchExecution.scala:351)
org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch$1$$anonfun$5$$anonfun$apply$9.apply(MicroBatchExecution.scala:347)
org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:323)
org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:59)
org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch$1$$anonfun$5.apply(MicroBatchExecution.scala:347)
org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch$1$$anonfun$5.apply(MicroBatchExecution.scala:339)
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
scala.collection.AbstractTraversable.map(Traversable.scala:104)
org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch$1.apply$mcZ$sp(MicroBatchExecution.scala:339)
org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch$1.apply(MicroBatchExecution.scala:335)
org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch$1.apply(MicroBatchExecution.scala:335)
org.apache.spark.sql.execution.streaming.MicroBatchExecution.withProgressLocked(MicroBatchExecution.scala:557)
org.apache.spark.sql.execution.streaming.MicroBatchExecution.org$apache$spark$sql$execution$streaming$MicroBatchExecution$$constructNextBatch(MicroBatchExecution.scala:335)
org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply$mcV$sp(MicroBatchExecution.scala:181)
org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:164)
org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:164)
org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:323)
org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:59)
org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1.apply$mcZ$sp(MicroBatchExecution.scala:164)
org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:56)
org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:158)
org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:281)
org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:191)


== Sink ==
0: 
1: [1] [2]
2: [3]
3: [4] [5]
4: [6] [7]
5: 
6: 
7: [8]
8: [11] [10] [9]
9: 
10: 
11: 
12: [12]
13: 
14: [14] [16] [15] [13]
15: 
16: [17]
17: 
18: 
19: [21] [19] [18] [20]
20: 
21: [23] [22] [26] [24] [25]
22: [27]
23: 
24: [28] [29]
25: [34] [33] [30] [31] [32]
26: [37] [35] [36]
27: [39] [40] [38]
28: [43] [41] [42]
29: 
30: [44]
31: [46] [45]
32: [48] [49] [47]


== Plan ==
== Parsed Logical Plan ==
SerializeFromObject [input[0, int, false] AS value#11115]
+- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#11114: int
   +- DeserializeToObject newInstance(class scala.Tuple2), obj#11113: scala.Tuple2
      +- Project [cast(key#11089 as string) AS key#11103, cast(value#11090 as string) AS value#11104]
         +- Project [key#11480 AS key#11089, value#11481 AS value#11090, topic#11482 AS topic#11091, partition#11483 AS partition#11092, offset#11484L AS offset#11093L, timestamp#11485 AS timestamp#11094, timestampType#11486 AS timestampType#11095]
            +- Streaming RelationV2 kafka[key#11480, value#11481, topic#11482, partition#11483, offset#11484L, timestamp#11485, timestampType#11486] (Options: [kafka.metadata.max.age.ms=1,failOnDataLoss=false,kafka.bootstrap.servers=127.0.0.1:40631,kafka.d...)

== Analyzed Logical Plan ==
value: int
SerializeFromObject [input[0, int, false] AS value#11115]
+- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#11114: int
   +- DeserializeToObject newInstance(class scala.Tuple2), obj#11113: scala.Tuple2
      +- Project [cast(key#11089 as string) AS key#11103, cast(value#11090 as string) AS value#11104]
         +- Project [key#11480 AS key#11089, value#11481 AS value#11090, topic#11482 AS topic#11091, partition#11483 AS partition#11092, offset#11484L AS offset#11093L, timestamp#11485 AS timestamp#11094, timestampType#11486 AS timestampType#11095]
            +- Streaming RelationV2 kafka[key#11480, value#11481, topic#11482, partition#11483, offset#11484L, timestamp#11485, timestampType#11486] (Options: [kafka.metadata.max.age.ms=1,failOnDataLoss=false,kafka.bootstrap.servers=127.0.0.1:40631,kafka.d...)

== Optimized Logical Plan ==
SerializeFromObject [input[0, int, false] AS value#11115]
+- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#11114: int
   +- DeserializeToObject newInstance(class scala.Tuple2), obj#11113: scala.Tuple2
      +- Project [cast(key#11480 as string) AS key#11103, cast(value#11481 as string) AS value#11104]
         +- Streaming RelationV2 kafka[key#11480, value#11481, topic#11482, partition#11483, offset#11484L, timestamp#11485, timestampType#11486] (Options: [kafka.metadata.max.age.ms=1,failOnDataLoss=false,kafka.bootstrap.servers=127.0.0.1:40631,kafka.d...)

== Physical Plan ==
*(1) SerializeFromObject [input[0, int, false] AS value#11115]
+- *(1) MapElements <function1>, obj#11114: int
   +- *(1) DeserializeToObject newInstance(class scala.Tuple2), obj#11113: scala.Tuple2
      +- *(1) Project [cast(key#11480 as string) AS key#11103, cast(value#11481 as string) AS value#11104]
         +- *(1) Project [key#11480, value#11481, topic#11482, partition#11483, offset#11484L, timestamp#11485, timestampType#11486]
            +- *(1) ScanV2 kafka[key#11480, value#11481, topic#11482, partition#11483, offset#11484L, timestamp#11485, timestampType#11486] (Options: [kafka.metadata.max.age.ms=1,failOnDataLoss=false,kafka.bootstrap.servers=127.0.0.1:40631,kafka.d...)

         
         
      at org.scalatest.Assertions$class.newAssertionFailedException(Assertions.scala:528)
      at org.scalatest.FunSuite.newAssertionFailedException(FunSuite.scala:1560)
      at org.scalatest.Assertions$class.fail(Assertions.scala:1089)
      at org.scalatest.FunSuite.fail(FunSuite.scala:1560)
      at org.apache.spark.sql.streaming.StreamTest$class.failTest$1(StreamTest.scala:453)
      at org.apache.spark.sql.streaming.StreamTest$class.executeAction$1(StreamTest.scala:719)
      at org.apache.spark.sql.streaming.StreamTest$$anonfun$liftedTree1$1$1.apply(StreamTest.scala:773)
      at org.apache.spark.sql.streaming.StreamTest$$anonfun$liftedTree1$1$1.apply(StreamTest.scala:760)
      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.liftedTree1$1(StreamTest.scala:760)
      at org.apache.spark.sql.streaming.StreamTest$class.testStream(StreamTest.scala:759)
      at org.apache.spark.sql.kafka010.KafkaSourceTest.testStream(KafkaMicroBatchSourceSuite.scala:49)
      at org.apache.spark.sql.streaming.StreamTest$class.runStressTest(StreamTest.scala:871)
      at org.apache.spark.sql.kafka010.KafkaSourceTest.runStressTest(KafkaMicroBatchSourceSuite.scala:49)
      at org.apache.spark.sql.kafka010.KafkaSourceStressSuite$$anonfun$21.apply$mcV$sp(KafkaMicroBatchSourceSuite.scala:1444)
      at org.apache.spark.sql.kafka010.KafkaSourceStressSuite$$anonfun$21.apply(KafkaMicroBatchSourceSuite.scala:1423)
      at org.apache.spark.sql.kafka010.KafkaSourceStressSuite$$anonfun$21.apply(KafkaMicroBatchSourceSuite.scala:1423)
      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:186)
      at org.apache.spark.SparkFunSuite.withFixture(SparkFunSuite.scala:103)
      at org.scalatest.FunSuiteLike$class.invokeWithFixture$1(FunSuiteLike.scala:183)
      at org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:196)
      at org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:196)
      at org.scalatest.SuperEngine.runTestImpl(Engine.scala:289)
      at org.scalatest.FunSuiteLike$class.runTest(FunSuiteLike.scala:196)
      at org.apache.spark.sql.kafka010.KafkaSourceTest.org$scalatest$BeforeAndAfterEach$$super$runTest(KafkaMicroBatchSourceSuite.scala:49)
      at org.scalatest.BeforeAndAfterEach$class.runTest(BeforeAndAfterEach.scala:221)
      at org.apache.spark.sql.kafka010.KafkaSourceTest.runTest(KafkaMicroBatchSourceSuite.scala:49)
      at org.scalatest.FunSuiteLike$$anonfun$runTests$1.apply(FunSuiteLike.scala:229)
      at org.scalatest.FunSuiteLike$$anonfun$runTests$1.apply(FunSuiteLike.scala:229)
      at org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:396)
      at org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:384)
      at scala.collection.immutable.List.foreach(List.scala:392)
      at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:384)
      at org.scalatest.SuperEngine.org$scalatest$SuperEngine$$runTestsInBranch(Engine.scala:379)
      at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:461)
      at org.scalatest.FunSuiteLike$class.runTests(FunSuiteLike.scala:229)
      at org.scalatest.FunSuite.runTests(FunSuite.scala:1560)
      at org.scalatest.Suite$class.run(Suite.scala:1147)
      at org.scalatest.FunSuite.org$scalatest$FunSuiteLike$$super$run(FunSuite.scala:1560)
      at org.scalatest.FunSuiteLike$$anonfun$run$1.apply(FunSuiteLike.scala:233)
      at org.scalatest.FunSuiteLike$$anonfun$run$1.apply(FunSuiteLike.scala:233)
      at org.scalatest.SuperEngine.runImpl(Engine.scala:521)
      at org.scalatest.FunSuiteLike$class.run(FunSuiteLike.scala:233)
      at org.apache.spark.SparkFunSuite.org$scalatest$BeforeAndAfterAll$$super$run(SparkFunSuite.scala:52)
      at org.scalatest.BeforeAndAfterAll$class.liftedTree1$1(BeforeAndAfterAll.scala:213)
      at org.scalatest.BeforeAndAfterAll$class.run(BeforeAndAfterAll.scala:210)
      at org.apache.spark.SparkFunSuite.run(SparkFunSuite.scala:52)
      at org.scalatest.Suite$class.callExecuteOnSuite$1(Suite.scala:1210)
      at org.scalatest.Suite$$anonfun$runNestedSuites$1.apply(Suite.scala:1257)
      at org.scalatest.Suite$$anonfun$runNestedSuites$1.apply(Suite.scala:1255)
      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:1255)
      at org.scalatest.tools.DiscoverySuite.runNestedSuites(DiscoverySuite.scala:30)
      at org.scalatest.Suite$class.run(Suite.scala:1144)
      at org.scalatest.tools.DiscoverySuite.run(DiscoverySuite.scala:30)
      at org.scalatest.tools.SuiteRunner.run(SuiteRunner.scala:45)
      at org.scalatest.tools.Runner$$anonfun$doRunRunRunDaDoRunRun$1.apply(Runner.scala:1340)
      at org.scalatest.tools.Runner$$anonfun$doRunRunRunDaDoRunRun$1.apply(Runner.scala:1334)
      at scala.collection.immutable.List.foreach(List.scala:392)
      at org.scalatest.tools.Runner$.doRunRunRunDaDoRunRun(Runner.scala:1334)
      at org.scalatest.tools.Runner$$anonfun$runOptionallyWithPassFailReporter$2.apply(Runner.scala:1011)
      at org.scalatest.tools.Runner$$anonfun$runOptionallyWithPassFailReporter$2.apply(Runner.scala:1010)
      at org.scalatest.tools.Runner$.withClassLoaderAndDispatchReporter(Runner.scala:1500)
      at org.scalatest.tools.Runner$.runOptionallyWithPassFailReporter(Runner.scala:1010)
      at org.scalatest.tools.Runner$.main(Runner.scala:827)
      at org.scalatest.tools.Runner.main(Runner.scala)