(camel) 01/02: CAMEL-21740: added error handling for corrupt replay ids (#17153)

[davsclaus]

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch camel-4.8.x
in repository https://gitbox.apache.org/repos/asf/camel.git

commit 9309dec3df89ef32d37024081f351530bd9f46c7
Author: jubar <j_b_...@proton.me>
AuthorDate: Sun Feb 16 11:12:08 2025 +0100

    CAMEL-21740: added error handling for corrupt replay ids (#17153)
---
 .../camel/component/salesforce/internal/client/PubSubApiClient.java | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git 
a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/PubSubApiClient.java
 
b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/PubSubApiClient.java
index ed3c911a0d0..8f7979fbf8c 100644
--- 
a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/PubSubApiClient.java
+++ 
b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/PubSubApiClient.java
@@ -73,6 +73,7 @@ public class PubSubApiClient extends ServiceSupport {
 
     public static final String PUBSUB_ERROR_AUTH_ERROR = 
"sfdc.platform.eventbus.grpc.service.auth.error";
     private static final String PUBSUB_ERROR_AUTH_REFRESH_INVALID = 
"sfdc.platform.eventbus.grpc.service.auth.refresh.invalid";
+    private static final String PUBSUB_ERROR_CORRUPTED_REPLAY_ID = 
"sfdc.platform.eventbus.grpc.subscription.fetch.replayid.corrupted";
 
     protected PubSubGrpc.PubSubStub asyncStub;
     protected PubSubGrpc.PubSubBlockingStub blockingStub;
@@ -344,6 +345,11 @@ public class PubSubApiClient extends ServiceSupport {
                             
session.attemptLoginUntilSuccessful(backoffIncrement, maxBackoff);
                             LOG.debug("logged in {}", consumer.getTopic());
                         }
+                        case PUBSUB_ERROR_CORRUPTED_REPLAY_ID -> {
+                            LOG.error("replay id is corrupt. try resubscribing 
with LATEST replay preset");
+                            replayId = null;
+                            initialReplayPreset = ReplayPreset.LATEST;
+                        }
                         default -> LOG.error("unexpected errorCode: {}", 
errorCode);
                     }
                 }