Merge pull request #2 from lamoda/feature/rdkafka-4

Support RDKafka 4
lamoda · Mar 2, 2021 · b656167 · b656167
2 parents 34215c6 + 005b531
commit b656167
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 1 deletion.
diff --git a/composer.json b/composer.json
@@ -12,7 +12,8 @@
   ],
   "require": {
     "php": ">=7.1",
-    "codeception/codeception": "~2.5"
+    "ext-rdkafka": "^3 || ^4",
+    "codeception/codeception": "^2.5 | ^3.0 | ^4.0"
   },
   "autoload": {
     "psr-4": {

diff --git a/src/Extension/KafkaModule.php b/src/Extension/KafkaModule.php
@@ -6,6 +6,7 @@
 
 use Codeception\Module;
 use Exception;
+use Lamoda\Codeception\Extension\MessageSerializer\ArrayMessageSerializer;
 use Lamoda\Codeception\Extension\MessageSerializer\MessageSerializerInterface;
 use RdKafka\Conf;
 use RdKafka\Consumer;
@@ -17,6 +18,7 @@
 class KafkaModule extends Module
 {
     protected const DEFAULT_PARTITION = 0;
+    protected const FLUSH_TIMEOUT_MS = 3000;
 
     /**
      * @var MessageSerializerInterface
@@ -83,6 +85,10 @@ public function putMessageInTopic(string $topicName, string $message, ?int $part
         $topic = $producer->newTopic($topicName, $this->topicConf);
 
         $topic->produce($partition ?? static::DEFAULT_PARTITION, 0, $message);
+
+        if (method_exists($producer, 'flush')) {
+            $producer->flush(self::FLUSH_TIMEOUT_MS);
+        }
     }
 
     public function putMessageListInTopic(string $topicName, array $messages, ?int $partition = null): void