Skip to content

Commit

Permalink
fix(topicdata): topic data access slow on sort newest (#1371)
Browse files Browse the repository at this point in the history
- Adding stop condition in newest sort to prevent extra polling 
- Fixing wrong number of records with pagination

close #192
  • Loading branch information
AlexisSouquiere authored and tchiotludo committed Apr 4, 2023
1 parent bd281de commit e1d7d79
Showing 1 changed file with 10 additions and 2 deletions.
12 changes: 10 additions & 2 deletions src/main/java/org/akhq/repositories/RecordRepository.java
Original file line number Diff line number Diff line change
Expand Up @@ -324,6 +324,12 @@ private List<Record> consumeNewest(Topic topic, Options options) {
filterMessageLength(current);
list.add(current);
}

// End of the partition, we can stop here
if (record.offset() == topicPartitionOffset.getEnd()) {
emptyPoll = 1;
break;
}
}
}
while (emptyPoll < 1);
Expand Down Expand Up @@ -394,8 +400,10 @@ private Optional<OffsetBound> getFirstOffsetForSortOldest(KafkaConsumer<byte[],
private Optional<EndOffsetBound> getOffsetForSortNewest(KafkaConsumer<byte[], byte[]> consumer, Partition partition, Options options, int pollSizePerPartition) {
return getFirstOffset(consumer, partition, options)
.map(first -> {
long last = partition.getLastOffset();
// Take end offset - 1 to get the last record offset
long last = partition.getLastOffset() - 1;

// If there is an after parameter in the request use this one
if (pollSizePerPartition > 0 && options.after.containsKey(partition.getId())) {
last = options.after.get(partition.getId()) - 1;
}
Expand All @@ -404,7 +412,7 @@ private Optional<EndOffsetBound> getOffsetForSortNewest(KafkaConsumer<byte[], by
consumer.close();
return null;
} else if (!(last - pollSizePerPartition < first)) {
first = last - pollSizePerPartition;
first = last - pollSizePerPartition + 1;
}

return EndOffsetBound.builder()
Expand Down

0 comments on commit e1d7d79

Please sign in to comment.