Support headers

extensions/eda/plugins/event_source/kafka.py

@@ -29,6 +29,7 @@
                Default to PLAIN.
     sasl_plain_username: Username for SASL PLAIN authentication
     sasl_plain_password: Password for SASL PLAIN authentication
+    with_headers: Include message headers. [True, False (default)]
 
 
 
@@ -64,6 +65,7 @@ async def main(  # pylint: disable=R0914
     offset = args.get("offset", "latest")
     encoding = args.get("encoding", "utf-8")
     security_protocol = args.get("security_protocol", "PLAINTEXT")
+    with_headers = args.get("with_headers", False)
 
     if offset not in ("latest", "earliest"):
         msg = f"Invalid offset option: {offset}"
@@ -110,6 +112,7 @@ async def main(  # pylint: disable=R0914
 
     try:
         async for msg in kafka_consumer:
+            event = {}
             data = None
             try:
                 value = msg.value.decode(encoding)
@@ -119,8 +122,22 @@ async def main(  # pylint: disable=R0914
             except UnicodeError:
                 logger.exception("Unicode Error")
 
+            if with_headers:
+                headers = None
+                try:
+                    headers_dict = {header[0]: header[1].decode(encoding) for header in msg.headers}
+                    headers_json = json.dumps(headers_dict)
+                    headers = json.loads(headers_json)
+                except UnicodeError:
+                    logger.exception("Unicode Error")
+
+                if headers:
+                   event["headers"] = headers
+
             if data:
-                await queue.put({"body": data})
+                event["body"] = data
+                await queue.put(event)
+
             await asyncio.sleep(0)
     finally:
         logger.info("Stopping kafka consumer")