github-actions[bot] commented on code in PR #66099:
URL: https://github.com/apache/doris/pull/66099#discussion_r3662567210
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaTaskInfo.java:
##########
@@ -126,31 +141,56 @@ public TRoutineLoadTask createRoutineLoadTask() throws
UserException {
@Override
public void updateAdaptiveTimeout(RoutineLoadJob routineLoadJob) {
- if (!isEof) {
+ adaptiveMinBatchIntervalSnapshotS =
Config.routine_load_adaptive_min_batch_interval_sec;
+ if (shouldUseAdaptiveBatch(routineLoadJob)) {
long maxBatchIntervalS =
Math.max(routineLoadJob.getMaxBatchIntervalS(),
- Config.routine_load_adaptive_min_batch_interval_sec);
+ adaptiveMinBatchIntervalSnapshotS);
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;
+ timeoutMs = realTimeoutSec * 1000;
} else {
- this.timeoutMs = routineLoadJob.getTimeout() * 1000;
+ timeoutMs = routineLoadJob.getTimeout() * 1000;
}
}
private void adaptiveBatchParam(TRoutineLoadTask tRoutineLoadTask,
RoutineLoadJob routineLoadJob) {
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);
- maxBatchRows = Math.max(maxBatchRows,
RoutineLoadJob.DEFAULT_MAX_BATCH_ROWS);
- maxBatchSize = Math.max(maxBatchSize,
RoutineLoadJob.DEFAULT_MAX_BATCH_SIZE);
+ if (shouldUseAdaptiveBatch(routineLoadJob)) {
+ maxBatchIntervalS = Math.max(maxBatchIntervalS,
adaptiveMinBatchIntervalSnapshotS);
+ maxBatchSize = Math.max(maxBatchSize,
ADAPTIVE_BATCH_BYTES_THRESHOLD);
}
tRoutineLoadTask.setMaxIntervalS(maxBatchIntervalS);
tRoutineLoadTask.setMaxBatchRows(maxBatchRows);
tRoutineLoadTask.setMaxBatchSize(maxBatchSize);
}
+ private boolean shouldUseAdaptiveBatch(RoutineLoadJob routineLoadJob) {
+ if (adaptiveMinBatchIntervalSnapshotS <= 0) {
+ throw new IllegalStateException(
+ "adaptive batch interval must be captured before beginning
the transaction");
+ }
+ if (isEof) {
+ return false;
+ }
+ if (DebugPointUtil.isEnable("KafkaTaskInfo.shouldUseAdaptiveBatch")) {
+ return true;
+ }
+ if (previousTaskExecutionTimeMs <= 0) {
+ return false;
+ }
+ long adaptiveBatchIntervalMs =
Math.max(routineLoadJob.getMaxBatchIntervalS(),
+ adaptiveMinBatchIntervalSnapshotS) * 1000L;
+ // Estimate the data volume in an adaptive interval from the previous
task throughput.
+ return reachesAdaptiveBytesThreshold(adaptiveBatchIntervalMs);
Review Comment:
[P1] Keep the row side of adaptive batching
This now admits only on projected bytes and also leaves `maxBatchRows` at
the user value. For example, the new `highRowsOnly` sample has 2,000,000 rows
in 20 seconds, so it projects 36,000,000 rows over 360 seconds and exceeds the
stated internal 20M-row threshold, but this return still rejects it because its
bytes are tiny. In the other direction, a byte-qualified task with the valid
200,000-row minimum is sent that unchanged row cap; BE stops as soon as
`left_rows` reaches zero, so the task can still end around the original short
interval instead of realizing the adaptive batch. The old adaptive branch
raised both row and byte limits. Please carry the previous total-row sample,
admit on projected 20M rows **or** 1 GiB, restore the 20M adaptive row cap, and
update the `highRowsOnly` oracle.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaTaskInfo.java:
##########
@@ -126,31 +141,56 @@ public TRoutineLoadTask createRoutineLoadTask() throws
UserException {
@Override
public void updateAdaptiveTimeout(RoutineLoadJob routineLoadJob) {
- if (!isEof) {
+ adaptiveMinBatchIntervalSnapshotS =
Config.routine_load_adaptive_min_batch_interval_sec;
+ if (shouldUseAdaptiveBatch(routineLoadJob)) {
long maxBatchIntervalS =
Math.max(routineLoadJob.getMaxBatchIntervalS(),
- Config.routine_load_adaptive_min_batch_interval_sec);
+ adaptiveMinBatchIntervalSnapshotS);
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;
+ timeoutMs = realTimeoutSec * 1000;
} else {
- this.timeoutMs = routineLoadJob.getTimeout() * 1000;
+ timeoutMs = routineLoadJob.getTimeout() * 1000;
}
}
private void adaptiveBatchParam(TRoutineLoadTask tRoutineLoadTask,
RoutineLoadJob routineLoadJob) {
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);
- maxBatchRows = Math.max(maxBatchRows,
RoutineLoadJob.DEFAULT_MAX_BATCH_ROWS);
- maxBatchSize = Math.max(maxBatchSize,
RoutineLoadJob.DEFAULT_MAX_BATCH_SIZE);
+ if (shouldUseAdaptiveBatch(routineLoadJob)) {
+ maxBatchIntervalS = Math.max(maxBatchIntervalS,
adaptiveMinBatchIntervalSnapshotS);
+ maxBatchSize = Math.max(maxBatchSize,
ADAPTIVE_BATCH_BYTES_THRESHOLD);
}
tRoutineLoadTask.setMaxIntervalS(maxBatchIntervalS);
tRoutineLoadTask.setMaxBatchRows(maxBatchRows);
tRoutineLoadTask.setMaxBatchSize(maxBatchSize);
}
+ private boolean shouldUseAdaptiveBatch(RoutineLoadJob routineLoadJob) {
+ if (adaptiveMinBatchIntervalSnapshotS <= 0) {
Review Comment:
[P1] Do not strand tasks on an accepted config value
`routine_load_adaptive_min_batch_interval_sec` is mutable and its default
config handler accepts `0` or a negative integer, so such an ADMIN SET reaches
this check. By then `RoutineLoadTaskScheduler` has already removed the task
from the queue and assigned its `beId`; `updateAdaptiveTimeout()` is outside
the allocation/begin/create recovery blocks, and the outer loop only logs this
exception. The task is never requeued, keeps `beId != -1`, has no
transaction/start time so it never times out, and continues consuming a
routine-load slot. Please separate the “snapshot captured” state from the
interval value (or reject non-positive values before applying them) and ensure
a failure here releases/requeues or deterministically pauses the task.
--
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]
