ACP2E-3387: [Cloud] Magento: queue message has been deleted

Author: aplapana

app/code/Magento/MysqlMq/Model/ResourceModel/Queue.php

@@ -7,13 +7,39 @@
 
 use Magento\Framework\DB\Select;
 use Magento\Framework\DB\Sql\Expression;
+use Magento\Framework\Model\ResourceModel\Db\Context;
 use Magento\MysqlMq\Model\QueueManagement;
 
 /**
  * Resource model for queue.
  */
 class Queue extends \Magento\Framework\Model\ResourceModel\Db\AbstractDb
 {
+    /**
+     *
+     */
+    private const CHUNK_SIZE = 10000;
+
+    /**
+     * @var int|mixed
+     */
+    private int $chunkSize;
+
+    /**
+     * @param Context $context
+     * @param null $connectionName
+     * @param int|null $chunkSize
+     */
+    public function __construct(Context $context, $connectionName = null, int $chunkSize = null)
+    {
+        parent::__construct($context, $connectionName);
+        if ($chunkSize) {
+            $this->chunkSize = $chunkSize;
+        } else {
+            $this->chunkSize = self::CHUNK_SIZE;
+        }
+    }
+
     /**
      * Model initialization
      *
@@ -170,13 +196,20 @@ public function getMessages($queueName, $limit = null)
     public function deleteMarkedMessages(): void
     {
         $connection = $this->getConnection();
-
         $select = $connection->select()
             ->from(['queue_message_status' => $this->getMessageStatusTable()], ['message_id'])
-            ->where('status = ?', QueueManagement::MESSAGE_STATUS_TO_BE_DELETED)
+            ->joinLeft(
+                ['message_status2' => $this->getMessageStatusTable()],
+                'queue_message_status.message_id = message_status2.message_id AND message_status2.status <> ' .
+                QueueManagement::MESSAGE_STATUS_TO_BE_DELETED,
+                []
+            )
+            ->where('queue_message_status.status = ?', QueueManagement::MESSAGE_STATUS_TO_BE_DELETED)
+            ->where('message_status2.message_id IS NULL')
             ->distinct();
+
         $messageIds = $connection->fetchCol($select);
-        foreach (array_chunk($messageIds, 10000) as $messageIdsChunk) {
+        foreach (array_chunk($messageIds, $this->chunkSize) as $messageIdsChunk) {
             $connection->delete($this->getMessageTable(), ['id IN (?)' => $messageIdsChunk]);
         }
     }

app/code/Magento/MysqlMq/Test/Unit/Model/ResourceModel/QueueTest.php

@@ -286,21 +286,15 @@ function ($arg1, $arg2) use ($tableNames) {
         $select->expects($this->once())
             ->method('from')->with(['queue_message_status' => $tableNames[0]], ['message_id'])->willReturnSelf();
         $select->expects($this->once())->method('where')
-            ->with('status <> ?', QueueManagement::MESSAGE_STATUS_TO_BE_DELETED)
+            ->with('status = ?', QueueManagement::MESSAGE_STATUS_TO_BE_DELETED)
             ->willReturnSelf();
         $select->expects($this->once())->method('distinct')->willReturnSelf();
-        $select->expects($this->once())
-            ->method('assemble')
-            ->willReturn(
-                'SELECT message_id FROM queue_message_status WHERE status = ' .
-                QueueManagement::MESSAGE_STATUS_TO_BE_DELETED
-            );
+        $connection->expects($this->once())->method('fetchCol')->with($select)->willReturn([1, 2]);
 
         $connection->expects($this->once())->method('delete')
             ->with(
                 $tableNames[1],
-                'id NOT IN (SELECT message_id FROM queue_message_status WHERE status = ' .
-                QueueManagement::MESSAGE_STATUS_TO_BE_DELETED . ')'
+                ['id IN (?)' => [1,2]]
             )->willReturn(2);
         $this->queue->deleteMarkedMessages();
     }