[kafka] change offset.

Author: makasim

pkg/rdkafka/Tests/RdKafkaConsumerTest.php

@@ -50,8 +50,7 @@ public function testShouldReceiveFromQueueAndReturnNullIfNoMessageInQueue()
         $kafkaConsumer = $this->createKafkaConsumerMock();
         $kafkaConsumer
             ->expects($this->once())
-            ->method('subscribe')
-            ->with(['dest'])
+            ->method('assign')
         ;
         $kafkaConsumer
             ->expects($this->once())
@@ -70,6 +69,36 @@ public function testShouldReceiveFromQueueAndReturnNullIfNoMessageInQueue()
         $this->assertNull($consumer->receive(1000));
     }
 
+    public function testShouldPassProperlyConfiguredTopicPartitionOnAssign()
+    {
+        $destination = new RdKafkaTopic('dest');
+
+        $kafkaMessage = new Message();
+        $kafkaMessage->err = RD_KAFKA_RESP_ERR__TIMED_OUT;
+
+        $kafkaConsumer = $this->createKafkaConsumerMock();
+        $kafkaConsumer
+            ->expects($this->once())
+            ->method('assign')
+        ;
+        $kafkaConsumer
+            ->expects($this->any())
+            ->method('consume')
+            ->willReturn($kafkaMessage)
+        ;
+
+        $consumer = new RdKafkaConsumer(
+            $kafkaConsumer,
+            $this->createContextMock(),
+            $destination,
+            $this->createSerializerMock()
+        );
+
+        $consumer->receive(1000);
+        $consumer->receive(1000);
+        $consumer->receive(1000);
+    }
+
     public function testShouldSubscribeOnFirstReceiveOnly()
     {
         $destination = new RdKafkaTopic('dest');
@@ -80,8 +109,7 @@ public function testShouldSubscribeOnFirstReceiveOnly()
         $kafkaConsumer = $this->createKafkaConsumerMock();
         $kafkaConsumer
             ->expects($this->once())
-            ->method('subscribe')
-            ->with(['dest'])
+            ->method('assign')
         ;
         $kafkaConsumer
             ->expects($this->any())
@@ -101,6 +129,38 @@ public function testShouldSubscribeOnFirstReceiveOnly()
         $consumer->receive(1000);
     }
 
+    public function testThrowOnOffsetChangeAfterSubscribing()
+    {
+        $destination = new RdKafkaTopic('dest');
+
+        $kafkaMessage = new Message();
+        $kafkaMessage->err = RD_KAFKA_RESP_ERR__TIMED_OUT;
+
+        $kafkaConsumer = $this->createKafkaConsumerMock();
+        $kafkaConsumer
+            ->expects($this->once())
+            ->method('assign')
+        ;
+        $kafkaConsumer
+            ->expects($this->any())
+            ->method('consume')
+            ->willReturn($kafkaMessage)
+        ;
+
+        $consumer = new RdKafkaConsumer(
+            $kafkaConsumer,
+            $this->createContextMock(),
+            $destination,
+            $this->createSerializerMock()
+        );
+
+        $consumer->receive(1000);
+
+        $this->expectException(\LogicException::class);
+        $this->expectExceptionMessage('The consumer has already subscribed.');
+        $consumer->setOffset(123);
+    }
+
     public function testShouldReceiveFromQueueAndReturnMessageIfMessageInQueue()
     {
         $destination = new RdKafkaTopic('dest');
@@ -114,8 +174,7 @@ public function testShouldReceiveFromQueueAndReturnMessageIfMessageInQueue()
         $kafkaConsumer = $this->createKafkaConsumerMock();
         $kafkaConsumer
             ->expects($this->once())
-            ->method('subscribe')
-            ->with(['dest'])
+            ->method('assign')
         ;
         $kafkaConsumer
             ->expects($this->once())