fix(topicdata): topic data access slow on sort newest (#1371)

- Adding stop condition in newest sort to prevent extra polling 
- Fixing wrong number of records with pagination

close #192

Author: AlexisSouquiere

src/main/java/org/akhq/repositories/RecordRepository.java

@@ -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);
@@ -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;
                 }
@@ -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()