3

We have a Platform events framework that takes the responsibility of processing a batch of n incoming platform events. This framework was working fine until around one/two months ago when we realized that we started to lose platform events. After some rounds of debugging and testing, we found that the behavior of the setResumeCheckpoint method has changed without any news in the documentation.

Let's put an easy example to understand it:

Each step represents a different transaction.

retries = 1
recordsToProcess = [account1, account2, account3];
  1. Starting from the first event, the account1 is processed fine; we inserted a new Account in DB and registered it as successful, no problems here. Going for the next event which is the account2 it fails for whatever reason and we save it in the pointer. Note that we do not throw any EventBus.RetryableException, we just let the transaction end. The result after the first transaction: accountsInserted = [account1] pointer = account2
  2. Start from the account2 fails, so it performs one retry by throwing the EventBus.RetryableException and after throwing it, SF rollbacks all the DML performed in the same transaction, the pointer is not manipulated so it should be the same. The result after the second transaction: accountsInserted = [] pointer = account2
  3. From the account2 again it fails, we reached the maximum retries so we are not going to throw an EventBus.RetryableException, we put the event pointer at the account3 with setResumeCheckpoint(account2). The result after the third transaction: accountsInserted = [account2] pointer = account3

Here is where we should expect a fourth transaction for the account3 but the setResumeCheckpoint is not making the trigger process the account3. In conclusion the account3 was never processed, we just lost that event.

Final result:

accountsInserted = [account1, account2];

Is this something expected? Where I can see this change documented? I have only found this document that does not explicitly say that setResumeCheckpoint and RetryableException are exclusive.

If you want to test this behavior, use this code block creating first the dummy__e platform event with a Number__c field of type number:

trigger dummy on dummy__e (after insert) {
    Boolean successEventProcessed = false;
    Integer retries = 1;
    
    for (dummy__e dummy : Trigger.new) {
        if (dummy.Number__c == 2) {
            
            if (successEventProcessed) {
                break;
            }
            
            if (EventBus.TriggerContext.currentContext().retries < retries ) {
                throw new EventBus.RetryableException('Retrying error');
            } else {
                EventBus.TriggerContext.currentContext().setResumeCheckpoint(dummy.replayId);
                break;
            }
        }
        
        EventBus.TriggerContext.currentContext().setResumeCheckpoint(dummy.replayId);
        successEventProcessed = true;
    }
}
2

0

You must log in to answer this question.

Browse other questions tagged .