Merge pull request #959 from Steveb-p/phprdkafka-4

Compatibility with Phprdkafka 4.0

Author: makasim

.travis.yml

@@ -78,7 +78,7 @@ script:
     - if [ "$PHPSTAN" = true ] && [ ! -z "${PKG_PHP_CHANGED_FILES}" ]; then docker run --workdir="/mqdev" -v "`pwd`:/mqdev" --rm enqueue/dev:latest php -d memory_limit=1024M bin/phpstan analyse -l 1 -c phpstan.neon -- ${PKG_PHP_CHANGED_FILES[@]} ; fi
     - if [ "$UNIT_TESTS" = true ]; then bin/phpunit --exclude-group=functional; fi
     - if [ "$FUNCTIONAL_TESTS" = true ]; then bin/test.sh --exclude-group=rdkafka; fi
-    - if [ "RDKAFKA_TESTS" = true ]; then bin/test.sh --group=rdkafka; fi
+    - if [ "$RDKAFKA_TESTS" = true ]; then bin/test.sh --group=rdkafka; fi
 
 notifications:
   webhooks:

composer.json

@@ -59,7 +59,7 @@
     "doctrine/doctrine-bundle": "~1.2|^2",
     "doctrine/mongodb-odm-bundle": "^3.5|^4",
     "alcaeus/mongo-php-adapter": "^1.0",
-    "kwn/php-rdkafka-stubs": "^1.0.2",
+    "kwn/php-rdkafka-stubs": "^1.0.2 | ^2.0",
     "friendsofphp/php-cs-fixer": "^2"
   },
   "autoload": {

docker-compose.yml

@@ -99,13 +99,13 @@ services:
       - '2181:2181'
 
   kafka:
-    image: 'wurstmeister/kafka:0.10.2.1'
+    image: 'wurstmeister/kafka'
     ports:
       - '9092:9092'
     environment:
       KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
-    volumes:
-      - '/var/run/docker.sock:/var/run/docker.sock'
+      KAFKA_ADVERTISED_HOST_NAME: kafka
+      KAFKA_ADVERTISED_PORT: 9092
 
   google-pubsub:
       image: 'google/cloud-sdk:latest'

pkg/rdkafka/RdKafkaConnectionFactory.php

@@ -28,6 +28,7 @@ class RdKafkaConnectionFactory implements ConnectionFactory
      *     'partitioner' => null,                          // https://arnaud-lb.github.io/php-rdkafka/phpdoc/rdkafka-topicconf.setpartitioner.html
      *     'log_level' => null,
      *     'commit_async' => false,
+     *     'shutdown_timeout' => -1,                       // https://github.com/arnaud-lb/php-rdkafka#proper-shutdown
      * ]
      *
      * or

pkg/rdkafka/RdKafkaContext.php

@@ -36,7 +36,7 @@ class RdKafkaContext implements Context
     private $conf;
 
     /**
-     * @var Producer
+     * @var RdKafkaProducer
      */
     private $producer;
 
@@ -50,9 +50,6 @@ class RdKafkaContext implements Context
      */
     private $rdKafkaConsumers;
 
-    /**
-     * @param array $config
-     */
     public function __construct(array $config)
     {
         $this->config = $config;
@@ -96,7 +93,23 @@ public function createTemporaryQueue(): Queue
      */
     public function createProducer(): Producer
     {
-        return new RdKafkaProducer($this->getProducer(), $this->getSerializer());
+        if (!isset($this->producer)) {
+            $producer = new VendorProducer($this->getConf());
+
+            if (isset($this->config['log_level'])) {
+                $producer->setLogLevel($this->config['log_level']);
+            }
+
+            $this->producer = new RdKafkaProducer($producer, $this->getSerializer());
+
+            // Once created RdKafkaProducer can store messages internally that need to be delivered before PHP shuts
+            // down. Otherwise, we are bound to lose messages in transit.
+            // Note that it is generally preferable to call "close" method explicitly before shutdown starts, since
+            // otherwise we might not have access to some objects, like database connections.
+            register_shutdown_function([$this->producer, 'flush'], $this->config['shutdown_timeout'] ?? -1);
+        }
+
+        return $this->producer;
     }
 
     /**
@@ -139,6 +152,11 @@ public function close(): void
         foreach ($kafkaConsumers as $kafkaConsumer) {
             $kafkaConsumer->unsubscribe();
         }
+
+        // Compatibility with phprdkafka 4.0.
+        if (isset($this->producer)) {
+            $this->producer->flush($this->config['shutdown_timeout'] ?? -1);
+        }
     }
 
     public function createSubscriptionConsumer(): SubscriptionConsumer
@@ -163,19 +181,6 @@ public static function getLibrdKafkaVersion(): string
         return "$major.$minor.$patch";
     }
 
-    private function getProducer(): VendorProducer
-    {
-        if (null === $this->producer) {
-            $this->producer = new VendorProducer($this->getConf());
-
-            if (isset($this->config['log_level'])) {
-                $this->producer->setLogLevel($this->config['log_level']);
-            }
-        }
-
-        return $this->producer;
-    }
-
     private function getConf(): Conf
     {
         if (null === $this->conf) {

pkg/rdkafka/RdKafkaProducer.php

@@ -111,4 +111,12 @@ public function getTimeToLive(): ?int
     {
         return null;
     }
+
+    public function flush(int $timeout): void
+    {
+        // Flush method is exposed in phprdkafka 4.0
+        if (method_exists($this->producer, 'flush')) {
+            $this->producer->flush($timeout);
+        }
+    }
 }

pkg/rdkafka/Tests/Spec/RdKafkaSendToAndReceiveFromTopicTest.php

@@ -19,9 +19,14 @@ public function test()
 
         $topic = $this->createTopic($context, uniqid('', true));
 
-        $consumer = $context->createConsumer($topic);
-
         $expectedBody = __CLASS__.time();
+        $producer = $context->createProducer();
+        $producer->send($topic, $context->createMessage($expectedBody));
+
+        // Calling close causes Producer to flush (wait for messages to be delivered to Kafka)
+        $context->close();
+
+        $consumer = $context->createConsumer($topic);
 
         $context->createProducer()->send($topic, $context->createMessage($expectedBody));
 
@@ -48,8 +53,6 @@ protected function createContext()
 
         $context = (new RdKafkaConnectionFactory($config))->createContext();
 
-        sleep(3);
-
         return $context;
     }
 }

pkg/rdkafka/composer.json

@@ -7,15 +7,15 @@
     "license": "MIT",
     "require": {
         "php": "^7.1.3",
-        "ext-rdkafka": "^3.0.3",
+        "ext-rdkafka": "^3.0.3|^4.0",
         "queue-interop/queue-interop": "^0.8"
     },
     "require-dev": {
         "phpunit/phpunit": "~7.5",
         "enqueue/test": "0.10.x-dev",
         "enqueue/null": "0.10.x-dev",
         "queue-interop/queue-spec": "^0.6",
-        "kwn/php-rdkafka-stubs": "^1.0.2"
+        "kwn/php-rdkafka-stubs": "^1.0.2 | ^2.0"
     },
     "support": {
         "email": "[email protected]",