ACP2E-2138: [Cloud]RabbitMQ Broken pipe or closed connection errors during indexing

dhorytskyi · dhorytskyi · commit b553bc368f28 · 2023-08-10T14:31:02.000-05:00
diff --git a/app/code/Magento/Indexer/Model/ProcessManager.php b/app/code/Magento/Indexer/Model/ProcessManager.php
@@ -7,6 +7,7 @@
 
 namespace Magento\Indexer\Model;
 
+use Magento\Framework\Amqp\ConfigPool as AmqpConfigPool;
 use Magento\Framework\App\ObjectManager;
 use Psr\Log\LoggerInterface;
 
@@ -18,7 +19,7 @@ class ProcessManager
     /**
      * Threads count environment variable name
      */
-    const THREADS_COUNT = 'MAGE_INDEXER_THREADS_COUNT';
+    public const THREADS_COUNT = 'MAGE_INDEXER_THREADS_COUNT';
 
     /** @var bool */
     private $failInChildProcess = false;
@@ -37,17 +38,24 @@ class ProcessManager
      */
     private $logger;
 
+    /**
+     * @var AmqpConfigPool
+     */
+    private AmqpConfigPool $amqpConfigPool;
+
     /**
      * @param \Magento\Framework\App\ResourceConnection $resource
      * @param \Magento\Framework\Registry $registry
      * @param int|null $threadsCount
      * @param LoggerInterface|null $logger
+     * @param AmqpConfigPool|null $amqpConfigPool
      */
     public function __construct(
         \Magento\Framework\App\ResourceConnection $resource,
         \Magento\Framework\Registry $registry = null,
         int $threadsCount = null,
-        LoggerInterface $logger = null
+        LoggerInterface $logger = null,
+        AmqpConfigPool $amqpConfigPool = null
     ) {
         $this->resource = $resource;
         if (null === $registry) {
@@ -60,6 +68,7 @@ public function __construct(
         $this->logger = $logger ?? ObjectManager::getInstance()->get(
             LoggerInterface::class
         );
+        $this->amqpConfigPool = $amqpConfigPool ?? ObjectManager::getInstance()->get(AmqpConfigPool::class);
     }
 
     /**
@@ -99,6 +108,8 @@ private function simpleThreadExecute($userFunctions)
     private function multiThreadsExecute($userFunctions)
     {
         $this->resource->closeConnection(null);
+        $this->amqpConfigPool->closeConnections();
+
         $threadNumber = 0;
         foreach ($userFunctions as $userFunction) {
             // phpcs:ignore Magento2.Functions.DiscouragedFunction
diff --git a/app/code/Magento/Indexer/Test/Unit/Model/ProcessManagerTest.php b/app/code/Magento/Indexer/Test/Unit/Model/ProcessManagerTest.php
@@ -7,9 +7,11 @@
 
 namespace Magento\Indexer\Test\Unit\Model;
 
+use Magento\Framework\Amqp\ConfigPool as AmqpConfigPool;
 use Magento\Framework\App\ResourceConnection;
 use Magento\Indexer\Model\ProcessManager;
 use PHPUnit\Framework\TestCase;
+use Psr\Log\LoggerInterface;
 
 /**
  * Class covers process manager execution test logic
@@ -28,12 +30,21 @@ class ProcessManagerTest extends TestCase
     public function testFailureInChildProcessHandleMultiThread(array $userFunctions, int $threadsCount): void
     {
         $connectionMock = $this->createMock(ResourceConnection::class);
+        $loggerMock = $this->createMock(LoggerInterface::class);
+        $amqpConfigPoolMock = $this->createMock(AmqpConfigPool::class);
         $processManager = new ProcessManager(
             $connectionMock,
             null,
-            $threadsCount
+            $threadsCount,
+            $loggerMock,
+            $amqpConfigPoolMock
         );
 
+        $connectionMock->expects($this->once())
+            ->method('closeConnection');
+        $amqpConfigPoolMock->expects($this->once())
+            ->method('closeConnections');
+
         try {
             $processManager->execute($userFunctions);
             $this->fail('Exception was not handled');
@@ -111,12 +122,21 @@ function () {
     public function testSuccessChildProcessHandleMultiThread(array $userFunctions, int $threadsCount): void
     {
         $connectionMock = $this->createMock(ResourceConnection::class);
+        $loggerMock = $this->createMock(LoggerInterface::class);
+        $amqpConfigPoolMock = $this->createMock(AmqpConfigPool::class);
         $processManager = new ProcessManager(
             $connectionMock,
             null,
-            $threadsCount
+            $threadsCount,
+            $loggerMock,
+            $amqpConfigPoolMock
         );
 
+        $connectionMock->expects($this->once())
+            ->method('closeConnection');
+        $amqpConfigPoolMock->expects($this->once())
+            ->method('closeConnections');
+
         try {
             $processManager->execute($userFunctions);
         } catch (\RuntimeException $exception) {
diff --git a/lib/internal/Magento/Framework/Amqp/Config.php b/lib/internal/Magento/Framework/Amqp/Config.php
@@ -169,11 +169,16 @@ private function createConnection(): AbstractConnection
      */
     public function getChannel()
     {
-        if (!isset($this->connection) || !isset($this->channel)) {
+        if (!isset($this->connection)) {
             $this->connection = $this->createConnection();
-
+        }
+        if (!isset($this->channel)
+            || !$this->channel->getConnection()
+            || !$this->channel->getConnection()->isConnected()
+        ) {
             $this->channel = $this->connection->channel();
         }
+
         return $this->channel;
     }
 
diff --git a/lib/internal/Magento/Framework/Amqp/ConfigPool.php b/lib/internal/Magento/Framework/Amqp/ConfigPool.php
@@ -43,4 +43,18 @@ public function get($connectionName)
         }
         return $this->pool[$connectionName];
     }
+
+    /**
+     * Close all opened connections.
+     *
+     * @return void
+     */
+    public function closeConnections(): void
+    {
+        foreach ($this->pool as $config) {
+            $connection = $config->getChannel()->getConnection();
+            $config->getChannel()->close();
+            $connection?->close();
+        }
+    }
 }
diff --git a/lib/internal/Magento/Framework/Amqp/Test/Unit/ConfigPoolTest.php b/lib/internal/Magento/Framework/Amqp/Test/Unit/ConfigPoolTest.php
@@ -10,18 +10,58 @@
 use Magento\Framework\Amqp\Config;
 use Magento\Framework\Amqp\ConfigFactory;
 use Magento\Framework\Amqp\ConfigPool;
+use PhpAmqpLib\Channel\AMQPChannel;
+use PhpAmqpLib\Connection\AbstractConnection;
+use PHPUnit\Framework\MockObject\MockObject;
 use PHPUnit\Framework\TestCase;
 
 class ConfigPoolTest extends TestCase
 {
+    /**
+     * @var ConfigFactory|MockObject
+     */
+    private $factory;
+
+    /**
+     * @var ConfigPool
+     */
+    private $model;
+
+    protected function setUp(): void
+    {
+        $this->factory = $this->createMock(ConfigFactory::class);
+        $this->model = new ConfigPool($this->factory);
+    }
+
     public function testGetConnection()
     {
-        $factory = $this->createMock(ConfigFactory::class);
         $config = $this->createMock(Config::class);
-        $factory->expects($this->once())->method('create')->with(['connectionName' => 'amqp'])->willReturn($config);
-        $model = new ConfigPool($factory);
-        $this->assertEquals($config, $model->get('amqp'));
+        $this->factory->expects($this->once())->method('create')->with(['connectionName' => 'amqp'])->willReturn($config);
+        $this->assertEquals($config, $this->model->get('amqp'));
         //test that object is cached
-        $this->assertEquals($config, $model->get('amqp'));
+        $this->assertEquals($config, $this->model->get('amqp'));
+    }
+
+    public function testCloseConnections(): void
+    {
+        $config = $this->createMock(Config::class);
+        $this->factory->method('create')
+            ->willReturn($config);
+        $this->model->get('amqp');
+
+        $channel = $this->createMock(AMQPChannel::class);
+        $config->expects($this->atLeastOnce())
+            ->method('getChannel')
+            ->willReturn($channel);
+        $connection = $this->createMock(AbstractConnection::class);
+        $channel->expects($this->atLeastOnce())
+            ->method('getConnection')
+            ->willReturn($connection);
+        $channel->expects($this->once())
+            ->method('close');
+        $connection->expects($this->once())
+            ->method('close');
+
+        $this->model->closeConnections();
     }
 }
diff --git a/lib/internal/Magento/Framework/Amqp/Test/Unit/ConfigTest.php b/lib/internal/Magento/Framework/Amqp/Test/Unit/ConfigTest.php
@@ -11,11 +11,20 @@
 use Magento\Framework\Amqp\Connection\Factory as ConnectionFactory;
 use Magento\Framework\Amqp\Connection\FactoryOptions;
 use Magento\Framework\App\DeploymentConfig;
+use PhpAmqpLib\Channel\AMQPChannel;
+use PhpAmqpLib\Connection\AbstractConnection;
 use PHPUnit\Framework\MockObject\MockObject;
 use PHPUnit\Framework\TestCase;
 
 class ConfigTest extends TestCase
 {
+    private const DEFAULT_CONFIG = [
+        Config::HOST => 'localhost',
+        Config::PORT => '5672',
+        Config::USERNAME => 'user',
+        Config::PASSWORD => 'pass',
+    ];
+
     /**
      * @var MockObject
      */
@@ -176,30 +185,94 @@ public function configDataProvider(): array
     {
         return [
             [
-                [
-                    Config::HOST => 'localhost',
-                    Config::PORT => '5672',
-                    Config::USERNAME => 'user',
-                    Config::PASSWORD => 'pass',
-                    Config::VIRTUALHOST => '/',
-                ],
+                self::DEFAULT_CONFIG,
                 [
                     'isSslEnabled' => false
                 ]
             ],
             [
-                [
-                    Config::HOST => 'localhost',
-                    Config::PORT => '5672',
-                    Config::USERNAME => 'user',
-                    Config::PASSWORD => 'pass',
-                    Config::VIRTUALHOST => '/',
-                    Config::SSL => ' true ',
-                ],
+                self::DEFAULT_CONFIG + [Config::SSL => ' true '],
                 [
                     'isSslEnabled' => true
                 ]
             ]
         ];
     }
+
+    public function testGetChannel(): void
+    {
+        $this->deploymentConfigMock->expects($this->once())
+            ->method('getConfigData')
+            ->with(Config::QUEUE_CONFIG)
+            ->willReturn([Config::AMQP_CONFIG => self::DEFAULT_CONFIG]);
+        $connectionMock = $this->createMock(AbstractConnection::class);
+        $this->connectionFactory->expects($this->once())
+            ->method('create')
+            ->willReturn($connectionMock);
+
+        $channelMock = $this->createMock(AMQPChannel::class);
+        $connectionMock->expects($this->once())
+            ->method('channel')
+            ->willReturn($channelMock);
+        $channelMock->expects($this->atLeastOnce())
+            ->method('getConnection')
+            ->willReturn($connectionMock);
+        $connectionMock->expects($this->atLeastOnce())
+            ->method('isConnected')
+            ->willReturn(true);
+
+        $this->assertEquals($channelMock, $this->amqpConfig->getChannel());
+        $this->assertEquals($channelMock, $this->amqpConfig->getChannel());
+    }
+
+    public function testGetChannelWithoutConnection(): void
+    {
+        $this->deploymentConfigMock->expects($this->once())
+            ->method('getConfigData')
+            ->with(Config::QUEUE_CONFIG)
+            ->willReturn([Config::AMQP_CONFIG => self::DEFAULT_CONFIG]);
+        $connectionMock = $this->createMock(AbstractConnection::class);
+        $this->connectionFactory->expects($this->once())
+            ->method('create')
+            ->willReturn($connectionMock);
+
+        $channel1Mock = $this->createMock(AMQPChannel::class);
+        $channel2Mock = $this->createMock(AMQPChannel::class);
+        $connectionMock->expects($this->exactly(2))
+            ->method('channel')
+            ->willReturnOnConsecutiveCalls($channel1Mock, $channel2Mock);
+        $this->amqpConfig->getChannel();
+        $channel1Mock->expects($this->once())
+            ->method('getConnection')
+            ->willReturn(null);
+
+        $this->assertEquals($channel2Mock, $this->amqpConfig->getChannel());
+    }
+
+    public function testGetChannelWithDisconnectedConnection(): void
+    {
+        $this->deploymentConfigMock->expects($this->once())
+            ->method('getConfigData')
+            ->with(Config::QUEUE_CONFIG)
+            ->willReturn([Config::AMQP_CONFIG => self::DEFAULT_CONFIG]);
+        $connectionMock = $this->createMock(AbstractConnection::class);
+        $this->connectionFactory->expects($this->once())
+            ->method('create')
+            ->willReturn($connectionMock);
+
+        $channel1Mock = $this->createMock(AMQPChannel::class);
+        $channel2Mock = $this->createMock(AMQPChannel::class);
+        $connectionMock->expects($this->exactly(2))
+            ->method('channel')
+            ->willReturnOnConsecutiveCalls($channel1Mock, $channel2Mock);
+        $this->amqpConfig->getChannel();
+        $channel1Mock->expects($this->atLeastOnce())
+            ->method('getConnection')
+            ->willReturn($connectionMock);
+        $connectionMock->expects($this->atLeastOnce())
+            ->method('isConnected')
+            ->willReturn(false);
+
+        $this->assertEquals($channel2Mock, $this->amqpConfig->getChannel());
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -169,11 +169,16 @@ private function createConnection(): AbstractConnection`
`169`	`169`	`*/`
`170`	`170`	`public function getChannel()`
`171`	`171`	`{`
`172`		`- if (!isset($this->connection) \|\| !isset($this->channel)) {`
	`172`	`+ if (!isset($this->connection)) {`
`173`	`173`	`$this->connection = $this->createConnection();`
`174`		`-`
	`174`	`+ }`
	`175`	`+ if (!isset($this->channel)`
	`176`	`+ \|\| !$this->channel->getConnection()`
	`177`	`+ \|\| !$this->channel->getConnection()->isConnected()`
	`178`	`+ ) {`
`175`	`179`	`$this->channel = $this->connection->channel();`
`176`	`180`	`}`
	`181`	`+`
`177`	`182`	`return $this->channel;`
`178`	`183`	`}`
`179`	`184`
Original file line number	Diff line number	Diff line change
`@@ -43,4 +43,18 @@ public function get($connectionName)`
`43`	`43`	`}`
`44`	`44`	`return $this->pool[$connectionName];`
`45`	`45`	`}`
	`46`	`+`
	`47`	`+ /**`
	`48`	`+ * Close all opened connections.`
	`49`	`+ *`
	`50`	`+ * @return void`
	`51`	`+ */`
	`52`	`+ public function closeConnections(): void`
	`53`	`+ {`
	`54`	`+ foreach ($this->pool as $config) {`
	`55`	`+ $connection = $config->getChannel()->getConnection();`
	`56`	`+ $config->getChannel()->close();`
	`57`	`+ $connection?->close();`
	`58`	`+ }`
	`59`	`+ }`
`46`	`60`	`}`