database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[branch-2.1](routine-load) dealing with the high watermark of Kafka may fallback (#37372)

Browse Source 
pick #35901
This commit is contained in:
hui lai
2024-07-07 18:15:54 +08:00
committed by GitHub
parent 960b02f293
commit af960f7c70
1 changed files with 8 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java
View File

@ -765,7 +765,14 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
List<Pair<Integer, Long>> tmp = KafkaUtil.getLatestOffsets(id, taskId, getBrokerList(),
getTopic(), getConvertedCustomProperties(), Lists.newArrayList(partitionIdToOffset.keySet()));
for (Pair<Integer, Long> pair : tmp) {
cachedPartitionWithLatestOffsets.put(pair.first, pair.second);
if (pair.second >= cachedPartitionWithLatestOffsets.getOrDefault(pair.first, Long.MIN_VALUE)) {
cachedPartitionWithLatestOffsets.put(pair.first, pair.second);
} else {
LOG.warn("Kafka offset fallback. partition: {}, cache offset: {}"
+ " get latest offset: {}, task {}, job {}",
pair.first, cachedPartitionWithLatestOffsets.getOrDefault(pair.first, Long.MIN_VALUE),
pair.second, taskId, id);
}
}
} catch (Exception e) {
// It needs to pause job when can not get partition meta.