Hello there michal-cesek! 👋

Thank you and congratulations 🎉 for opening your very first issue in this project! 💖

In case you want to claim this issue, please comment down below! We will try to get back to you as soon as we can. 👀 Hello @michal-cesek 
thank you for raising this!

Currently we assume that `value` is a json (obj/string/integer). If value is not valid json ( like string `123notajson` ) we set it to null. So, there can be 2 cases then `value` is null in filter: 
- kafka-record's value is null 
- kafka-record's value can't be represented as json after deserialization 

This is why you see false-positive filters. 

We will discuss it internally and think how to fix this. For now, as workaround you can use `valueAsText` variable in filter that will be null only if kafka-record's value is null.

cc @Haarolean  @iliax thank you for taking care of this. In addition, I wanted to point out that there is inconsistency in the UI. The second screenshot shows an active filter and the search box input filled with the code -> Ok. The third screenshot shows the same active filter, but the search box is not prefilled -> Not Ok / Confusing

![screenshot](https://user-images.githubusercontent.com/2612817/236838774-027c6452-2ff7-48fe-94b6-31dedb55e1cc.png)

cc @Haarolean  @michal-cesek we have a huge messages view refactoring planned -- #3504 
Topic message filter is displayed but not applied
### Issue submitter TODO list

- [X] I've looked up my issue in [FAQ](https://docs.kafka-ui.provectus.io/faq/common-problems)
- [X] I've searched for an already existing issues [here](https://github.com/provectus/kafka-ui/issues)
- [X] I've tried running `master`-labeled docker image and the issue still persists there
- [X] I'm running a supported version of the application which is listed [here](https://github.com/provectus/kafka-ui/blob/master/SECURITY.md)

### Describe the bug (actual behavior)

The active message filter component is displayed but no filtering is applied to the topic message list

### Expected behavior

The filter is applied to the topic message list

### Your installation details

1. b0c367c - v0.6.2

### Steps to reproduce

**1.** Set up the filter by using the "Add filters" functionality

![filter](https://user-images.githubusercontent.com/2612817/234866765-50adf24c-bd12-4f59-9a9b-1479a619d4d0.png)

![ok](https://user-images.githubusercontent.com/2612817/234863606-9e6fcf2e-121a-443b-9b2c-14518fe3aeb7.png)

**2.** Leave the "Messages" view. E.g. click on the "Brokers" link from the left panel

**3.** Return to the view from the first step

**RESULT:** The filter is not applied and the unexpected topic message is shown

![notok](https://user-images.githubusercontent.com/2612817/234864596-f827732d-d877-40b0-8328-d72a8b673824.png)
 

### Screenshots

_No response_

### Logs

_No response_

### Additional context

_No response_
kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/MessageFilters.java
kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/MessageFilters.java
kafka-ui-api/src/test/java/com/provectus/kafka/ui/emitter/MessageFiltersTest.java
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 e48501f6a75..6e9f8a8bbe3 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
@@ -39,41 +39,42 @@ static Predicate<TopicMessageDTO> containsStringFilter(String string) {
   }
 
   static Predicate<TopicMessageDTO> groovyScriptFilter(String script) {
-    var compiledScript = compileScript(script);
+    var engine = getGroovyEngine();
+    var compiledScript = compileScript(engine, script);
     var jsonSlurper = new JsonSlurper();
     return new Predicate<TopicMessageDTO>() {
       @SneakyThrows
       @Override
       public boolean test(TopicMessageDTO msg) {
-        var bindings = getGroovyEngine().createBindings();
+        var bindings = engine.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()));
+        bindings.put("key", parseToJsonOrReturnAsIs(jsonSlurper, msg.getKey()));
+        bindings.put("value", parseToJsonOrReturnAsIs(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));
+              "Unexpected script result: %s, Boolean should be returned instead".formatted(result));
         }
       }
     };
   }
 
   @Nullable
-  private static Object parseToJsonOrReturnNull(JsonSlurper parser, @Nullable String str) {
+  private static Object parseToJsonOrReturnAsIs(JsonSlurper parser, @Nullable String str) {
     if (str == null) {
       return null;
     }
     try {
       return parser.parseText(str);
     } catch (Exception e) {
-      return null;
+      return str;
     }
   }
 
@@ -86,9 +87,9 @@ private static synchronized GroovyScriptEngineImpl getGroovyEngine() {
     return GROOVY_ENGINE;
   }
 
-  private static CompiledScript compileScript(String script) {
+  private static CompiledScript compileScript(GroovyScriptEngineImpl engine, String script) {
     try {
-      return getGroovyEngine().compile(script);
+      return engine.compile(script);
     } catch (ScriptException e) {
       throw new ValidationException("Script syntax error: " + e.getMessage());
     }

diff --git a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/emitter/MessageFiltersTest.java b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/emitter/MessageFiltersTest.java
index 94a377c9c84..4e9f5034cd2 100644
--- a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/emitter/MessageFiltersTest.java
+++ b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/emitter/MessageFiltersTest.java
@@ -118,10 +118,18 @@ void canCheckKeyAsJsonObjectIfItCanBeParsedToJson() {
     }
 
     @Test
-    void keySetToNullIfKeyCantBeParsedToJson() {
-      var f = groovyScriptFilter("key == null");
+    void keySetToKeyStringIfCantBeParsedToJson() {
+      var f = groovyScriptFilter("key == \"not json\"");
       assertTrue(f.test(msg().key("not json")));
-      assertFalse(f.test(msg().key("{ \"k\" : \"v\" }")));
+    }
+
+    @Test
+    void keyAndKeyAsTextSetToNullIfRecordsKeyIsNull() {
+      var f = groovyScriptFilter("key == null");
+      assertTrue(f.test(msg().key(null)));
+
+      f = groovyScriptFilter("keyAsText == null");
+      assertTrue(f.test(msg().key(null)));
     }
 
     @Test
@@ -132,10 +140,18 @@ void canCheckValueAsJsonObjectIfItCanBeParsedToJson() {
     }
 
     @Test
-    void valueSetToNullIfKeyCantBeParsedToJson() {
-      var f = groovyScriptFilter("value == null");
+    void valueSetToContentStringIfCantBeParsedToJson() {
+      var f = groovyScriptFilter("value == \"not json\"");
       assertTrue(f.test(msg().content("not json")));
-      assertFalse(f.test(msg().content("{ \"k\" : \"v\" }")));
+    }
+
+    @Test
+    void valueAndValueAsTextSetToNullIfRecordsContentIsNull() {
+      var f = groovyScriptFilter("value == null");
+      assertTrue(f.test(msg().content(null)));
+
+      f = groovyScriptFilter("valueAsText == null");
+      assertTrue(f.test(msg().content(null)));
     }
 
     @Test
@@ -185,4 +201,4 @@ private TopicMessageDTO msg() {
         .partition(1);
   }
 
-}
\ No newline at end of file
+}
