Asynchronous Service Processing in Pega
In this blog article, we will see how Pega can process incoming service requests in an asynchronous manner. We will also deep dive into the entire architecture behind the asynchronous service processing. Let’s get started.
Business scenario Say ABC is a banking organization that provide loans to customer. ABC uses a Pega application for sales processing. A legacy Java application maintains the loan status for each consumer loans. When there is some loan status update in the Java application, it needs to communicate the same to the Pega applications to update the loan case.
Since it is a very old legacy Java application, the only option is to use the SOAP web service to make a call to Pega application.
So a Service SOAP is hosted in the Pega application can accept the SOAP request from the Java application, update the loan case and then returns the response to the Java application. Here the Java application waits for Pega service to process the entire request and receives the processing response. We can call it as synchronous processing.
Think about the error situation, say a lock issue occurred and service SOAP failed to update the case!!
There are two main ways to handle this situation
Option 1: Service can respond with status message as fail and so the Java application can try to resend the request again! By this option, the Java application should be able to resend the same request again and again until the processing turn success.
OR
Option 2: In case of service error, Pega application can queue the request to process at a later point of time and send the queue item ID back to Java application for tracking purposes. Here Pega takes ownership in processing the request. Java application need not send the request, again and again, instead, they can just check the status of Queue item ID to verify if it is processed.
Option 2 falls under asynchronous processing 🙂 where Pega queues the request to process at a later point of time.
The sequential steps for the asynchronous service processing for our use case is below
Step 1: When the loan status change, the Java application sends a SOAP request to Pega application with the loan number and status.
Step 2: The SOAP service queues ( after error) the request to asynchronous processing. Queuing here refers to persisting the mapped request in the queue table and assigning the queue item to agent processing.
Step 3: SOAP service returns back the Queue Item ID in the SOAP response to the Java application.
Step 4: Asynchronous processing (agent processing) will take care of updating the loan cases and updates the status.
Step 5: The Java application can request the status of the Queue Item ID in separate service request.
Step 6: The SOAP service responds with the queue status.
There may be some other use cases as well, where you want to process the service requests asynchronously every time irrespective of error situation.
For example
a) The Java application do not need the response at that point of time (not urgent may be), can also be kind of fire and forget concept.
b) If the Pega processing involves multiple system updates which can take a lot of time for processing then the Java application need to wait long for the response. In this situation, instead can poll back at a later point in time
In this tutorial, we will focus on asynchronous processing only in the state of error.
What are the pre-requisites for this tutorial?
a) A loan processing case type is created with simple data model – loan number and loan status. A couple of cases were created with different loan number and status.
b) Service SOAP rule is created, that accepts simple request input as – Loan Number and Loan status and updates the loan case using service activity as shown below.
c) The update from service SOAP should work perfectly fine
Try to simulate the request for a valid loan number to see if the service works as expected.
Service returned pyStatusMessage = Success and also the case is updated with status Inactive.
All pre-requisites are ready ☺️
How to configure asynchronous process for service SOAP?
Step 1: Update the execution mode (!!)
Open the created service SOAP – service tab under the Processing options block.
Note: the execution mode dropdown can vary based on your Pega version.
By default the execution mode will be set to – Execute synchronously.
You see one-way operation. It is used to process the request and send only the HTTP response code.
Execute synchronously – Process the request and return SOAP response.
Execute synchronously (one-way) – Process the request and return only the HTTP response code.
Execute asynchronously ( queue for execution) – Queue the request and return the Queue item ID as response.
One-way operation (queue for operation) – queue the request and return only the HTTP response code – like the fire and forget concept.
Now a big question – How to queue for execution only when processing an error??
Important note: In lower versions, you get an option to execute asynchronously ( queue for error) option.
You need to set the option – Execute synchronously so that in case of success processing, your service behaves in synchronous way.
Step 2: Update the Fault tab – Queue when condition.
Note: For the Service rules, you may have the Queue when condition in response tab!
In the When key, you can specify a when condition rule.
So when the when condition satisfies to true, then the request will be queued and behaves in asynchronous ways, simple right.
Step 2.1: As per the current use case from service activity,
.pyStatusMessage = Success -> In case of success processing
.pyStatusMessage = Fail -> In case of Fail processing
Create a when rule with pyStatusMessage = Fail and specify in the fault tab
You can leave the other tabs empty, unless you want some custom messages to be sent. By default, we will get the QueueItemID in the FaultString for SOAP service.
Important note: For other service types, you will have the option in response tab and you can use the parameter – pxQueueItemID to map it to response properties.
I am going to keep it empty and save the rule.
There you see an error message – You need to specify a Request Processor name in the service tab.
Step 3: Create a new service request processor data instance.
Records -> Integration resources -> Service Request Processor -> Create New
Service package name – You should specify the same service package name under which you created the service SOAP rule.
What is a service request data instance?
We saw before that queuing results means persisting the request details into queue table and a standard agent processes the queue entry.
This is the entire architecture behind asynchronous processing.
Service request data instance will provide the default configuration settings for queuing and de-queing.
Queuing options –
Here you can specify, one or more queue class name based on conditions.
The default queue class is – System-Queue-ExecutionRequest-Service-Default.
You can also extend the class System-Queue-ExecutionRequest-Service- to create custom queue classes.
We will see some interesting information about this at the end of this post. For now, we will use the default class.
Dequeuing options –
– You can specify the maximum number of execution attempts before the queue item goes to broken-process state.
– You can keep in queue(in persisted state) after max attempts tried. This is very handy to re-queue and debugging purpose.
– You can also keep in queue after successful execution. This is used when the Java application requests the status of the Queue ID. It needs to be there in queue (table). If unchecked, pega removes the queue item from DB table.
As shown below is my configuration for UpdateLoanQueue instance.
Specify the instance in the service rule form and save the rule.
Now we specified the queuing configurations, but who is going to process the queue instances??
You can find the responsible standard agents under Pega integration services ruleset.
By the way, similar to service asynchronous processing, we also have the connect asynchronous processing, which is out of topic for this tutorial! But the same concept applies there.
The ProcessServiceQueue agent takes care of processing the default service queue instances.
From Admin studio, make sure the above agent is up and running.
It is time to test the asynchronous queuing part
Step 1: Create a case with Loan number = L11111 and status = Active.
Step 2: Keep the case in edit mode (means you have locked the case)
Step 3: Run the service SOAP by initializing the context.
Step 4: You see the Queue Item ID in response.
SYSTEM-QUEUE-EXECUTIONREQUEST-SERVICE-DEFAULT 160715944468400050D86EDF8122FA7FF0A93BDC7F02C1F8
This is the Queue ID that is sent back to the Java application.
Step 5: From the app explorer, verify the instances for – System-Queue-ExecutionRequest-Service-Default
It seems the queue entry has resulted in broken process error!!!
Click and open the instance to see the error details.
Yeah, because the security tab of the activity is checked for authentication and so the OOTB agent fail at the time of processing.
Usually this can skipped by either updating the agent schedule to bypass authentication or to uncheck the configuration in activity rule form.
Here the best option is to uncheck the authentication in activity rule form.
Save the activity rule form.
Step 6: Switch to admin studio and requeue the broken queue item.
Step 7: verify the queue status. It is success now 🙂
Also the case is updated right to status Inactive.
The next step, we need to do additional configuration for the Java application to get the Queue item status.
How to create service to send the Queue item status?
Step 1: Create a new Service SOAP rule
Step 1.1: Start the service SOAP wizard
Designer studio -> Integration -> Services -> Service wizard Use Invoke existing activity rules option.
Step 1.2: Select activity class – @baseclass and activity name – GetExecutionRequest.
Step 1.3: Use the default Item ID as request parameter.
Step 1.4: use the same service package as of the previous service package.
Your service SOAP should be successfully created.
Step 2: Verify the configuration.
Open the newly created service SOAP and verify the request response tabs
Request
Response
Here we need to send the Queue item status as response.
Map the response parameter ExecutionStatus to Param.pxExecutionStatus.
Save the service SOAP rule.
Step 3: Run the service SOAP rule manually by providing the right Queue Item ID.
Yesss…. Here you see the Execution status as Success.
We have successfully achieved our use case of asynchronous processing of service requests.
One final discussion before we close this topic – can we create custom queue class for asynchronous processing?
As per documentation, yes we can have custom queue classes. But… I don’t see the standard agent processing works well.
At one place it was mentioned OOTB agent will pick the queue entries but in other place it was mentioned to create new agents to process custom queue entries!!
(I verified both options, but the custom queue entries never got picked up!)
Also an article where advanced agents are used to process the custom queue items!!
Things to remember
– Most of the service and connect rules support asynchronous processing mode.
– You can queue the service requests every time asynchronously or only during error scenario and send the Queue item ID back to calling application
– For error scenario, you can use the Queue when condition either in faults tab (SOAP service) or in response tab (other service rule form). This will automatically queue the instance for asynchronous processing.
– Service Request processor data instance is used to specify the queueing and dequeuing options.
– System-Queue-ExecutionRequest-Service-Default is the default class that can hold the queue items and is mapped to pr_sys_queue_execreq_svc table with Queue ID as primary key.
– Pega-IntSvcs Agent hold the standard agent rules responsible for connect and service asynchronous processing.
– You can create an additional service rules to use the GetExecutionRequest activity to send the status of queue item back to calling application.
Hope you guys understood some basics about asynchronous processing in Pega. There are very good supporting articles in Pega. Please have a look at that as well 🙂
