Subscribing to a Channel with an Invalid ReplayId?

Question

I am currently in process of building a CDC application in my environment that captures changes and then places them on an Azure service bus for later processing.

The way I have the application configured currently is that every time it connects to Salesforce via CDC, it reads the last recorded ReplayId from a database. This way it can always pickup where it left off in case of failure. As new events are captured, it records the ReplayId as being processed, so that it can be used next time the app connects.

However, the problem I am running into is, what happens if the app is down for a few days, and the ReplayId is no longer valid (the event is not available)? This is an unlikely scenario, but something I am trying to resolve incase it does occur.

From all of my testing, when I subscribe to my channel with an "invalid" ReplayId, I dont get any messages or warnings. Instead, it appears that I am successfully connected, but changes stop being captured. If I manually update the ReplayId to something I know is valid, the application starts receiving messages as expected.

What can I do to get around this? I tried subscribing to various metadata channels, but I didnt see any messages there warning me that my event is no longer available... My only thought is that I need to make an initial subscription to the channel with the ReplayId of -2 (get everything), and use this to get the oldest ReplayId available to compare with my stored value. If my stored value is lower than the Id I get back, I will know that my stored Id is out of date.

Are there any better solutions to this problem? If anyone has any ideas, I would greatly appreciate it. Thanks!

Answer 1

Best practice is to store replay ID value in your target system. You can then use it on resubscription to retrieve events from SF that are missing from your target. This works for short failure windows.

If your failure window is so long that the retention period has elapsed and the event aged out of the messaging destination (channel), then replay ID is useless. You can ask for all events with -2 or only new events with -1...but either way, you might have some data "loss". Your event and possibly other events are no longer in the channel. Frequent monitoring and high reliability on subscriber end is where you want to invest effort.

Higher-level licenses allow for retention periods longer than 24 hours.

Depending on the volume of data, frequency of events and other process parameters, you could get the CDC piece out of the box with Heroku Connect. It does imply a Postgres DB on Heroku + licensing cost of HC and some operational overhead but it'll be done for you. Going from Postgres/Heroku to Azure is an easy (easier!) problem to solve.

Answer 2

I was finally able to solve this issue!

Despite all of the documentation stating that a message should be broadcast to the user if an invalid ReplayId is used during a connection, I could never get a message to come through when Subscribing to the metadata channels. Alas, I had to use AddListener instead of Subscribe.

This is the code that allowed my to receive the invalid ReplayId message:

_client.GetChannel(ChannelFields.META_SUBSCRIBE).AddListener(new MetaListener(_logger, _tableStorageService));

Once I added this, all I needed to do was code up a Listener to watch for the Invalid ReplayId message:

class MetaListener : IMessageListener
{
        private readonly ILogger<SalesforceStreamingService> _logger;
        private TableStorageService _tableStorageService;
        public MetaListener(ILogger<SalesforceStreamingService> logger, TableStorageService tableStorageService)
        {
            _logger = logger;
            _tableStorageService = tableStorageService;
        }

        public void OnMessage(IClientSessionChannel channel, IMessage message)
        {
            _logger.LogInformation($"Recieving Metadata information: {message}");
            var msg = message.ToString();
            if(msg.Contains("400::The replayId") && msg.Contains("you provided was invalid."))
            {
                _logger.LogError($"ReplayId has been detected as invalid. Setting ReplayId to -2 for next connection.");
                _tableStorageService.ReplayIdValid = false;
            }
        }
}

With the listener working, all I had to add was a check during my client connection to see if the ReplayIdValid == false. If so, I just use -2 instead of my stored ReplayId.

This solved my issue.

If anyone else is in a similar situation, you can find the full list of CDC error codes here: https://developer.salesforce.com/docs/atlas.en-us.api_streaming.meta/api_streaming/streaming_error_codes.htm

Hope this helps!