JMP-2204: align timeout handling (#15)

* adjust timeout handling

* fix tests

* adapt interface

* add poll, change poll logic (#16)

* add poll, change poll logic

* fix cs

Author: nick-zh

src/Conf/KafkaConfiguration.php

@@ -20,24 +20,17 @@ class KafkaConfiguration extends RdKafkaConf
      */
     protected $topicSubscriptions;
 
-    /**
-     * @var int
-     */
-    protected $timeout;
-
     /**
      * @param string[] $brokers
      * @param array|TopicSubscription[] $topicSubscriptions
-     * @param integer $timeout
      * @param mixed[] $config
      */
-    public function __construct(array $brokers, array $topicSubscriptions, int $timeout, array $config = [])
+    public function __construct(array $brokers, array $topicSubscriptions, array $config = [])
     {
         parent::__construct();
 
         $this->brokers = $brokers;
         $this->topicSubscriptions = $topicSubscriptions;
-        $this->timeout = $timeout;
 
         $this->initializeConfig($config);
     }
@@ -58,14 +51,6 @@ public function getTopicSubscriptions(): array
         return $this->topicSubscriptions;
     }
 
-    /**
-     * @return integer
-     */
-    public function getTimeout(): int
-    {
-        return $this->timeout;
-    }
-
     /**
      * @return string[]
      */

src/Consumer/AbstractKafkaConsumer.php

@@ -81,18 +81,19 @@ public function getConfiguration(): array
      * Consumes a message and returns it
      * In cases of errors / timeouts an exception is thrown
      *
+     * @param integer $timeoutMs
      * @return KafkaConsumerMessageInterface
      * @throws KafkaConsumerConsumeException
      * @throws KafkaConsumerEndOfPartitionException
      * @throws KafkaConsumerTimeoutException
      */
-    public function consume(): KafkaConsumerMessageInterface
+    public function consume(int $timeoutMs = 10000): KafkaConsumerMessageInterface
     {
         if (false === $this->isSubscribed()) {
             throw new KafkaConsumerConsumeException(KafkaConsumerConsumeException::NOT_SUBSCRIBED_EXCEPTION_MESSAGE);
         }
 
-        if (null === $rdKafkaMessage = $this->kafkaConsume($this->kafkaConfiguration->getTimeout())) {
+        if (null === $rdKafkaMessage = $this->kafkaConsume($timeoutMs)) {
             throw new KafkaConsumerEndOfPartitionException(
                 rd_kafka_err2str(RD_KAFKA_RESP_ERR__PARTITION_EOF),
                 RD_KAFKA_RESP_ERR__PARTITION_EOF
@@ -118,17 +119,18 @@ public function consume(): KafkaConsumerMessageInterface
      * Queries the broker for metadata on a certain topic
      *
      * @param string $topicName
+     * @param integer $timeoutMs
      * @return RdKafkaMetadataTopic
      * @throws RdKafkaException
      */
-    public function getMetadataForTopic(string $topicName): RdKafkaMetadataTopic
+    public function getMetadataForTopic(string $topicName, int $timeoutMs = 10000): RdKafkaMetadataTopic
     {
         $topic = $this->consumer->newTopic($topicName);
         return $this->consumer
             ->getMetadata(
                 false,
                 $topic,
-                $this->kafkaConfiguration->getTimeout()
+                $timeoutMs
             )
             ->getTopics()
             ->current();
@@ -138,28 +140,28 @@ public function getMetadataForTopic(string $topicName): RdKafkaMetadataTopic
      * Get the earliest offset for a certain timestamp for topic partitions
      *
      * @param array|RdKafkaTopicPartition[] $topicPartitions
-     * @param integer                       $timeout
+     * @param integer                       $timeoutMs
      * @return array|RdKafkaTopicPartition[]
      */
-    public function offsetsForTimes(array $topicPartitions, int $timeout): array
+    public function offsetsForTimes(array $topicPartitions, int $timeoutMs): array
     {
-        return $this->consumer->offsetsForTimes($topicPartitions, $timeout);
+        return $this->consumer->offsetsForTimes($topicPartitions, $timeoutMs);
     }
 
     /**
      * Queries the broker for the first offset of a given topic and partition
      *
      * @param string  $topic
      * @param integer $partition
-     * @param integer $timeout
+     * @param integer $timeoutMs
      * @return integer
      */
-    public function getFirstOffsetForTopicPartition(string $topic, int $partition, int $timeout): int
+    public function getFirstOffsetForTopicPartition(string $topic, int $partition, int $timeoutMs): int
     {
         $lowOffset = 0;
         $highOffset = 0;
 
-        $this->consumer->queryWatermarkOffsets($topic, $partition, $lowOffset, $highOffset, $timeout);
+        $this->consumer->queryWatermarkOffsets($topic, $partition, $lowOffset, $highOffset, $timeoutMs);
 
         return $lowOffset;
     }
@@ -169,15 +171,15 @@ public function getFirstOffsetForTopicPartition(string $topic, int $partition, i
      *
      * @param string  $topic
      * @param integer $partition
-     * @param integer $timeout
+     * @param integer $timeoutMs
      * @return integer
      */
-    public function getLastOffsetForTopicPartition(string $topic, int $partition, int $timeout): int
+    public function getLastOffsetForTopicPartition(string $topic, int $partition, int $timeoutMs): int
     {
         $lowOffset = 0;
         $highOffset = 0;
 
-        $this->consumer->queryWatermarkOffsets($topic, $partition, $lowOffset, $highOffset, $timeout);
+        $this->consumer->queryWatermarkOffsets($topic, $partition, $lowOffset, $highOffset, $timeoutMs);
 
         return $highOffset;
     }
@@ -200,8 +202,8 @@ protected function getConsumerMessage(RdKafkaMessage $message): KafkaConsumerMes
     }
 
     /**
-     * @param integer $timeout
+     * @param integer $timeoutMs
      * @return null|RdKafkaMessage
      */
-    abstract protected function kafkaConsume(int $timeout): ?RdKafkaMessage;
+    abstract protected function kafkaConsume(int $timeoutMs): ?RdKafkaMessage;
 }

src/Consumer/KafkaConsumerBuilder.php

@@ -47,11 +47,6 @@ final class KafkaConsumerBuilder implements KafkaConsumerBuilderInterface
      */
     private $consumerType = self::CONSUMER_TYPE_HIGH_LEVEL;
 
-    /**
-     * @var int
-     */
-    private $timeout = 1000;
-
     /**
      * @var callable
      */
@@ -170,20 +165,6 @@ public function withAdditionalConfig(array $config): KafkaConsumerBuilderInterfa
         return $that;
     }
 
-    /**
-     * Set the timeout for all consumer actions
-     *
-     * @param integer $timeout
-     * @return KafkaConsumerBuilderInterface
-     */
-    public function withTimeout(int $timeout): KafkaConsumerBuilderInterface
-    {
-        $that = clone $this;
-        $that->timeout = $timeout;
-
-        return $that;
-    }
-
     /**
      * Set the consumer group
      *
@@ -321,7 +302,6 @@ public function build(): KafkaConsumerInterface
         $kafkaConfig = new KafkaConfiguration(
             $this->brokers,
             $this->topics,
-            $this->timeout,
             $this->config
         );
 

src/Consumer/KafkaConsumerBuilderInterface.php

@@ -58,14 +58,6 @@ public function withSubscription(
      */
     public function withAdditionalConfig(array $config): self;
 
-    /**
-     * Set the timeout for all consumer actions
-     *
-     * @param integer $timeout
-     * @return KafkaConsumerBuilderInterface
-     */
-    public function withTimeout(int $timeout): self;
-
     /**
      * Set the consumer group
      *

src/Consumer/KafkaConsumerInterface.php

@@ -39,9 +39,10 @@ public function isSubscribed(): bool;
      * Consumes a message and returns it
      * In cases of errors / timeouts a KafkaConsumerConsumeException is thrown
      *
+     * @param integer $timeoutMs
      * @return KafkaConsumerMessageInterface
      */
-    public function consume(): KafkaConsumerMessageInterface;
+    public function consume(int $timeoutMs = 10000): KafkaConsumerMessageInterface;
 
     /**
      * Commits the offset to the broker for the given message(s)
@@ -62,36 +63,37 @@ public function getConfiguration(): array;
      * Queries the broker for metadata on a certain topic
      *
      * @param string $topicName
+     * @param integer $timeoutMs
      * @return RdKafkaMetadataTopic
      */
-    public function getMetadataForTopic(string $topicName): RdKafkaMetadataTopic;
+    public function getMetadataForTopic(string $topicName, int $timeoutMs = 10000): RdKafkaMetadataTopic;
 
     /**
      * Get the earliest offset for a certain timestamp for topic partitions
      *
      * @param array|RdKafkaTopicPartition[] $topicPartitions
-     * @param integer                       $timeout
+     * @param integer                       $timeoutMs
      * @return array|RdKafkaTopicPartition[]
      */
-    public function offsetsForTimes(array $topicPartitions, int $timeout): array;
+    public function offsetsForTimes(array $topicPartitions, int $timeoutMs): array;
 
     /**
      * Queries the broker for the first offset of a given topic and partition
      *
      * @param string  $topic
      * @param integer $partition
-     * @param integer $timeout
+     * @param integer $timeoutMs
      * @return integer
      */
-    public function getFirstOffsetForTopicPartition(string $topic, int $partition, int $timeout): int;
+    public function getFirstOffsetForTopicPartition(string $topic, int $partition, int $timeoutMs): int;
 
     /**
      * Queries the broker for the last offset of a given topic and partition
      *
      * @param string  $topic
      * @param integer $partition
-     * @param integer $timeout
+     * @param integer $timeoutMs
      * @return integer
      */
-    public function getLastOffsetForTopicPartition(string $topic, int $partition, int $timeout): int;
+    public function getLastOffsetForTopicPartition(string $topic, int $partition, int $timeoutMs): int;
 }

src/Consumer/KafkaHighLevelConsumer.php

@@ -143,14 +143,14 @@ public function getAssignment(): array
      * Gets the commited offset for a TopicPartition for the configured consumer group
      *
      * @param array|RdKafkaTopicPartition[] $topicPartitions
-     * @param integer                       $timeout
+     * @param integer                       $timeoutMs
      * @return array|RdKafkaTopicPartition[]
      * @throws KafkaConsumerRequestException
      */
-    public function getCommittedOffsets(array $topicPartitions, int $timeout): array
+    public function getCommittedOffsets(array $topicPartitions, int $timeoutMs): array
     {
         try {
-            return $this->consumer->getCommittedOffsets($topicPartitions, $timeout);
+            return $this->consumer->getCommittedOffsets($topicPartitions, $timeoutMs);
         } catch (RdKafkaException $e) {
             throw new KafkaConsumerRequestException($e->getMessage(), $e->getCode());
         }
@@ -178,13 +178,13 @@ public function close(): void
     }
 
     /**
-     * @param integer $timeout
+     * @param integer $timeoutMs
      * @return RdKafkaMessage|null
      * @throws RdKafkaException
      */
-    protected function kafkaConsume(int $timeout): ?RdKafkaMessage
+    protected function kafkaConsume(int $timeoutMs): ?RdKafkaMessage
     {
-        return $this->consumer->consume($timeout);
+        return $this->consumer->consume($timeoutMs);
     }
 
     /**

src/Consumer/KafkaHighLevelConsumerInterface.php

@@ -36,10 +36,10 @@ public function getAssignment(): array;
      * Gets the commited offset for a TopicPartition for the configured consumer group
      *
      * @param array|RdKafkaTopicPartition[] $topicPartitions
-     * @param integer                       $timeout
+     * @param integer                       $timeoutMs
      * @return array|RdKafkaTopicPartition[]
      */
-    public function getCommittedOffsets(array $topicPartitions, int $timeout): array;
+    public function getCommittedOffsets(array $topicPartitions, int $timeoutMs): array;
 
     /**
      * Get current offset positions of the consumer

src/Consumer/KafkaLowLevelConsumer.php

@@ -141,12 +141,12 @@ public function unsubscribe(): void
     }
 
     /**
-     * @param integer $timeout
+     * @param integer $timeoutMs
      * @return null|RdKafkaMessage
      */
-    protected function kafkaConsume(int $timeout): ?RdKafkaMessage
+    protected function kafkaConsume(int $timeoutMs): ?RdKafkaMessage
     {
-        return $this->queue->consume($timeout);
+        return $this->queue->consume($timeoutMs);
     }
 
     /**