[ https://issues.apache.org/jira/browse/SPARK-38046?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Jungtaek Lim reassigned SPARK-38046: ------------------------------------ Assignee: Boyang Jerry Peng > Fix KafkaSource/KafkaMicroBatch flaky test due to non-deterministic timing > -------------------------------------------------------------------------- > > Key: SPARK-38046 > URL: https://issues.apache.org/jira/browse/SPARK-38046 > Project: Spark > Issue Type: Improvement > Components: Tests > Affects Versions: 3.2.0 > Reporter: Boyang Jerry Peng > Assignee: Boyang Jerry Peng > Priority: Major > > There is a test call "compositeReadLimit" > > [https://github.com/apache/spark/blob/master/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala#L460] > > that is flaky. The problem is because the Kakfa connector is always getting > the actual system time and not advancing it manually, thus leaving room for > non-deterministic behaviors especially since the source determines if > "maxTriggerDelayMs" is satisfied by comparing the last trigger time with the > current system time. One can simply "sleep" at points in the test to > generate different outcomes. > > Example output when test fails: > > {code:java} > - compositeReadLimit *** FAILED *** (7 seconds, 862 milliseconds) > == Results == > !== Correct Answer - 0 == == Spark Answer - 14 == > !struct<> struct<value:int> > ! [112] > ! [113] > ! [114] > ! [115] > ! [116] > ! [117] > ! [118] > ! [119] > ! [120] > ! [16] > ! [17] > ! [18] > ! [19] > ! [20] > > > == Progress == > > StartStream(ProcessingTimeTrigger(100),org.apache.spark.sql.streaming.util.StreamManualClock@30075210,Map(),null) > AssertOnQuery(<condition>, ) > CheckAnswer: > [1],[10],[100],[101],[102],[103],[104],[105],[106],[107],[11],[108],[109],[110],[111],[12],[13],[14],[15] > AdvanceManualClock(100) > AssertOnQuery(<condition>, ) > => CheckNewAnswer: > Assert(<condition>, ) > AdvanceManualClock(100) > AssertOnQuery(<condition>, ) > CheckAnswer: > [1],[10],[100],[101],[102],[103],[104],[105],[106],[107],[11],[108],[109],[110],[111],[112],[113],[114],[115],[116],[12],[117],[118],[119],[120],[121],[13],[14],[15],[16],[17],[18],[19],[2],[20],[21],[22],[23],[24] > AdvanceManualClock(100) > AssertOnQuery(<condition>, ) > CheckNewAnswer: > Assert(<condition>, ) > AdvanceManualClock(100) > AssertOnQuery(<condition>, ) > CheckAnswer: > [1],[10],[100],[101],[102],[103],[104],[105],[106],[107],[11],[108],[109],[110],[111],[112],[113],[114],[115],[116],[117],[118],[119],[12],[120],[121],[122],[123],[124],[125],[126],[127],[128],[13],[14],[15],[16],[17],[18],[19],[2],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30] > > == Stream == > Output Mode: Append > Stream state: {KafkaSourceV1[Subscribe[topic-41]]: > {"topic-41":{"2":1,"1":11,"0":21}}} > Thread state: alive > Thread stack trace: java.lang.Object.wait(Native Method) > org.apache.spark.util.ManualClock.waitTillTime(ManualClock.scala:67) > > org.apache.spark.sql.streaming.util.StreamManualClock.waitTillTime(StreamManualClock.scala:34) > > org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:76) > > org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:222) > > org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:350) > > org.apache.spark.sql.execution.streaming.StreamExecution$$Lambda$3081/1859014229.apply$mcV$sp(Unknown > Source) > scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23) > org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:857) > > org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:325) > > org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:252) > > > == Sink == > 0: [100] [101] [102] [103] [104] [105] [106] [107] [108] [109] [110] [111] > [1] [10] [11] [12] [13] [14] [15] > 1: [112] [113] [114] [115] [116] [117] [118] [119] [120] [16] [17] [18] > [19] [20] > > > == Plan == > == Parsed Logical Plan == > WriteToDataSourceV2 > org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@48d73c40 > +- SerializeFromObject [input[0, int, false] AS value#7455] > +- MapElements > org.apache.spark.sql.kafka010.KafkaMicroBatchSourceSuiteBase$$Lambda$6114/1322446561@4df69693, > class scala.Tuple2, [StructField(_1,StringType,true), > StructField(_2,StringType,true)], obj#7454: int > +- DeserializeToObject newInstance(class scala.Tuple2), obj#7453: > scala.Tuple2 > +- Project [cast(key#7429 as string) AS key#7443, cast(value#7430 > as string) AS value#7444] > +- Project [key#7501 AS key#7429, value#7502 AS value#7430, > topic#7503 AS topic#7431, partition#7504 AS partition#7432, offset#7505L AS > offset#7433L, timestamp#7506 AS timestamp#7434, timestampType#7507 AS > timestampType#7435] > +- LogicalRDD [key#7501, value#7502, topic#7503, > partition#7504, offset#7505L, timestamp#7506, timestampType#7507], true > > == Analyzed Logical Plan == > WriteToDataSourceV2 > org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@48d73c40 > +- SerializeFromObject [input[0, int, false] AS value#7455] > +- MapElements > org.apache.spark.sql.kafka010.KafkaMicroBatchSourceSuiteBase$$Lambda$6114/1322446561@4df69693, > class scala.Tuple2, [StructField(_1,StringType,true), > StructField(_2,StringType,true)], obj#7454: int > +- DeserializeToObject newInstance(class scala.Tuple2), obj#7453: > scala.Tuple2 > +- Project [cast(key#7429 as string) AS key#7443, cast(value#7430 > as string) AS value#7444] > +- Project [key#7501 AS key#7429, value#7502 AS value#7430, > topic#7503 AS topic#7431, partition#7504 AS partition#7432, offset#7505L AS > offset#7433L, timestamp#7506 AS timestamp#7434, timestampType#7507 AS > timestampType#7435] > +- LogicalRDD [key#7501, value#7502, topic#7503, > partition#7504, offset#7505L, timestamp#7506, timestampType#7507], true > > == Optimized Logical Plan == > WriteToDataSourceV2 > org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@48d73c40 > +- SerializeFromObject [input[0, int, false] AS value#7455] > +- MapElements > org.apache.spark.sql.kafka010.KafkaMicroBatchSourceSuiteBase$$Lambda$6114/1322446561@4df69693, > class scala.Tuple2, [StructField(_1,StringType,true), > StructField(_2,StringType,true)], obj#7454: int > +- DeserializeToObject newInstance(class scala.Tuple2), obj#7453: > scala.Tuple2 > +- Project [cast(key#7501 as string) AS key#7443, cast(value#7502 > as string) AS value#7444] > +- LogicalRDD [key#7501, value#7502, topic#7503, > partition#7504, offset#7505L, timestamp#7506, timestampType#7507], true > > == Physical Plan == > WriteToDataSourceV2 > org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@48d73c40, > org.apache.spark.sql.execution.datasources.v2.DataSourceV2Strategy$$Lambda$3704/659759030@5d8e6616 > +- *(1) SerializeFromObject [input[0, int, false] AS value#7455] > +- *(1) MapElements > org.apache.spark.sql.kafka010.KafkaMicroBatchSourceSuiteBase$$Lambda$6114/1322446561@4df69693, > obj#7454: int > +- *(1) DeserializeToObject newInstance(class scala.Tuple2), > obj#7453: scala.Tuple2 > +- *(1) Project [cast(key#7501 as string) AS key#7443, > cast(value#7502 as string) AS value#7444] > +- *(1) Scan ExistingRDD > kafka[key#7501,value#7502,topic#7503,partition#7504,offset#7505L,timestamp#7506,timestampType#7507] > (StreamTest.scala:466) > org.scalatest.exceptions.TestFailedException: > at > org.scalatest.Assertions.newAssertionFailedException(Assertions.scala:530) > at > org.scalatest.Assertions.newAssertionFailedException$(Assertions.scala:529) > at org.scalatest.FunSuite.newAssertionFailedException(FunSuite.scala:1560) > at org.scalatest.Assertions.fail(Assertions.scala:1091) > at org.scalatest.Assertions.fail$(Assertions.scala:1087) > at org.scalatest.FunSuite.fail(FunSuite.scala:1560) > at > org.apache.spark.sql.streaming.StreamTest.failTest$1(StreamTest.scala:466) > at org.apache.spark.sql.streamin2022-01-07 10:25:45,899 INFO logger - > [Controller id=0 epoch=1] Changed partition topic-42-0 from NewPartition to > OnlinePartition with state LeaderAndIsr(leader=0, leaderEpoch=0, isr=List(0), > zkVersion=0){code} -- This message was sent by Atlassian Jira (v8.20.1#820001) --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org