(camel) branch main updated: CAMEL-23391: camel-salesforce: Improve error resilience in subscription failure handling (#22840)

[davsclaus]

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

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


The following commit(s) were added to refs/heads/main by this push:
     new faeb2c295379 CAMEL-23391: camel-salesforce: Improve error resilience 
in subscription failure handling (#22840)
faeb2c295379 is described below

commit faeb2c2953799441bc8c9d3879fc7c6a59757fd9
Author: Shailendra Pandey <shailendra.25.v...@gmail.com>
AuthorDate: Thu Apr 30 19:20:32 2026 +0530

    CAMEL-23391: camel-salesforce: Improve error resilience in subscription 
failure handling (#22840)
    
    * camel-salesforce: Improve error resilience in subscription failure 
handling
    
    * Address review: remove redundant ORG_LIMIT_ERROR, drop silent retry on 
missing error, restore comments
---
 .../internal/streaming/SubscriptionHelper.java         | 18 ++++++++++++++++--
 1 file changed, 16 insertions(+), 2 deletions(-)

diff --git 
a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
 
b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
index 6035fb66676e..cfdd259e1e66 100644
--- 
a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
+++ 
b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
@@ -29,6 +29,7 @@ import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Stream;
 
@@ -50,6 +51,7 @@ import org.cometd.client.BayeuxClient;
 import org.cometd.client.BayeuxClient.State;
 import org.cometd.client.http.jetty.JettyHttpClientTransport;
 import org.cometd.client.transport.ClientTransport;
+import org.cometd.common.TransportException;
 import org.eclipse.jetty.client.Request;
 import org.eclipse.jetty.http.HttpCookie;
 import org.eclipse.jetty.http.HttpCookieStore;
@@ -82,6 +84,9 @@ public class SubscriptionHelper extends ServiceSupport {
     private static final String SERVER_TOO_BUSY_ERROR = "503::";
     private static final String AUTHENTICATION_INVALID = "401::Authentication 
invalid";
     private static final String INVALID_REPLAY_ID_PATTERN = "400::The replayId 
\\{.*} you provided was invalid.*";
+    private static final String CHANNEL_INVALID_PATTERN = "400::The channel.*";
+    private static final String DENIED_BY_SEC_POLICY = 
"403:denied_by_security_policy";
+    private static final String AUTHORIZATION_ERROR = "403::";
 
     BayeuxClient client;
 
@@ -219,7 +224,8 @@ public class SubscriptionHelper extends ServiceSupport {
         boolean abort = true;
 
         LOG.warn(msg);
-        if (isTemporaryError(message)) {
+        if (isTemporaryError(message) || error.equals(AUTHENTICATION_INVALID) 
|| error.startsWith(DENIED_BY_SEC_POLICY)
+                || error.startsWith(AUTHORIZATION_ERROR)) {
 
             // retry after delay
             final long backoff = handshakeBackoff.getAndAdd(backoffIncrement);
@@ -255,6 +261,9 @@ public class SubscriptionHelper extends ServiceSupport {
             for (var consumer : consumers) {
                 subscribe(consumer);
             }
+        } else if (error.matches(CHANNEL_INVALID_PATTERN)) {
+            LOG.warn("Channel invalid for channel {}, removing from 
subscription list", channelName);
+            channelsToSubscribe.remove(channelName);
         }
 
         if (abort && client != null) {
@@ -465,7 +474,12 @@ public class SubscriptionHelper extends ServiceSupport {
 
     private static boolean isTemporaryError(Message message) {
         String failureReason = getFailureReason(message);
-        return failureReason != null && 
failureReason.startsWith(SERVER_TOO_BUSY_ERROR);
+        if (failureReason != null && 
failureReason.startsWith(SERVER_TOO_BUSY_ERROR)) {
+            return true;
+        }
+        Exception exception = getFailure(message);
+        return exception instanceof IOException || exception instanceof 
TransportException
+                || exception instanceof TimeoutException;
     }
 
     private static String getFailureReason(Message message) {