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

Author: BaDos

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

@@ -0,0 +1,103 @@
+<?php
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+declare(strict_types=1);
+
+namespace Magento\MessageQueue\Setup;
+
+use Magento\Framework\Setup\ConfigOptionsListInterface;
+use Magento\Framework\Setup\Option\SelectConfigOption;
+use Magento\Framework\App\DeploymentConfig;
+use Magento\Framework\Config\Data\ConfigData;
+use Magento\Framework\Config\File\ConfigFilePool;
+
+/**
+ * Deployment configuration consumers options needed for Setup application
+ */
+class ConfigOptionsList implements ConfigOptionsListInterface
+{
+    /**
+     * Input key for the option
+     */
+    const INPUT_KEY_QUEUE_CONSUMERS_WAIT_FOR_MESSAGES ='consumers-wait-for-messages';
+
+    /**
+     * Path to the value in the deployment config
+     */
+    const CONFIG_PATH_QUEUE_CONSUMERS_WAIT_FOR_MESSAGES = 'queue/consumers_wait_for_messages';
+
+    /**
+     * Default value
+     */
+    const DEFULT_CONSUMERS_WAIT_FOR_MESSAGES = 1;
+
+    private $selectOptions = [0, 1];
+
+    /**
+     * @inheritdoc
+     */
+    public function getOptions()
+    {
+        return [
+            new SelectConfigOption(
+                self::INPUT_KEY_QUEUE_CONSUMERS_WAIT_FOR_MESSAGES,
+                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',
+                self::DEFULT_CONSUMERS_WAIT_FOR_MESSAGES
+            ),
+        ];
+    }
+
+    /**
+     * @inheritdoc
+     * @SuppressWarnings(PHPMD.UnusedFormalParameter)
+     */
+    public function createConfig(array $data, DeploymentConfig $deploymentConfig)
+    {
+        $configData = new ConfigData(ConfigFilePool::APP_ENV);
+
+        if (!$this->isDataEmpty($data, self::INPUT_KEY_QUEUE_CONSUMERS_WAIT_FOR_MESSAGES)) {
+            $configData->set(
+                self::CONFIG_PATH_QUEUE_CONSUMERS_WAIT_FOR_MESSAGES,
+                (int)$data[self::INPUT_KEY_QUEUE_CONSUMERS_WAIT_FOR_MESSAGES]
+            );
+        }
+
+        return[$configData];
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function validate(array $options, DeploymentConfig $deploymentConfig)
+    {
+        $errors = [];
+
+        if (!$this->isDataEmpty($options, self::INPUT_KEY_QUEUE_CONSUMERS_WAIT_FOR_MESSAGES)
+            && !in_array($options[self::INPUT_KEY_QUEUE_CONSUMERS_WAIT_FOR_MESSAGES], $this->selectOptions)) {
+            $errors[] = 'You can use only 1 or 0 for ' . self::INPUT_KEY_QUEUE_CONSUMERS_WAIT_FOR_MESSAGES . ' option';
+        }
+
+        return $errors;
+    }
+
+    /**
+     * Check if data ($data) with key ($key) is empty
+     *
+     * @param array $data
+     * @param string $key
+     * @return bool
+     */
+    private function isDataEmpty(array $data, $key)
+    {
+        if (isset($data[$key]) && $data[$key] !== '') {
+            return false;
+        }
+
+        return true;
+    }
+}

lib/internal/Magento/Framework/MessageQueue/CallbackInvoker.php

@@ -8,6 +8,7 @@
 
 use Magento\Framework\MessageQueue\PoisonPill\PoisonPillCompareInterface;
 use Magento\Framework\MessageQueue\PoisonPill\PoisonPillReadInterface;
+use Magento\Framework\App\DeploymentConfig;
 
 /**
  * Class CallbackInvoker to invoke callbacks for consumer classes
@@ -29,16 +30,24 @@ class CallbackInvoker implements CallbackInvokerInterface
      */
     private $poisonPillCompare;
 
+    /**
+     * @var DeploymentConfig
+     */
+    private $deploymentConfig;
+
     /**
      * @param PoisonPillReadInterface $poisonPillRead
      * @param PoisonPillCompareInterface $poisonPillCompare
+     * @param DeploymentConfig $deploymentConfig
      */
     public function __construct(
         PoisonPillReadInterface $poisonPillRead,
-        PoisonPillCompareInterface $poisonPillCompare
+        PoisonPillCompareInterface $poisonPillCompare,
+        DeploymentConfig $deploymentConfig
     ) {
         $this->poisonPillRead = $poisonPillRead;
         $this->poisonPillCompare = $poisonPillCompare;
+        $this->deploymentConfig = $deploymentConfig;
     }
 
     /**
@@ -56,13 +65,29 @@ public function invoke(QueueInterface $queue, $maxNumberOfMessages, $callback)
             do {
                 $message = $queue->dequeue();
                 // phpcs:ignore Magento2.Functions.DiscouragedFunction
-            } while ($message === null && (sleep(1) === 0));
+            } while ($message === null && $this->isWaitingNextMessage() && (sleep(1) === 0));
+
+            if ($message === null) {
+                break;
+            }
+
             if (false === $this->poisonPillCompare->isLatestVersion($this->poisonPillVersion)) {
                 $queue->reject($message);
                 // phpcs:ignore Magento2.Security.LanguageConstruct.ExitUsage
                 exit(0);
             }
+
             $callback($message);
         }
     }
+
+    /**
+     * Checks if consumers should wait for message from the queue
+     *
+     * @return bool
+     */
+    private function isWaitingNextMessage(): bool
+    {
+        return $this->deploymentConfig->get('queue/consumers_wait_for_messages', 1) === 1;
+    }
 }