github-actions[bot] commented on code in PR #66099:
URL: https://github.com/apache/doris/pull/66099#discussion_r3664815939
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaTaskInfo.java:
##########
@@ -72,12 +79,21 @@ public KafkaTaskInfo(KafkaTaskInfo kafkaTaskInfo,
Map<Integer, Long> partitionId
kafkaTaskInfo.getTimeoutMs(), kafkaTaskInfo.getBeId(),
isMultiTable,
kafkaTaskInfo.getLastScheduledTime(),
kafkaTaskInfo.getIsEof());
this.partitionIdToOffset = partitionIdToOffset;
+ this.previousTaskBytes = kafkaTaskInfo.previousTaskBytes;
Review Comment:
[P2] Do not reuse an older sample after an executed task fails
This copy also runs when an executed task is renewed after abort or timeout,
but Kafka rejects every ABORTED attachment in `checkCommitInfo`, so
`handleTaskByTxnCommitAttachment` never replaces these fields. For example,
after a 100 MiB/20 s task selects the 360-second mode, a small successor can
consume data and then hit a transient commit failure; its rollback attachment
contains that task's own bytes/load time, yet the replacement inherits the
older 100 MiB/20 s pair and selects 360 seconds again. Repeated failures can
keep low-throughput retries on stale adaptive limits. Please invalidate the
sample when an executed attempt is renewed without an accepted attachment, or
separately retain a validated per-task bytes/time sample without applying its
uncommitted progress, and cover abort/timeout renewal.
--
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]
