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

[fix](routine-load) fix routine load lag is negative (#34113)

Browse Source 
* [fix](routine-load) fix routine load lag is negative (#33846)

* fix merge error
This commit is contained in:
HHoflittlefish777
2024-04-25 17:24:41 +08:00
committed by GitHub
parent e3ed861e4b
commit b873be6588
2 changed files with 17 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaProgress.java
View File

@ -85,6 +85,10 @@ public class KafkaProgress extends RoutineLoadProgress {
return partitionIdToOffset.get(kafkaPartition);
}
public Map<Integer, Long> getOffsetByPartition() {
return partitionIdToOffset;
}
public boolean containsPartition(Integer kafkaPartition) {
return partitionIdToOffset.containsKey(kafkaPartition);
}

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

@ -285,16 +285,27 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
return false;
}
private void updateProgressAndOffsetsCache(RLTaskTxnCommitAttachment attachment) {
((KafkaProgress) attachment.getProgress()).getOffsetByPartition().entrySet().stream()
.forEach(entity -> {
if (cachedPartitionWithLatestOffsets.containsKey(entity.getKey())
&& cachedPartitionWithLatestOffsets.get(entity.getKey()) < entity.getValue() + 1) {
cachedPartitionWithLatestOffsets.put(entity.getKey(), entity.getValue() + 1);
}
});
this.progress.update(attachment);
}
@Override
protected void updateProgress(RLTaskTxnCommitAttachment attachment) throws UserException {
super.updateProgress(attachment);
this.progress.update(attachment);
updateProgressAndOffsetsCache(attachment);
}
@Override
protected void replayUpdateProgress(RLTaskTxnCommitAttachment attachment) {
super.replayUpdateProgress(attachment);
this.progress.update(attachment);
updateProgressAndOffsetsCache(attachment);
}
@Override