Merge pull request #891 from Steveb-p/allow-kafka-consume-headers

Allow reading headers from Kafka Message headers

Author: makasim

pkg/rdkafka/RdKafkaConsumer.php

@@ -169,6 +169,12 @@ private function doReceive(int $timeout): ?RdKafkaMessage
                 $message->setPartition($kafkaMessage->partition);
                 $message->setKafkaMessage($kafkaMessage);
 
+                // Merge headers passed from Kafka with possible earlier serialized payload headers. Prefer Kafka's.
+                // Note: Requires phprdkafka >= 3.1.0
+                if (isset($kafkaMessage->headers)) {
+                    $message->setHeaders(array_merge($message->getHeaders(), $kafkaMessage->headers));
+                }
+
                 return $message;
             default:
                 throw new \LogicException($kafkaMessage->errstr(), $kafkaMessage->err);