From a97d65f5903f03ac5cd182bcab7c006f8935ac81 Mon Sep 17 00:00:00 2001
From: tchiotludo <tchiot.ludo@gmail.com>
Date: Mon, 21 Jun 2021 22:18:38 +0200
Subject: [PATCH] feat(topic-data): search for tombstones in a topic

close #691
---
 .../akhq/repositories/RecordRepository.java   | 30 +++++++++++++++++--
 1 file changed, 27 insertions(+), 3 deletions(-)

diff --git a/src/main/java/org/akhq/repositories/RecordRepository.java b/src/main/java/org/akhq/repositories/RecordRepository.java
index 6825bdf60..5f35bf7bc 100644
--- a/src/main/java/org/akhq/repositories/RecordRepository.java
+++ b/src/main/java/org/akhq/repositories/RecordRepository.java
@@ -637,15 +637,21 @@ private static boolean searchFilter(BaseOptions options, Record record) {
             return search(options.getSearch(), Arrays.asList(record.getKey(), record.getValue()));
         } else {
             if (options.getSearchByKey() != null) {
-                if (!search(options.getSearchByKey(), Collections.singletonList(record.getKey()))) return false;
+                if (!search(options.getSearchByKey(), Collections.singletonList(record.getKey()))) {
+                    return false;
+                }
             }
 
             if (options.getSearchByValue() != null) {
-                if (!search(options.getSearchByValue(), Collections.singletonList(record.getValue()))) return false;
+                if (!search(options.getSearchByValue(), Collections.singletonList(record.getValue()))) {
+                    return false;
+                }
             }
 
             if (options.getSearchByHeaderKey() != null) {
-                if (!search(options.getSearchByHeaderKey(), record.getHeaders().keySet())) return false;
+                if (!search(options.getSearchByHeaderKey(), record.getHeaders().keySet())) {
+                    return false;
+                }
             }
 
             if (options.getSearchByHeaderValue() != null) {
@@ -667,6 +673,12 @@ private static boolean search(Search searchFilter, Collection<String> stringsToS
     }
 
     private static boolean containsAll(String search, Collection<String> in) {
+        if (search.equals("null")) {
+            return in
+                .stream()
+                .allMatch(Objects::isNull);
+        }
+
         String[] split = search.toLowerCase().split("\\s");
         for (String s : in) {
             if(s != null) {
@@ -682,6 +694,12 @@ private static boolean containsAll(String search, Collection<String> in) {
     }
 
     private static boolean equalsAll(String search, Collection<String> in) {
+        if (search.equals("null")) {
+            return in
+                .stream()
+                .allMatch(Objects::isNull);
+        }
+
         String[] split = search.toLowerCase().split("\\s");
         for (String s : in) {
             if(s != null) {
@@ -697,6 +715,12 @@ private static boolean equalsAll(String search, Collection<String> in) {
     }
 
     private static boolean notContainsAll(String search, Collection<String> in) {
+        if (search.equals("null")) {
+            return in
+                .stream()
+                .noneMatch(Objects::isNull);
+        }
+
         String[] split = search.toLowerCase().split("\\s");
         for (String s : in) {
             if(s != null) {