Repository: provectus/kafka-ui
Issue: Topic messages: Newest first
Labels: type/enhancement
scope/backend
scope/frontend

Change sorting of topic messages - sort by offset in descending order (latest messages first).

Issue comments:
Current state of this task:
Need to figure out a way to read a set of topic messages from newer ones to older ones on BE, and then update existing FE request to properly work with new API.
Waiting fronted...

Pull request review comments:

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ConsumingService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ConsumingService.java
index fceb80aa9ef..a6351aae019 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ConsumingService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ConsumingService.java
@@ -14,12 +14,14 @@
 import com.provectus.kafka.ui.util.OffsetsSeekForward;
 import java.time.Duration;
 import java.util.Collection;
+import java.util.Comparator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
 import org.apache.commons.lang3.StringUtils;
@@ -136,6 +138,15 @@ static class RecordEmitter
 
     private static final Duration POLL_TIMEOUT_MS = Duration.ofMillis(1000L);
 
+    private static final Comparator<ConsumerRecord<?, ?>> PARTITION_COMPARING =
+        Comparator.comparing(
+            ConsumerRecord::partition,
+            Comparator.nullsFirst(Comparator.naturalOrder())
+        );
+    private static final Comparator<ConsumerRecord<?, ?>> REVERED_COMPARING =
+        PARTITION_COMPARING.thenComparing(ConsumerRecord::offset).reversed();
+
+
     private final Supplier<KafkaConsumer<Bytes, Bytes>> consumerSupplier;
     private final OffsetsSeek offsetsSeek;
 
@@ -146,7 +157,16 @@ public void accept(FluxSink<ConsumerRecord<Bytes, Bytes>> sink) {
         while (!sink.isCancelled() && !waitingOffsets.endReached()) {
           ConsumerRecords<Bytes, Bytes> records = consumer.poll(POLL_TIMEOUT_MS);
           log.info("{} records polled", records.count());
-          for (ConsumerRecord<Bytes, Bytes> record : records) {
+
+          final Iterable<ConsumerRecord<Bytes, Bytes>> iterable;
+          if (offsetsSeek.getConsumerPosition().getSeekDirection().equals(SeekDirection.FORWARD)) {
+            iterable = records;
+          } else {
+            iterable = StreamSupport.stream(records.spliterator(), false)
+                .sorted(REVERED_COMPARING).collect(Collectors.toList());
+          }
+
+          for (ConsumerRecord<Bytes, Bytes> record : iterable) {
             if (!sink.isCancelled() && !waitingOffsets.endReached()) {
               sink.next(record);
               waitingOffsets.markPolled(record);
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/OffsetsSeek.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/OffsetsSeek.java
index 9d082c67bc8..6383a83121d 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/OffsetsSeek.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/OffsetsSeek.java
@@ -23,6 +23,10 @@ public OffsetsSeek(String topic, ConsumerPosition consumerPosition) {
     this.consumerPosition = consumerPosition;
   }
 
+  public ConsumerPosition getConsumerPosition() {
+    return consumerPosition;
+  }
+
   public WaitingOffsets assignAndSeek(Consumer<Bytes, Bytes> consumer) {
     SeekType seekType = consumerPosition.getSeekType();
     log.info("Positioning consumer for topic {} with {}", topic, consumerPosition);


Test patch: