github-actions[bot] commented on code in PR #66099:
URL: https://github.com/apache/doris/pull/66099#discussion_r3656317781
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaTaskInfo.java:
##########
@@ -141,16 +159,41 @@ private void adaptiveBatchParam(TRoutineLoadTask
tRoutineLoadTask, RoutineLoadJo
long maxBatchIntervalS = routineLoadJob.getMaxBatchIntervalS();
long maxBatchRows = routineLoadJob.getMaxBatchRows();
long maxBatchSize = routineLoadJob.getMaxBatchSizeBytes();
- if (!isEof) {
+ if (shouldUseAdaptiveBatch(routineLoadJob)) {
Review Comment:
[P2] Reuse one adaptive decision across `beginTxn`
`updateAdaptiveTimeout()` evaluates this predicate before `beginTxn`, but
task creation evaluates it again here, and the predicate reads the mutable
adaptive-interval config. For example, consider a preceding non-EOF 30-second
task that processed 1M one-byte Kafka messages, each yielding one row. With the
interval set to 360, the first check projects 12M rows, so the transaction and
plan get the normal 300-second timeout. If the supported runtime config is
raised to 720 before this call, the same sample projects 24M rows and this path
emits a 720-second task; at that rate the 20M-message limit is reached only
after about 600 seconds and the 1-GiB limit is not reached, so the
transaction/query can expire first. The old `!isEof` branch stayed adaptive in
both phases for this sample. Please compute one mode/interval/limit snapshot
before starting the transaction and reuse it for the transaction, plan, and
Thrift 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]
