[kafka] add ability to set offset.

makasim · makasim · commit 04d4530efd66 · 2018-01-05T00:43:16.000+02:00
diff --git a/pkg/rdkafka/RdKafkaConsumer.php b/pkg/rdkafka/RdKafkaConsumer.php
@@ -6,6 +6,7 @@
 use Interop\Queue\PsrConsumer;
 use Interop\Queue\PsrMessage;
 use RdKafka\KafkaConsumer;
+use RdKafka\TopicPartition;
 
 class RdKafkaConsumer implements PsrConsumer
 {
@@ -36,6 +37,11 @@ class RdKafkaConsumer implements PsrConsumer
      */
     private $commitAsync;
 
+    /**
+     * @var int|null
+     */
+    private $offset;
+
     /**
      * @param KafkaConsumer  $consumer
      * @param RdKafkaContext $context
@@ -49,6 +55,7 @@ public function __construct(KafkaConsumer $consumer, RdKafkaContext $context, Rd
         $this->topic = $topic;
         $this->subscribed = false;
         $this->commitAsync = false;
+        $this->offset = null;
 
         $this->setSerializer($serializer);
     }
@@ -69,6 +76,15 @@ public function setCommitAsync($async)
         $this->commitAsync = (bool) $async;
     }
 
+    public function setOffset($offset)
+    {
+        if ($this->subscribed) {
+            throw new \LogicException('The consumer has already subscribed.');
+        }
+
+        $this->offset = $offset;
+    }
+
     /**
      * {@inheritdoc}
      */
@@ -83,7 +99,11 @@ public function getQueue()
     public function receive($timeout = 0)
     {
         if (false == $this->subscribed) {
-            $this->consumer->subscribe([$this->topic->getTopicName()]);
+            $this->consumer->assign([new TopicPartition(
+                $this->getQueue()->getQueueName(),
+                $this->getQueue()->getPartition(),
+                $this->offset
+            )]);
 
             $this->subscribed = true;
         }
