org.scalatest.exceptions.TestFailedException: Assert on query failed: == Progress == StartStream(ProcessingTime(1),org.apache.spark.util.SystemClock@3db1548c,Map()) AssertOnQuery(<condition>, ) AddKafkaData(topics = Set(topic-22), data = WrappedArray(1, 2, 3), message = ) CheckAnswer: [2],[3],[4] => AssertOnQuery(<condition>, ) == Stream == Output Mode: Append Stream state: {KafkaSource[Subscribe[topic-22]]: {"topic-22":{"2":0,"4":1,"1":1,"3":1,"0":1}}} Thread state: alive == Sink == 0: 1: [2] 2: [3] 3: [4] == Plan == == Parsed Logical Plan == SerializeFromObject [input[0, int, false] AS value#4414] +- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#4413: int +- DeserializeToObject newInstance(class scala.Tuple2), obj#4412: scala.Tuple2 +- Project [cast(key#4479 as string) AS key#4403, cast(value#4480 as string) AS value#4404] +- LogicalRDD [key#4479, value#4480, topic#4481, partition#4482, offset#4483L, timestamp#4484, timestampType#4485] == Analyzed Logical Plan == value: int SerializeFromObject [input[0, int, false] AS value#4414] +- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#4413: int +- DeserializeToObject newInstance(class scala.Tuple2), obj#4412: scala.Tuple2 +- Project [cast(key#4479 as string) AS key#4403, cast(value#4480 as string) AS value#4404] +- LogicalRDD [key#4479, value#4480, topic#4481, partition#4482, offset#4483L, timestamp#4484, timestampType#4485] == Optimized Logical Plan == SerializeFromObject [input[0, int, false] AS value#4414] +- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#4413: int +- DeserializeToObject newInstance(class scala.Tuple2), obj#4412: scala.Tuple2 +- Project [cast(key#4479 as string) AS key#4403, cast(value#4480 as string) AS value#4404] +- LogicalRDD [key#4479, value#4480, topic#4481, partition#4482, offset#4483L, timestamp#4484, timestampType#4485] == Physical Plan == *SerializeFromObject [input[0, int, false] AS value#4414] +- *MapElements <function1>, obj#4413: int +- *DeserializeToObject newInstance(class scala.Tuple2), obj#4412: scala.Tuple2 +- *Project [cast(key#4479 as string) AS key#4403, cast(value#4480 as string) AS value#4404] +- Scan ExistingRDD[key#4479,value#4480,topic#4481,partition#4482,offset#4483L,timestamp#4484,timestampType#4485]
sbt.ForkMain$ForkError: org.scalatest.exceptions.TestFailedException:
Assert on query failed:
== Progress ==
StartStream(ProcessingTime(1),org.apache.spark.util.SystemClock@3db1548c,Map())
AssertOnQuery(<condition>, )
AddKafkaData(topics = Set(topic-22), data = WrappedArray(1, 2, 3), message = )
CheckAnswer: [2],[3],[4]
=> AssertOnQuery(<condition>, )
== Stream ==
Output Mode: Append
Stream state: {KafkaSource[Subscribe[topic-22]]: {"topic-22":{"2":0,"4":1,"1":1,"3":1,"0":1}}}
Thread state: alive
== Sink ==
0:
1: [2]
2: [3]
3: [4]
== Plan ==
== Parsed Logical Plan ==
SerializeFromObject [input[0, int, false] AS value#4414]
+- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#4413: int
+- DeserializeToObject newInstance(class scala.Tuple2), obj#4412: scala.Tuple2
+- Project [cast(key#4479 as string) AS key#4403, cast(value#4480 as string) AS value#4404]
+- LogicalRDD [key#4479, value#4480, topic#4481, partition#4482, offset#4483L, timestamp#4484, timestampType#4485]
== Analyzed Logical Plan ==
value: int
SerializeFromObject [input[0, int, false] AS value#4414]
+- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#4413: int
+- DeserializeToObject newInstance(class scala.Tuple2), obj#4412: scala.Tuple2
+- Project [cast(key#4479 as string) AS key#4403, cast(value#4480 as string) AS value#4404]
+- LogicalRDD [key#4479, value#4480, topic#4481, partition#4482, offset#4483L, timestamp#4484, timestampType#4485]
== Optimized Logical Plan ==
SerializeFromObject [input[0, int, false] AS value#4414]
+- MapElements <function1>, class scala.Tuple2, [StructField(_1,StringType,true), StructField(_2,StringType,true)], obj#4413: int
+- DeserializeToObject newInstance(class scala.Tuple2), obj#4412: scala.Tuple2
+- Project [cast(key#4479 as string) AS key#4403, cast(value#4480 as string) AS value#4404]
+- LogicalRDD [key#4479, value#4480, topic#4481, partition#4482, offset#4483L, timestamp#4484, timestampType#4485]
== Physical Plan ==
*SerializeFromObject [input[0, int, false] AS value#4414]
+- *MapElements <function1>, obj#4413: int
+- *DeserializeToObject newInstance(class scala.Tuple2), obj#4412: scala.Tuple2
+- *Project [cast(key#4479 as string) AS key#4403, cast(value#4480 as string) AS value#4404]
+- Scan ExistingRDD[key#4479,value#4480,topic#4481,partition#4482,offset#4483L,timestamp#4484,timestampType#4485]
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:315)
at org.apache.spark.sql.streaming.StreamTest$class.verify$1(StreamTest.scala:286)
at org.apache.spark.sql.streaming.StreamTest$$anonfun$testStream$1.apply(StreamTest.scala:438)
at org.apache.spark.sql.streaming.StreamTest$$anonfun$testStream$1.apply(StreamTest.scala:326)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35)
at org.apache.spark.sql.streaming.StreamTest$class.testStream(StreamTest.scala:326)
at org.apache.spark.sql.kafka010.KafkaSourceTest.testStream(KafkaSourceSuite.scala:40)
at org.apache.spark.sql.kafka010.KafkaSourceSuite$$anonfun$7.apply$mcV$sp(KafkaSourceSuite.scala:444)
at org.apache.spark.sql.kafka010.KafkaSourceSuite$$anonfun$7.apply(KafkaSourceSuite.scala:428)
at org.apache.spark.sql.kafka010.KafkaSourceSuite$$anonfun$7.apply(KafkaSourceSuite.scala:428)
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:40)
at org.scalatest.BeforeAndAfterEach$class.runTest(BeforeAndAfterEach.scala:255)
at org.apache.spark.sql.kafka010.KafkaSourceTest.runTest(KafkaSourceSuite.scala:40)
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.tools.Framework.org$scalatest$tools$Framework$$runSuite(Framework.scala:357)
at org.scalatest.tools.Framework$ScalaTestTask.execute(Framework.scala:502)
at sbt.ForkMain$Run$2.call(ForkMain.java:296)
at sbt.ForkMain$Run$2.call(ForkMain.java:286)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)