Shrohilla/kafka header bindings

src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaProducerAsyncCollector.cs

@@ -2,8 +2,10 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
+using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using Newtonsoft.Json.Linq;
 
 namespace Microsoft.Azure.WebJobs.Extensions.Kafka
 {
@@ -32,14 +34,60 @@ public Task AddAsync(T item, CancellationToken cancellationToken)
 
             object messageToSend = item;
 
-            if (item.GetType() == typeof(string) || item.GetType() == typeof(byte[]))
+            if (item.GetType() == typeof(string))
+            {
+                messageToSend = convertToKafkaEventData(item);
+            }
+
+            if (item.GetType() == typeof(byte[]))
             {
                 messageToSend = new KafkaEventData<T>(item);
             }
 
             return entity.SendAndCreateEntityIfNotExistsAsync(messageToSend, functionInstanceId, cancellationToken);
         }
 
+        private object convertToKafkaEventData(T item)
+        {
+            try
+            {
+                return buildKafkaDataEvent(item);
+            }
+            catch (Exception)
+            {
+                return new KafkaEventData<T>(item);
+            }
+        }
+
+        private object buildKafkaDataEvent(T item)
+        {
+            JObject dataObj = JObject.Parse(item.ToString());
+            if (dataObj == null)
+            {
+                return new KafkaEventData<T>(item);
+            }
+
+            if (dataObj.ContainsKey("Offset") && dataObj.ContainsKey("Partition") && dataObj.ContainsKey("Topic")
+                && dataObj.ContainsKey("Timestamp") && dataObj.ContainsKey("Value") && dataObj.ContainsKey("Headers"))
+            {
+                return buildKafkaEventData(dataObj);
+            }
+
+            return new KafkaEventData<T>(item);
+        }
+
+        private KafkaEventData<string> buildKafkaEventData(JObject dataObj)
+        {
+            KafkaEventData<string> messageToSend = new KafkaEventData<string>((string)dataObj["Value"]);
+            messageToSend.Timestamp = (DateTime)dataObj["Timestamp"];
+            messageToSend.Partition = (int)dataObj["Partition"];
+            JArray headerList = (JArray)dataObj["Headers"];
+            foreach ( JObject header in headerList) {
+                messageToSend.Headers.Add((string)header["Key"], Encoding.Unicode.GetBytes((string)header["Value"]));
+            }
+            return messageToSend;
+        }
+
         public Task FlushAsync(CancellationToken cancellationToken = default(CancellationToken))
         {
             // Batching not supported.