github-actions[bot] commented on code in PR #66099:
URL: https://github.com/apache/doris/pull/66099#discussion_r3695387305
##########
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 = maxBatchRows;
+ for (Map.Entry<Integer, Long> entry : partitionIdToOffset.entrySet()) {
+ Long latestOffset =
cachedPartitionWithLatestOffsets.get(entry.getKey());
Review Comment:
[P2] Reset and fence the lag cache when changing Kafka source
`cachedPartitionWithLatestOffsets` survives the supported paused-job changes
to `topic` or `brokerList`, and every cache update keeps the larger value with
`Math.max`. If the old source cached partition 0 above `max_batch_rows` and the
new source uses the same partition at a lower offset, `hasMoreDataToConsume()`
treats `newOffset < oldCachedOffset` as fresh data and skips a broker refresh;
this line then counts the old watermark as the new task's lag and can keep
selecting the 360-second/20M-row/1GiB adaptive limits until the new source
catches up to the old source's offset space. Please clear the latest-offset
cache when the source identity changes and cover pause/ALTER/resume with
overlapping partition IDs.
Clearing alone also races with both refresh paths: they snapshot
topic/broker under the job lock, perform Kafka I/O after unlocking, and merge
without checking that the source is still current, so a pre-ALTER response can
repopulate the cleared map. Please invalidate the cache and reject late results
with a source generation/identity check (without holding the lock across the
RPC), and cover that interleaving too.
--
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]
