Repository: provectus/kafka-ui
Issue: UI: Display smart filters filtered stats
Labels: type/enhancement
scope/backend
scope/frontend
status/accepted

When polling messages from topic (and filtering is enabled) we should show number of errors that appeared during filtration process - > this number placed in `TopicMessageConsuming.filterApplyErrors`
UI: 
1. during polling process - show value of `TopicMessageConsuming.filterApplyErrors` if it is > 0
2. when polling is finished BE will send `TopicMessageEvent` with `type = DONE` with final stats of polling process - this stats should be rendered. (also this event should be logically treated as polling end by UI)


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/AbstractEmitter.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/AbstractEmitter.java
index ddb5dc3666c..35d8a3bef34 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/AbstractEmitter.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/AbstractEmitter.java
@@ -64,6 +64,18 @@ protected void sendPhase(FluxSink<TopicMessageEventDTO> sink, String name) {
   protected void sendConsuming(FluxSink<TopicMessageEventDTO> sink,
                                ConsumerRecords<Bytes, Bytes> records,
                                long elapsed) {
-    consumingStats.sendConsumingEvt(sink, records, elapsed);
+    consumingStats.sendConsumingEvt(sink, records, elapsed, getFilterApplyErrors(sink));
+  }
+
+  protected void sendFinishStatsAndCompleteSink(FluxSink<TopicMessageEventDTO> sink) {
+    consumingStats.sendFinishEvent(sink, getFilterApplyErrors(sink));
+    sink.complete();
+  }
+
+  protected Number getFilterApplyErrors(FluxSink<?> sink) {
+    return sink.contextView()
+        .<MessageFilterStats>getOrEmpty(MessageFilterStats.class)
+        .<Number>map(MessageFilterStats::getFilterApplyErrors)
+        .orElse(0);
   }
 }
\ No newline at end of file
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 d2012355db6..fe1644676a0 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
@@ -80,7 +80,7 @@ public void accept(FluxSink<TopicMessageEventDTO> sink) {
           log.debug("sink is cancelled after partitions poll iteration");
         }
       }
-      sink.complete();
+      sendFinishStatsAndCompleteSink(sink);
       log.debug("Polling finished");
     } catch (Exception e) {
       log.error("Error occurred while consuming records", e);
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/ConsumingStats.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/ConsumingStats.java
index 830eb87320b..bd0754fc687 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/ConsumingStats.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/ConsumingStats.java
@@ -16,7 +16,8 @@ class ConsumingStats {
 
   void sendConsumingEvt(FluxSink<TopicMessageEventDTO> sink,
                         ConsumerRecords<Bytes, Bytes> polledRecords,
-                        long elapsed) {
+                        long elapsed,
+                        Number filterApplyErrors) {
     for (ConsumerRecord<Bytes, Bytes> rec : polledRecords) {
       for (Header header : rec.headers()) {
         bytes +=
@@ -27,15 +28,28 @@ void sendConsumingEvt(FluxSink<TopicMessageEventDTO> sink,
     }
     this.records += polledRecords.count();
     this.elapsed += elapsed;
-    final TopicMessageConsumingDTO consuming = new TopicMessageConsumingDTO()
-        .bytesConsumed(this.bytes)
-        .elapsedMs(this.elapsed)
-        .isCancelled(sink.isCancelled())
-        .messagesConsumed(this.records);
     sink.next(
         new TopicMessageEventDTO()
             .type(TopicMessageEventDTO.TypeEnum.CONSUMING)
-            .consuming(consuming)
+            .consuming(createConsumingStats(sink, filterApplyErrors))
     );
   }
+
+  void sendFinishEvent(FluxSink<TopicMessageEventDTO> sink, Number filterApplyErrors) {
+    sink.next(
+        new TopicMessageEventDTO()
+            .type(TopicMessageEventDTO.TypeEnum.DONE)
+            .consuming(createConsumingStats(sink, filterApplyErrors))
+    );
+  }
+
+  private TopicMessageConsumingDTO createConsumingStats(FluxSink<TopicMessageEventDTO> sink,
+                                                        Number filterApplyErrors) {
+    return new TopicMessageConsumingDTO()
+        .bytesConsumed(this.bytes)
+        .elapsedMs(this.elapsed)
+        .isCancelled(sink.isCancelled())
+        .filterApplyErrors(filterApplyErrors.intValue())
+        .messagesConsumed(this.records);
+  }
 }
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 69d9801b705..502a8c9a26c 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
@@ -54,7 +54,7 @@ public void accept(FluxSink<TopicMessageEventDTO> sink) {
           }
         }
       }
-      sink.complete();
+      sendFinishStatsAndCompleteSink(sink);
       log.info("Polling finished");
     } catch (Exception e) {
       log.error("Error occurred while consuming records", e);
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/MessageFilterStats.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/MessageFilterStats.java
new file mode 100644
index 00000000000..3b6df3cdea4
--- /dev/null
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/MessageFilterStats.java
@@ -0,0 +1,16 @@
+package com.provectus.kafka.ui.emitter;
+
+import java.util.concurrent.atomic.AtomicLong;
+import lombok.AccessLevel;
+import lombok.Getter;
+
+public class MessageFilterStats {
+
+  @Getter(AccessLevel.PACKAGE)
+  private final AtomicLong filterApplyErrors = new AtomicLong();
+
+  public final void incrementApplyErrors() {
+    filterApplyErrors.incrementAndGet();
+  }
+
+}
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/MessageFilters.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/MessageFilters.java
index ea0707f1029..e48501f6a75 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/MessageFilters.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/MessageFilters.java
@@ -9,6 +9,7 @@
 import javax.script.CompiledScript;
 import javax.script.ScriptEngineManager;
 import javax.script.ScriptException;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.codehaus.groovy.jsr223.GroovyScriptEngineImpl;
@@ -40,25 +41,26 @@ static Predicate<TopicMessageDTO> containsStringFilter(String string) {
   static Predicate<TopicMessageDTO> groovyScriptFilter(String script) {
     var compiledScript = compileScript(script);
     var jsonSlurper = new JsonSlurper();
-    return msg -> {
-      var bindings = getGroovyEngine().createBindings();
-      bindings.put("partition", msg.getPartition());
-      bindings.put("offset", msg.getOffset());
-      bindings.put("timestampMs", msg.getTimestamp().toInstant().toEpochMilli());
-      bindings.put("keyAsText", msg.getKey());
-      bindings.put("valueAsText", msg.getContent());
-      bindings.put("headers", msg.getHeaders());
-      bindings.put("key", parseToJsonOrReturnNull(jsonSlurper, msg.getKey()));
-      bindings.put("value", parseToJsonOrReturnNull(jsonSlurper, msg.getContent()));
-      try {
+    return new Predicate<TopicMessageDTO>() {
+      @SneakyThrows
+      @Override
+      public boolean test(TopicMessageDTO msg) {
+        var bindings = getGroovyEngine().createBindings();
+        bindings.put("partition", msg.getPartition());
+        bindings.put("offset", msg.getOffset());
+        bindings.put("timestampMs", msg.getTimestamp().toInstant().toEpochMilli());
+        bindings.put("keyAsText", msg.getKey());
+        bindings.put("valueAsText", msg.getContent());
+        bindings.put("headers", msg.getHeaders());
+        bindings.put("key", parseToJsonOrReturnNull(jsonSlurper, msg.getKey()));
+        bindings.put("value", parseToJsonOrReturnNull(jsonSlurper, msg.getContent()));
         var result = compiledScript.eval(bindings);
         if (result instanceof Boolean) {
           return (Boolean) result;
+        } else {
+          throw new ValidationException(
+              String.format("Unexpected script result: %s, Boolean should be returned instead", result));
         }
-        return false;
-      } catch (Exception e) {
-        log.trace("Error executing filter script '{}' on message '{}' ", script, msg, e);
-        return false;
       }
     };
   }
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/AdminClientServiceImpl.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/AdminClientServiceImpl.java
index 7b93b1420e0..9291198df73 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/AdminClientServiceImpl.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/AdminClientServiceImpl.java
@@ -37,6 +37,7 @@ private Mono<ReactiveAdminClient> createAdminClient(KafkaCluster cluster) {
       properties
           .put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.getBootstrapServers());
       properties.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, clientTimeout);
+      properties.putIfAbsent(AdminClientConfig.CLIENT_ID_CONFIG, "kafka-ui-app");
       return AdminClient.create(properties);
     })
         .flatMap(ReactiveAdminClient::create)
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/MessagesService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/MessagesService.java
index ecfeda01227..d0bf94c98b6 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/MessagesService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/MessagesService.java
@@ -2,6 +2,7 @@
 
 import com.provectus.kafka.ui.emitter.BackwardRecordEmitter;
 import com.provectus.kafka.ui.emitter.ForwardRecordEmitter;
+import com.provectus.kafka.ui.emitter.MessageFilterStats;
 import com.provectus.kafka.ui.emitter.MessageFilters;
 import com.provectus.kafka.ui.emitter.TailingEmitter;
 import com.provectus.kafka.ui.exception.TopicNotFoundException;
@@ -172,8 +173,10 @@ private Flux<TopicMessageEventDTO> loadMessagesImpl(KafkaCluster cluster,
           recordDeserializer
       );
     }
+    MessageFilterStats filterStats = new MessageFilterStats();
     return Flux.create(emitter)
-        .filter(getMsgFilter(query, filterQueryType))
+        .contextWrite(ctx -> ctx.put(MessageFilterStats.class, filterStats))
+        .filter(getMsgFilter(query, filterQueryType, filterStats))
         .takeWhile(createTakeWhilePredicate(seekDirection, limit))
         .subscribeOn(Schedulers.boundedElastic())
         .share();
@@ -186,7 +189,9 @@ private Predicate<TopicMessageEventDTO> createTakeWhilePredicate(
         : new ResultSizeLimiter(limit);
   }
 
-  private Predicate<TopicMessageEventDTO> getMsgFilter(String query, MessageFilterTypeDTO filterQueryType) {
+  private Predicate<TopicMessageEventDTO> getMsgFilter(String query,
+                                                       MessageFilterTypeDTO filterQueryType,
+                                                       MessageFilterStats filterStats) {
     if (StringUtils.isEmpty(query)) {
       return evt -> true;
     }
@@ -194,7 +199,13 @@ private Predicate<TopicMessageEventDTO> getMsgFilter(String query, MessageFilter
     return evt -> {
       // we only apply filter for message events
       if (evt.getType() == TopicMessageEventDTO.TypeEnum.MESSAGE) {
-        return messageFilter.test(evt.getMessage());
+        try {
+          return messageFilter.test(evt.getMessage());
+        } catch (Exception e) {
+          filterStats.incrementApplyErrors();
+          log.trace("Error applying filter '{}' for message {}", query, evt.getMessage());
+          return false;
+        }
       }
       return true;
     };
diff --git a/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml b/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
index a1ba87d85af..1034fc32025 100644
--- a/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
+++ b/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
@@ -2408,6 +2408,8 @@ components:
           type: boolean
         messagesConsumed:
           type: integer
+        filterApplyErrors:
+          type: integer
 
 
     TopicMessage:


Test patch:
diff --git a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/RecordEmitterTest.java b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/RecordEmitterTest.java
index 3289d177d2b..350386fd7e9 100644
--- a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/RecordEmitterTest.java
+++ b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/RecordEmitterTest.java
@@ -120,17 +120,17 @@ void pollNothingOnEmptyTopic() {
         RECORD_DESERIALIZER
     );
 
-    StepVerifier.create(
-        Flux.create(forwardEmitter)
-            .filter(m -> m.getType().equals(TopicMessageEventDTO.TypeEnum.MESSAGE))
-            .take(100)
-    ).expectNextCount(0).expectComplete().verify();
-
-    StepVerifier.create(
-        Flux.create(backwardEmitter)
-            .filter(m -> m.getType().equals(TopicMessageEventDTO.TypeEnum.MESSAGE))
-            .take(100)
-    ).expectNextCount(0).expectComplete().verify();
+    StepVerifier.create(Flux.create(forwardEmitter))
+        .expectNextMatches(m -> m.getType().equals(TopicMessageEventDTO.TypeEnum.PHASE))
+        .expectNextMatches(m -> m.getType().equals(TopicMessageEventDTO.TypeEnum.DONE))
+        .expectComplete()
+        .verify();
+
+    StepVerifier.create(Flux.create(backwardEmitter))
+        .expectNextMatches(m -> m.getType().equals(TopicMessageEventDTO.TypeEnum.PHASE))
+        .expectNextMatches(m -> m.getType().equals(TopicMessageEventDTO.TypeEnum.DONE))
+        .expectComplete()
+        .verify();
   }
 
   @Test