Repository: provectus/kafka-ui
Issue: [UI] Offset & timestamp filters do not work
Labels: type/bug
scope/frontend
status/accepted
status/confirmed

On the Messages page when Offset/Timestamp fields **are filled** we need to pass seekType argument with OFFSET/TIMESTAMP values correspondingly. 

When Offset/Timestamp are **not filled** we should pass seekType=LATEST if "Newest first" mode is on and seekType=BEGINNING when mode is "Oldest first".

Issue comments:

Pull request review comments:

Patch that resolved the issue:
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 c651bd5e56b..996f8b9f70c 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
@@ -17,6 +17,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.InterruptException;
 import org.apache.kafka.common.utils.Bytes;
 import reactor.core.publisher.FluxSink;
 
@@ -85,6 +86,9 @@ public void accept(FluxSink<TopicMessageEventDTO> sink) {
       }
       sendFinishStatsAndCompleteSink(sink);
       log.debug("Polling finished");
+    } catch (InterruptException kafkaInterruptException) {
+      log.debug("Polling finished due to thread interruption");
+      sink.complete();
     } catch (Exception e) {
       log.error("Error occurred while consuming records", e);
       sink.error(e);
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/ForwardRecordEmitter.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/ForwardRecordEmitter.java
index 132b4790297..9fadb149d48 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/ForwardRecordEmitter.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/ForwardRecordEmitter.java
@@ -9,6 +9,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.errors.InterruptException;
 import org.apache.kafka.common.utils.Bytes;
 import reactor.core.publisher.FluxSink;
 
@@ -59,6 +60,9 @@ public void accept(FluxSink<TopicMessageEventDTO> sink) {
       }
       sendFinishStatsAndCompleteSink(sink);
       log.debug("Polling finished");
+    } catch (InterruptException kafkaInterruptException) {
+      log.debug("Polling finished due to thread interruption");
+      sink.complete();
     } catch (Exception e) {
       log.error("Error occurred while consuming records", e);
       sink.error(e);
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/TailingEmitter.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/TailingEmitter.java
index 025dd874cfa..06cd8dad998 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/TailingEmitter.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/TailingEmitter.java
@@ -41,6 +41,7 @@ public void accept(FluxSink<TopicMessageEventDTO> sink) {
       sink.complete();
       log.debug("Tailing finished");
     } catch (InterruptException kafkaInterruptException) {
+      log.debug("Tailing finished due to thread interruption");
       sink.complete();
     } catch (Exception e) {
       log.error("Error consuming {}", consumerPosition, e);
diff --git a/kafka-ui-react-app/src/components/Topics/Topic/Messages/Filters/Filters.tsx b/kafka-ui-react-app/src/components/Topics/Topic/Messages/Filters/Filters.tsx
index 9f54e56d5a4..0e3d3d46614 100644
--- a/kafka-ui-react-app/src/components/Topics/Topic/Messages/Filters/Filters.tsx
+++ b/kafka-ui-react-app/src/components/Topics/Topic/Messages/Filters/Filters.tsx
@@ -219,6 +219,15 @@ const Filters: React.FC<FiltersProps> = ({
         default:
           props.seekType = currentSeekType;
       }
+
+      if (offset && currentSeekType === SeekType.OFFSET) {
+        props.seekType = SeekType.OFFSET;
+      }
+
+      if (timestamp && currentSeekType === SeekType.TIMESTAMP) {
+        props.seekType = SeekType.TIMESTAMP;
+      }
+
       props.seekTo = selectedPartitions.map(({ value }) => {
         const offsetProperty =
           seekDirection === SeekDirection.FORWARD ? 'offsetMin' : 'offsetMax';


Test patch: