raghav-reglobe opened a new issue, #70568: URL: https://github.com/apache/doris/issues/70568
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/doris/issues?q=is%3Aissue) and found no similar issues. ### Version Observed on a build of master at 1503e390275c (the FE reports `doris-4.1.0`), cloud mode, with patches that do not touch `fe/fe-core/src/main/java/org/apache/doris/load/routineload`. The lines involved are the same on master at ee73acc6131e (2026-11-13): `RoutineLoadTaskScheduler.java:105-116` and `:157-160`, `kafka/KafkaRoutineLoadJob.java:265-266`, `RoutineLoadTaskInfo.java:165-167` and `:183-186`. ### What's Wrong? A Kafka routine load job whose task has nothing to consume, and has not committed yet, asks Kafka for the latest offsets on every pass of the task scheduler, with no delay. Each lookup goes FE → BE → broker. Measured with the master FE's counter `routine_load_get_meta_count`: | state of the job | lookups | |---|---| | new job on an empty topic, never loaded a row | 13,781 in the first 39 s (about 353 a second, 1.08 ms each) | | a job that had loaded 23 rows, after `PAUSE` and `RESUME` with nothing new in the topic | 3,847 in about 16 s (246 a second at the end) | | the same job `RUNNING` at the end of the topic after a task had committed | 5 to 19 a minute | | the same job `PAUSED` | 5 to 8 a minute | It does not stop by itself: it lasts until a message arrives and a task commits. One job cost little in my test (about 0.1 core on the two BEs together), but every job in that state does it, and it comes back after every resume and, as I read the code, after every change of the master FE, because the tasks are made again. Why, as I read it: 1. A task made when the job starts or resumes is created with `isEof = false`, `delaySchedule = false` (`KafkaRoutineLoadJob.java:265-266`). 2. `RoutineLoadTaskScheduler.process()` delays a task only if `needDedalySchedule()` is true, that is `delaySchedule || isEof` (`:105-116`). For this task it is false, so `scheduleOneTask` runs. 3. `hasMoreDataToConsume()` finds no offset below its cached latest offset and asks Kafka through a BE (`KafkaUtil.getLatestOffsets`). There is nothing new, so the task goes back to the queue (`:157-160`) with neither flag set. 4. The two flags are recomputed only when a task's transaction commits (`RoutineLoadTaskInfo.java:183-186`). A task with nothing to consume never commits, so step 2 repeats on the next pass, 1 ms later. The delay to `max_batch_interval` that the scheduler already has for a task that ended at EOF is the behaviour I expected here too; this branch does not reach it. ### What You Expected? A task that finds nothing to consume waits `max_batch_interval` before the next check, like a task that committed at EOF. ### How to Reproduce? 1. Create any table and an empty Kafka topic. 2. Create a routine load job on it with default batch properties. What I ran, names changed and the column list shortened: ```sql CREATE ROUTINE LOAD test_db.job_idle ON t COLUMNS(...) PROPERTIES ("format" = "json") FROM KAFKA ("kafka_broker_list" = "<brokers>", "kafka_topic" = "<an empty topic>", "property.group.id" = "job_idle", "property.kafka_default_offsets" = "OFFSET_BEGINNING"); ``` 3. On the master FE read the counter twice, ten seconds apart: ``` curl -s http://<master-fe>:8030/metrics | grep routine_load_get_meta_count ``` It rises by thousands. `PAUSE ROUTINE LOAD` stops it. 4. The second case: let the job load a few rows, `PAUSE` it, `RESUME` it with nothing new in the topic, and read the counter again. ### Anything Else? A fix I have not built or tested: in the branch at `RoutineLoadTaskScheduler.java:157-160`, mark the task for delayed scheduling before it goes back to the queue (`setDelaySchedule(true)`). `scheduleOneTask` has just set `lastScheduledTime`, so the existing check at `:108-113` then holds the task for `max_batch_interval`, and the flag is recomputed at the task's next commit, so a topic with data is not slowed. I can send it as a PR with a unit test. ### Are you willing to submit PR? - [x] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- 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]
