Merge pull request #1101 from nick-zh/remove-set-default-topic-conf

[rdkafka] remove topic conf, deprecated

Author: makasim

.travis.yml

@@ -68,6 +68,7 @@ install:
     - echo "memory_limit=2048M" >> ~/.phpenv/versions/$(phpenv version-name)/etc/conf.d/travis.ini
     - php ./bin/fix-symfony-version.php "$SYMFONY_VERSION"
     - composer install
+    - sed -i 's/525568/16777471/' vendor/kwn/php-rdkafka-stubs/stubs/constants.php
     - if [ "$PREPARE_CONTAINER" = true ]; then docker --version; fi
     - if [ "$PREPARE_CONTAINER" = true ]; then docker-compose --version; fi
     - if [ "$PREPARE_CONTAINER" = true ]; then bin/dev -b; fi

pkg/rdkafka/RdKafkaConnectionFactory.php

@@ -39,6 +39,10 @@ class RdKafkaConnectionFactory implements ConnectionFactory
      */
     public function __construct($config = 'kafka:')
     {
+        if (version_compare(RdKafkaContext::getLibrdKafkaVersion(), '1.0.0', '<')) {
+            throw new \RuntimeException('You must install librdkafka:1.0.0 or higher');
+        }
+
         if (empty($config) || 'kafka:' === $config) {
             $config = [];
         } elseif (is_string($config)) {

pkg/rdkafka/RdKafkaContext.php

@@ -19,7 +19,6 @@
 use RdKafka\Conf;
 use RdKafka\KafkaConsumer;
 use RdKafka\Producer as VendorProducer;
-use RdKafka\TopicConf;
 
 class RdKafkaContext implements Context
 {
@@ -184,20 +183,18 @@ public static function getLibrdKafkaVersion(): string
     private function getConf(): Conf
     {
         if (null === $this->conf) {
-            $topicConf = new TopicConf();
+            $this->conf = new Conf();
 
             if (isset($this->config['topic']) && is_array($this->config['topic'])) {
                 foreach ($this->config['topic'] as $key => $value) {
-                    $topicConf->set($key, $value);
+                    $this->conf->set($key, $value);
                 }
             }
 
             if (isset($this->config['partitioner'])) {
-                $topicConf->setPartitioner($this->config['partitioner']);
+                $this->conf->set('partitioner', $this->config['partitioner']);
             }
 
-            $this->conf = new Conf();
-
             if (isset($this->config['global']) && is_array($this->config['global'])) {
                 foreach ($this->config['global'] as $key => $value) {
                     $this->conf->set($key, $value);
@@ -219,8 +216,6 @@ private function getConf(): Conf
             if (isset($this->config['stats_cb'])) {
                 $this->conf->setStatsCb($this->config['stats_cb']);
             }
-
-            $this->conf->setDefaultTopicConf($topicConf);
         }
 
         return $this->conf;

pkg/rdkafka/RdKafkaProducer.php

@@ -47,7 +47,8 @@ public function send(Destination $destination, Message $message): void
         // 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', '>=')
+            // Since we are forcing to use at least librdkafka:1.0.0, no need to check the lib version anymore
+            if (false !== phpversion('rdkafka')
                 && version_compare(phpversion('rdkafka'), '3.1.0', '<=')) {
                 trigger_error(
                     'Phprdkafka <= 3.1.0 is incompatible with librdkafka 1.0.0 when calling `producev`. '.

pkg/rdkafka/Tests/Spec/RdKafkaSendToAndReceiveFromTopicTest.php

@@ -30,7 +30,8 @@ public function test()
 
         $context->createProducer()->send($topic, $context->createMessage($expectedBody));
 
-        $message = $consumer->receive(10000); // 10 sec
+        // Initial balancing can take some time, so we want to make sure the timeout is high enough
+        $message = $consumer->receive(15000); // 15 sec
 
         $this->assertInstanceOf(Message::class, $message);
         $consumer->acknowledge($message);
@@ -47,7 +48,7 @@ protected function createContext()
                 'enable.auto.commit' => 'false',
             ],
             'topic' => [
-                'auto.offset.reset' => 'beginning',
+                'auto.offset.reset' => 'earliest',
             ],
         ];