Fix DBAL Consumer duplicating messages when rejecting with requeue

Author: Steveb-p

pkg/dbal/DbalConsumer.php

@@ -100,16 +100,14 @@ public function reject(Message $message, bool $requeue = false): void
     {
         InvalidMessageException::assertMessageInstanceOf($message, DbalMessage::class);
 
+        $this->acknowledge($message);
+
         if ($requeue) {
             $message = clone $message;
             $message->setRedelivered(false);
 
             $this->getContext()->createProducer()->send($this->queue, $message);
-
-            return;
         }
-
-        $this->deleteMessage($message->getDeliveryId());
     }
 
     protected function getContext(): DbalContext

pkg/dbal/Tests/Functional/DbalConsumerTest.php

@@ -144,6 +144,31 @@ public function testShouldDeleteExpiredMessage()
         $this->assertSame(0, $this->getQuerySize());
     }
 
+    public function testShouldRemoveOriginalMessageThatHaveBeenRejectedWithRequeu()
+    {
+        $context = $this->context;
+        $queue = $context->createQueue(__METHOD__);
+
+        $consumer = $context->createConsumer($queue);
+
+        // guard
+        $this->assertSame(0, $this->getQuerySize());
+
+        $producer = $context->createProducer();
+
+        /** @var DbalMessage $message */
+        $message = $context->createMessage(__CLASS__);
+        $producer->send($queue, $message);
+
+        $this->assertCount(1, $this->getQuerySize());
+
+        $message = $consumer->receive(100); // 100ms
+
+        $this->assertInstanceOf(DbalMessage::class, $message);
+        $consumer->reject($message, true);
+        $this->assertCount(1, $this->getQuerySize());
+    }
+
     private function getQuerySize(): int
     {
         return (int) $this->context->getDbalConnection()