fix: NPE in KafkaSteps when consuming records with null header values (#795)

* Initial plan

* Initial plan

* Fix NPE when consuming Kafka records with null header values

Co-authored-by: denouche <3081451+denouche@users.noreply.github.com>

* Fix NullPointerException when Kafka header value is null

Co-authored-by: denouche <3081451+denouche@users.noreply.github.com>

* Use HashMap collector to support null header values

Co-authored-by: denouche <3081451+denouche@users.noreply.github.com>

* fix tests

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: denouche <3081451+denouche@users.noreply.github.com>

Author: 2 people

tzatziki-spring-kafka/src/main/java/com/decathlon/tzatziki/steps/KafkaSteps.java

@@ -417,7 +417,7 @@ private ProducerRecord<String, GenericRecord> mapToAvroRecord(Schema schema, Str
 
         ProducerRecord<String, GenericRecord> producerRecord = new ProducerRecord<>(topic, messageKey, genericRecordMessage);
         ((Map<String, String>) avroRecord.get("headers"))
-                .forEach((key, value) -> producerRecord.headers().add(key, value.getBytes(UTF_8)));
+                .forEach((key, value) -> producerRecord.headers().add(key, value != null ? value.getBytes(UTF_8) : null));
 
         return producerRecord;
     }
@@ -479,7 +479,7 @@ public ProducerRecord<String, String> mapToJsonRecord(String topic, Map<?, Objec
         String messageKey = (String) jsonRecord.get("key");
         ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, messageKey, Mapper.toJson(jsonRecord.get("value")));
         ((Map<String, String>) jsonRecord.get("headers"))
-                .forEach((key, value) -> producerRecord.headers().add(key, value.getBytes(UTF_8)));
+                .forEach((key, value) -> producerRecord.headers().add(key, value!=null ? value.getBytes(UTF_8):null));
 
         return producerRecord;
     }
@@ -544,7 +544,9 @@ private List<Map<String, Object>> consumerRecordsToMaps(ConsumerRecords<?, ?> re
         return StreamSupport.stream(records.spliterator(), false)
                 .map(record -> {
                     Map<String, String> headers = Stream.of(record.headers().toArray())
-                            .collect(Collectors.toMap(Header::key, header -> new String(header.value())));
+                            .collect(HashMap::new,
+                                    (map, header) -> map.put(header.key(), header.value() != null ? new String(header.value()) : null),
+                                    HashMap::putAll);
                     Map<String, Object> value = Mapper.read(record.value().toString());
                     String messageKey = record.key() != null ? String.valueOf(record.key()) : "";
                     return Map.of("value", value, "headers", headers, "key", messageKey);

tzatziki-spring-kafka/src/test/resources/com/decathlon/tzatziki/steps/kafka.feature

@@ -426,6 +426,44 @@ Feature: to interact with a spring boot service having a connection to a kafka q
       """
     And the exposed-users topic contains 1 user
 
+  Scenario: we can handle null header values in messages
+    When this user is consumed from the json-users-with-key topic:
+      """yml
+      headers:
+        uuid: some-id
+        nullable-header: null
+      value:
+        id: 1
+        name: bob
+      key: a-key
+      """
+    Then we have received 1 message on the topic json-users-with-key
+
+  Scenario: we can handle null header values sent on a topic
+    When this json message is published on the json-users topic:
+      """yml
+      headers:
+        uuid: some-id
+        nullable-header: null
+        empty-header: ""
+      value:
+        id: 1
+        name: bob
+      key: a-key
+      """
+    Then the json-users topic contains only this json message:
+      """yml
+      headers:
+        uuid: some-id
+        nullable-header: null
+        empty-header: ""
+      value:
+        id: 1
+        name: bob
+      key: a-key
+      """
+    And the json-users topic contains 1 json message
+
   Scenario: we can assert that no message has been sent to a topic
     * the exposed-users topic contains 0 user