From 32adfe912334594b3598d09146118cde3ddbfa98 Mon Sep 17 00:00:00 2001
From: Dmitrii Bocharov <bdshadow@gmail.com>
Date: Wed, 3 Nov 2021 08:28:54 +0100
Subject: [PATCH] Fix NPE in CloudEventDeserializer when deserializing header
 with null value (#415)

Signed-off-by: Dmitrii Bocharov <dmitrii.bocharov@embedit.cz>
---
 .../BaseGenericBinaryMessageReaderImpl.java   |  3 +++
 .../kafka/CloudEventDeserializerTest.java     | 21 +++++++++++++++++++
 2 files changed, 24 insertions(+)

diff --git a/core/src/main/java/io/cloudevents/core/message/impl/BaseGenericBinaryMessageReaderImpl.java b/core/src/main/java/io/cloudevents/core/message/impl/BaseGenericBinaryMessageReaderImpl.java
index 38b1506f..cf0b00cc 100644
--- a/core/src/main/java/io/cloudevents/core/message/impl/BaseGenericBinaryMessageReaderImpl.java
+++ b/core/src/main/java/io/cloudevents/core/message/impl/BaseGenericBinaryMessageReaderImpl.java
@@ -56,6 +56,9 @@ public abstract class BaseGenericBinaryMessageReaderImpl<HK, HV> extends BaseBin
         // This implementation avoids to use visitAttributes and visitExtensions
         // in order to complete the visit in one loop
         this.forEachHeader((key, value) -> {
+            if (value == null) {
+                return;
+            }
             if (isContentTypeHeader(key)) {
                 visitor.withContextAttribute(CloudEventV1.DATACONTENTTYPE, toCloudEventsValue(value));
             } else if (isCloudEventsHeader(key)) {
diff --git a/kafka/src/test/java/io/cloudevents/kafka/CloudEventDeserializerTest.java b/kafka/src/test/java/io/cloudevents/kafka/CloudEventDeserializerTest.java
index ca92b425..6ab7606b 100644
--- a/kafka/src/test/java/io/cloudevents/kafka/CloudEventDeserializerTest.java
+++ b/kafka/src/test/java/io/cloudevents/kafka/CloudEventDeserializerTest.java
@@ -23,6 +23,7 @@ import io.cloudevents.core.mock.MyCloudEventData;
 import io.cloudevents.core.test.Data;
 import io.cloudevents.rw.CloudEventDataMapper;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.Headers;
 import org.junit.jupiter.api.Test;
 
 import java.util.HashMap;
@@ -60,6 +61,26 @@ public class CloudEventDeserializerTest {
         );
     }
 
+    @Test
+    public void deserializerShouldWorkWithNullableManuallyDefinedHeaders() {
+        String topic = "test";
+        CloudEvent testCloudEvent = Data.V1_MIN;
+        CloudEventDeserializer cloudEventDeserializer = new CloudEventDeserializer();
+
+        // Serialize the event first
+        ProducerRecord<Void, byte[]> inRecord = KafkaMessageFactory
+            .createWriter(topic)
+            .writeBinary(testCloudEvent);
+
+        // add optional subject header with null value
+        Headers headers = inRecord.headers();
+        headers.add("ce_subject", null);
+        CloudEvent outEvent = cloudEventDeserializer.deserialize(topic, headers, inRecord.value());
+
+        assertThat(outEvent)
+            .isEqualTo(testCloudEvent);
+    }
+
     private void testDeserialize(CloudEventDeserializer deserializer, CloudEvent input, CloudEvent expected) {
         String topic = "test";