Overview

The Message Queue API has been designed with the intention of 3rd party application integration and high volume processing. One of these 3rd parties is our own Business Process Engine. The Message Queue API has been tested with the Informatica BPM system, but is open to any 3rd party application which can read and write to/from an Apache MQ Server.

In order to be most flexible we tried to minimize any efforts in setting up queues and linking applications together.

Queue Communication Overview

Product 360 produces messages as well as receives messages from Apache MQ. This picture shows the general communication streams between an external application and the Product 360 Application Servers by using Message Queues.

Each queue in this picture is an own queue within the Apache MQ Server instance, so a single server will host all those queues.

The queues allow us to decouple the external applications from the product 360 servers and therefor they form also a buffer for peak load scenarios. Each side of the equation can process the messages in their own speed, typically as fast as possible without crashing. If there are more messages within the queue, the queue will safely have them persisted as long as it's needed. In case the Apache Message queue would reach it's limit it will automatically make sure that the producer will slow down. All this making sure that no module is overloaded at any time.

Of course, this holds only true in case there are no synchronous REST api calls. None, or just a very minimum. Each rest call is a synchronous call which uses a http thread within the application server. Client users also need those http threads and they are a limited resource. So by executing huge amounts of synchronous requests which might execute some performance intensive logic, it can happen that a server becomes unresponsive for users. This is something which should be avoided. Using the Message Queue API instead of the synchronous REST api bypasses this.

Let's follow the basic flow:

1. A user modifies something, either manually or with a job.

   1. A BPM trigger is configured for this modification. A new trigger event message is created with the configured payload. This event is sent to the queue which is configured in the trigger configuration.
      There can be multiple queues configured for the trigger framework - if more than one, the user can actually chose in which queue the event message should be sent! The setting is within the server.properties file, it's called: infa.bpm.trigger.queue.ids. It defaults to "bpm".
      Also multiple triggers which send events to multiple queues is possible. One for the BPM server, another one for a 3rd party system - even with different payloads.

   2. A DQ trigger is configured for object created or changed. This trigger will write directly a corresponding DQ message to the batch api queue. As the server which changed the item is not necessarily the same server which processes the Message Queue API. Also this way it's guaranteed that all our batching capabilities are used

2. Informatica BPM (or a 3rd party system) consumes the Trigger Event messages and starts the corresponding workflow process.

3. Based on the workflows logic, a batch api request is sent to the batch api queue. Currently only Data Quality and Merge requests are supported for the batch api.

4. The batch api consumer consumes the message from the queue.

5. The batch api framework groups and accumulates multiple messages for a specific amount of time or number of messages in order to optimize the data quality and merge executions. Please refer to Batch API Queue for details on the algorithm and the possible configurations.

6. The results of the data quality and merge executions are split into multiple responses so that every requests receives his own response. This renders the batch processing fully transparent to the external application. The responses are sent to the queue which is defined in the requests as "responseQueue". In this picture the "bpm_response" - but it could be a different one!

7. Informatica BPM or a 3rd party application reads the response and processes it by continuing or starting a new workflow process instance.

8. The workflow might also execute other "service api" calls by means of request message. For example enter/leave a workflow status, assigning an item into a task etc. Please have a look at Service API Queue for details on how to compose those messages.

9. The Service API Consumer reads the service api request message and executes it against our standard service api. As this consumer is implemented in a fully generic way he basically is able to execute most of the available service api calls. Not all of them are fully tested and supported. The ones which are are listed on the Service API Queue page

10. The response from the Service api is sent to the response queue, in our example also "bpm_response". This can also be a custom queue!

11. The workflow might also call the object api to obtain a full picture of an object, or to modify the object. In terms of single item processing the object api should be used instead of the list api as the object api is backed by an in memory cache.

12. The object api consumer directly calls the internal object api, bypassing the Service API completely. As the object api is optimized for high throughput the extra marshalling/unmarshalling can be avoided this way. When the object api queue is used, also no further thread pools are involved in the object api - the consumer threads do all the work. In case the object api is executed via rest, an internal thread pool is processing the request. This is to prevent overload scenarios of the application server.

13. The response from the object api is being sent to the response queue, in our example also "bpm_response". This can also be a custom queue!

14. Although discouraged and potentially dangerous, the workflow might also still call the service or object api directly. Please be aware that this will block an http thread during the time of the execution. Depending on the request and the number of concurrent workflows this might lead to an overload in the Jetty Server which will render the server unresponsive for web-clients! As there is a limited number of consumer threads for the Service API Consumer the way via message queue can not overload the system.

General Queue Configuration

All queues which are summarized in "Message Queue API" are configured in the server.properties configuration file.

Each queue has an unique id, the queue id. The ID is name after the queue. prefix. So in this example it's "batchapi".

• type: The type can not be changed as we currently only support ActiveMQ.

• writer.count: The number of threads which are used to write messages into the queue

• consumer.count: The number of threads which are used to consume messages from the queue

• url: the url to the Apache MQ Server

• username: the username which should be used for the apache active mq server

• password: the password which should be used for the apache active mq server

• message.format: Either XML or JSON. Communication with the Informatica BPM Server requires here XML. If other scenarios allow JSON, use it.

• name: the name of the Queue within the Apache Active MQ server. If the queue is not available in Apache Active MQ, it will automatically be created

• label: the human readable label of the queue. It's used in the trigger configuration UI to chose the the target queue of the trigger event.

Custom Queues

Customers can easily create own queues by just providing the configuration like described above in the server.properties file. Let's assume we have a downstream syndication system which needs to receive all modifications a user does.

1. Add the "syndication" queue to the server.properties file:

   queue.syndication.type = ${queue.default.type}

   queue.syndication.writer.count = ${queue.default.writer.count}

   queue.syndication.consumer.count = ${queue.default.consumer.count}

   queue.syndication.url = ${queue.default.url}

   queue.syndication.username = ${queue.default.username}

   queue.syndication.password = ${queue.default.password}

   queue.syndication.message.format = JSON

   queue.syndication.name = CUSTOM_SYNDICATION

   queue.syndication.label = Syndication

   We chose JSON as format since it's more efficient than xml and the syndication system can handle it just well.

2. Add the "syndication" queue to the trigger queues:
   infa.bpm.trigger.queue.ids = bpm,syndication

3. Chose the "Syndication" queue in the trigger configuration of the "Business Process management" perspective

   

4. Open the Apache Active MQ Console and see the content of the new CUSTOM_SYNDICATION queue

   

5. Have a look at the content of the message. It looks something like this

   Payload in new Syndication Queue when configured with

   {

   "entityId": "1185@1",

   "entity": "Article",

   "data": {

   "catalog.label": "MASTER",

   "triggerConfiguration": "ItemChanged",

   "triggerIdentifier": "hlr.persistence.trigger.entityChanged",

   "user.label": "Administrator",

   "userGroup.label": ""

   },

   "entityItemChange": {

   "_module": "UI",

   "_entity": "Article",

   "_identifier": "Item_A",

   "_user": {

   "_internalId": "1",

   "_entityId": 2600,

   "_externalId": "'Administrator'"

   },

   "_revision": {

   "_internalId": "1",

   "_entityId": 5600,

   "_externalId": "'root'"

   },

   "_container": {

   "_internalId": "1",

   "_entityId": 2900,

   "_externalId": "'MASTER'"

   },

   "_entityItem": {

   "_internalId": "1185@1",

   "_entityId": 1000,

   "_externalId": "'Item_A'@'MASTER'"

   },

   "_eventTimestamp": "2022-06-09T10:42:25.420Z",

   "_changeType": "CHANGED",

   "_changedFields": [

   "Article.ManufacturerName"

   ],

   "_changedEntities": [

   "Article"

   ],

   "_changeSummary": {

   "article": {

   "_changeType": "CHANGED",

   "_mainSupplierProxy": {

   "_entityId": 2800,

   "_internalId": "3",

   "_externalId": "'Heiler Product Manager'"

   },

   "manufacturerName": {

   "_old": "Mercedes",

   "_current": "BMW"

   }

   }

   }

   }

   }

General Headers

These headers apply to all queues, unless stated otherwise on the queue page.

Default Queues