Kdjykuj

I am trying to implement Bulk Operations in Magento 2.3.1 using the docs here as a reference.

When I call scheduleBulk on the BulkManagementInterface the operation is saved and can be viewed from the Bulk Action log but I get the following exception and my consumer never processes the operation:

Message queue topic "example.topic" is not configured.

I have the following files in app/etc:

di.xml

<?xml version="1.0"?>

<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework:ObjectManager/etc/config.xsd">

    <type name="MagentoFrameworkMessageQueueMergerFactory">

        <arguments>

            <argument name="mergers" xsi:type="array">

                <item name="ExampleConsumer" xsi:type="string">VendorModuleModelMerger</item>

            </argument>

        </arguments>

    </type>

</config>

communication.xml

<?xml version="1.0"?>

<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework:Communication/etc/communication.xsd">

    <topic name="example.topic" request="MagentoAsynchronousOperationsApiDataOperationInterface">

        <handler name="processExampleOperation" type="VendorModuleModelConsumer" method="processOperation"/>

    </topic>

</config>

queue_consumer.xml

<?xml version="1.0"?>

<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/consumer.xsd">

    <consumer name="ExampleConsumer" queue="example_queue" connection="db" handler="VendorModuleModelConsumer::processOperation"/>

</config>

queue_publisher.xml

<?xml version="1.0"?>

<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/publisher.xsd">

    <publisher topic="example.topic">

        <connection name="db" exchange="example.exchange" />

    </publisher>

</config>

queue_topology.xml

<?xml version="1.0"?>

<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/topology.xsd">

    <exchange name="example.exchange type="topic" connection="db">

        <binding id="ExampleBinding" topic="example.topic" destinationType="queue" destination="example_queue"/>

    </exchange>

</config>

Here is a snippet where I call scheduleBulk:

$bulkUuid = $this->_identityService->generateId();

$bulkDescription = 'Example Bulk Operation';

$operations = ;

$serializedData = [

    'entity_id' => $id,

    'entity_link' => $this->_urlBuilder->getUrl('vendor/controller/' . $id),

    'meta_information' => 'Meta Information',

];

$data = [

    'data' => [

        'bulk_uuid' => $bulkUuid,

        'topic_name' => 'example.topic',

        'serialized_data' => $this->_jsonHelper->jsonEncode($serializedData),

        'status' => OperationInterface::STATUS_TYPE_OPEN,

    ]

];



$operation = $this->_operationFactory->create($data);

$operations = $operation;





$userId = $this->_userContext->getUserId();

$result = $this->_bulkManagement->scheduleBulk($bulkUuid, $operations, $bulkDescription, $userId);



if (!$result) {

    throw new MagentoFrameworkExceptionLocalizedException(

        __('Something went wrong while processing the request.')

    );

}

When I run bin/magento queue:consumers:list my consumer is listed, and when I run start it is started.

The exception is thrown from

MagentoFrameworkMessageQueueConfig in the getPublisherConfigByTopic
function on line 184

. If I log $this->queueConfigData->get() at the top of that function it is an empty array.

Is there an issue with my configuration above that would be preventing it from being loaded?

Note: I have modified the sample code above and replaced with example data.