Fix NPE in CloudEventDeserializer when deserializing header with null value (#415)

Signed-off-by: Dmitrii Bocharov <dmitrii.bocharov@embedit.cz>

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)) {

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";