MC-19250: The stuck deployment on the Cloud because of consumers

Author: BaDos

app/code/Magento/MessageQueue/Setup/ConfigOptionsList.php

@@ -46,7 +46,7 @@ public function getOptions()
                 SelectConfigOption::FRONTEND_WIZARD_SELECT,
                 $this->selectOptions,
                 self::CONFIG_PATH_QUEUE_CONSUMERS_WAIT_FOR_MESSAGES,
-                'Should consumers wait for message from the queue? 1 - Yes, 0 - No',
+                'Should consumers wait for a message from the queue? 1 - Yes, 0 - No',
                 self::DEFULT_CONSUMERS_WAIT_FOR_MESSAGES
             ),
         ];

dev/tests/integration/testsuite/Magento/Framework/MessageQueue/UseCase/QueueTestCaseAbstract.php

@@ -45,7 +45,7 @@ class QueueTestCaseAbstract extends \PHPUnit\Framework\TestCase
     /**
      * @var PublisherConsumerController
      */
-    private $publisherConsumerController;
+    protected $publisherConsumerController;
 
     protected function setUp()
     {

dev/tests/integration/testsuite/Magento/Framework/MessageQueue/UseCase/WaitAndNotWaitMessagesTest.php

@@ -6,7 +6,10 @@
 namespace Magento\Framework\MessageQueue\UseCase;
 
 use Magento\Framework\App\DeploymentConfig\FileReader;
+use Magento\TestModuleAsyncAmqp\Model\AsyncTestData;
+use Magento\Framework\App\DeploymentConfig\Writer;
 use Magento\Framework\Config\File\ConfigFilePool;
+use Magento\Framework\Filesystem;
 
 class WaitAndNotWaitMessagesTest extends QueueTestCaseAbstract
 {
@@ -15,25 +18,113 @@ class WaitAndNotWaitMessagesTest extends QueueTestCaseAbstract
      */
     private $reader;
 
+    /**
+     * @var Filesystem
+     */
+    private $filesystem;
+
+    /**
+     * @var ConfigFilePool
+     */
+    private $configFilePool;
+
     /**
      * @var array
      */
     private $config;
 
+    /**
+     * @var \Magento\TestModuleAsyncAmqp\Model\AsyncTestData
+     */
+    protected $msgObject;
+
+    /**
+     * {@inheritdoc}
+     */
+    protected $consumers = ['mixed.sync.and.async.queue.consumer'];
+
+    /**
+     * @var string[]
+     */
+    protected $messages = ['message1', 'message2', 'message3'];
+
+    /**
+     * @var int|null
+     */
+    protected $maxMessages = 4;
+
     /**
      * @inheritdoc
      */
     protected function setUp()
     {
         parent::setUp();
+        $this->msgObject = $this->objectManager->create(AsyncTestData::class);
         $this->reader = $this->objectManager->get(FileReader::class);
-
+        $this->filesystem = $this->objectManager->get(Filesystem::class);
         $this->config = $this->loadConfig();
     }
 
-    public function testDefaultConfiguration()
+    /**
+     * Check if consumers wait for messages from the queue
+     */
+    public function testWaitForMessages()
     {
         $this->assertArraySubset(['queue' => ['consumers_wait_for_messages' => 1]], $this->config);
+
+        foreach ($this->messages as $msg) {
+            $this->publishMessage($msg);
+        }
+
+        $this->waitForAsynchronousResult(count($this->messages), $this->logFilePath);
+
+        foreach ($this->messages as $item) {
+            $this->assertContains($item, file_get_contents($this->logFilePath));
+        }
+
+        $this->publishMessage('message4');
+        $this->waitForAsynchronousResult(count($this->messages) + 1, $this->logFilePath);
+        $this->assertContains('message4', file_get_contents($this->logFilePath));
+    }
+
+    /**
+     * Check if consumers do not wait for messages from the queue and die
+     */
+    public function testNotWaitForMessages(): void
+    {
+        $this->publisherConsumerController->stopConsumers();
+
+        $config = $this->config;
+        $config['queue']['consumers_wait_for_messages'] = 0;
+        $this->writeConfig($config);
+
+        $this->assertArraySubset(['queue' => ['consumers_wait_for_messages' => 0]], $this->loadConfig());
+        foreach ($this->messages as $msg) {
+            $this->publishMessage($msg);
+        }
+
+        $this->publisherConsumerController->startConsumers();
+        $this->waitForAsynchronousResult(count($this->messages), $this->logFilePath);
+
+        foreach ($this->messages as $item) {
+            $this->assertContains($item, file_get_contents($this->logFilePath));
+        }
+
+        // Checks that consumers do not wait 4th message and die
+        $this->assertArraySubset(
+            ['mixed.sync.and.async.queue.consumer' => []],
+            $this->publisherConsumerController->getConsumersProcessIds()
+        );
+    }
+
+    /**
+     * @param string $msg
+     */
+    private function publishMessage(string $msg): void
+    {
+        $this->msgObject->setValue($msg);
+        $this->msgObject->setTextFilePath($this->logFilePath);
+        $this->publisher->publish('multi.topic.queue.topic.c', $this->msgObject);
     }
 
     /**
@@ -43,4 +134,22 @@ private function loadConfig(): array
     {
         return $this->reader->load(ConfigFilePool::APP_ENV);
     }
+
+    /**
+     * @param array $config
+     */
+    private function writeConfig(array $config): void
+    {
+        $writer = $this->objectManager->get(Writer::class);
+        $writer->saveConfig([ConfigFilePool::APP_ENV => $config], true);
+    }
+
+    /**
+     * @inheritdoc
+     */
+    protected function tearDown()
+    {
+        parent::tearDown();
+        $this->writeConfig($this->config);
+    }
 }