Retry Long-Lived RPC Calls After an Error Occurs

When an error occurs, a long-lived RPC call, including PublishStream, Subscribe, and ManagedSubscribe (beta), stops running and the Pub/Sub API service closes the streaming connection. To create a new streaming connection, you must retry the RPC call. How you retry the long-lived RPC call depends on the error received in the gRPC StatusRuntimeException, which you can catch in your client.

Because errors are sometimes transient and can go away after a period of time, we recommend that you limit the number of retry attempts and that you retry the RPC calls with exponential backoff. Increasing the time between calls ensures that you don’t keep receiving the same error and exhaust all your retry attempts.

Tip

To inspect a Pub/Sub API error code value, get the gRPC StatusRuntimeException Trailers section of the exception by calling getTrailers() on the exception. The Pub/Sub API error code provides information about the cause of the failure. For a list of error codes, see Error Codes. For example, the error code in this example is sfdc.platform.eventbus.grpc.subscription.fetch.replayid.corrupted. It means that the specified Replay ID value is invalid. One possible cause is that it’s an old Replay ID outside the 72-hour retention period.

1=== GRPC Exception ===
2io.grpc.StatusRuntimeException: INVALID_ARGUMENT: The Replay ID validation failed. Ensure that the Replay ID is valid. rpcId: 2f6b4cee-3525-49d1-8fdb-0bd3d662062f
3=== Trailers ===
4[Trailer] = content-type [Value] = application/grpc
5[Trailer] = rpc-id [Value] = 2f6b4cee-3525-49d1-8fdb-0bd3d662062f
6[Trailer] = error-code [Value] = sfdc.platform.eventbus.grpc.subscription.fetch.replayid.corrupted
7[Trailer] = type [Value] = Subscribe

How to Retry the Subscribe RPC Method

To retry the Subscribe RPC Method after an error occurs, use a ReplayPreset option to specify from which position in the event bus to resubscribe depending on the error received.

Here are examples of how to retry the Subscribe call based on the error code.

If a temporary server error is returned, retry the Subscribe call with the CUSTOM ReplayPreset enum value. Pass in the Replay ID value that you saved from the last processed event to receive events after the one that the client last processed. The subscription resumes from where it left off, after that event with the specified Replay ID.
If the Replay ID that you passed in with the CUSTOM ReplayPreset enum value is corrupted, retry the Subscribe call with the LATEST ReplayPreset enum value to receive new events only, or use the EARLIEST option to receive all events that are stored. Because it can slow down performance when a large volume of events is stored in the event bus, use the EARLIEST option sparingly.
If you aren’t interested in the events that were missed during client shutdown, retry the Subscribe call by using the LATEST ReplayPreset enum value to get new events.

For more details about the Subscribe call options in the ReplayPreset enum value, see Replaying an Event Stream.

Example

This code snippet shows how to get the error code from the trailers portion of StatusRuntimeException. It then checks for some common error codes and follows a strategy to determine which point in the stream to resubscribe from. This example uses a scheduler to schedule the Subscribe RPC call after a delay, if necessary. The implementation of the scheduler isn’t shown in this example.

1public static String ERROR_REPLAY_ID_VALIDATION_FAILED = "fetch.replayid.validation.failed";
2public static String ERROR_REPLAY_ID_INVALID = "fetch.replayid.corrupted";
3public static String ERROR_SERVICE_UNAVAILABLE = "service.unavailable";
4
5// ...
6
7@Override
8public void onError(Throwable t) {
9    printStatusRuntimeException("Error during Subscribe", (Exception) t);
10    logger.info("Retries remaining: " + retriesLeft.get());
11    if (retriesLeft.get() == 0) {
12        logger.info("Exhausted all retries. Closing Subscription.");
13        isActive.set(false);
14    } else {
15        retriesLeft.decrementAndGet();
16        Metadata trailers = ((StatusRuntimeException)t).getTrailers() != null ? ((StatusRuntimeException)t).getTrailers() : null;
17        String errorCode = (trailers != null && trailers.get(Metadata.Key.of("error-code", Metadata.ASCII_STRING_MARSHALLER)) != null) ?
18                trailers.get(Metadata.Key.of("error-code", Metadata.ASCII_STRING_MARSHALLER)) : null;
19
20        // Closing the old stream for sanity
21        serverStream.onCompleted();
22
23        ReplayPreset retryReplayPreset = ReplayPreset.LATEST;
24        ByteString retryReplayId = null;
25        long retryDelay = getBackoffWaitTime();
26
27        // Retry strategies that can be implemented based on the error type.
28        if(errorCode != null && !errorCode.isEmpty()) {
29            if (errorCode.contains(ERROR_SERVICE_UNAVAILABLE)) {
30                logger.info("Service currently unavailable. Trying again with stored Replay.");
31                retryDelay = SERVICE_UNAVAILABLE_WAIT_BEFORE_RETRY_SECONDS * 1000;
32                retryReplayPreset = ReplayPreset.CUSTOM;
33                retryReplayId = getStoredReplay();
34            } else if (errorCode.contains(ERROR_REPLAY_ID_VALIDATION_FAILED) || errorCode.contains(ERROR_REPLAY_ID_INVALID)) {
35                logger.info("Invalid or no replayId provided in FetchRequest for CUSTOM Replay. Trying again with EARLIEST Replay.");
36                retryReplayPreset = ReplayPreset.EARLIEST;
37            } else {
38                logger.info("Retrying with LATEST Replay.");
39            }
40        } else {
41            logger.info("Unknown error. Retrying with LATEST Replay.");
42        }
43        logger.info(String.format("Retrying in %s ms.", retryDelay));
44        retryScheduler.schedule(new RetryRequestSender(retryReplayPreset, retryReplayId), retryDelay, TimeUnit.MILLISECONDS);
45    }
46}

How to Retry the ManagedSubscribe RPC Method (Beta)

The process of retrying the ManagedSubscribe RPC Method (Beta) is simpler than retrying the Subscribe RPC method. If a client retries the ManagedSubscribe call after sending commit requests to store the Replay ID of a processed event on the server, the subscription resumes after the event with the committed Replay ID. The client doesn’t manage the Replay ID’s storage.

Also, the ManagedSubscribe RPC method uses event replay options that you set in the ManagedEventSubscription metadata by using Metadata API or Tooling API. These replay options are used if no Replay ID was committed or the committed Replay ID is invalid.

If no Replay ID was committed on the server and you retry the ManagedSubscribe call, the subscription restarts from the value specified in the defaultReplay field of ManagedEventSubscription. If defaultReplay is set to LATEST, the subscription receives new events only. If defaultReplay is set to EARLIEST, the subscription starts from the earliest events stored in the event bus.
If the committed Replay ID is invalid and you retry the ManagedSubscribe call, the subscription restarts from the value specified in the errorRecoveryReplay field of ManagedEventSubscription. If errorRecoveryReplay is set to LATEST, the subscription receives new events only. If errorRecoveryReplay is set to EARLIEST, the subscription starts from the earliest events stored in the event bus.

For more information, see ManagedEventSubscription (Beta) in the Metadata API Developer Guide.

How to Retry Publishing Failed Events

The PublishStream RPC Method uses bidirectional streaming. It can send a stream of publish requests while receiving a stream of publish responses from the server. The server returns a PublishResponse for each PublishRequest when publishing is complete for a batch of events. A PublishResponse contains a list of PublishResult objects, each containing the publishing result for one event.

To preserve the order of event batches, wait for the PublishResponse to be received for a published batch of events before sending a new PublishRequest for a new batch. Be aware that this method can slow down the speed of your publishing process.

Publish a batch of events in PublishRequest, and then wait for the PublishResponse to be returned before sending a new PublishRequest.
In PublishResponse, inspect each PublishResult for each event.
If a PublishResult has an error for one or more events, republish the events.
Repeat this process to publish a new batch of events, and then inspect new publish results.

If the order of event batches isn’t important, you can inspect the publish results in each PublishResponse as it’s received. Then republish failed events whose PublishResult contains an error. However, this method can cause one batch to finish publishing before you republish a failed batch of events.

Correlating Publish Results with Published Events

To determine which PublishResult corresponds to which event published, correlate the failed publish result with the event ID (ProducerEvent.Id) by using the correlationKey field that’s returned in PublishResult, which is part of PublishResponse. The correlationKey is set to the value that you set in ProducerEvent.id in PublishRequest when you publish the event. To identify the event published, match PublishResult.correlationKey with ProducerEvent.id (the event ID). You can then republish the event. It’s advisable to limit the number of retries so that your publish calls don’t execute too many times in case the publishing keeps failing.

To view the messages and fields of the pub-sub-api service, see pubsub_api.proto in GitHub. For more information about the correlation key, see Identify an Event with ProducerEvent.id and PublishResponse.correlationKey.

Example

This code snippet shows how to retry publishing failed events. It limits the retries up to a certain number set in the constant MAX_NUM_OF_RETRIES_PER_BATCH value. The example uses a private helper method, validatePublishResponse(), to get a list of PublishResults that have errors (PublishResult.error is true), which represents the failed events. Next, it gets the correlation key from each PublishResult and matches it to a ProducerEvent ID by using a map holding correlation keys and ProducerEvents. Finally, it retries publishing the failed events.

1while (true) {
2    batchCount++;
3    logger.info("publishing " + NUM_EVENTS_PER_BATCH + " events for batch " + batchCount);
4
5    // Assume that each event will have a unique ID (i.e., correlationKey) assigned to it.
6    // Store the corresponding ProducerEvent so that we can look up specific events by
7    // correlationKey when a partial publish failure occurs.
8    Map<String, ProducerEvent> eventsByCorrelationKey = new HashMap<>(NUM_EVENTS_PER_BATCH);
9    for (int i = 0; i < NUM_EVENTS_PER_BATCH; i++) {
10        ProducerEvent producerEvent = generateProducerEvent(i);
11        eventsByCorrelationKey.put(producerEvent.getId(), producerEvent);
12    }
13
14    PublishRequest publishRequest = PublishRequest.newBuilder().setTopicName(busTopicName)
15            .addAllEvents(eventsByCorrelationKey.values()).build();
16
17    // Publish batch of events.
18    requestObserver.onNext(publishRequest);
19
20    // Get the failed publishing results by using a helper method, validatePublishResponse()
21    List<PublishResult> failedEvents = validatePublishResponse(
22        errorStatuses, batchFinishedLatchRef.get(),
23        eventsByCorrelationKey.size(), publishResponses, batchCount);
24
25    // Loop to retry any failed events
26    for (int i = 1; i < MAX_NUM_OF_RETRIES_PER_BATCH + 1; i++) {
27        if (failedEvents == null || failedEvents.size() == 0) {
28            break;
29        }
30
31        logger.info("retry attempt " + i + ": retrying " + failedEvents.size() +
32            " event(s) from batch " + batchCount);
33
34        batchFinishedLatchRef.set(new CountDownLatch(1));
35
36        // Use the correlationKey provided in each PublishResult to look up the original
37        // corresponding ProducerEvent
38        List<ProducerEvent> eventsToRetry = Lists.newArrayListWithExpectedSize(failedEvents.size());
39        for (int j = 0; j < failedEvents.size(); j++) {
40            String correlationKey = failedEvents.get(j).getCorrelationKey();
41            ProducerEvent producerEvent = eventsByCorrelationKey.get(correlationKey);
42            eventsToRetry.add(j, producerEvent);
43        }
44
45        // Create a retry PublishRequest that contains all events that must be retried
46        PublishRequest retryPublishRequest = PublishRequest.newBuilder().setTopicName(busTopicName)
47                .addAllEvents(eventsToRetry).build();
48
49        // Retry publishing of the failed events
50        requestObserver.onNext(retryPublishRequest);
51
52        failedEvents = validatePublishResponse(errorStatuses, batchFinishedLatchRef.get(), eventsToRetry.size(),
53                publishResponses, batchCount);
54    }

The process of republishing failed events with the unary Publish RPC Method is similar to that of the PublishStream RPC. You can correlate events with a PublishResult by using the correlation key. The only difference between the two RPCs is that Publish RPC has no long-lived connection and the call ends after it executes.

Note