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...
Topic messages: Newest first
Change sorting of topic messages - sort by offset in descending order (latest messages first).
kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/BackwardRecordEmitter.java kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/OffsetsSeek.java
kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/BackwardRecordEmitter.java kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/OffsetsSeek.java
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/BackwardRecordEmitter.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/BackwardRecordEmitter.java
index 088366c238e..8bf2bab547f 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/BackwardRecordEmitter.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/BackwardRecordEmitter.java
@@ -64,6 +64,13 @@ public void accept(FluxSink<ConsumerRecord<Bytes, Bytes>> sink) {
 
               log.debug("{} records polled", records.count());
               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) {
                 if (!sink.isCancelled() && !waitingOffsets.beginReached()) {
                   sink.next(msg);
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 17a496d53bb..6cc275c4c81 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
@@ -110,6 +110,11 @@ public List<TopicPartition> topicPartitions() {
           .collect(Collectors.toList());
     }
 
+    public void markPolled(int partition) {
+      endOffsets.remove(partition);
+      beginOffsets.remove(partition);
+    }
+
     public void markPolled(ConsumerRecord<?, ?> rec) {
       Long endWaiting = endOffsets.get(rec.partition());
       if (endWaiting != null && endWaiting <= rec.offset()) {
