raghav-reglobe opened a new pull request, #70573: URL: https://github.com/apache/doris/pull/70573
### Background A routine load job reads its data source (a Kafka topic or a Kinesis stream) through tasks. The FE's `RoutineLoadTaskScheduler` takes a task from its queue, asks whether the source has anything new for it (for Kafka, the latest offsets, fetched through a BE), and either submits the task to a BE or puts it back in the queue. To avoid many small transactions it holds a task back for `max_batch_interval` when the task's last run ended at EOF. ### What problem does this PR solve? Issue Number: close #70568 Related PR: None PR Type: Bugfix Problem Summary: A routine load task that has nothing to consume and has not committed yet asks the data source for its offsets on every scheduler pass, 1 ms apart: several hundred lookups a second through a BE to the broker, for as long as nothing arrives. It happens for a new job on an empty topic and again after every resume with nothing new. The scheduler now marks such a task for delay before it puts it back, so it looks again once per `max_batch_interval`, as a task that ended at EOF already does. After an upgrade an idle job stops the lookups; a message that arrives at a job that has not committed yet is picked up within `max_batch_interval` and no longer within a millisecond. ### Details - **Frequency:** Always, for a job whose task finds nothing to consume before the task's first commit: a new job on a topic with nothing after its start offsets, and every job after a `RESUME` (and, as I read `KafkaRoutineLoadJob.divideRoutineLoadJob`, after a change of the master FE, because the tasks are made again) when nothing new has arrived. - **Root cause:** `RoutineLoadTaskScheduler.process()` holds a task back only if `needDedalySchedule()` is true, that is `delaySchedule || isEof`. Both flags are computed when a task commits (`RoutineLoadTaskInfo.handleTaskByTxnCommitAttachment`). A task made by `divideRoutineLoadJob` starts with both false and `lastScheduledTime = -1`. When `hasMoreDataToConsume()` is false the task goes back to the queue with neither flag set, so the next pass, one millisecond later, asks again. A task with nothing to consume never commits, so nothing ever sets a flag. - **Why existing tests missed it:** `RoutineLoadTaskSchedulerTest` never ran the scheduler over a task with nothing to consume, and the defect changes no result: it shows only in the counter `routine_load_get_meta_count` and in load on the BE and the broker. - **Fix:** in the branch of `scheduleOneTask` that handles "nothing to consume", call `setDelaySchedule(true)` before the task is put back. `scheduleOneTask` has just set `lastScheduledTime`, so the check that already exists in `process()` holds the task for `max_batch_interval`. The flag is computed again at the task's next commit. - **Other paths and branches:** Kinesis jobs use the same scheduler and get the same delay. Until its next commit a task delayed this way is treated as a task at EOF is today, in every path: a Kafka task that is renewed copies the flag (`unprotectRenewTask`), so a retry after a failed submit of such a task also waits for what is left of `max_batch_interval`. The job's own periodic checks of the source (a few lookups a minute, also while the job is paused) are not touched. - **Behavior change:** a task that finds nothing to consume before its first commit waits `max_batch_interval` before it looks again. So the first message that arrives at such a job is loaded up to `max_batch_interval` later than before (measured below: 0.6 to 0.7 s before, 5.6 s and 10.0 s after, with `max_batch_interval` = 10). After a commit this delay already applied. ### Testing Unit test, new: `RoutineLoadTaskSchedulerTest.testTaskWithNothingToConsumeWaitsForMaxBatchInterval` runs the scheduler five passes over a task as a new job makes it, with a data source that never has anything: the task is asked once and is then held; after `max_batch_interval` it is asked once more. Before the fix it is asked on every pass (by reading; I did not run the test against master). On this branch (master at ee73acc6131 plus this commit), on a build machine of ours: the FE builds, `mvn checkstyle:check` on fe-core passes, and the eleven routine load test classes pass: 121 tests, 0 failures, 0 errors. Manual test on a single-node cluster with a Kafka of its own (FE, BE and a one-broker Kafka 3.9 in one pod), built from a master snapshot of early November (1503e390275c) with unrelated patches, without and with this commit. One job, `max_batch_interval` = 10, lookups counted with the FE metric `routine_load_get_meta_count` over 40 s windows: | state of the job | without the fix | with the fix | |---|---|---| | new job on an empty topic | 622 a second | 18 a minute | | `RUNNING` at the end of the topic, after a commit | 18 a minute | 18 a minute | | `PAUSED` | 9 a minute | 6 a minute | | resumed with nothing new | 660 a second | 18 a minute | | a message sent to the idle job lands after | 0.7 s (new job), 0.6 s (resumed) | 5.6 s (new job), 10.0 s (resumed) | The issue has the same defect measured on a real cluster (353 a second on a new job, 246 a second after a resume). There is no regression suite in this PR: the effect is a lookup rate, which the suites under `regression-test/suites/load_p0/routine_load` have no way to read that I know of. Tell me if you want one and where it should look. ### Release note Fix: a routine load job with nothing to consume no longer asks Kafka for its offsets hundreds of times a second before its first commit (a new job on an empty topic, or a resumed job with nothing new). Such a job now looks again once per `max_batch_interval`. ### Check List (For Author) - Behavior changed: Yes (see Details) - Does this need documentation: No - Pre-submission status: Partially verified: upstream CI has not run on this head, and the manual run was on our own build line (a master snapshot of early November with unrelated patches), not on a build of this branch. ### AI Disclosure - Tool: Claude Code (Claude Fable 5.2). - Generated: the code, the test and this description. - Human review: The author directed the work and reviewed the test results. An automated multi-agent review of the code is running, and this line will carry its outcome before the PR is marked ready for review. The author did not read the code line by line. 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
