BE: Support sending null headers (#651)

Co-authored-by: Roman Zabaluev <[email protected]>

Author: AkashDeepSinghJassal

api/src/main/java/io/kafbat/ui/serdes/ProducerRecordCreator.java

@@ -31,7 +31,7 @@ public ProducerRecord<byte[], byte[]> create(String topic,
 
   private Iterable<Header> createHeaders(Map<String, String> clientHeaders) {
     RecordHeaders headers = new RecordHeaders();
-    clientHeaders.forEach((k, v) -> headers.add(new RecordHeader(k, v.getBytes())));
+    clientHeaders.forEach((k, v) -> headers.add(new RecordHeader(k, v == null ? null : v.getBytes())));
     return headers;
   }
 

api/src/test/java/io/kafbat/ui/service/SendAndReadTests.java

@@ -19,6 +19,7 @@
 import io.kafbat.ui.serdes.builtin.StringSerde;
 import io.kafbat.ui.serdes.builtin.sr.SchemaRegistrySerde;
 import java.time.Duration;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -425,6 +426,25 @@ void topicMessageMetadataJson() {
         });
   }
 
+  @Test
+  void headerValueNullPresentTest() {
+    new SendAndReadSpec()
+        .withKeySchema(JSON_SCHEMA)
+        .withValueSchema(JSON_SCHEMA)
+        .withMsgToSend(
+            new CreateTopicMessageDTO()
+                .key(JSON_SCHEMA_RECORD)
+                .keySerde(SchemaRegistrySerde.name())
+                .content(JSON_SCHEMA_RECORD)
+                .valueSerde(SchemaRegistrySerde.name())
+                .headers(Collections.singletonMap("header123", null))
+        )
+        .doAssert(polled -> {
+          assertThat(polled.getHeaders().get("header123")).isNull();
+        });
+  }
+
+
   @Test
   void noKeyAndNoContentPresentTest() {
     new SendAndReadSpec()