Convert Kafka Header values to Strings (#6507)

* Convert Kafka Header values to Strings

Signed-off-by: Kondaka <krishkdk@amazon.com>

* Addressed review comments

Signed-off-by: Kondaka <krishkdk@amazon.com>

Signed-off-by: Kondaka <krishkdk@amazon.com>

* Fixed header value parsing logic

Signed-off-by: Kondaka <krishkdk@amazon.com>

* Addressed review comments. Created a new class for header extraction

Signed-off-by: Kondaka <krishkdk@amazon.com>

* Addressed review comments

Signed-off-by: Kondaka <krishkdk@amazon.com>

* Fixed license header failures

Signed-off-by: Kondaka <krishkdk@amazon.com>

---------

Signed-off-by: Kondaka <krishkdk@amazon.com>

Author: kkondaka

data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java

@@ -27,8 +27,6 @@
 import org.apache.kafka.common.errors.RebalanceInProgressException;
 import org.apache.kafka.common.utils.ExponentialBackoff;
 import org.apache.kafka.common.errors.RecordDeserializationException;
-import org.apache.kafka.common.header.Header;
-import org.apache.kafka.common.header.Headers;
 import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSet;
 import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSetManager;
 import org.opensearch.dataprepper.model.buffer.Buffer;
@@ -504,12 +502,8 @@ private <T> Record<Event> getRecord(ConsumerRecord<String, T> consumerRecord, in
         if (kafkaKeyMode == KafkaKeyMode.INCLUDE_AS_METADATA) {
             eventMetadata.setAttribute("kafka_key", key);
         }
-        Headers headers = consumerRecord.headers();
-        if (headers != null) {
-            Map<String, byte[]> headerData = new HashMap<>();
-            for (Header header: headers) {
-                headerData.put(header.key(), header.value());
-            }
+        Map<String, Object> headerData = KafkaHeadersExtractor.extractMessageHeaders(consumerRecord.headers());
+        if (headerData != null) {
             eventMetadata.setAttribute("kafka_headers", headerData);
         }
         final long receivedTimeStamp = getRecordTimeStamp(consumerRecord, now.toEpochMilli());

data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaHeadersExtractor.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper.plugins.kafka.consumer;
+
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+public class KafkaHeadersExtractor {
+    public static Map<String, Object> extractMessageHeaders(Headers headers) {
+        if (headers == null) {
+            return null;
+        }
+        Map<String, Object> headerData = new HashMap<>();
+        for (Header header : headers) {
+            byte[] headerValue = header.value();
+            if (headerValue == null) {
+                headerData.put(header.key(), null);
+                continue;
+            }
+            String strValue = new String(headerValue, StandardCharsets.UTF_8);
+            if (Arrays.equals(headerValue, strValue.getBytes(StandardCharsets.UTF_8))
+                    && isPrintableString(strValue)) {
+                headerData.put(header.key(), strValue);
+            } else {
+                headerData.put(header.key(), headerValue);
+            }
+        }
+        return headerData;
+    }
+
+    private static boolean isPrintableString(String value) {
+        for (int i = 0; i < value.length(); i++) {
+            char c = value.charAt(i);
+            if (Character.isISOControl(c) && c != '\t' && c != '\n' && c != '\r') {
+                return false;
+            }
+        }
+        return true;
+    }
+}