Call flush method on shutdown / context closing

This allows compatibility with phprdkafka 4.0

Author: Steveb-p

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

RdKafkaContext.php

@@ -36,7 +36,7 @@ class RdKafkaContext implements Context
     private $conf;
 
     /**
-     * @var Producer
+     * @var RdKafkaProducer
      */
     private $producer;
 
@@ -96,7 +96,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 +155,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 +184,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) {

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);
+        }
+    }
 }

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;
     }
 }

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]",