Update to rdkafka 4.0.0 (#40)

* add ext-rdkafka 4.0 capabilities

Author: nick-zh

README.md

@@ -14,7 +14,7 @@ To read more about the functions used in this lib, check out the [documentation]
 
 ## Requirements
 - php: ^7.1
-- ext-rdkafka: ^3.1.2
+- ext-rdkafka: ^4.0.0
 
 ## Installation
 ```composer require jobcloud/messaging-lib "~4.0"```

composer.json

@@ -17,12 +17,12 @@
   ],
   "require": {
     "php": "^7.3",
-    "ext-rdkafka": "^3.1.2",
+    "ext-rdkafka": "^4.0.0",
     "ext-json": "*",
     "flix-tech/avro-serde-php": "^1.3"
   },
   "require-dev": {
-    "phpunit/phpunit": "^8.2.3|^7.5.13",
+    "phpunit/phpunit": "~8.3.5",
     "squizlabs/php_codesniffer": "^3.4.2",
     "phpstan/phpstan": "^0.11.12",
     "php-mock/php-mock-phpunit": "^2.4",

docker/dev/php/Dockerfile

@@ -6,7 +6,7 @@ ARG HOST_USER_ID
 ARG HOST_USER
 
 # PHP: Copy configuration files & remove dist files
-COPY dev/php/files/php/ /phpIni
+RUN mkdir /phpIni
 COPY dev/php/files/bin/ /usr/local/bin/
 
 # SYS: Install required packages
@@ -35,7 +35,7 @@ RUN composer global require hirak/prestissimo
 
 # PHP: Install php extensions
 RUN pecl channel-update pecl.php.net && \
-    pecl install rdkafka xdebug && \
+    pecl install rdkafka && \
     docker-php-ext-install pcntl && \
     php-ext-enable rdkafka pcntl
 

docker/dev/php/files/php/20-xdebug.ini

@@ -1,5 +1,3 @@
 parameters:
     level: 7
     paths: [ src ]
-    ignoreErrors:
-        - '#Class RdKafka\\TopicPartition constructor invoked with 3 parameters, 0 required#'

phpstan.neon

@@ -18,6 +18,7 @@
 use RdKafka\KafkaConsumer as RdKafkaHighLevelConsumer;
 use RdKafka\Metadata\Topic as RdKafkaMetadataTopic;
 use RdKafka\Message as RdKafkaMessage;
+use RdKafka\TopicPartition as RdKafkaTopicPartition;
 
 abstract class AbstractKafkaConsumer implements KafkaConsumerInterface
 {
@@ -136,6 +137,54 @@ public function getMetadataForTopic(RdKafkaConsumerTopic $topic): RdKafkaMetadat
             ->current();
     }
 
+    /**
+     * Get the earliest offset for a certain timestamp for topic partitions
+     *
+     * @param array|RdKafkaTopicPartition[] $topicPartitions
+     * @param integer                       $timeout
+     * @return array
+     */
+    public function offsetsForTimes(array $topicPartitions, int $timeout): array
+    {
+        return $this->consumer->offsetsForTimes($topicPartitions, $timeout);
+    }
+
+    /**
+     * Queries the broker for the first offset of a given topic and partition
+     *
+     * @param string  $topic
+     * @param integer $partition
+     * @param integer $timeout
+     * @return integer
+     */
+    public function getFirstOffsetForTopicPartition(string $topic, int $partition, int $timeout): int
+    {
+        $lowOffset = 0;
+        $highOffset = 0;
+
+        $this->consumer->queryWatermarkOffsets($topic, $partition, $lowOffset, $highOffset, $timeout);
+
+        return $lowOffset;
+    }
+
+    /**
+     * Queries the broker for the last offset of a given topic and partition
+     *
+     * @param string  $topic
+     * @param integer $partition
+     * @param integer $timeout
+     * @return integer
+     */
+    public function getLastOffsetForTopicPartition(string $topic, int $partition, int $timeout): int
+    {
+        $lowOffset = 0;
+        $highOffset = 0;
+
+        $this->consumer->queryWatermarkOffsets($topic, $partition, $lowOffset, $highOffset, $timeout);
+
+        return $highOffset;
+    }
+
     /**
      * @param RdKafkaMessage $message
      * @return KafkaConsumerMessageInterface

src/Kafka/Consumer/AbstractKafkaConsumer.php

@@ -63,6 +63,11 @@ final class KafkaConsumerBuilder implements KafkaConsumerBuilderInterface
      */
     private $consumeCallback;
 
+    /**
+     * @var callable
+     */
+    private $logCallback;
+
     /**
      * @var callable
      */
@@ -247,6 +252,20 @@ public function withConsumeCallback(callable $consumeCallback): KafkaConsumerBui
         return $that;
     }
 
+    /**
+     * Callback for log related events
+     *
+     * @param callable $logCallback
+     * @return KafkaConsumerBuilderInterface
+     */
+    public function withLogCallback(callable $logCallback): KafkaConsumerBuilderInterface
+    {
+        $that = clone $this;
+        $that->logCallback = $logCallback;
+
+        return $that;
+    }
+
     /**
      * Set callback that is being called on offset commits
      *
@@ -293,7 +312,6 @@ public function build(): KafkaConsumerInterface
 
         //set additional config
         $this->config['group.id'] = $this->consumerGroup;
-        $this->config['enable.auto.offset.store'] = false;
 
         //create config
         $kafkaConfig = new KafkaConfiguration(
@@ -309,6 +327,7 @@ public function build(): KafkaConsumerInterface
         //create RdConsumer
 
         if (self::CONSUMER_TYPE_LOW_LEVEL === $this->consumerType) {
+            $this->config['enable.auto.offset.store'] = false;
             if (null !== $this->consumeCallback) {
                 throw new KafkaConsumerBuilderException(
                     sprintf(
@@ -328,6 +347,8 @@ public function build(): KafkaConsumerInterface
             );
         }
 
+        $this->config['enable.auto.commit'] = false;
+
         $rdKafkaConsumer = new RdKafkaHighLevelConsumer($kafkaConfig);
 
         return new KafkaHighLevelConsumer($rdKafkaConsumer, $kafkaConfig, $this->decoder);
@@ -349,6 +370,10 @@ private function registerCallbacks(KafkaConfiguration $conf): void
             $conf->setConsumeCb($this->consumeCallback);
         }
 
+        if (null !== $this->logCallback) {
+            $conf->setLogCb($this->logCallback);
+        }
+
         if (null !== $this->offsetCommitCallback) {
             $conf->setOffsetCommitCb($this->rebalanceCallback);
         }

src/Kafka/Consumer/KafkaConsumerBuilder.php

@@ -124,6 +124,14 @@ public function withOffsetCommitCallback(callable $offsetCommitCallback): self;
      */
     public function withDecoder(DecoderInterface $decoder): self;
 
+    /**
+     * Callback for log related events
+     *
+     * @param callable $consumeCallback
+     * @return KafkaConsumerBuilderInterface
+     */
+    public function withLogCallback(callable $consumeCallback): KafkaConsumerBuilderInterface;
+
     /**
      * Returns your consumer instance
      *

src/Kafka/Consumer/KafkaConsumerBuilderInterface.php

@@ -9,6 +9,7 @@
 use Jobcloud\Messaging\Kafka\Message\KafkaConsumerMessageInterface;
 use RdKafka\Metadata\Topic as RdKafkaMetadataTopic;
 use RdKafka\ConsumerTopic as RdKafkaConsumerTopic;
+use RdKafka\TopicPartition as RdKafkaTopicPartition;
 
 interface KafkaConsumerInterface extends ConsumerInterface
 {
@@ -65,4 +66,33 @@ public function getConfiguration(): array;
      * @return RdKafkaMetadataTopic
      */
     public function getMetadataForTopic(RdKafkaConsumerTopic $topic): RdKafkaMetadataTopic;
+
+    /**
+     * Get the earliest offset for a certain timestamp for topic partitions
+     *
+     * @param array|RdKafkaTopicPartition[] $topicPartitions
+     * @param integer                       $timeout
+     * @return array
+     */
+    public function offsetsForTimes(array $topicPartitions, int $timeout): array;
+
+    /**
+     * Queries the broker for the first offset of a given topic and partition
+     *
+     * @param string  $topic
+     * @param integer $partition
+     * @param integer $timeout
+     * @return integer
+     */
+    public function getFirstOffsetForTopicPartition(string $topic, int $partition, int $timeout): int;
+
+    /**
+     * Queries the broker for the last offset of a given topic and partition
+     *
+     * @param string  $topic
+     * @param integer $partition
+     * @param integer $timeout
+     * @return integer
+     */
+    public function getLastOffsetForTopicPartition(string $topic, int $partition, int $timeout): int;
 }

src/Kafka/Consumer/KafkaConsumerInterface.php

@@ -156,6 +156,27 @@ public function getCommittedOffsets(array $topicPartitions, int $timeout): array
         }
     }
 
+    /**
+     * Get current offset positions of the consumer
+     *
+     * @param array|RdKafkaTopicPartition[] $topicPartitions
+     * @return array
+     */
+    public function getOffsetPositions(array $topicPartitions): array
+    {
+        return $this->consumer->getOffsetPositions($topicPartitions);
+    }
+
+    /**
+     * Close the consumer connection
+     *
+     * @return void;
+     */
+    public function close(): void
+    {
+        $this->consumer->close();
+    }
+
     /**
      * @param integer $timeout
      * @return RdKafkaMessage|null