[FE] Fix invalid SeekType value for the offset and timestamp fields (#3057)

* SeekType fix the value during the offset and timestamp fields

* BE: thread interruption logging added to Emitters

Co-authored-by: iliax <[email protected]>

Author: Mgrdich

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);

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);

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);

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';