github-actions[bot] commented on code in PR #65166:
URL: https://github.com/apache/doris/pull/65166#discussion_r3523352582
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java:
##########
@@ -706,6 +706,7 @@ public class SchemaTable extends Table {
.column("CURRENT_ABORT_TASK_NUM",
ScalarType.createType(PrimitiveType.INT))
.column("IS_ABNORMAL_PAUSE",
ScalarType.createType(PrimitiveType.BOOLEAN))
.column("COMPUTE_GROUP",
ScalarType.createStringType())
+ .column("LAST_TASK_SCHEDULE_TIME",
ScalarType.createStringType())
Review Comment:
This exposes a master-only runtime value through a schema table that can be
served by a follower. `lastTaskScheduleTimeMs` is transient and only
`RoutineLoadTaskScheduler` updates it; that scheduler is started from
`Env.startMasterOnlyDaemonThreads()`. But `SchemaScanNode.finalizeForNereids()`
routes only the `tables` schema table to the master FE, so `routine_load_jobs`
still fetches from the local FE. A user connected to a follower can therefore
see `LAST_TASK_SCHEDULE_TIME = ''` even after the master has scheduled the job,
while `SHOW ROUTINE LOAD` is forwarded to the master and shows the timestamp.
Please either route this schema table to the master when exposing this field,
or persist/replay the timestamp so follower reads are authoritative.
##########
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)
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]
