MC-18477: Deadlocks when consumers run from cron

BaDos · BaDos · commit 15a194ec2e1a · 2019-07-31T10:59:19.000-05:00
diff --git a/app/code/Magento/MessageQueue/Console/StartConsumerCommand.php b/app/code/Magento/MessageQueue/Console/StartConsumerCommand.php
@@ -64,7 +64,7 @@ public function __construct(
     }
 
     /**
-     * {@inheritdoc}
+     * @inheritdoc
      */
     protected function execute(InputInterface $input, OutputInterface $output)
     {
@@ -79,13 +79,13 @@ protected function execute(InputInterface $input, OutputInterface $output)
 
         $singleThread = $input->getOption(self::OPTION_SINGLE_THREAD);
 
-        if ($singleThread && $this->lockManager->isLocked(md5($consumerName))) {
+        if ($singleThread && $this->lockManager->isLocked(md5($consumerName))) { //phpcs:ignore
             $output->writeln('<error>Consumer with the same name is running</error>');
             return \Magento\Framework\Console\Cli::RETURN_FAILURE;
         }
 
         if ($singleThread) {
-            $this->lockManager->lock(md5($consumerName));
+            $this->lockManager->lock(md5($consumerName)); //phpcs:ignore
         }
 
         $this->appState->setAreaCode($areaCode ?? 'global');
@@ -94,14 +94,14 @@ protected function execute(InputInterface $input, OutputInterface $output)
         $consumer->process($numberOfMessages);
 
         if ($singleThread) {
-            $this->lockManager->unlock(md5($consumerName));
+            $this->lockManager->unlock(md5($consumerName)); //phpcs:ignore
         }
 
         return \Magento\Framework\Console\Cli::RETURN_SUCCESS;
     }
 
     /**
-     * {@inheritdoc}
+     * @inheritdoc
      */
     protected function configure()
     {
@@ -136,13 +136,13 @@ protected function configure()
             self::OPTION_SINGLE_THREAD,
             null,
             InputOption::VALUE_NONE,
-            'Does not allow to run a consumer if it was run before.'
+            'This option prevents running multiple copies of one consumer simultaneously.'
         );
         $this->addOption(
             self::PID_FILE_PATH,
             null,
             InputOption::VALUE_REQUIRED,
-            'The file path for saving PID'
+            'The file path for saving PID (This option is deprecated, use --single-thread instead)'
         );
         $this->setHelp(
             <<<HELP
@@ -164,7 +164,7 @@ protected function configure()
 
     <comment>%command.full_name% someConsumer --area-code='adminhtml'</comment>
     
-To do not allow to run a consumer if it was run before:
+To do not run multiple copies of one consumer simultaneously:
 
     <comment>%command.full_name% someConsumer --single-thread'</comment>
 
diff --git a/app/code/Magento/MessageQueue/Model/Cron/ConsumersRunner.php b/app/code/Magento/MessageQueue/Model/Cron/ConsumersRunner.php
@@ -71,6 +71,7 @@ class ConsumersRunner
      * @param ConsumerConfigInterface $consumerConfig The consumer config provider
      * @param DeploymentConfig $deploymentConfig The application deployment configuration
      * @param ShellInterface $shellBackground The shell command line wrapper for executing command in background
+     * @param LockManagerInterface $lockManager The lock manager
      * @param ConnectionTypeResolver $mqConnectionTypeResolver Consumer connection resolver
      * @param LoggerInterface $logger Logger
      */
@@ -146,20 +147,22 @@ private function canBeRun(ConsumerConfigItemInterface $consumerConfig, array $al
             return false;
         }
 
-        if ($this->lockManager->isLocked(md5($consumerName))) {
+        if ($this->lockManager->isLocked(md5($consumerName))) { //phpcs:ignore
             return false;
         }
 
         $connectionName = $consumerConfig->getConnection();
         try {
             $this->mqConnectionTypeResolver->getConnectionType($connectionName);
         } catch (\LogicException $e) {
-            $this->logger->info(sprintf(
-                'Consumer "%s" skipped as required connection "%s" is not configured. %s',
-                $consumerName,
-                $connectionName,
-                $e->getMessage()
-            ));
+            $this->logger->info(
+                sprintf(
+                    'Consumer "%s" skipped as required connection "%s" is not configured. %s',
+                    $consumerName,
+                    $connectionName,
+                    $e->getMessage()
+                )
+            );
             return false;
         }
 
diff --git a/app/code/Magento/MessageQueue/Test/Unit/Console/StartConsumerCommandTest.php b/app/code/Magento/MessageQueue/Test/Unit/Console/StartConsumerCommandTest.php
@@ -131,16 +131,15 @@ public function testExecute(
 
         $this->lockManagerMock->expects($this->exactly($isLockedExpects))
             ->method('isLocked')
-            ->with(md5($consumerName))
+            ->with(md5($consumerName)) //phpcs:ignore
             ->willReturn($isLocked);
 
         $this->lockManagerMock->expects($this->exactly($lockExpects))
             ->method('lock')
-            ->with(md5($consumerName));
+            ->with(md5($consumerName)); //phpcs:ignore
         $this->lockManagerMock->expects($this->exactly($unlockExpects))
             ->method('unlock')
-            ->with(md5($consumerName));
-
+            ->with(md5($consumerName)); //phpcs:ignore
 
         $this->assertEquals(
             $expectedReturn,
diff --git a/app/code/Magento/MessageQueue/Test/Unit/Model/Cron/ConsumersRunnerTest.php b/app/code/Magento/MessageQueue/Test/Unit/Model/Cron/ConsumersRunnerTest.php
@@ -15,6 +15,9 @@
 use Symfony\Component\Process\PhpExecutableFinder;
 use Magento\Framework\Lock\LockManagerInterface;
 
+/**
+ * Unit tests for ConsumersRunner.
+ */
 class ConsumersRunnerTest extends \PHPUnit\Framework\TestCase
 {
     /**
@@ -90,11 +93,13 @@ public function testRunDisabled()
     {
         $this->deploymentConfigMock->expects($this->once())
             ->method('get')
-            ->willReturnMap([
-                ['cron_consumers_runner/cron_run', true, false],
-                ['cron_consumers_runner/max_messages', 10000, 10000],
-                ['cron_consumers_runner/consumers', [], []],
-            ]);
+            ->willReturnMap(
+                [
+                    ['cron_consumers_runner/cron_run', true, false],
+                    ['cron_consumers_runner/max_messages', 10000, 10000],
+                    ['cron_consumers_runner/consumers', [], []],
+                ]
+            );
 
         $this->consumerConfigMock->expects($this->never())
             ->method('getConsumers');
@@ -131,11 +136,13 @@ public function testRun(
 
         $this->deploymentConfigMock->expects($this->exactly(3))
             ->method('get')
-            ->willReturnMap([
-                ['cron_consumers_runner/cron_run', true, true],
-                ['cron_consumers_runner/max_messages', 10000, $maxMessages],
-                ['cron_consumers_runner/consumers', [], $allowedConsumers],
-            ]);
+            ->willReturnMap(
+                [
+                    ['cron_consumers_runner/cron_run', true, true],
+                    ['cron_consumers_runner/max_messages', 10000, $maxMessages],
+                    ['cron_consumers_runner/consumers', [], $allowedConsumers],
+                ]
+            );
 
         /** @var ConsumerConfigInterface|MockObject $firstCunsumer */
         $consumer = $this->getMockBuilder(ConsumerConfigItemInterface::class)
@@ -154,7 +161,7 @@ public function testRun(
 
         $this->lockManagerMock->expects($this->exactly($isRunExpects))
             ->method('isLocked')
-            ->with(md5($consumerName))
+            ->with(md5($consumerName)) //phpcs:ignore
             ->willReturn($isLocked);
 
         $this->shellBackgroundMock->expects($this->exactly($shellBackgroundExpects))
diff --git a/dev/tests/integration/testsuite/Magento/MessageQueue/Model/Cron/ConsumersRunnerTest.php b/dev/tests/integration/testsuite/Magento/MessageQueue/Model/Cron/ConsumersRunnerTest.php
@@ -97,16 +97,18 @@ protected function setUp()
 
         $this->shellMock->expects($this->any())
             ->method('execute')
-            ->willReturnCallback(function ($command, $arguments) {
-                $command = vsprintf($command, $arguments);
-                $params = \Magento\TestFramework\Helper\Bootstrap::getInstance()->getAppInitParams();
-                $params['MAGE_DIRS']['base']['path'] = BP;
-                $params = 'INTEGRATION_TEST_PARAMS="' . urldecode(http_build_query($params)) . '"';
-                $command = str_replace('bin/magento', 'dev/tests/integration/bin/magento', $command);
-                $command = $params . ' ' . $command;
-
-                return exec("{$command} >/dev/null &");
-            });
+            ->willReturnCallback(
+                function ($command, $arguments) {
+                    $command = vsprintf($command, $arguments);
+                    $params = \Magento\TestFramework\Helper\Bootstrap::getInstance()->getAppInitParams();
+                    $params['MAGE_DIRS']['base']['path'] = BP;
+                    $params = 'INTEGRATION_TEST_PARAMS="' . urldecode(http_build_query($params)) . '"';
+                    $command = str_replace('bin/magento', 'dev/tests/integration/bin/magento', $command);
+                    $command = $params . ' ' . $command;
+
+                    return exec("{$command} >/dev/null &"); //phpcs:ignore
+                }
+            );
     }
 
     /**
@@ -120,16 +122,16 @@ public function testSpecificConsumerAndRerun()
         $config = $this->config;
         $config['cron_consumers_runner'] = ['consumers' => [$specificConsumer], 'max_messages' => 0];
         $this->writeConfig($config);
-        $this->reRunConsumersAndCheckPidFiles($specificConsumer);
-        $this->reRunConsumersAndCheckPidFiles($specificConsumer);
-        $this->assertTrue($this->lockManager->isLocked(md5($specificConsumer)));
+        $this->reRunConsumersAndCheckLocks($specificConsumer);
+        $this->reRunConsumersAndCheckLocks($specificConsumer);
+        $this->assertTrue($this->lockManager->isLocked(md5($specificConsumer))); //phpcs:ignore
     }
 
     /**
      * @param string $specificConsumer
      * @return void
      */
-    private function reRunConsumersAndCheckPidFiles($specificConsumer)
+    private function reRunConsumersAndCheckLocks($specificConsumer)
     {
         $this->consumersRunner->run();
 
@@ -139,9 +141,9 @@ private function reRunConsumersAndCheckPidFiles($specificConsumer)
             $consumerName = $consumer->getName();
 
             if ($consumerName === $specificConsumer) {
-                $this->assertTrue($this->lockManager->isLocked(md5($consumerName)));
+                $this->assertTrue($this->lockManager->isLocked(md5($consumerName))); //phpcs:ignore
             } else {
-                $this->assertFalse($this->lockManager->isLocked(md5($consumerName)));
+                $this->assertFalse($this->lockManager->isLocked(md5($consumerName))); //phpcs:ignore
             }
         }
     }
@@ -163,7 +165,7 @@ public function testCronJobDisabled()
         sleep(20);
 
         foreach ($this->consumerConfig->getConsumers() as $consumer) {
-            $this->assertFalse($this->lockManager->isLocked(md5($consumer->getName())));
+            $this->assertFalse($this->lockManager->isLocked(md5($consumer->getName()))); //phpcs:ignore
         }
     }
 
@@ -193,7 +195,7 @@ protected function tearDown()
     {
         foreach ($this->consumerConfig->getConsumers() as $consumer) {
             foreach ($this->getConsumerProcessIds($consumer->getName()) as $consumerProcessId) {
-                exec("kill {$consumerProcessId}");
+                exec("kill {$consumerProcessId}"); //phpcs:ignore
             }
         }
 
@@ -213,6 +215,7 @@ protected function tearDown()
      */
     private function getConsumerProcessIds($consumer)
     {
+        //phpcs:ignore
         exec("ps ax | grep -v grep | grep 'queue:consumers:start {$consumer}' | awk '{print $1}'", $output);
         return $output;
     }
