Get Asynchronous Publish Results with Apex Publish Callbacks

Learning Objectives

After completing this unit, you’ll be able to:

• Write an Apex publish callback class that handles failed events.
  
• Publish an event with a callback instance.
  
• Write a test class for the Apex publish callback class.
  

Use a Publish Callback Class and Inspect the Final Result

To catch the asynchronous publishing errors, Vijay from Cloud Kicks decides to use Apex publish callbacks. He implements a callback to capture the events that failed to publish and retry publishing them.

To write an Apex publish callback class, Vijay implements the EventBus.EventPublishFailureCallback interface. The implemented onFailure method contains logic to get the events that failed to publish and republishes them. The custom code limits the event republishing to two attempts. A map holds the UUID values of each published event and maps it to the order record ID. This mapping is used to populate the event Order_Id__c field.

Note

If no limit is specified in the onFailure method for republishing events in the custom code, the system limits republishing with a callback instance up to 10 times. See Apex Publish Callback Limits in the Platform Events Developer Guide.

Add the Class in the Developer Console

1. In the Developer Console, click New | Apex Class.
   
2. For the class name, enter FailureCallbackWithCorrelation.
   
3. Replace the default code with this class body.
   

public class FailureCallbackWithCorrelation implements EventBus.EventPublishFailureCallback {
    public static final Integer MAX_RETRIES = 2;
    private Integer retryCounter = 0;
    private Map<String,String> uuidMap;
    

    // Callback constructor
    public FailureCallbackWithCorrelation(Map<String,String> uuidMap) {
        this.uuidMap = uuidMap;
    }
    

    public void onFailure(EventBus.FailureResult result) {
        List<String> eventUuids = result.getEventUuids();
        Map<String,String> newUuidMap = new Map<String,String>();
        

        if (retryCounter < MAX_RETRIES) {
            // Try to re-publish the failed events
            List<Order_Event__e> events = new List<Order_Event__e>();
            for (String uuid : eventUuids) {
                // Create a new event with the contents of the failed event
                Order_Event__e event = (Order_Event__e)
                    Order_Event__e.sObjectType.newSObject(null, true);
                // Fill event with the right order record Id
                event.Order_Id__c = uuidMap.get(uuid);
                events.add(event);
                

                // Use a new map since the new event will have a different uuid
                newUuidMap.put(event.EventUuid, event.Order_Id__c);
            }
            // Replace old uuid map because we no longer need its contents
            uuidMap = newUuidMap;
            

            // Republish with the same callback passed in again as 'this'
            System.debug('Republish ' + eventUuids.size() + ' failed event(s).');
            EventBus.publish(events, this);
            System.debug('Republish event for Order with Ids: ' +
                         String.join(uuidMap.values(), ', '));
            

            // Increase counter
            retryCounter++;
        } else {
            // Retry exhausted, log an error instead
            System.debug(eventUuids.size() + ' event(s) failed to publish after ' +
                         MAX_RETRIES + ' retries ' +
                         'for Order with Ids: ' + String.join(uuidMap.values(), ', '));
        }
    }
    

    // Getter methods so we can validate this in the unit test
    public Integer getRetryCounter() {
        return retryCounter;
    }
    public Map<String,String> getUuidMap() {
        return uuidMap;
    }
}

To execute the publish callback, publish some order events. But before that, let's make sure the debug logs are collected for the Automated Process user.

Note

The FailureCallbackWithCorrelation class republishes events in the onFailure method. For simplicity, only the event Order_Id__c field is populated. The order ID is taken from the map that maps the EventUuid field with the order ID. Typically, your event contains more fields. To populate the remaining fields, we recommend you write an efficient SOQL query on the associated Order records using the IDs to get the additional fields. For more information about efficient SOQL queries, see Running Apex within Governor Execution Limits in the Platform Events Developer Guide. We don't recommend storing those fields in the map because it increases the size of the callback instance and may slow down performance.

Add a Trace Flag for Automated Process for Debug Logs

A publish callback runs under the Automated Process user. To collect debug logs for the callback’s execution, add a trace flag for Automated Process.

Add a trace flag Entry for the Automated Process User:

1. From Setup, in the Quick Find box, enter Debug Logs, then click Debug Logs.
   
2. Click New.
   
3. For Traced Entity Type, select Automated Process.
   
4. Select the time period to collect logs. The start and expiration dates default to the current date and time. To extend the expiration date, click the end date input box, and select the next day from the calendar.
   
5. For Debug Level, click New Debug Level. Enter a name, such as CustomDebugLevel, and accept the defaults.
   
6. Save your work.
   

When the callback is invoked, it’s logged in the debug log. Logging for the callback requires the System debug log level to be set to at least Info. The default System debug log level works because it is Debug, which is more granular than Info. When the callback is invoked, the debug log line looks as follows.

CODE_UNIT_STARTED [EXTERNAL]|platform.event.publish.callbacks.tasks.apex.ApexCallbackMethodInvoker

Note

All records that are created in a callback have their system user fields, such as CreatedById and OwnerId, set to Automated Process. You can explicitly set the OwnerId to another value. For example, to assign a task to a specific user, set the task OwnerId to that user’s ID.

Create an Event with an EventUuid Field Value

Before publishing an event, let's create an event with a pre-populated EventUuid field value. The EventUuid field uniquely identifies an event message. You can use the EventUuid field used to match the events returned in the publish callback with the events that were published in the EventBus.publish call.

To have the system generate an EventUuid field value in each event object, use the SObjectType.newSObject(recordTypeId, loadDefaults)Apex method to create the event object, as follows.

Order_Event__e event = (Order_Event__e)Order_Event__e.sObjectType.newSObject(null, true);
// The EventUuid value is returned after object creation
System.debug('EventUuid: ' + event.EventUuid);
// Debug output
// EventUuid: 19bd382e-8964-43de-ac01-d5d82dd0bf78

Publish Events with a Callback Instance

After creating events using the newSObject method, you can publish the events with a callback instance.

This code snippet creates two events and publishes them. The EventBus.publish call references an instance of the callback class, FailureCallbackWithCorrelation, which was defined earlier. It then checks for any immediate errors returned by the publish call. If there are any asynchronous errors, they cause the onFailure method in the FailureCallbackWithCorrelation class to run.

List<Order_Event__e> eventList = new List<Order_Event__e>();
Map<String,String> uuidMap = new Map<String,String>();
// Create event objects with prepopulated EventUuid fields.
Order_Event__e event1 = (Order_Event__e)Order_Event__e.sObjectType.newSObject(null, true);
event1.Order_Id__c='Order1 ID';
System.debug('event1 EventUuid: ' + event1.EventUuid);
// Map event UUID -> Order Id so we can look up later
uuidMap.put(event1.EventUuid, event1.Order_Id__c);
Order_Event__e event2 = (Order_Event__e)Order_Event__e.sObjectType.newSObject(null, true);
event2.Order_Id__c='Order2 ID';
System.debug('event2 EventUuid: ' + event2.EventUuid);
// Map event UUID -> Order Id so we can look up later
uuidMap.put(event2.EventUuid, event2.Order_Id__c);
// Add event objects to the list.
eventList.add(event1);
eventList.add(event2);
// Publish events with an instance of the failure callback.
FailureCallbackWithCorrelation cb = new FailureCallbackWithCorrelation(uuidMap);
List<Database.SaveResult> results = EventBus.publish(eventList, cb);
// Inspect synchronous publishing result for each event.
for (Database.SaveResult sr : results) {
    if (sr.isSuccess()) {
        System.debug('Successfully published event.');
    } else {
        for(Database.Error err : sr.getErrors()) {
            System.debug('Error returned: ' +
                        err.getStatusCode() +
                        ' - ' +
                        err.getMessage());
        }
    }
}

When you run the code snippet above in the Developer Console, you see the log output of the code snippet. The log output of the trigger is in Setup, in the Debug Logs page for the Automated Process user. However, in this case, because the publishing most likely succeeds, the onFailure method isn't called and you won't see a log for the callback execution.

Because you can't simulate a system error, you can't cause the onFailure method to run. However, you can use a test class to simulate an error and test the onFailure. The test class is shown in the next section.

Test the Publish Callback Class

Vijay writes a test that simulates the failed publishing of a test event, which triggers the execution of the onFailure callback method.

This method simulates a failed publishing of a test event or a batch of test events in an Apex test.

Test.getEventBus().fail();

In an Apex test, event messages are published synchronously in the test event bus. The Test.getEventBus().fail() method causes the publishing of events to fail immediately after the call, and event messages are removed from the test event bus. This method causes the onFailure method in the callback class to be invoked. When the event messages fail to publish, none of the Apex triggers defined on the platform event receive any failed events.

To simulate successful event delivery, call the Test.getEventBus().deliver(); method or have your events delivered after Test.stopTest(). Event messages are delivered immediately after each of those statements. Successfully delivered events cause the execution of the onSuccess method in the callback class. The onSuccess method isn't covered in this Trailhead module, but you can learn more about it in Deliver Test Event Messages in the Platform Events Developer Guide.

Test Class Example

Add this class in your Trailhead Playground in the same way you added the callback class earlier.

This example class, FailureCallbackWithCorrelationTest, is a test class for the FailureCallbackWithCorrelation class. This test class shows how to test the failed publishing of test event messages in the test event bus. The test first creates an event and publishes it with the callback. Next, the test fails the delivery of the event twice in a loop. The test verifies that the callback retries publishing the event twice by checking that the retryCounter variable has been increased.

@isTest
public class FailureCallbackWithCorrelationTest {
   

    @isTest
    static void testFailedEventsWithFail() {
        // Create test event
        Order_Event__e event = (Order_Event__e)Order_Event__e.sObjectType.newSObject(
            null, true);
        event.Order_Id__c='dummyOrderId';
        

        // Populate map
        Map<String,String> uuidMap = new Map<String,String>();
        uuidMap.put(event.EventUuid, 'dummyOrderId');
        

        // Create callback
        FailureCallbackWithCorrelation cb = new FailureCallbackWithCorrelation(uuidMap);
        

        // Make sure retry counter is 0
        Assert.areEqual(0, cb.getRetryCounter(),
            'Newly created callback should have retry counter at 0');
        

        // Publish an event with callback
        EventBus.publish(event, cb);
		

        // If we fail all publish attempts, callback should run MAX_RETRIES times.
        // For each attempt, the callback should republish the event,
        //    increase the counter, and update the map
        String prevUuid = event.EventUuid;
        for (Integer i = 1; i <= FailureCallbackWithCorrelation.MAX_RETRIES; i++) {
            Test.getEventBus().fail();
            Assert.areEqual(i, cb.getRetryCounter(), 'Retry counter should be ' + i);
            Assert.areEqual(1, cb.getUuidMap().size(), 'Map size should be 1');
            String currUuid = (new List<String>(cb.getUuidMap().keySet())).get(0);
            Assert.areNotEqual(prevUuid, currUuid,
                'Map should be updated with newly created event Uuid');
            Assert.areEqual('dummyOrderId', cb.getUuidMap().get(currUuid),
                'Map value should be the original Order Id');
            prevUuid = currUuid;
        }
        

        // If we publish another failed event, callback should not retry.
        Order_Event__e event2 = (Order_Event__e)Order_Event__e.sObjectType.newSObject(
            null, true);
        event2.Order_Id__c='dummyOrderId';
        EventBus.publish(event, cb);
        Test.getEventBus().fail();
        Assert.areEqual(FailureCallbackWithCorrelation.MAX_RETRIES, cb.getRetryCounter(),
                        'Retry counter should still be ' +
                         FailureCallbackWithCorrelation.MAX_RETRIES);
    }
}

Run the Test Class in the Developer Console

1. In the Developer Console, click Test | New Run.
   
2. In the Test Classes column, select FailureCallbackWithCorrelationTest.
   
3. Select the test, and then click Run.
   
4. Double-click the completed test run to open the results in detail view. You can verify that the test succeeded.
   
5. To verify that the callback onFailure method was invoked, open the debug log. Click the Logs tab and double-click the latest log entry. The debug log shows that the events were republished twice and failed publishing. This is an excerpt of some operations logged in the debug log.
   

15:23:26.0 (253774507)|METHOD_ENTRY|[30]||eventbus.TestBroker.fail()
15:23:26.0 (303029804)|CODE_UNIT_STARTED|[EXTERNAL]|platform.event.publish.callbacks.tasks.apex.ApexCallbackMethodInvoker
…
15:23:26.0 (322897823)|USER_DEBUG|[34]|DEBUG|Republish 1 failed event(s).
…
15:23:26.0 (326264814)|DML_BEGIN|[35]|Op:Insert|Type:Order_Event__e|Rows:1
…
15:23:26.0 (337113675)|USER_DEBUG|[36]|DEBUG|Republish event for Order with Ids: dummyOrderId
…
15:23:26.0 (347183100)|CODE_UNIT_FINISHED|platform.event.publish.callbacks.tasks.apex.ApexCallbackMethodInvoker
…
15:23:26.0 (356601358)|METHOD_ENTRY|[30]||eventbus.TestBroker.fail()
15:23:26.0 (360100652)|CODE_UNIT_STARTED|[EXTERNAL]|platform.event.publish.callbacks.tasks.apex.ApexCallbackMethodInvoker
…
15:23:26.0 (364909736)|USER_DEBUG|[34]|DEBUG|Republish 1 failed event(s).
15:23:26.0 (366375150)|DML_BEGIN|[35]|Op:Insert|Type:Order_Event__e|Rows:1
…
15:23:26.0 (379841318)|USER_DEBUG|[36]|DEBUG|Republish event for Order with Ids: dummyOrderId
15:23:26.0 (379895350)|VARIABLE_ASSIGNMENT|[40]|this.retryCounter|2|0x361f18cc
15:23:26.0 (386397600)|CODE_UNIT_FINISHED|platform.event.publish.callbacks.tasks.apex.ApexCallbackMethodInvoker
…
15:23:26.0 (402851235)|USER_DEBUG|[43]|DEBUG|1 event(s) failed to publish after 2 retries for Order with Ids: dummyOrderId
15:23:26.0 (404102455)|CODE_UNIT_FINISHED|platform.event.publish.callbacks.tasks.apex.ApexCallbackMethodInvoker

For more information and for code examples, see Test Apex Publish Callbacks in the Platform Events Developer Guide.

Publish Callback Best Practices

Keep in mind these best practices when implementing Apex publish callbacks.

• Don’t republish the same event object that is created with SObjectType.newSObject.
• Publish a list of events instead of individual events with a callback.
  
• Don’t use the SObject generic type when publishing a list of events. Instead, use the specific platform event type as the data type in the list.
  
• To reduce the callback instance size, keep the map of event UUIDs small in the callback. A small callback instance size ensures better performance and helps avoid hitting the cumulative usage limit of all publish callbacks. The FailureCallbackWithCorrelation class in this unit maps the event UUID to a record ID that you can query to populate the remaining event fields. Alternatively, if you want to save the entire event as the map value, make sure that the event doesn't have many fields and the field sizes are small.
  
• Be mindful when tracking successful publishes through the onSuccess method. Because most publish calls typically succeed, tracking successful publishes likely isn’t a concern. If a large volume of events is published, processing successful publishes can have performance and Apex limit impacts. We recommend that you process successful publishes only when absolutely necessary. The performance impact doesn't apply to processing failed publishes through the onFailure method because failed publish calls occur rarely.
  

For more information about these best practices, see Publish Callback Best Practices in the Platform Events Developer Guide.

You now know how to use Apex publish callbacks and how to test them. Publishing events is only half the story. Writing subscribers that receive events is the other half. In the next unit, you learn best practices for writing robust and resilient platform event triggers.

Resources

• Salesforce Help: Developer Console
• Salesforce Help: Create a Test Run
• Platform Events Developer Guide: Get the Result of Asynchronous Platform Event Publishing with Apex Publish Callbacks
• Platform Events Developer Guide: Test Apex Publish Callbacks
• Platform Events Developer Guide: Publish Callback Best Practices
• Platform Events Developer Guide: Apex Publish Callback Limits