fix(topicdata): fix null key and value (#1261)

Co-authored-by: alozano3 <[email protected]>
tchiotludo · Apr 4, 2023 · 2ab8506 · 2ab8506
1 parent 1c11ac1
commit 2ab8506
Showing 1 changed file with 3 additions and 2 deletions.
diff --git a/src/main/java/org/akhq/repositories/RecordRepository.java b/src/main/java/org/akhq/repositories/RecordRepository.java
@@ -39,6 +39,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import jakarta.inject.Inject;
@@ -606,7 +607,7 @@ public RecordMetadata produce(
                 SchemaSerializer keySerializer = serializerFactory.createSerializer(clusterId, keySchemaId.get());
                 keyAsBytes = keySerializer.serialize(key.get());
             } else {
-                keyAsBytes = key.get().getBytes();
+                keyAsBytes = key.filter(Predicate.not(String::isEmpty)).map(String::getBytes).orElse(null);
             }
         } else {
             try {
@@ -622,7 +623,7 @@ public RecordMetadata produce(
             SchemaSerializer valueSerializer = serializerFactory.createSerializer(clusterId, valueSchemaId.get());
             valueAsBytes = valueSerializer.serialize(value.get());
         } else {
-            valueAsBytes = value.map(String::getBytes).orElse(null);
+            valueAsBytes = value.filter(Predicate.not(String::isEmpty)).map(String::getBytes).orElse(null);
         }
 
         return produce(clusterId, topic, valueAsBytes, headers, keyAsBytes, partition, timestamp);