github-actions[bot] commented on code in PR #65166:
URL: https://github.com/apache/doris/pull/65166#discussion_r3523353961
##########
regression-test/suites/load_p0/routine_load/test_routine_load_job_info_system_table.groovy:
##########
@@ -139,9 +150,93 @@ suite("test_routine_load_job_info_system_table","p0") {
log.info("compute group res: ${computeGroupRes}".toString())
assertTrue(computeGroupRes.size() > 0)
assertNotNull(computeGroupRes[0][1])
+
+ sql """
+ CREATE TABLE IF NOT EXISTS ${scheduleTableName}
+ (
+ k00 INT NOT NULL,
+ k01 VARCHAR(32) NULL
+ )
+ DUPLICATE KEY(k00)
+ DISTRIBUTED BY HASH(k00) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ );
+ """
+
+ adminClient.createTopics([new NewTopic(scheduleTopic, 1, (short)
1)]).all().get()
+ producer = RoutineLoadTestUtils.createKafkaProducer(kafkaBroker)
+ RoutineLoadTestUtils.sendTestDataToKafka(producer,
[scheduleTopic], ["1|a"])
+ producer.flush()
+
+ sql """
+ CREATE ROUTINE LOAD ${scheduleJobName} on ${scheduleTableName}
+ COLUMNS(k00,k01),
+ COLUMNS TERMINATED BY "|"
+ PROPERTIES
+ (
+ "max_batch_interval" = "5",
+ "max_batch_rows" = "300000",
+ "max_batch_size" = "209715200"
+ )
+ FROM KAFKA
+ (
+ "kafka_broker_list" = "${kafkaBroker}",
+ "kafka_topic" = "${scheduleTopic}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING"
+ );
+ """
+
+ count = 0
+ def lastTaskScheduleTimeRes = sql """
+ SELECT JOB_NAME, LAST_TASK_SCHEDULE_TIME
+ FROM information_schema.routine_load_jobs
+ WHERE JOB_NAME = '${scheduleJobName}'
+ AND LAST_TASK_SCHEDULE_TIME != ''
+ """
+ while (lastTaskScheduleTimeRes.size() == 0) {
+ if (count >= 60) {
+ Assert.fail("LAST_TASK_SCHEDULE_TIME was not updated for
${scheduleJobName}")
+ }
+ def state = sql "show routine load for ${scheduleJobName}"
+ def rowCount = sql "SELECT count(*) FROM ${scheduleTableName}"
+ log.info("routine load state:
${state[0][8].toString()}".toString())
+ log.info("row count: ${rowCount}".toString())
+ log.info("last task schedule time res:
${lastTaskScheduleTimeRes}".toString())
+ sleep(1000)
+ lastTaskScheduleTimeRes = sql """
+ SELECT JOB_NAME, LAST_TASK_SCHEDULE_TIME
+ FROM information_schema.routine_load_jobs
+ WHERE JOB_NAME = '${scheduleJobName}'
+ AND LAST_TASK_SCHEDULE_TIME != ''
+ """
+ count++
+ }
+ log.info("last task schedule time res:
${lastTaskScheduleTimeRes}".toString())
+ assertTrue(lastTaskScheduleTimeRes.size() > 0)
+ assertNotNull(lastTaskScheduleTimeRes[0][1])
+ assertTrue(lastTaskScheduleTimeRes[0][1].toString().length() > 0)
+
+ // Verify SHOW ROUTINE LOAD also includes LastTaskScheduleTime
+ def showRoutineLoadRes = sql "SHOW ROUTINE LOAD FOR
${scheduleJobName}"
+ log.info("show routine load res: ${showRoutineLoadRes}".toString())
+ assertTrue(showRoutineLoadRes.size() > 0)
+ // LastTaskScheduleTime should be the last column (index 23, after
ComputeGroup at index 22)
+ def lastTaskScheduleTimeFromShow = showRoutineLoadRes[0][23]
+ log.info("LastTaskScheduleTime from SHOW ROUTINE LOAD:
${lastTaskScheduleTimeFromShow}".toString())
+ assertNotNull(lastTaskScheduleTimeFromShow)
+ assertTrue(lastTaskScheduleTimeFromShow.toString().length() > 0)
+ // Verify consistency between SHOW ROUTINE LOAD and system table
+ assertEquals(lastTaskScheduleTimeRes[0][1].toString(),
lastTaskScheduleTimeFromShow.toString())
Review Comment:
The exact equality here can race with the scheduler.
`lastTaskScheduleTimeRes` is captured from `information_schema`, then the test
runs a separate `SHOW ROUTINE LOAD`; meanwhile `scheduleOneTask()` updates the
job timestamp before `hasMoreDataToConsume()`, and the no-data path requeues
the task for later scheduling. That means a healthy job can advance
`LastTaskScheduleTime` between these two reads and fail this assertion even
though both paths are working. Please avoid comparing exact values across
separate live reads, or pause/stop the job and wait for scheduling to quiesce
before comparing a stable timestamp.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
