Salesforce Pub/Sub Connector | Error "io.grpc.StatusRuntimeException: UNAVAILABLE: RST_STREAM closed stream. HTTP/2 error code: NO_ERROR" when using the subscribe channel listener

SYMPTOM

You have a mule application that uses the "Subscribe Channel Listener" operation of the Salesforce Pub/Sub connector and the below error is reported intermittently.

ERROR SourceDataSupplier [grpc-default-executor-493] [event: ]: An error has occurred while receiving event. Root exception was: io.grpc.StatusRuntimeException: UNAVAILABLE: RST_STREAM closed stream. HTTP/2 error code: NO_ERROR
	at io.grpc.Status.asRuntimeException(Status.java:535)
	at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:478)
	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:562)
	at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:70)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:743)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:722)
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:750)
. Arguments: {sourceDataSupplier=com.mulesoft.connector.salesforce.pubsub.internal.service.source.SourceDataSupplier@599932cb, topic=/event/Order_Platform_Event__e}.
io.grpc.StatusRuntimeException: UNAVAILABLE: RST_STREAM closed stream. HTTP/2 error code: NO_ERROR
	at io.grpc.Status.asRuntimeException(Status.java:535) ~[grpc-api-1.45.0.jar:1.45.0]
	at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:478) ~[grpc-stub-1.45.0.jar:1.45.0]
	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:562) ~[grpc-core-1.45.0.jar:1.45.0]
	at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:70) ~[grpc-core-1.45.0.jar:1.45.0]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:743) ~[grpc-core-1.45.0.jar:1.45.0]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:722) ~[grpc-core-1.45.0.jar:1.45.0]
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37) ~[grpc-core-1.45.0.jar:1.45.0]
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133) ~[grpc-core-1.45.0.jar:1.45.0]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_332]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_332]
	at java.lang.Thread.run(Thread.java:750) ~[?:1.8.0_332]
[04-24 23:24:59.938] ERROR SubscribeChannelSource [grpc-default-executor-493] [event: ]: An error has occurred while processing events. Root exception was: io.grpc.StatusRuntimeException: UNAVAILABLE: RST_STREAM closed stream. HTTP/2 error code: NO_ERROR
	at io.grpc.Status.asRuntimeException(Status.java:535)
	at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:478)
	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:562)
	at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:70)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:743)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:722)
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:750)
. Arguments: {connection=com.mulesoft.connector.salesforce.pubsub.internal.connection.PubSubConnection@4b352b91 (sessionId: , instanceUrl: , tenantId:, apiVersion: 54.0, partnerConnection: com.sforce.soap.partner.PartnerConnection@6addb5b5, metadataConnection: com.sforce.soap.metadata.MetadataConnection@75fb7863).}.
io.grpc.StatusRuntimeException: UNAVAILABLE: RST_STREAM closed stream. HTTP/2 error code: NO_ERROR
	at io.grpc.Status.asRuntimeException(Status.java:535) ~[grpc-api-1.45.0.jar:1.45.0]
	at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:478) ~[grpc-stub-1.45.0.jar:1.45.0]
	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:562) ~[grpc-core-1.45.0.jar:1.45.0]
	at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:70) ~[grpc-core-1.45.0.jar:1.45.0]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:743) ~[grpc-core-1.45.0.jar:1.45.0]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:722) ~[grpc-core-1.45.0.jar:1.45.0]
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37) ~[grpc-core-1.45.0.jar:1.45.0]
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133) ~[grpc-core-1.45.0.jar:1.45.0]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_332]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_332]
	at java.lang.Thread.run(Thread.java:750) ~[?:1.8.0_332]
[04-24 23:24:59.940] ERROR DefaultSystemExceptionStrategy [grpc-default-executor-493] [event: ]: 
********************************************************************************
Message               : UNAVAILABLE: RST_STREAM closed stream. HTTP/2 error code: NO_ERROR
Element               : (None)
Element DSL           : (None)
Error type            : (None)
FlowStack             : (None)
--------------------------------------------------------------------------------
Root Exception stack trace:
io.grpc.StatusRuntimeException: UNAVAILABLE: RST_STREAM closed stream. HTTP/2 error code: NO_ERROR
	at io.grpc.Status.asRuntimeException(Status.java:535)
	at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:478)
	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:562)
	at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:70)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:743)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:722)
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:750)

********************************************************************************

[04-24 23:24:59.941] WARN  ExtensionMessageSource [grpc-default-executor-493] [event: ]: Message source 'subscribe-channel-listener' on flow '' threw exception. Attempting to reconnect...
org.mule.runtime.api.connection.ConnectionException: UNAVAILABLE: RST_STREAM closed stream. HTTP/2 error code: NO_ERROR
	at com.mulesoft.connector.salesforce.pubsub.internal.source.SubscribeChannelSource.handleWhenExceptionOccurred(SubscribeChannelSource.java:298) ~[mule4-salesforce-pubsub-connector-1.0.5-mule-plugin.jar:?]
	at com.mulesoft.connector.salesforce.pubsub.internal.source.SubscribeChannelSource.lambda$onStart$4(SubscribeChannelSource.java:156) ~[mule4-salesforce-pubsub-connector-1.0.5-mule-plugin.jar:?]
	at com.mulesoft.connector.salesforce.pubsub.internal.service.source.SourceDataSupplier.onError(SourceDataSupplier.java:104) ~[mule4-salesforce-pubsub-connector-1.0.5-mule-plugin.jar:?]
	at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:478) ~[grpc-stub-1.45.0.jar:1.45.0]
	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:562) ~[grpc-core-1.45.0.jar:1.45.0]
	at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:70) ~[grpc-core-1.45.0.jar:1.45.0]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:743) ~[grpc-core-1.45.0.jar:1.45.0]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:722) ~[grpc-core-1.45.0.jar:1.45.0]
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37) ~[grpc-core-1.45.0.jar:1.45.0]
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133) ~[grpc-core-1.45.0.jar:1.45.0]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_332]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_332]
	at java.lang.Thread.run(Thread.java:750) ~[?:1.8.0_332]
Caused by: io.grpc.StatusRuntimeException: UNAVAILABLE: RST_STREAM closed stream. HTTP/2 error code: NO_ERROR
	at io.grpc.Status.asRuntimeException(Status.java:535) ~[grpc-api-1.45.0.jar:1.45.0]
	... 10 more

CAUSE

The error is not directly from the Mule server. It can be due to network disconnect or the underlying infrastructure. 
So it is caused either by some network disconnect or the underlying infrastructure of Salesforce PubSub. 

SOLUTION

These errors should be intermittent and resubscribing will fix things. So make sure there is a reconnection strategy enabled either at the operation level or at the connector level and you will see the below logs that it is successfully subscribed when you enable the debug logging for the Salesforce Pub/Sub connector.

DEBUG PubSubConnection [[MuleRuntime].uber.295: [].uber@org.mule.runtime.module.extension.internal.runtime.source.ExtensionMessageSource.lambda$null$17:435 @3ff51a41] [event: ]: gRPC channel created successfully

DEBUG AsyncPubSubClientImpl [[MuleRuntime].uber.295: [].uber@org.mule.runtime.module.extension.internal.runtime.source.ExtensionMessageSource.lambda$null$17:435 @3ff51a41] [event: ]: Listener started from LATEST and listening to 100 events on channel: /event/Platform_Event__e..

But say if you are thinking that it might be losing events at the time of reconnecting, the suggestion is not to use the Latest Replay option, but rather go for Replay id from object store. There is an example here. However, when going for this option you should pay attention to sequential vs parallel processing of events when you create the Replay Id storing strategy and also may want to take into account using an Idempotent Message Validator to avoid potential duplicate event processing caused by the storing strategy.