[raghav-reglobe]

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 Kafka routine load task that has not run since it was made and finds nothing to consume asks the broker for its offsets on every scheduler pass, 1 ms apart: several hundred lookups a second through a BE, 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 as waiting for data and looks at it again once per max_batch_interval, as it already does for a task that ended at EOF. The mark comes off as soon as a look finds data.

Details

• Frequency: Always, for a task that has not run since divideRoutineLoadJob made it and that finds nothing to consume: a new job with nothing after its start offsets (a new job starts at the END of its topic by default, so that is any topic where nothing arrives right away), and every job after a RESUME when nothing new has arrived. By reading, it is the same whenever a job's tasks are made again: an auto resume, a change of the topic's partitions on a running job, a change of the master FE. I have not run those three.
• Root cause: RoutineLoadTaskScheduler.process() holds a task back only if needDedalySchedule() is true, that is delaySchedule || isEof. Both flags are set when a task's transaction ends: at a commit (RoutineLoadTaskInfo.handleTaskByTxnCommitAttachment), and delaySchedule also on the task that renews an aborted one. A task made by divideRoutineLoadJob starts with both false. When hasMoreDataToConsume() is false it 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:
  • RoutineLoadTaskInfo gets a state of its own, waitingForData. delaySchedule and isEof are not touched: the end of a task's transaction decides them, and Kafka uses delaySchedule for the read-committed case.
  • scheduleOneTask sets it when a task that has never been scheduled (lastScheduledTime == -1, as divideRoutineLoadJob makes it) finds nothing to consume, and takes it off as soon as a look finds data.
  • process() holds a task that is at EOF, delayed or waiting for data until max_batch_interval has passed since it was last scheduled.
  • A marked task is never renewed: it has no transaction and was never submitted. A task made by a renewal carries the earlier task's time, so it is never marked. (The field's comment and one test name speak of a task that renews a marked one; that case cannot occur.)
• What this PR does not change, on purpose: a task that renews one whose commit was not at EOF, and that then finds nothing, still asks on every pass until data arrives. test_routine_load_eof expects such a task to be scheduled again as soon as data arrives (at least 20 commits in 30 s at one message a second), which leaves no room for a wait of max_batch_interval there. The first version of this PR (head 92dc170) held that task as well and failed that suite's scenario when I ran it (5 commits; table below). The lookups in that case are measured below. If you want them bounded too, a short fixed wait on that path would do it; I left it out because it changes how fast a busy job picks up its next batch, which is yours to decide. This PR closes the issue although that case remains; say if you want the issue kept open for it.
• Other paths and branches:
  • Kinesis: a Kinesis task never reaches the changed branch, because KinesisRoutineLoadJob.hasMoreDataToConsume always returns true.
  • A task that is renewed after a timeout or a failed submit carries the earlier task's time, so it is never marked; it also keeps offsets at which a look already found data.
  • The job's own periodic checks of the source are not touched: every 10 s one partition lookup for a job that is RUNNING or NEED_SCHEDULE and one lag lookup for every job that is not stopped or cancelled, a PAUSED one included. That is why an idle job still makes 12 lookups a minute plus one per max_batch_interval.
  • process() now passes a held task whose job has been removed on to scheduleOneTask, which abandons it. Before, the missing job failed the wait check, and the task was dropped with an "interrupted" log and a stack trace.
• Behavior change: a task that has not run since it was made and finds nothing waits max_batch_interval before it looks again. So a message that arrives at a new or resumed job after the job's first empty look is loaded up to max_batch_interval later than before (measured below: 0.6 to 0.7 s before, 9.0 s and 8.8 s after, with max_batch_interval = 10). A job whose last commit was at EOF already waited that long before this change. The same wait applies to a task that is made while a job runs (after a change of the topic's partitions, an auto resume or a change of the master FE) when its first look finds nothing. While a task is held, its own checks run at each look, once per max_batch_interval, and no longer on every pass: the check of the job's compute group, and the data check that also notices an unreachable broker. So an idle job is paused for such a fault up to one interval later than before, and a fault that clears within the hold does not pause it. For a job that names no partitions, the job's own check every 10 s still logs the fault and raises routine_load_get_meta_fail_count.
• One suite changed: test_routine_load_task_exception_recovery creates its job on an empty topic at the default max_batch_interval of 60 s and sends afterwards. When the job's first look falls before the send, the task now waits 60 s, and the suite's wait of about 62 s can run out. It gets "max_batch_interval" = "5". The local review read every suite that creates a routine load job (73 files) for the same pattern and found no other one that can fail from it.

Testing

Unit tests, in RoutineLoadTaskSchedulerTest (none sleeps, none can block on an empty queue):

• testNewTaskWithNothingToConsumeWaitsForMaxBatchInterval: a task as a new job makes it, with nothing to consume, is asked once in five passes and then held; a held pass does not move its time; after max_batch_interval it is asked once more; delaySchedule and isEof stay false; no BE is asked for a slot.
• testWaitingTaskGoesOnWhenDataArrives: data that arrives while the task waits is seen at its next look; the mark comes off; with no BE free the task is then tried on every pass.
• testNewTaskWithDataIsNotMarked.
• testTaskRenewedAfterCommitNotAtEofIsNotHeld: a task that carries an earlier task's time and has neither flag is asked on every pass.
• testTaskRenewingWaitingTaskStartsWithoutMark.
• testWaitingTaskOfRemovedJobIsAbandoned.

On this branch, on a build machine of ours: the FE builds, mvn checkstyle:check on fe-core reports 0 violations, and the eleven routine load test classes pass: 126 tests, 0 failures, 0 errors. The six new tests use the new field, so they do not compile on master or on the first version of this PR; I did not run them there in any other form.

Manual tests on single-node clusters with a Kafka of their own (FE, BE and a one-broker Kafka 3.9 in one pod), built from a master snapshot of early November (1503e39) with unrelated patches: without the change, with the first version of this PR, and with this version.

Lookups, counted with the FE metric routine_load_get_meta_count over 40 s windows (20 s for PAUSED), one job, max_batch_interval = 10:

state of the job	without the change	with this PR
new job on an empty topic	622 a second	19 a minute
RUNNING at the end of the topic, after a commit	18 a minute	18 a minute
PAUSED	3 in 20 s	2 in 20 s
resumed with nothing new	660 a second	17 a minute
a message sent to the idle job lands after	0.7 s (new job), 0.6 s (resumed)	9.0 s (new job), 8.8 s (resumed)

The PAUSED row is the job's own lag check and is not changed by this PR.

The scenario of test_routine_load_eof (one message a second, max_batch_interval = 5; the debug point RoutineLoadTaskInfo.judgeEof makes every commit count as not at EOF), run by hand with the suite's three windows:

	without the change	first version of this PR	this version	the suite needs
60 s as it is, commits	13	13	13	at most 20
30 s with the debug point, commits	25	5	21	at least 20
60 s with it off, commits	13	13	14	at most 20
the 30 s window four more times	14, 22, 24, 20	5, 6, 5, 6	20, 23, 25, 23
after a commit not at EOF, nothing arriving: lookups	751 a second	0.4 a second	739 a second

The last row is the case this PR leaves as it is. The fourth row shows that the suite's bound is close to what this scenario gives on my setup even without the change.

I have not run the regression suites themselves; I have no setup for the regression framework. A suite for the lookup rate is possible, since test_routine_load_metrics already reads this metric from the FE. Tell me if you want one in this PR.

Release note

Fix: a Kafka routine load job whose task has not run yet and has nothing to consume (a new job with nothing after its start offsets, or a resumed job with nothing new) no longer asks the broker for its offsets hundreds of times a second. Such a task now looks again once per max_batch_interval, so data that arrives after its first empty look is loaded up to max_batch_interval later (60 s by default).

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, the regression suites were not run, and the manual runs were 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); the local review ran on Claude Opus 5.5.
• Generated: the code, the tests and this description.
• Human review: The author directed the work and reviewed the description, the review findings and the test results. The first version was reviewed by an automated multi-agent review together with the author, who did not read the code line by line; that review found the problem with the task renewed after a commit, which this version removes. A second review of the same kind, of this version, ended in an approval with no blocker and no major finding; its findings on wording are taken into this description, and its findings on test coverage and one optional improvement are in its comment below.

🤖 Generated with Claude Code

[raghav-reglobe]

Local pipeline review — ✅ PASS

schema: doris-repo-review/v1
status: PASS
pr: apache/doris#70573
commit: 8a371547fd64b49067d61f916ee5d1f1d7aa03b2
base: 89d4f2826e2c9edbba01f5f70fc1d243e89f17b5
reviewed_at: 2026-11-14T09:21+05:30
reviewer: raghav-reglobe
model: claude-opus-5-5
effort: xhigh
findings: {blocker: 0, major: 0, minor: 4, nit: 4}
rounds: 1
converged: true

Notes for maintainers

• F-05 (observability, a behaviour change not declared at the reviewed commit; the PR description now declares it) fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java:174-176 — while a new task is held for lack of data, its own checks (the job's compute group, and the data check that notices an unreachable broker) run at each look, once per max_batch_interval, not on every pass; so an idle job is paused for such a fault up to one interval later, and a fault that clears within the hold does not pause it. No row is delayed or lost by it. Checked by reading, not run.
• F-01 fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java:164-165 — the wait also applies when a job's tasks are made again by an auto resume or a change of the topic's partitions, and a new job starts at the end of its topic by default, so any quiet topic gives the empty first look; the release note did not name the cost (data that arrives after the first empty look loads up to max_batch_interval later). The description and its release note are corrected; the code comment still names three triggers only.
• F-02 fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java:109-114 — optional: a task held only for lack of data could be released from the job's offset cache, which the job refreshes every 10 s, with no extra lookup; that would cut the wait from one interval to about 10 s.
• F-03, F-04, F-08 fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadTaskSchedulerTest.java:361-369 — test gaps: no test runs the two production builders that set and carry lastScheduledTime == -1, on which the fix keys; the "held pass does not move the time" check can pass within one millisecond; no test runs the edited gate with a task held by isEof or delaySchedule.
• F-06, F-07 — close #70568 closes the issue while one case the description names stays as it is (a task renewed after a commit that was not at EOF still asks on every pass); and the new field's comment and one test name describe a renewal of a marked task, which cannot occur, and say a commit decides delaySchedule, which an abort renewal also sets.

_{Reviewed locally with the doris-repo-review pipeline. Repository policy may accept this receipt for the matching commit; it is not a human Apache approval.}