0ct0pu5/kafka-ui
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

#122 Fixed compacted topics backward reading (#607)

Browse source
This commit is contained in:
German Osin 2021-07-02 14:52:40 +03:00 • committed by GitHub
parent 09ebd03e71
commit ca476d3373
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 12 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/BackwardRecordEmitter.java
View file

@ -64,6 +64,13 @@ public class BackwardRecordEmitter
log.debug("{} records polled", records.count()); log.debug("{} records polled", records.count());
log.debug("{} records sent", partitionRecords.size()); log.debug("{} records sent", partitionRecords.size());
// This is workaround for case when partition begin offset is less than
// real minimal offset, usually appear in compcated topics
if (records.count() > 0 && partitionRecords.isEmpty()) {
waitingOffsets.markPolled(entry.getKey().partition());
}
for (ConsumerRecord<Bytes, Bytes> msg : partitionRecords) { for (ConsumerRecord<Bytes, Bytes> msg : partitionRecords) {
if (!sink.isCancelled() && !waitingOffsets.beginReached()) { if (!sink.isCancelled() && !waitingOffsets.beginReached()) {
sink.next(msg); sink.next(msg);

5
kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/OffsetsSeek.java
View file

@ -110,6 +110,11 @@ public abstract class OffsetsSeek {
.collect(Collectors.toList()); .collect(Collectors.toList());
} }
public void markPolled(int partition) {
endOffsets.remove(partition);
beginOffsets.remove(partition);
}
public void markPolled(ConsumerRecord<?, ?> rec) { public void markPolled(ConsumerRecord<?, ?> rec) {
Long endWaiting = endOffsets.get(rec.partition()); Long endWaiting = endOffsets.get(rec.partition());
if (endWaiting != null && endWaiting <= rec.offset()) { if (endWaiting != null && endWaiting <= rec.offset()) {