MC-42514: [Magento Cloud] Broken pipe error when processing bulk actions

- Introduced prefetch count value to limit the number of unacknowledged messages for each consumer

Author: viktym

lib/internal/Magento/Framework/Amqp/Queue.php

@@ -14,8 +14,6 @@
 use Psr\Log\LoggerInterface;
 
 /**
- * Class Queue
- *
  * @api
  * @since 103.0.0
  */
@@ -41,24 +39,34 @@ class Queue implements QueueInterface
      */
     private $logger;
 
+    /**
+     * The prefetch value is used to specify how many messages that are being sent to the consumer at the same time.
+     * @see https://www.rabbitmq.com/consumer-prefetch.html
+     * @var int
+     */
+    private $prefetchCount;
+
     /**
      * Initialize dependencies.
      *
      * @param Config $amqpConfig
      * @param EnvelopeFactory $envelopeFactory
      * @param string $queueName
      * @param LoggerInterface $logger
+     * @param int $prefetchCount
      */
     public function __construct(
         Config $amqpConfig,
         EnvelopeFactory $envelopeFactory,
         $queueName,
-        LoggerInterface $logger
+        LoggerInterface $logger,
+        $prefetchCount = 100
     ) {
         $this->amqpConfig = $amqpConfig;
         $this->queueName = $queueName;
         $this->envelopeFactory = $envelopeFactory;
         $this->logger = $logger;
+        $this->prefetchCount = (int)$prefetchCount;
     }
 
     /**
@@ -144,6 +152,7 @@ public function subscribe($callback)
 
         $channel = $this->amqpConfig->getChannel();
         // @codingStandardsIgnoreStart
+        $channel->basic_qos(0, $this->prefetchCount, false);
         $channel->basic_consume($this->queueName, '', false, false, false, false, $callbackConverter);
         // @codingStandardsIgnoreEnd
         while (count($channel->callbacks)) {

lib/internal/Magento/Framework/Amqp/Test/Unit/QueueTest.php

@@ -0,0 +1,74 @@
+<?php
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+declare(strict_types=1);
+
+namespace Magento\Framework\Amqp\Test\Unit;
+
+use Magento\Framework\Amqp\Config;
+use Magento\Framework\Amqp\Queue;
+use Magento\Framework\MessageQueue\EnvelopeFactory;
+use PhpAmqpLib\Channel\AMQPChannel;
+use PHPUnit\Framework\MockObject\MockObject;
+use PHPUnit\Framework\TestCase;
+use Psr\Log\LoggerInterface;
+
+class QueueTest extends TestCase
+{
+    private const PREFETCH_COUNT = 100;
+    /**
+     * @var Config|MockObject
+     */
+    private $config;
+
+    /**
+     * @var EnvelopeFactory|MockObject
+     */
+    private $envelopeFactory;
+
+    /**
+     * @var LoggerInterface|MockObject
+     */
+    private $logger;
+
+    /**
+     * @var Queue
+     */
+    private $model;
+
+    protected function setUp(): void
+    {
+        $this->config = $this->createMock(Config::class);
+        $this->envelopeFactory = $this->createMock(EnvelopeFactory::class);
+        $this->logger = $this->createMock(LoggerInterface::class);
+
+        $this->model = new Queue(
+            $this->config,
+            $this->envelopeFactory,
+            'testQueue',
+            $this->logger,
+            self::PREFETCH_COUNT
+        );
+    }
+
+    /**
+     * Test verifies that prefetch value is used to specify how many messages
+     * are being sent to the consumer at the same time.
+     */
+    public function testSubscribe()
+    {
+        $callback = function () {
+        };
+        $amqpChannel = $this->createMock(AMQPChannel::class);
+        $amqpChannel->expects($this->once())
+            ->method('basic_qos')
+            ->with(0, self::PREFETCH_COUNT, false);
+        $this->config->expects($this->once())
+            ->method('getChannel')
+            ->willReturn($amqpChannel);
+
+        $this->model->subscribe($callback);
+    }
+}