Apply Best Practices for Writing Platform Event Triggers

trigger ResumeEventProcessingTrigger on Order_Event__e (after insert) {
    for (Order_Event__e event : Trigger.New) {
        // Process the event message.
        // ...
        

        // Set the Replay ID of the last successfully processed event message.
        // If a limit is hit, the trigger refires and processing starts with the
        // event after the last one processed (the set Replay ID).
       

        EventBus.TriggerContext.currentContext().setResumeCheckpoint(event.replayId);
    }
}

trigger ResendEventsTrigger on Order_Event__e (after insert) {
    try {
       for (Order_Event__e event : Trigger.New) {
         // Process platform event.
         // ...
       }
    } catch(Exception e) {
         System.debug('Caught exception: ' + e.getMessage() +
           ' Stack trace:' + e.getStackTrace());
         System.debug('Retrying the platform event trigger');
        // Retry the trigger up to 4 times and resend batch of events
        if (EventBus.TriggerContext.currentContext().retries < 4) {
            throw new EventBus.RetryableException(
                     'Retrying the platform event trigger.');
  }
}

trigger EventTriggerTemplate on Order_Event__e (after insert) {
    Integer processedEventsCounter = 0;
    Boolean shouldContinueToProcess = true;
    

    for (Order_Event__e event : Trigger.New) {
        try {
            // Process event message.
            // ....
            

            // Set Replay ID after which to resume event processing
            // in new trigger execution.
            EventBus.TriggerContext.currentContext().setResumeCheckpoint(
                event.ReplayId);
            System.debug('Processed event with Replay ID: ' + event.ReplayId);
            processedEventsCounter++;
            

            if (shouldContinueToProcess != true) {
                // Resume after the last successfully processed event message
                // after the trigger stops running.
                // Exit for loop.
                break;
            }
        } catch(Exception e) {
            // This catch block works only for non-limit exceptions
            // because limit exceptions cannot be caught.
            //
            // If no events have been processed, throw new EventBus.RetryableException
            if (processedEventsCounter == 0) {
                 // Only throw RetryableException when the first event
                // is processed but before setResumeCheckpoint is called.
                if (EventBus.TriggerContext.currentContext().retries < 4) {
                	throw new EventBus.RetryableException();
                }
            } else {
                 // Else log the exception in the debug log. The trigger will
                 // resume from the checkpoint next time.
                 System.debug('An exception occurred: ' + e.getMessage());
                // Rethrow exception - trigger exits and resumes with this event
                // as the first event in the batch of the new invocation.
                throw e;
            }
        }
    }
}

public class PlatformEventsConstants {
	public static final Integer MAX_RETRIES = 3;
}

public class PlatformEventsTestUtil {
    private static Boolean shouldForceException = false;
    private static Boolean isCatchableException = false;
    private static Boolean isContinuous = false;
    private static Integer runCounter = 0;
    private static Integer exceptionThrownAtRun = 0;
    

    public class MyCustomException extends Exception {}
    

    // Throws a limit exception at a specific trigger run.
    // The test method calls this utility method.
    public static void throwLimitExceptionAtRun(Integer exThrownAtRun) {
        enableExceptionForTesting(false, exThrownAtRun, false);
    }
    

    // Throws a catchable exception at a specific trigger run.
    // The test method calls this utility method.
    public static void throwCatchableExceptionAtRun(Integer exThrownAtRun) {
        enableExceptionForTesting(true, exThrownAtRun, false);
    }
    

    // Throws a catchable exception starting at a specific trigger run
    // and keeps throwing it again.
    // The test method calls this utility method.
    public static void throwCatchableExceptionContinuousAtRun(Integer exThrownAtRun) {
        enableExceptionForTesting(true, exThrownAtRun, true);
    }
    

    // Sets up the test parameters.
    private static void enableExceptionForTesting(Boolean isCatchableEx,
                                                 Integer exThrownAtRun,
                                                 Boolean isCont) {
        if (Test.isRunningTest()) {
            shouldForceException = true;
            isCatchableException = isCatchableEx;
            isContinuous = isCont;
            runCounter = 0;
            exceptionThrownAtRun = exThrownAtRun;
        }
    }
    

    // Enables testing of exceptions in an Apex trigger.
    // Runs only in test context.
    // The Apex trigger calls this method.
    public static void forceExceptionForTesting() {
        if (Test.isRunningTest() && shouldForceException) {
            runCounter++;
            System.debug('runCounter=' + runCounter);
            

            Boolean shouldThrowException = (runCounter == exceptionThrownAtRun) ||
                (runCounter >= exceptionThrownAtRun && isContinuous);
            

            if (shouldThrowException) {
                if (isCatchableException) {
                    System.debug('Throwing a catchable exception.');
                    throw new MyCustomException();
                } else {
                    System.debug('Throwing a limit exception.');
                    throw new System.LimitException();
                }
            }
        }
    }
    

    // Cleans up the test parameters. The test method calls this method.
    public static void cleanUp() {
        shouldForceException = false;
        isCatchableException = false;
        isContinuous = false;
        runCounter = 0;
        exceptionThrownAtRun = 0;
    }
    

    public static Integer getRunCounter() {
        return runCounter;
    }
}

trigger OrderEventTrigger on Order_Event__e (after insert) {
    private Integer processedEventsCounter = 0;
    

    for (Order_Event__e event : Trigger.New) {
        try {
            // Enable testing exceptions. Runs only in test context.
            PlatformEventsTestUtil.forceExceptionForTesting();
            

            // Process event message.
            // ...
            

            // Set Replay ID after which to resume event processing
            // in a new trigger execution.
            EventBus.TriggerContext.currentContext().setResumeCheckpoint(
                event.ReplayId);
            System.debug('Processed event with Replay ID: ' + event.ReplayId);
            processedEventsCounter++;
           

        } catch(Exception e) {
            // This catch block works only for non-limit exceptions
            // because limit exceptions cannot be caught.
            //
            // If no events have been processed, throw new EventBus.RetryableException
            if (processedEventsCounter == 0) {
                // Only throw RetryableException when the first event
                // is processed but before setResumeCheckpoint is called.
                if (EventBus.TriggerContext.currentContext().retries <
                        PlatformEventsConstants.MAX_RETRIES) {
                    throw new EventBus.RetryableException();
                }
            }
            // This block is reached when catchable exception is received and
            // one of these statements is true:
            // - At least one event was processed.
            // - No event was processed but the number of retries
            //   for RetryableException is exhausted.
            //
            // Log the exception in the debug log. The trigger will
            // resume from the checkpoint if a checkpoint was set.
            System.debug('An exception occurred in OrderEventTrigger: ' + e.getTypeName());
            

            // Rethrow exception: Trigger exits and resumes with this event
            // as the first event in the batch of the new invocation.
            throw e;
        }
    }
}

@isTest
public class OrderEventTriggerTest {
    // This test causes the trigger to hit a non-catchable limit exception
    // on the third event.
    // Expected: The trigger resumes where it left off and processes the next events.
    @isTest
    static void testResumeAfterLimitException() {
        PlatformEventsTestUtil.throwLimitExceptionAtRun(3);
        try {
            // Publish 5 events.
            // First event batch should have 2 successfully processed events
            // because the exception occurs on the 3rd event.
            // Verify that trigger resumes and processes remaining events.
            publishAndDeliverEvents(5, 2);
        } finally {
            // Clean up test parameters so they don't affect other tests
            PlatformEventsTestUtil.cleanUp();
        }
    }
    

    // This test causes the trigger to hit a catchable exception on the second event.
    // Expected: The trigger resumes where it left off and processes the next events.
    @isTest
    static void testResumeAfterCatchableExceptionOnSecondEvent() {
        PlatformEventsTestUtil.throwCatchableExceptionAtRun(2);
        try {
            // Publish 5 events.
            // First event batch should have 1 successfully processed event
            // and setResumeCheckpoint was called.
            // The exception occurs on the 2nd event.
            // Verify that trigger resumes and processes remaining events.
            publishAndDeliverEvents(5, 1);
        } finally {
            // Clean up test parameters so they don't affect other tests
            PlatformEventsTestUtil.cleanUp();
        }
    }
    

	// This test causes the trigger to hit a catchable exception on the first event.
	// Expected: The trigger is retried with the entire batch of events.
    @isTest
    static void testRetryAfterCatchableExceptionOnFirstEvent() {
        PlatformEventsTestUtil.throwCatchableExceptionAtRun(1);
        try {
            // Publish 5 events.
            // First batch of events delivered is empty because
            // the execution jumps to the catch block.
            // Because setResumeCheckpoint has not been called yet,
            // the trigger will be retried with EventBus.RetryableException
            // with the entire batch of events.
            publishAndDeliverEvents(5, 0);
        } finally {
            // Clean up test parameters so they don't affect other tests
            PlatformEventsTestUtil.cleanUp();
        }
    }
    

    // This test causes the trigger to hit a catchable exception on the first event and
    // exhausts the number of retries.
    // Expected: The trigger is retried with the entire batch of events MAX_RETRIES + 1 times.
    // After the retries are exhausted, the trigger rethrows the exception.
    @isTest
    static void testRetryAfterCatchableExceptionOnFirstEventMaxRetriesReached() {
        PlatformEventsTestUtil.throwCatchableExceptionContinuousAtRun(1);
        try {
            publishEvents(2);
            

            // The trigger should run (max retry + 1) times
            for (Integer i = 1; i <= PlatformEventsConstants.MAX_RETRIES + 1; i++) {
                Test.getEventBus().deliver();
                Assert.areEqual(i, PlatformEventsTestUtil.getRunCounter(),
                                'Trigger should have retried.');
            }
            

            // Calling deliver more times won't cause the trigger to run,
            // since we're not throwing the Retryable exception anymore.
            Test.getEventBus().deliver();
            Assert.areEqual(PlatformEventsConstants.MAX_RETRIES + 1,
                            PlatformEventsTestUtil.getRunCounter(),
                            'Trigger should NOT have retried.');
        } finally {
            // Clean up test parameters so they don't affect other tests
            PlatformEventsTestUtil.cleanUp();
        }
    }
    

    // Helper method to publish events.
    private static void publishEvents(Integer totalEvents) {
        List<Order_Event__e> eventList = new List<Order_Event__e>();
        

        // Create test events. Number of test events to create is totalEvents.
        for(Integer i = 0; i < totalEvents; i++) {
            Order_Event__e event = (Order_Event__e)Order_Event__e.sObjectType.newSObject(
                null, true);
            event.Order_Id__c='dummyOrderId' + i;
            eventList.add(event);
        }
        

        // Publish all events
        EventBus.publish(eventList);
    }
    

    // Helper method to publish events and call deliver twice to verify
    // that the trigger resumes after an exception.
    private static void publishAndDeliverEvents(Integer totalEvents,
                                                Integer expectedFirstDeliveryCount) {
        publishEvents(totalEvents);
        

        // Deliver the batch of events
        Test.getEventBus().deliver();
        

        // Verify that the trigger processed expectedFirstDeliveryNum of events
        // because it processed only that many events before the exception
        // was hit.
        Assert.areEqual(expectedFirstDeliveryCount,
                        getTriggerPosition(), 'Unexpected number of events processed.');
                

        // Call deliver() again to deliver the remaining events in the batch
        // and cause the trigger resume after the exception.
        Test.getEventBus().deliver();
               

        // Verify that the trigger resumes and remaining events are processed
        // for the remaining events, so there are a total of totalEvents processed events.
        Assert.areEqual(totalEvents,
                        getTriggerPosition(),
                        'Unexpected number of events processed.');
                                                    

        getTriggerPosition();
    }
    

    // Helper method to return the ReplayId of the last processed event.
    // In test context, replayIDs start from 1 and are incremented.
    private static Integer getTriggerPosition() {
        Integer position = 0;
        

        EventBusSubscriber[] subscribers =
            [SELECT Name, Type, Position, Retries, LastError
             FROM EventBusSubscriber WHERE Topic='Order_Event__e'];
        

        for (EventBusSubscriber sub : subscribers) {
            if (sub.Name == 'OrderEventTrigger') {
                System.debug('sub.Position='+sub.Position);
                position = sub.Position;
            }
        }
        

        return position;
    }
}