Maintain backward compatibility

TiMESPLiNTER · TiMESPLiNTER · commit 733caddf5493 · 2019-09-25T11:15:29.000+02:00
diff --git a/pkg/rdkafka/RdKafkaContext.php b/pkg/rdkafka/RdKafkaContext.php
@@ -138,6 +138,17 @@ public function purgeQueue(Queue $queue): void
         throw PurgeQueueNotSupportedException::providerDoestNotSupportIt();
     }
 
+    public static function getLibrdKafkaVersion(): string
+    {
+        if (!defined('RD_KAFKA_VERSION')) {
+            throw new \RuntimeException('RD_KAFKA_VERSION constant is not defined. Phprdkafka is probably not installed');
+        }
+        $major = (RD_KAFKA_VERSION & 0xFF000000) >> 24;
+        $minor = (RD_KAFKA_VERSION & 0x00FF0000) >> 16;
+        $patch = (RD_KAFKA_VERSION & 0x0000FF00) >> 8;
+        return "$major.$minor.$patch";
+    }
+
     private function getProducer(): VendorProducer
     {
         if (null === $this->producer) {
diff --git a/pkg/rdkafka/RdKafkaProducer.php b/pkg/rdkafka/RdKafkaProducer.php
@@ -42,7 +42,24 @@ public function send(Destination $destination, Message $message): void
         $key = $message->getKey() ?: $destination->getKey() ?: null;
 
         $topic = $this->producer->newTopic($destination->getTopicName(), $destination->getConf());
-        $topic->producev($partition, 0 /* must be 0 */, $payload, $key, $message->getHeaders());
+
+        // Note: Topic::producev method exists in phprdkafka > 3.1.0
+        // Headers in payload are maintained for backwards compatibility with apps that might run on lower phprdkafka version
+        if (method_exists($topic, 'producev')) {
+            // Phprdkafka <= 3.1.0 will fail calling `producev` on librdkafka >= 1.0.0 causing segfault
+            if (version_compare(RdKafkaContext::getLibrdKafkaVersion(), '1.0.0', '>=')
+                && version_compare(phpversion('rdkafka'), '3.1.0', '<=')) {
+                trigger_error(
+                    'Phprdkafka <= 3.1.0 is incompatible with librdkafka 1.0.0 when calling `producev`',
+                    E_USER_WARNING
+                );
+            } else {
+                $topic->producev($partition, 0 /* must be 0 */, $payload, $key, $message->getHeaders());
+                return;
+            }
+        }
+
+        $topic->produce($partition, 0 /* must be 0 */, $payload, $key);
     }
 
     /**
