github-actions[bot] commented on code in PR #66099:
URL: https://github.com/apache/doris/pull/66099#discussion_r3697590124
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java:
##########
@@ -914,6 +914,25 @@ public void
replayModifyProperties(AlterRoutineLoadJobOperationLog log) {
}
}
+ boolean isTaskLagGreaterThanMaxBatchRows(Map<Integer, Long>
partitionIdToOffset) {
+ long remainingRows = RoutineLoadJob.DEFAULT_MAX_BATCH_ROWS;
Review Comment:
[P2] Honor configured row limits above the internal floor
The current PR contract says adaptive batching starts only when task lag
exceeds `max_batch_rows`, but this always gates at `DEFAULT_MAX_BATCH_ROWS`
(20M) even though `adaptiveBatchParam()` sends `max(configuredRows, 20M)` to
BE. For example, with `max_batch_rows=50M`, `max_batch_interval=10`, a large
byte cap, and 25M tiny queued records, this returns true and raises the
interval/transaction timeout to 360/3600 seconds while BE's row limit remains
50M; slow or sustained consumption can therefore keep the transaction open
hundreds of seconds even though lag never crossed the configured admission
target. Please compare against `Math.max(getMaxBatchRows(),
DEFAULT_MAX_BATCH_ROWS)` and add equality/one-above coverage for a configured
cap above 20M.
--
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]
