This is an automated email from the ASF dual-hosted git repository.
sollhui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new cd0efa38f72 [fix](job) Avoid adaptive batching for Kafka tasks with
low lag (#66099)
cd0efa38f72 is described below
commit cd0efa38f7224acbe83ebee03b99535fdfb3b6dc
Author: hui lai <[email protected]>
AuthorDate: Tue Aug 4 17:55:41 2026 +0800
[fix](job) Avoid adaptive batching for Kafka tasks with low lag (#66099)
### What problem does this PR solve?
Kafka Routine Load previously enabled adaptive batching whenever a task
was not at EOF. For a continuously written but lightly backlogged topic,
this could repeatedly increase a short user-configured batch interval to
the adaptive interval, delaying offset commits and increasing ingestion
latency.
Using bytes or observed processing speed is not a reliable admission
criterion. Records may be small while the topic contains many pending
records, and processing may also slow down because the cluster is under
pressure.
This PR changes Kafka adaptive batching to use task-level Kafka lag:
- Calculate the pending offset lag only for the partitions assigned to
the current task.
- Enable adaptive batching only when the aggregate task lag is greater
than the effective adaptive row limit: `max(max_batch_rows,
RoutineLoadJob.DEFAULT_MAX_BATCH_ROWS)`.
- Keep the configured batch parameters when lag is unknown,
non-positive, or no greater than that threshold.
- Ignore lag from partitions that are not assigned to the task.
- Capture the adaptive decision and
`routine_load_adaptive_min_batch_interval_sec` once in
`updateAdaptiveTimeout()`, before `beginTxn()`.
- Reuse the captured values when constructing the Thrift task so that
the transaction timeout and BE batch parameters remain consistent during
the same scheduling attempt.
- When adaptive batching is enabled, retain the existing behavior of
raising the limits to at least `RoutineLoadJob.DEFAULT_MAX_BATCH_ROWS`
and `RoutineLoadJob.DEFAULT_MAX_BATCH_SIZE`.
---
.../routineload/kafka/KafkaRoutineLoadJob.java | 19 +++
.../load/routineload/kafka/KafkaTaskInfo.java | 16 ++-
.../load/routineload/KafkaRoutineLoadJobTest.java | 160 +++++++++++++++++++++
.../test_routine_load_adaptive_param.groovy | 16 ++-
4 files changed, 200 insertions(+), 11 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
index 8b824747979..502202aa8cc 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
@@ -914,6 +914,25 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
}
}
+ boolean isTaskLagGreaterThanMaxBatchRows(Map<Integer, Long>
partitionIdToOffset) {
+ long remainingRows = Math.max(maxBatchRows,
RoutineLoadJob.DEFAULT_MAX_BATCH_ROWS);
+ for (Map.Entry<Integer, Long> entry : partitionIdToOffset.entrySet()) {
+ Long latestOffset =
cachedPartitionWithLatestOffsets.get(entry.getKey());
+ if (latestOffset == null) {
+ continue;
+ }
+ long partitionLag = latestOffset - entry.getValue();
+ if (partitionLag <= 0) {
+ continue;
+ }
+ if (partitionLag > remainingRows) {
+ return true;
+ }
+ remainingRows -= partitionLag;
+ }
+ return false;
+ }
+
// check if given partitions has more data to consume.
// 'partitionIdToOffset' to the offset to be consumed.
public boolean hasMoreDataToConsume(UUID taskId, Map<Integer, Long>
partitionIdToOffset) throws UserException {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaTaskInfo.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaTaskInfo.java
index 21aec901571..98aabe63926 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaTaskInfo.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaTaskInfo.java
@@ -23,6 +23,7 @@ import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Table;
import org.apache.doris.common.Config;
import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.common.util.DebugUtil;
import org.apache.doris.load.routineload.RLTaskTxnCommitAttachment;
import org.apache.doris.load.routineload.RoutineLoadJob;
@@ -60,6 +61,9 @@ public class KafkaTaskInfo extends RoutineLoadTaskInfo {
// <partitionId, offset to be consumed>
private Map<Integer, Long> partitionIdToOffset;
+ private int adaptiveMinBatchInterval;
+ private boolean isAdaptiveBatch;
+
public KafkaTaskInfo(UUID id, long jobId,
long timeoutMs, Map<Integer, Long>
partitionIdToOffset, boolean isMultiTable,
long lastScheduledTime, boolean isEof) {
@@ -126,9 +130,13 @@ public class KafkaTaskInfo extends RoutineLoadTaskInfo {
@Override
public void updateAdaptiveTimeout(RoutineLoadJob routineLoadJob) {
- if (!isEof) {
+ adaptiveMinBatchInterval =
Config.routine_load_adaptive_min_batch_interval_sec;
+ KafkaRoutineLoadJob kafkaRoutineLoadJob = (KafkaRoutineLoadJob)
routineLoadJob;
+ isAdaptiveBatch =
DebugPointUtil.isEnable("KafkaTaskInfo.shouldUseAdaptiveBatch")
+ ||
kafkaRoutineLoadJob.isTaskLagGreaterThanMaxBatchRows(partitionIdToOffset);
+ if (isAdaptiveBatch) {
long maxBatchIntervalS =
Math.max(routineLoadJob.getMaxBatchIntervalS(),
- Config.routine_load_adaptive_min_batch_interval_sec);
+ adaptiveMinBatchInterval);
long timeoutSec = maxBatchIntervalS *
Config.routine_load_task_timeout_multiplier;
long realTimeoutSec = Math.max(timeoutSec,
Config.routine_load_task_min_timeout_sec);
this.timeoutMs = realTimeoutSec * 1000;
@@ -141,8 +149,8 @@ public class KafkaTaskInfo extends RoutineLoadTaskInfo {
long maxBatchIntervalS = routineLoadJob.getMaxBatchIntervalS();
long maxBatchRows = routineLoadJob.getMaxBatchRows();
long maxBatchSize = routineLoadJob.getMaxBatchSizeBytes();
- if (!isEof) {
- maxBatchIntervalS = Math.max(maxBatchIntervalS,
Config.routine_load_adaptive_min_batch_interval_sec);
+ if (isAdaptiveBatch) {
+ maxBatchIntervalS = Math.max(maxBatchIntervalS,
adaptiveMinBatchInterval);
maxBatchRows = Math.max(maxBatchRows,
RoutineLoadJob.DEFAULT_MAX_BATCH_ROWS);
maxBatchSize = Math.max(maxBatchSize,
RoutineLoadJob.DEFAULT_MAX_BATCH_SIZE);
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
index ebdad39a274..77570bb0309 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
@@ -46,6 +46,7 @@ import
org.apache.doris.nereids.trees.plans.commands.load.LoadProperty;
import org.apache.doris.nereids.trees.plans.commands.load.LoadSeparator;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.thrift.TResourceInfo;
+import org.apache.doris.thrift.TRoutineLoadTask;
import com.google.common.base.Joiner;
import com.google.common.collect.Lists;
@@ -390,6 +391,165 @@ public class KafkaRoutineLoadJobTest {
}
}
+ @Test
+ public void testAdaptiveBatchUsesTaskLagThreshold() {
+ RoutineLoadManager routineLoadManager =
Mockito.mock(RoutineLoadManager.class);
+ Env env = Mockito.mock(Env.class);
+ int previousAdaptiveIntervalSec =
Config.routine_load_adaptive_min_batch_interval_sec;
+
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+ Config.routine_load_adaptive_min_batch_interval_sec = 360;
+ envStatic.when(Env::getCurrentEnv).thenReturn(env);
+
Mockito.when(env.getRoutineLoadManager()).thenReturn(routineLoadManager);
+
+ KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L,
"kafka_routine_load_job", 1L,
+ 1L, "127.0.0.1:9020", "topic1", UserIdentity.ADMIN);
+ Deencapsulation.setField(routineLoadJob, "maxBatchIntervalS", 20L);
+ Deencapsulation.setField(routineLoadJob, "maxBatchRows", 200000L);
+ Deencapsulation.setField(routineLoadJob, "maxBatchSizeBytes", 100L
* 1024 * 1024);
+
Mockito.when(routineLoadManager.getJob(1L)).thenReturn(routineLoadJob);
+
+ Map<Integer, Long> taskProgress = Maps.newHashMap();
+ taskProgress.put(1, 10L);
+ taskProgress.put(2, 20L);
+
+ KafkaTaskInfo taskWithUnknownLag = new KafkaTaskInfo(new UUID(1,
1), 1L, 20000,
+ taskProgress, false, 1000, false);
+ taskWithUnknownLag.updateAdaptiveTimeout(routineLoadJob);
+ TRoutineLoadTask unknownLagThriftTask = new TRoutineLoadTask();
+ Deencapsulation.invoke(
+ taskWithUnknownLag, "adaptiveBatchParam",
unknownLagThriftTask, routineLoadJob);
+ Assert.assertEquals(20L, unknownLagThriftTask.getMaxIntervalS());
+
+ Map<Integer, Long> latestOffsets = Maps.newHashMap();
+ latestOffsets.put(1, 10_000_010L);
+ latestOffsets.put(2, 10_000_020L);
+ latestOffsets.put(3, 100_000_000L);
+ Deencapsulation.setField(routineLoadJob,
"cachedPartitionWithLatestOffsets", latestOffsets);
+
+ KafkaTaskInfo taskAtLagThreshold = new KafkaTaskInfo(new UUID(1,
2), 1L, 20000,
+ taskProgress, false, 1000, false);
+ taskAtLagThreshold.updateAdaptiveTimeout(routineLoadJob);
+ latestOffsets.put(2, 10_000_021L);
+ TRoutineLoadTask thresholdThriftTask = new TRoutineLoadTask();
+ Deencapsulation.invoke(
+ taskAtLagThreshold, "adaptiveBatchParam",
thresholdThriftTask, routineLoadJob);
+ Assert.assertEquals(20L, thresholdThriftTask.getMaxIntervalS());
+ Assert.assertEquals(200000L,
thresholdThriftTask.getMaxBatchRows());
+ Assert.assertEquals(100L * 1024 * 1024,
thresholdThriftTask.getMaxBatchSize());
+ Assert.assertEquals(routineLoadJob.getTimeout() * 1000L,
taskAtLagThreshold.getTimeoutMs());
+
+ KafkaTaskInfo taskAboveLagThreshold = new KafkaTaskInfo(new
UUID(1, 3), 1L, 20000,
+ taskProgress, false, 1000, true);
+ taskAboveLagThreshold.updateAdaptiveTimeout(routineLoadJob);
+ TRoutineLoadTask adaptiveThriftTask = new TRoutineLoadTask();
+ Deencapsulation.invoke(
+ taskAboveLagThreshold, "adaptiveBatchParam",
adaptiveThriftTask, routineLoadJob);
+ Assert.assertEquals(360L, adaptiveThriftTask.getMaxIntervalS());
+ Assert.assertEquals(RoutineLoadJob.DEFAULT_MAX_BATCH_ROWS,
+ adaptiveThriftTask.getMaxBatchRows());
+ Assert.assertEquals(RoutineLoadJob.DEFAULT_MAX_BATCH_SIZE,
+ adaptiveThriftTask.getMaxBatchSize());
+ Assert.assertEquals(360L *
Config.routine_load_task_timeout_multiplier * 1000,
+ taskAboveLagThreshold.getTimeoutMs());
+
+ Deencapsulation.setField(routineLoadJob, "maxBatchRows",
50_000_000L);
+ latestOffsets.put(1, 25_000_010L);
+ latestOffsets.put(2, 25_000_020L);
+ KafkaTaskInfo taskAtConfiguredLagThreshold = new KafkaTaskInfo(new
UUID(1, 4), 1L, 20000,
+ taskProgress, false, 1000, false);
+ taskAtConfiguredLagThreshold.updateAdaptiveTimeout(routineLoadJob);
+ latestOffsets.put(2, 25_000_021L);
+ TRoutineLoadTask configuredThresholdThriftTask = new
TRoutineLoadTask();
+ Deencapsulation.invoke(taskAtConfiguredLagThreshold,
"adaptiveBatchParam",
+ configuredThresholdThriftTask, routineLoadJob);
+ Assert.assertEquals(20L,
configuredThresholdThriftTask.getMaxIntervalS());
+ Assert.assertEquals(50_000_000L,
configuredThresholdThriftTask.getMaxBatchRows());
+ Assert.assertEquals(routineLoadJob.getTimeout() * 1000L,
+ taskAtConfiguredLagThreshold.getTimeoutMs());
+
+ KafkaTaskInfo taskAboveConfiguredLagThreshold = new
KafkaTaskInfo(new UUID(1, 5), 1L, 20000,
+ taskProgress, false, 1000, false);
+
taskAboveConfiguredLagThreshold.updateAdaptiveTimeout(routineLoadJob);
+ TRoutineLoadTask configuredAdaptiveThriftTask = new
TRoutineLoadTask();
+ Deencapsulation.invoke(taskAboveConfiguredLagThreshold,
"adaptiveBatchParam",
+ configuredAdaptiveThriftTask, routineLoadJob);
+ Assert.assertEquals(360L,
configuredAdaptiveThriftTask.getMaxIntervalS());
+ Assert.assertEquals(50_000_000L,
configuredAdaptiveThriftTask.getMaxBatchRows());
+ } finally {
+ Config.routine_load_adaptive_min_batch_interval_sec =
previousAdaptiveIntervalSec;
+ }
+ }
+
+ @Test
+ public void testAdaptiveBatchUsesCapturedIntervalAcrossConfigChange() {
+ RoutineLoadManager routineLoadManager =
Mockito.mock(RoutineLoadManager.class);
+ Env env = Mockito.mock(Env.class);
+ int previousAdaptiveIntervalSec =
Config.routine_load_adaptive_min_batch_interval_sec;
+
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+ Config.routine_load_adaptive_min_batch_interval_sec = 360;
+ envStatic.when(Env::getCurrentEnv).thenReturn(env);
+
Mockito.when(env.getRoutineLoadManager()).thenReturn(routineLoadManager);
+
+ KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L,
"kafka_routine_load_job", 1L,
+ 1L, "127.0.0.1:9020", "topic1", UserIdentity.ADMIN);
+ Deencapsulation.setField(routineLoadJob, "maxBatchIntervalS", 30L);
+ Deencapsulation.setField(routineLoadJob, "maxBatchRows", 200000L);
+ Deencapsulation.setField(routineLoadJob, "maxBatchSizeBytes", 100L
* 1024 * 1024);
+
Mockito.when(routineLoadManager.getJob(1L)).thenReturn(routineLoadJob);
+
+ Map<Integer, Long> taskProgress = Maps.newHashMap();
+ taskProgress.put(1, 10L);
+ Map<Integer, Long> latestOffsets = Maps.newHashMap();
+ latestOffsets.put(1, 20_000_011L);
+ Deencapsulation.setField(routineLoadJob,
"cachedPartitionWithLatestOffsets", latestOffsets);
+
+ KafkaTaskInfo scheduledTask = new KafkaTaskInfo(new UUID(1, 6),
1L, 20000,
+ taskProgress, false, 1000, false);
+ scheduledTask.updateAdaptiveTimeout(routineLoadJob);
+ long adaptiveTimeoutMs = 360L *
Config.routine_load_task_timeout_multiplier * 1000L;
+ Assert.assertEquals(adaptiveTimeoutMs,
scheduledTask.getTimeoutMs());
+
+ Config.routine_load_adaptive_min_batch_interval_sec = 720;
+ TRoutineLoadTask scheduledThriftTask = new TRoutineLoadTask();
+ Deencapsulation.invoke(scheduledTask, "adaptiveBatchParam",
scheduledThriftTask, routineLoadJob);
+ Assert.assertEquals(360L, scheduledThriftTask.getMaxIntervalS());
+ Assert.assertEquals(RoutineLoadJob.DEFAULT_MAX_BATCH_ROWS,
scheduledThriftTask.getMaxBatchRows());
+ Assert.assertEquals(RoutineLoadJob.DEFAULT_MAX_BATCH_SIZE,
scheduledThriftTask.getMaxBatchSize());
+ Assert.assertEquals(adaptiveTimeoutMs,
scheduledTask.getTimeoutMs());
+
+ KafkaTaskInfo nextSchedulingAttempt = new KafkaTaskInfo(new
UUID(1, 7), 1L, 20000,
+ taskProgress, false, 1000, false);
+ nextSchedulingAttempt.updateAdaptiveTimeout(routineLoadJob);
+ TRoutineLoadTask nextThriftTask = new TRoutineLoadTask();
+ Deencapsulation.invoke(nextSchedulingAttempt,
"adaptiveBatchParam", nextThriftTask, routineLoadJob);
+ Assert.assertEquals(720L, nextThriftTask.getMaxIntervalS());
+ Assert.assertEquals(720L *
Config.routine_load_task_timeout_multiplier * 1000L,
+ nextSchedulingAttempt.getTimeoutMs());
+
+ for (int nonPositiveInterval : new int[] {0, -1}) {
+ Config.routine_load_adaptive_min_batch_interval_sec =
nonPositiveInterval;
+ KafkaTaskInfo nonPositiveConfigTask = new KafkaTaskInfo(new
UUID(1, 8), 1L, 20000,
+ taskProgress, false, 1000, false);
+ nonPositiveConfigTask.updateAdaptiveTimeout(routineLoadJob);
+ TRoutineLoadTask nonPositiveConfigThriftTask = new
TRoutineLoadTask();
+ Deencapsulation.invoke(nonPositiveConfigTask,
"adaptiveBatchParam",
+ nonPositiveConfigThriftTask, routineLoadJob);
+ Assert.assertEquals(30L,
nonPositiveConfigThriftTask.getMaxIntervalS());
+ Assert.assertEquals(RoutineLoadJob.DEFAULT_MAX_BATCH_ROWS,
+ nonPositiveConfigThriftTask.getMaxBatchRows());
+ Assert.assertEquals(RoutineLoadJob.DEFAULT_MAX_BATCH_SIZE,
+ nonPositiveConfigThriftTask.getMaxBatchSize());
+ long normalTimeoutMs = Math.max(30L *
Config.routine_load_task_timeout_multiplier,
+ Config.routine_load_task_min_timeout_sec) * 1000L;
+ Assert.assertEquals(normalTimeoutMs,
nonPositiveConfigTask.getTimeoutMs());
+ }
+ } finally {
+ Config.routine_load_adaptive_min_batch_interval_sec =
previousAdaptiveIntervalSec;
+ }
+ }
+
@Test
public void testProcessTimeOutTasks() throws Exception {
RoutineLoadManager routineLoadManager =
Mockito.mock(RoutineLoadManager.class);
diff --git
a/regression-test/suites/load_p0/routine_load/test_routine_load_adaptive_param.groovy
b/regression-test/suites/load_p0/routine_load/test_routine_load_adaptive_param.groovy
index 49d901c31c9..8652d018780 100644
---
a/regression-test/suites/load_p0/routine_load/test_routine_load_adaptive_param.groovy
+++
b/regression-test/suites/load_p0/routine_load/test_routine_load_adaptive_param.groovy
@@ -65,31 +65,33 @@ suite("test_routine_load_adaptive_param","nonConcurrent") {
);
"""
- def injection = "RoutineLoadTaskInfo.judgeEof"
+ def eofInjection = "RoutineLoadTaskInfo.judgeEof"
+ def adaptiveBatchInjection = "KafkaTaskInfo.shouldUseAdaptiveBatch"
try {
- GetDebugPoint().enableDebugPointForAllFEs(injection)
+ GetDebugPoint().enableDebugPointForAllFEs(eofInjection)
+
GetDebugPoint().enableDebugPointForAllFEs(adaptiveBatchInjection)
RoutineLoadTestUtils.sendTestDataToKafka(producer,
kafkaCsvTpoics)
RoutineLoadTestUtils.waitForTaskFinish(runSql, job, tableName,
0)
logger.info("---test adaptively increase---")
RoutineLoadTestUtils.sendTestDataToKafka(producer,
kafkaCsvTpoics)
- // Drive data each round so an isEof=false task keeps being
scheduled. The converged
- // adaptive timeout (3600) lives on the renewed idle task
(txnId == -1), so both checks
+ // Drive data each round so a task keeps being scheduled. The
converged adaptive timeout
+ // (3600) lives on the renewed idle task (txnId == -1), so
both checks
// poll by value (task timeout col, and the committed txn's
persisted timeout looked up
// by task-UUID label) instead of racing a sub-second running
task.
RoutineLoadTestUtils.checkTaskTimeoutWithData(runSql,
producer, kafkaCsvTpoics, job, "3600")
RoutineLoadTestUtils.checkTxnTimeoutMatchesTaskTimeout(runSql,
producer, kafkaCsvTpoics, job, "3600000")
RoutineLoadTestUtils.waitForTaskFinish(runSql, job, tableName,
2)
} finally {
- GetDebugPoint().disableDebugPointForAllFEs(injection)
+
GetDebugPoint().disableDebugPointForAllFEs(adaptiveBatchInjection)
+ GetDebugPoint().disableDebugPointForAllFEs(eofInjection)
}
logger.info("---test restore adaptively---")
RoutineLoadTestUtils.sendTestDataToKafka(producer, kafkaCsvTpoics)
RoutineLoadTestUtils.waitForTaskFinish(runSql, job, tableName, 4)
- // After EOF the adaptive timeout only converges when an isEof
task is scheduled with
- // data, so keep feeding small batches until the task timeout
restores to the job timeout.
+ // Keep feeding small batches until the low task lag restores the
timeout to the job timeout.
RoutineLoadTestUtils.checkTaskTimeoutWithData(runSql, producer,
kafkaCsvTpoics, job, "100")
} finally {
sql "stop routine load for ${job}"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
