camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dhira...@apache.org
Subject camel git commit: CAMEL-10238: Refactored to handle handshake failure, added increment and max backoff properties
Date Tue, 30 Aug 2016 05:49:17 GMT
Repository: camel
Updated Branches:
  refs/heads/camel-2.17.x 1c07ce242 -> 3e5ef769e


CAMEL-10238: Refactored to handle handshake failure, added increment and max backoff properties

Conflicts:
	components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceComponent.java
	components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java
	components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
	components/camel-salesforce/camel-salesforce-component/src/test/resources/log4j2.properties


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/3e5ef769
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/3e5ef769
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/3e5ef769

Branch: refs/heads/camel-2.17.x
Commit: 3e5ef769e2d8d13338729566fdb128955b4f94bf
Parents: 1c07ce2
Author: Dhiraj Bokde <dhirajsb@yahoo.com>
Authored: Mon Aug 29 22:32:56 2016 -0700
Committer: Dhiraj Bokde <dhirajsb@yahoo.com>
Committed: Mon Aug 29 22:45:22 2016 -0700

----------------------------------------------------------------------
 .../salesforce/SalesforceEndpointConfig.java    |  36 +++-
 .../internal/streaming/SubscriptionHelper.java  | 165 +++++++++++--------
 2 files changed, 129 insertions(+), 72 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/3e5ef769/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java
b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java
index 2bb6306..92ca010 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java
@@ -73,6 +73,10 @@ public class SalesforceEndpointConfig implements Cloneable {
     // default maximum authentication retries on failed authentication or expired session
     public static final int DEFAULT_MAX_AUTHENTICATION_RETRIES = 4;
 
+    // default increment and limit for Streaming connection restart attempts
+    public static final long DEFAULT_BACKOFF_INCREMENT = 1000L;
+    public static final long DEFAULT_MAX_BACKOFF = 30000L;
+
     // general properties
     @UriParam
     private String apiVersion = DEFAULT_VERSION;
@@ -145,6 +149,14 @@ public class SalesforceEndpointConfig implements Cloneable {
     @UriParam
     private SalesforceHttpClient httpClient;
 
+    // Streaming connection restart attempt backoff interval increment
+    @UriParam
+    private long backoffIncrement = DEFAULT_BACKOFF_INCREMENT;
+
+    // Streaming connection restart attempt maximum backoff interval
+    @UriParam
+    private long maxBackoff = DEFAULT_MAX_BACKOFF;
+
     public SalesforceEndpointConfig copy() {
         try {
             final SalesforceEndpointConfig copy = (SalesforceEndpointConfig) super.clone();
@@ -485,6 +497,28 @@ public class SalesforceEndpointConfig implements Cloneable {
         return httpClient;
     }
 
+    public long getBackoffIncrement() {
+        return backoffIncrement;
+    }
+
+    /**
+     * Backoff interval increment for Streaming connection restart attempts for failures
beyond CometD auto-reconnect.
+     */
+    public void setBackoffIncrement(long backoffIncrement) {
+        this.backoffIncrement = backoffIncrement;
+    }
+
+    public long getMaxBackoff() {
+        return maxBackoff;
+    }
+
+    /**
+     * Maximum backoff interval for Streaming connection restart attempts for failures beyond
CometD auto-reconnect.
+     */
+    public void setMaxBackoff(long maxBackoff) {
+        this.maxBackoff = maxBackoff;
+    }
+
     public Map<String, Object> toValueMap() {
 
         final Map<String, Object> valueMap = new HashMap<String, Object>();
@@ -511,7 +545,7 @@ public class SalesforceEndpointConfig implements Cloneable {
         valueMap.put(JOB_ID, jobId);
         valueMap.put(BATCH_ID, batchId);
         valueMap.put(RESULT_ID, resultId);
-        
+
         // add analytics API properties
         valueMap.put(REPORT_ID, reportId);
         valueMap.put(INCLUDE_DETAILS, includeDetails);

http://git-wip-us.apache.org/repos/asf/camel/blob/3e5ef769/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
----------------------------------------------------------------------
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 1e3e8a4..cbe84ab 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
@@ -20,6 +20,7 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicLong;
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.SECONDS;
@@ -66,6 +67,8 @@ public class SubscriptionHelper extends ServiceSupport {
     private final long timeout = 60 * 1000L;
 
     private final Map<SalesforceConsumer, ClientSessionChannel.MessageListener> listenerMap;
+    private final long maxBackoff;
+    private final long backoffIncrement;
 
     private ClientSessionChannel.MessageListener handshakeListener;
     private ClientSessionChannel.MessageListener connectListener;
@@ -77,6 +80,7 @@ public class SubscriptionHelper extends ServiceSupport {
     private volatile Exception connectException;
 
     private volatile boolean reconnecting;
+    private final AtomicLong restartBackoff;
 
     public SubscriptionHelper(SalesforceComponent component) throws Exception {
         this.component = component;
@@ -86,6 +90,10 @@ public class SubscriptionHelper extends ServiceSupport {
 
         // create CometD client
         this.client = createClient();
+
+        restartBackoff = new AtomicLong(0);
+        backoffIncrement = component.getConfig().getBackoffIncrement();
+        maxBackoff = component.getConfig().getMaxBackoff();
     }
 
     @Override
@@ -108,6 +116,10 @@ public class SubscriptionHelper extends ServiceSupport {
                         LOG.warn("Handshake failure: {}", message);
                         handshakeError = (String) message.get(ERROR_FIELD);
                         handshakeException = getFailure(message);
+
+                        // restart if handshake fails for any reason
+                        restartClient();
+
                     } else if (!listenerMap.isEmpty()) {
                         reconnecting = true;
                     }
@@ -129,12 +141,12 @@ public class SubscriptionHelper extends ServiceSupport {
                         connectException = getFailure(message);
 
                         if (connectError != null) {
-                            // refresh oauth token, if it's a 401 error
-                            if (connectError.startsWith("401::")) {
+                            // refresh oauth token, if it's a 403 error
+                            if (connectError.startsWith("403::")) {
                                 try {
                                     session.login(null);
                                 } catch (SalesforceException e) {
-                                    LOG.error("Error renewing OAuth token on Connect 401:
{} ", e.getMessage(), e);
+                                    LOG.error("Error renewing OAuth token on Connect 403:
" + e.getMessage(), e);
                                 }
                             }
                         }
@@ -166,74 +178,7 @@ public class SubscriptionHelper extends ServiceSupport {
             disconnectListener = new ClientSessionChannel.MessageListener() {
                 @Override
                 public void onMessage(ClientSessionChannel clientSessionChannel, Message
message) {
-
-                    // launch an async task to reconnect
-                    final SalesforceHttpClient httpClient = component.getConfig().getHttpClient();
-
-                    httpClient.getExecutor().execute(new Runnable() {
-                        @Override
-                        public void run() {
-
-                            boolean abort = false;
-                            // wait for disconnect
-                            while (!client.isDisconnected()) {
-                                try {
-                                    Thread.sleep(DISCONNECT_INTERVAL);
-                                } catch (InterruptedException e) {
-                                    LOG.error("Aborting reconnect on interrupt!");
-                                    abort = true;
-                                }
-                            }
-
-                            if (!abort) {
-
-                                LOG.info("Reconnecting on unexpected disconnect from Salesforce...");
-                                final long backoffIncrement = client.getBackoffIncrement();
-                                final long maxBackoff = client.getMaxBackoff();
-
-                                long backoff = backoffIncrement;
-                                String msg = String.format("Failed to reconnect, exceeded
maximum backoff %s msecs", maxBackoff);
-                                Exception lastError = new SalesforceException(msg, null);
-
-                                // retry until interrupted, or handshook or connect backoff
exceeded
-                                while (!abort && !client.isHandshook() &&
backoff < maxBackoff) {
-
-                                    try {
-                                        // reset client
-                                        doStop();
-
-                                        // register listeners and restart
-                                        doStart();
-
-                                    } catch (Exception e) {
-                                        LOG.error("Error reconnecting to Salesforce: {}",
e.getMessage(), e);
-                                        lastError = e;
-                                    }
-
-                                    if (!client.isHandshook()) {
-                                        LOG.debug("Pausing for {} msecs after reconnect failure",
backoff);
-                                        try {
-                                            Thread.sleep(backoff);
-                                        } catch (InterruptedException e) {
-                                            LOG.error("Aborting reconnect on interrupt!");
-                                            abort = true;
-                                        }
-                                        backoff += backoffIncrement;
-                                    }
-                                }
-
-                                if (client.isHandshook()) {
-                                    LOG.info("Successfully reconnected to Salesforce!");
-                                } else if (!abort) {
-                                    // notify all consumers
-                                    String abortMsg = "Aborting Salesforce reconnect due
to: " + lastError.getMessage();
-                                    for (SalesforceConsumer consumer : listenerMap.keySet())
{
-                                        consumer.handleException(abortMsg, new SalesforceException(abortMsg,
lastError));
-                                    }
-                                }
-                            }
-                        }
-                    });
+                    restartClient();
                 }
             };
         }
@@ -263,6 +208,84 @@ public class SubscriptionHelper extends ServiceSupport {
         }
     }
 
+    // launch an async task to restart
+    private void restartClient() {
+
+        // launch a new restart command
+        final SalesforceHttpClient httpClient = component.getConfig().getHttpClient();
+        httpClient.getExecutor().execute(new Runnable() {
+            @Override
+            public void run() {
+
+                LOG.info("Restarting on unexpected disconnect from Salesforce...");
+                boolean abort = false;
+
+                // wait for disconnect
+                LOG.debug("Waiting to disconnect...");
+                while (!client.isDisconnected()) {
+                    try {
+                        Thread.sleep(DISCONNECT_INTERVAL);
+                    } catch (InterruptedException e) {
+                        LOG.error("Aborting restart on interrupt!");
+                        abort = true;
+                    }
+                }
+
+                if (!abort) {
+
+                    // update restart attempt backoff
+                    final long backoff = restartBackoff.getAndAdd(backoffIncrement);
+                    if (backoff > maxBackoff) {
+                        LOG.error("Restart aborted after exceeding {} msecs backoff", maxBackoff);
+                        abort = true;
+                    } else {
+
+                        // pause before restart attempt
+                        LOG.debug("Pausing for {} msecs before restart attempt", backoff);
+                        try {
+                            Thread.sleep(backoff);
+                        } catch (InterruptedException e) {
+                            LOG.error("Aborting restart on interrupt!");
+                            abort = true;
+                        }
+                    }
+
+                    if (!abort) {
+                        Exception lastError = new SalesforceException("Unknown error", null);
+                        try {
+                            // reset client
+                            doStop();
+
+                            // register listeners and restart
+                            doStart();
+
+                        } catch (Exception e) {
+                            LOG.error("Error restarting: " + e.getMessage(), e);
+                            lastError = e;
+                        }
+
+                        if (client.isHandshook()) {
+                            LOG.info("Successfully restarted!");
+                            // reset backoff interval
+                            restartBackoff.set(client.getBackoffIncrement());
+                        } else {
+                            LOG.error("Failed to restart after pausing for {} msecs", backoff);
+                            if ((backoff + backoffIncrement) > maxBackoff) {
+                                // notify all consumers
+                                String abortMsg = "Aborting restart attempt due to: " + lastError.getMessage();
+                                SalesforceException ex = new SalesforceException(abortMsg,
lastError);
+                                for (SalesforceConsumer consumer : listenerMap.keySet())
{
+                                    consumer.handleException(abortMsg, ex);
+                                }
+                            }
+                        }
+                    }
+                }
+
+            }
+        });
+    }
+
     @SuppressWarnings("unchecked")
     private Exception getFailure(Message message) {
         Exception exception = null;


Mime
View raw message