"Object already exists for the key" Error with Salesforce Connector Events Listener in Multiple CloudHub Workers

SYMPTOM

You are using the Replay Channel Listener of Salesforce Connector for Mule 4 and deployed the application to CloudHub with multiple workers. The following error "Object already exists for the key" is seen in the application logs:

Message source 'replay-channel' on flow 'sfdc-events9Flow' threw exception. Attempting to reconnect...
org.mule.runtime.api.connection.ConnectionException: Object already exists for the key bkt@my.com/u/notifications/mychannel in partition StreamingObjectStore in store APP_sfdc-self__defaultPersistentObjectStore
	at org.mule.extension.salesforce.internal.service.streaming.SalesforceBayeuxMessageListener.onMessage(SalesforceBayeuxMessageListener.java:54)
	at org.cometd.common.AbstractClientSession$AbstractSessionChannel.notifyOnMessage(AbstractClientSession.java:600)
	at org.cometd.common.AbstractClientSession$AbstractSessionChannel.notifyMessageListeners(AbstractClientSession.java:591)
	at org.cometd.common.AbstractClientSession.notifyListeners(AbstractClientSession.java:297)
	at org.cometd.common.AbstractClientSession.receive(AbstractClientSession.java:263)
	at org.cometd.client.BayeuxClient.processMessage(BayeuxClient.java:829)
	at org.cometd.client.BayeuxClient.processMessages(BayeuxClient.java:625)
	at org.cometd.client.BayeuxClient.access$3100(BayeuxClient.java:100)
	at org.cometd.client.BayeuxClient$MessageTransportListener.onMessages(BayeuxClient.java:1197)
	at org.cometd.client.transport.LongPollingTransport$2.onComplete(LongPollingTransport.java:239)

CAUSE

The Replay Channel Listener is not designed to run in multiple CloudHub workers when the replay option FROM_LAST_REPLAY_ID is used or the option "Replay failed events if any or resume from last replay id" is checked. Salesforce Connector internally uses the CloudHub Object Store v2 to keep track of the failed events and the highest processed replayId.

If multiple workers of the same CloudHub application attempt to update the object store concurrently, it will hit the same object store key and throw the above exception in the logs.

For more details, refer to Limitations of using Object Store v2 (OSv2) in CloudHub applications.

SOLUTION

Use one CloudHub worker to handle the Salesforce events listener portion and have it propagate the events to other microservices.