Doris-Breakwater commented on issue #70568: URL: https://github.com/apache/doris/issues/70568#issuecomment-6850450463
Breakwater-GitHub-Analysis-Slot: slot_a3d75383692c The reported loop is supported by the source: this is a missing backoff on the FE's no-data scheduling path. The proposed `setDelaySchedule(true)` before requeueing is a reasonable minimal fix to submit with regression coverage. This assessment is based on source inspection; I have not reproduced the workload, built a patch, or run tests. **Verified in code** I checked `1503e390275c7c2bcf814907776491b8cbf3ccc0` and `ee73acc6131e6c6c752eb59e252404590be6ad77`. There are no differences between them in the routine-load implementation, its tests, or `KafkaUtil.java`. 1. `divideRoutineLoadJob()` creates tasks with `lastScheduledTime = -1` and `isEof = false`; `delaySchedule` defaults to false. The production task scheduler runs with a 1 ms daemon interval. Its interval gate applies only when `delaySchedule || isEof` is true. On a false result from `hasMoreDataToConsume()`, it requeues the same task without changing either flag. This return occurs before BE allocation and `beginTxn()`, so this idle task cannot obtain a transaction attachment that would change its flags. See [task creation](https://github.com/apache/doris/blob/ee73acc6131e6c6c752eb59e252404590be6ad77/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java#L252-L280) and [scheduler gate and requeue](https://github.com/apache/doris/blob/ee73acc6131e6c6c752eb59e252404590be6ad77/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java#L75-L166). 2. At the end of all assigned partitions, `KafkaRoutineLoadJob.hasMoreDataToConsume()` refreshes latest offsets and returns false again. The cache has no time-based freshness gate here. The request reaches the BE's latest-offset handler and ultimately calls `query_watermark_offsets()` for each requested partition. Thus this path really does generate repeated external metadata work, rather than just repeated local cache checks. See [FE offset check](https://github.com/apache/doris/blob/ee73acc6131e6c6c752eb59e252404590be6ad77/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java#L974-L1058) and [BE watermark queries](https://github.com/apache/doris/blob/1503e390275c7c2bcf814907776491b8cbf3ccc0/be/src/load/routine_load/data_consumer.cpp#L479-L512). 3. Pause/resume clears the job's task list and creates fresh tasks, explaining recurrence even after the job previously consumed rows. More generally, the vulnerable state is **any queued task with both flags false and no available data**, not just a job that has never loaded a row. Recurrence after FE leadership changes still needs runtime verification; this investigation does not establish that every failover follows the same path. **Fix considerations** Setting `delaySchedule` in the no-data branch uses the existing timestamp and interval gate without delaying the initial check of a fresh task. Repeated empty checks would then be separated by `max_batch_interval`, measured from the preceding scheduling attempt's start, subject to scheduler availability. There is one qualification to “a topic with data is not slowed”: data arriving after an empty check can wait until the interval expires. Once the task runs, attachment handling recomputes `delaySchedule` and `isEof`; renewal carries those values forward. Preserve the existing EOF and `read_committed` zero-row-with-lag behavior. See [flag recomputation](https://github.com/apache/doris/blob/ee73acc6131e6c6c752eb59e252404590be6ad77/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskInfo.java#L165-L204) and [renewal](https://github.com/apache/doris/blob/ee73acc6131e6c6c752eb59e252404590be6ad77/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java#L418-L429). The change would throttle metadata requests; the scheduler would still rotate delayed tasks through its queue at the existing daemon cadence. **Suggested PR validation** - Add a scheduler test with a real task whose flags initially are false. After one no-data result, verify requeueing and no transaction/BE submission; repeated passes before the deadline must not invoke the data check or advance `lastScheduledTime`. - Move the timestamp beyond the deadline and verify exactly one new check, including another empty result. Avoid wall-clock sleeps in the unit test. - Verify a fresh task with available data schedules immediately, and an idle task resumes ingestion after its deadline. Check attachment recomputation and renewal alongside the existing EOF and `read_committed` delay tests. - Repeat the empty-topic and pause/resume reproductions, then publish data and confirm progress. Include multiple tasks/jobs and FE leadership change as integration checks. **Remaining evidence and metric interpretation** The report already provides enough detail for a targeted PR. For before/after validation, please include sanitized `SHOW ROUTINE LOAD` / task output with effective `max_batch_interval`, partition count, task concurrency and progress, plus timestamped metadata-count, failure-count and latency-counter deltas under the same workload. If errors or unexpected pauses occur, include the corresponding FE/BE log excerpts. A query profile is not needed for this scheduler path. `routine_load_get_meta_count` is shared by Kafka metadata operations; it is not a per-job or exact broker-request counter. Separately, the job scheduler refreshes lag for every non-final job, including PAUSED jobs, at its default 10-second cadence. That supplies a concrete explanation for the small paused baseline and means acceptance should be removal of the rapid task-driven requests, rather than a counter that stays flat. See [counter accounting](https://github.com/apache/doris/blob/ee73acc6131e6c6c752eb59e252404590be6ad77/fe/fe-core/src/main/java/org/apache/doris/datasource/kafka/KafkaUtil.java#L233-L317) and [lag refresh eligibility](https://github.com/apache/doris/blob/ee73acc6131e6c6c752eb59e252404590be6ad77/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java#L910-L921). The measured rates and CPU cost remain reporter observations, not independently reproduced results. -- 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]
