Repository: provectus/kafka-ui
Issue: Message display crash on message with custom header
Labels: type/bug

**Describe the bug**
When opening a topic containing messages with custom headers application crashes:
```
23:05:01.764 [boundedElastic-1] ERROR org.springframework.boot.autoconfigure.web.reactive.error.AbstractErrorWebExceptionHandler - [ff384bf8]  500 Server Error for HTTP GET "/api/clusters/local/topics/dlt/messages?limit=100"
java.lang.NullPointerException: null
	at java.lang.String.<init>(String.java:630) ~[?:?]
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
	|_ checkpoint ⇢ Handler com.provectus.kafka.ui.controller.MessagesController#getTopicMessages(String, String, SeekType, List, Integer, String, ServerWebExchange) [DispatcherHandler]
	|_ checkpoint ⇢ com.provectus.kafka.ui.config.CustomWebFilter [DefaultWebFilterChain]
	|_ checkpoint ⇢ com.provectus.kafka.ui.config.ReadOnlyModeFilter [DefaultWebFilterChain]
	|_ checkpoint ⇢ org.springframework.security.web.server.authorization.AuthorizationWebFilter [DefaultWebFilterChain]
	|_ checkpoint ⇢ org.springframework.security.web.server.authorization.ExceptionTranslationWebFilter [DefaultWebFilterChain]
	|_ checkpoint ⇢ org.springframework.security.web.server.authentication.logout.LogoutWebFilter [DefaultWebFilterChain]
	|_ checkpoint ⇢ org.springframework.security.web.server.savedrequest.ServerRequestCacheWebFilter [DefaultWebFilterChain]
	|_ checkpoint ⇢ org.springframework.security.web.server.context.SecurityContextServerWebExchangeWebFilter [DefaultWebFilterChain]
	|_ checkpoint ⇢ org.springframework.security.web.server.context.ReactorContextWebFilter [DefaultWebFilterChain]
	|_ checkpoint ⇢ org.springframework.security.web.server.header.HttpHeaderWriterWebFilter [DefaultWebFilterChain]
	|_ checkpoint ⇢ org.springframework.security.config.web.server.ServerHttpSecurity$ServerWebExchangeReactorContextWebFilter [DefaultWebFilterChain]
	|_ checkpoint ⇢ org.springframework.security.web.server.WebFilterChainProxy [DefaultWebFilterChain]
	|_ checkpoint ⇢ HTTP GET "/api/clusters/local/topics/dlt/messages?limit=100" [ExceptionHandlingWebHandler]
Stack trace:
		at java.lang.String.<init>(String.java:630) ~[?:?]
		at com.provectus.kafka.ui.util.ClusterUtil.lambda$mapToTopicMessage$15(ClusterUtil.java:203) ~[classes!/:?]
		at java.util.Iterator.forEachRemaining(Iterator.java:133) ~[?:?]
		at com.provectus.kafka.ui.util.ClusterUtil.mapToTopicMessage(ClusterUtil.java:203) ~[classes!/:?]
		at com.provectus.kafka.ui.service.ConsumingService.lambda$loadMessages$2(ConsumingService.java:61) ~[classes!/:?]
		at reactor.core.publisher.FluxMap$MapConditionalSubscriber.onNext(FluxMap.java:199) [reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
		at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.onNext(FluxSubscribeOn.java:151) [reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
		at reactor.core.publisher.FluxCreate$BufferAsyncSink.drain(FluxCreate.java:793) [reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
		at reactor.core.publisher.FluxCreate$BufferAsyncSink.next(FluxCreate.java:718) [reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
		at reactor.core.publisher.FluxCreate$SerializedSink.next(FluxCreate.java:153) [reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
		at com.provectus.kafka.ui.service.ConsumingService$RecordEmitter.accept(ConsumingService.java:146) [classes!/:?]
		at com.provectus.kafka.ui.service.ConsumingService$RecordEmitter.accept(ConsumingService.java:128) [classes!/:?]
		at reactor.core.publisher.FluxCreate.subscribe(FluxCreate.java:94) [reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
		at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.run(FluxSubscribeOn.java:194) [reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
		at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) [reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
		at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) [reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
		at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
		at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) [?:?]
		at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
		at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
		at java.lang.Thread.run(Thread.java:830) [?:?]
23:05:01.772 [boundedElastic-1] DEBUG org.springframework.http.codec.json.Jackson2JsonEncoder - [ff384bf8] Encoding [class ErrorResponse {
    code: 5000
    message: Unexpected internal error
    timestamp: 163036470 (truncated)...]
```


**Set up**
I'm running kafka, zookeeper and kafka-ui as a docker compose


**Steps to Reproduce**
Steps to reproduce the behaviour:

1.  publish a message to a topic with custom headers: `echo '1:{"id":25,"name":"pippo"}' | kafkacat -b kafka:9092 -t test.topic -H "name=value" -K: -P
2. open the topic messages ui

**Expected behavior**
Message being displayed in the topic messages list

Issue comments:
Hi, @rlogiacco. Thanks for creating this issue. According to the stack trace, I assume you are using version 0.1. I checked this part of the code and only one possible situation could be there if your header value is null. I created PR to fix this. Could you please check the version with tag master, after we'll merge it?

Pull request review comments:

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/ClusterUtil.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/ClusterUtil.java
index ccd8dd47ad4..3c21232539f 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/ClusterUtil.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/ClusterUtil.java
@@ -281,9 +281,15 @@ public static int convertToIntServerStatus(ServerStatus serverStatus) {
 
   public static TopicMessage mapToTopicMessage(ConsumerRecord<Bytes, Bytes> consumerRecord,
                                                RecordSerDe recordDeserializer) {
+
     Map<String, String> headers = new HashMap<>();
     consumerRecord.headers().iterator()
-        .forEachRemaining(header -> headers.put(header.key(), new String(header.value())));
+        .forEachRemaining(header ->
+            headers.put(
+                header.key(),
+                header.value() != null ? new String(header.value()) : null
+            )
+    );
 
     TopicMessage topicMessage = new 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 39f57c797e4..e7147dea1cc 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
@@ -15,6 +15,7 @@
 import com.provectus.kafka.ui.util.OffsetsSeekForward;
 import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -29,6 +30,8 @@
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
 import org.apache.kafka.common.serialization.BytesDeserializer;
 import org.apache.kafka.common.utils.Bytes;
 import org.junit.jupiter.api.AfterAll;
@@ -57,13 +60,21 @@ static void generateMsgs() throws Exception {
         for (int i = 0; i < MSGS_PER_PARTITION; i++) {
           long ts = System.currentTimeMillis() + i;
           var value = "msg_" + partition + "_" + i;
-          var metadata =
-              producer.send(new ProducerRecord<>(TOPIC, partition, ts, null, value)).get();
-          SENT_RECORDS.add(new Record(
-              value,
-              new TopicPartition(metadata.topic(), metadata.partition()),
-              metadata.offset(),
-              ts)
+          var metadata = producer.send(
+              new ProducerRecord<>(
+                  TOPIC, partition, ts, null, value, List.of(
+                      new RecordHeader("name", null),
+                      new RecordHeader("name2", "value".getBytes())
+                  )
+              )
+          ).get();
+          SENT_RECORDS.add(
+              new Record(
+                  value,
+                  new TopicPartition(metadata.topic(), metadata.partition()),
+                  metadata.offset(),
+                  ts
+              )
           );
         }
       }