Fix queue consuming from stored offset

L3o-pold · L3o-pold · commit d3d797584e5a · 2019-06-03T10:27:28.000+02:00
diff --git a/src/Queue/KafkaQueue.php b/src/Queue/KafkaQueue.php
@@ -150,7 +150,7 @@ public function pop($queue = null)
                 $topicConf->set('auto.offset.reset', 'largest');
 
                 $this->topics[$queue] = $this->consumer->newTopic($queue, $topicConf);
-                $this->topics[$queue]->consumeQueueStart(0, RD_KAFKA_OFFSET_BEGINNING, $this->queues[$queue]);
+                $this->topics[$queue]->consumeQueueStart(0, RD_KAFKA_OFFSET_STORED, $this->queues[$queue]);
             }
 
             $message = $this->queues[$queue]->consume(1000);

Original file line number	Diff line number	Diff line change
`@@ -150,7 +150,7 @@ public function pop($queue = null)`
`150`	`150`	`$topicConf->set('auto.offset.reset', 'largest');`
`151`	`151`
`152`	`152`	`$this->topics[$queue] = $this->consumer->newTopic($queue, $topicConf);`
`153`		`- $this->topics[$queue]->consumeQueueStart(0, RD_KAFKA_OFFSET_BEGINNING, $this->queues[$queue]);`
	`153`	`+ $this->topics[$queue]->consumeQueueStart(0, RD_KAFKA_OFFSET_STORED, $this->queues[$queue]);`
`154`	`154`	`}`
`155`	`155`
`156`	`156`	`$message = $this->queues[$queue]->consume(1000);`