Make memcached client maxReconnectDelay configurable via sys prop "msm.maxReconnectDelay"

Martin Grotzke · Martin Grotzke · commit 8de1011a10b7 · 2013-12-02T16:08:33.000+01:00
The maxReconnectDelay is specified in seconds.
diff --git a/core/src/main/java/de/javakaffee/web/msm/Configurations.java b/core/src/main/java/de/javakaffee/web/msm/Configurations.java
@@ -26,11 +26,18 @@
  */
 public final class Configurations {
 
+    /**
+     * The TTL for {@link NodeAvailabilityCache} entries, in millis.
+     */
     public static final String NODE_AVAILABILITY_CACHE_TTL_KEY = "msm.nodeAvailabilityCacheTTL";
+    /**
+     * The max reconnect delay for the MemcachedClient, in seconds.
+     */
+    public static final String MAX_RECONNECT_DELAY_KEY = "msm.maxReconnectDelay";
 
     private static final Log LOG = LogFactory.getLog(Configurations.class);
 
-    public static int getSystemPropertyInt(final String propName, final int defaultValue) {
+    public static int getSystemProperty(final String propName, final int defaultValue) {
         final String value = System.getProperty(propName);
         if(value != null) {
             try {
@@ -42,4 +49,16 @@ public static int getSystemPropertyInt(final String propName, final int defaultV
         return defaultValue;
     }
 
+    public static long getSystemProperty(final String propName, final long defaultValue) {
+        final String value = System.getProperty(propName);
+        if(value != null) {
+            try {
+                return Long.parseLong(value);
+            } catch(final NumberFormatException e) {
+                LOG.warn("Could not parse configured value for system property '" + propName + "': " + value);
+            }
+        }
+        return defaultValue;
+    }
+
 }
diff --git a/core/src/main/java/de/javakaffee/web/msm/CouchbaseClientFactory.java b/core/src/main/java/de/javakaffee/web/msm/CouchbaseClientFactory.java
@@ -30,7 +30,7 @@ public class CouchbaseClientFactory implements MemcachedClientFactory.CouchbaseC
     @Override
     public CouchbaseClient createCouchbaseClient(final MemcachedNodesManager memcachedNodesManager,
             final String memcachedProtocol, final String username, String password, final long operationTimeout,
-            final Statistics statistics ) {
+            final long maxReconnectDelay, final Statistics statistics ) {
         try {
             // CouchbaseClient does not accept null for password
             if(password == null)
@@ -40,6 +40,7 @@ public CouchbaseClient createCouchbaseClient(final MemcachedNodesManager memcach
             // And: https://code.google.com/p/spymemcached/wiki/Examples#Establishing_a_Membase_Connection
             final CouchbaseConnectionFactoryBuilder factory = newCouchbaseConnectionFactoryBuilder();
             factory.setOpTimeout(operationTimeout);
+            factory.setMaxReconnectDelay(maxReconnectDelay);
             factory.setFailureMode(FailureMode.Redistribute);
             return new CouchbaseClient(factory.buildCouchbaseConnection(memcachedNodesManager.getCouchbaseBucketURIs(), username, password));
         } catch (final Exception e) {
diff --git a/core/src/main/java/de/javakaffee/web/msm/MemcachedClientFactory.java b/core/src/main/java/de/javakaffee/web/msm/MemcachedClientFactory.java
@@ -36,20 +36,20 @@ public class MemcachedClientFactory {
     static interface CouchbaseClientFactory {
         MemcachedClient createCouchbaseClient(MemcachedNodesManager memcachedNodesManager,
                 String memcachedProtocol, String username, String password, long operationTimeout,
-                Statistics statistics );
+                long maxReconnectDelay, Statistics statistics );
     }
 
     protected MemcachedClient createMemcachedClient(final MemcachedNodesManager memcachedNodesManager,
             final String memcachedProtocol, final String username, final String password, final long operationTimeout,
-            final Statistics statistics ) {
+            final long maxReconnectDelay, final Statistics statistics ) {
         try {
             final ConnectionType connectionType = ConnectionType.valueOf(memcachedNodesManager.isCouchbaseBucketConfig(), username, password);
             if (connectionType.isCouchbaseBucketConfig()) {
-                return createCouchbaseClient(memcachedNodesManager, memcachedProtocol, username, password, operationTimeout,
+                return createCouchbaseClient(memcachedNodesManager, memcachedProtocol, username, password, operationTimeout, maxReconnectDelay,
                         statistics);
             }
             final ConnectionFactory connectionFactory = createConnectionFactory(memcachedNodesManager, connectionType, memcachedProtocol,
-                    username, password, operationTimeout, statistics);
+                    username, password, operationTimeout, maxReconnectDelay, statistics);
             return new MemcachedClient(connectionFactory, memcachedNodesManager.getAllMemcachedAddresses());
         } catch (final Exception e) {
             throw new RuntimeException("Could not create memcached client", e);
@@ -58,37 +58,58 @@ protected MemcachedClient createMemcachedClient(final MemcachedNodesManager memc
 
     protected MemcachedClient createCouchbaseClient(final MemcachedNodesManager memcachedNodesManager,
             final String memcachedProtocol, final String username, final String password, final long operationTimeout,
-            final Statistics statistics) {
+            final long maxReconnectDelay, final Statistics statistics) {
         try {
             final CouchbaseClientFactory factory = Class.forName("de.javakaffee.web.msm.CouchbaseClientFactory").asSubclass(CouchbaseClientFactory.class).newInstance();
-            return factory.createCouchbaseClient(memcachedNodesManager, memcachedProtocol, username, password, operationTimeout, statistics);
+            return factory.createCouchbaseClient(memcachedNodesManager, memcachedProtocol, username, password, operationTimeout, maxReconnectDelay, statistics);
         } catch (final Exception e) {
             throw new RuntimeException(e);
         }
     }
 
     protected ConnectionFactory createConnectionFactory(final MemcachedNodesManager memcachedNodesManager,
-            final ConnectionType connectionType, final String memcachedProtocol, final String username, final String password, final long operationTimeout, final Statistics statistics ) {
+            final ConnectionType connectionType, final String memcachedProtocol, final String username, final String password, final long operationTimeout,
+            final long maxReconnectDelay, final Statistics statistics ) {
         if (PROTOCOL_BINARY.equals( memcachedProtocol )) {
             if (connectionType.isSASL()) {
                 final AuthDescriptor authDescriptor = new AuthDescriptor(new String[]{"PLAIN"}, new PlainCallbackHandler(username, password));
                 return memcachedNodesManager.isEncodeNodeIdInSessionId()
                         ? new SuffixLocatorBinaryConnectionFactory( memcachedNodesManager,
-                                memcachedNodesManager.getSessionIdFormat(), statistics, operationTimeout,
+                                memcachedNodesManager.getSessionIdFormat(), statistics, operationTimeout, maxReconnectDelay,
                                 authDescriptor)
                         : new ConnectionFactoryBuilder().setProtocol(ConnectionFactoryBuilder.Protocol.BINARY)
                                 .setAuthDescriptor(authDescriptor)
-                                .setOpTimeout(operationTimeout).build();
+                                .setOpTimeout(operationTimeout)
+                                .setMaxReconnectDelay(maxReconnectDelay)
+                                .build();
             }
             else {
                 return memcachedNodesManager.isEncodeNodeIdInSessionId() ? new SuffixLocatorBinaryConnectionFactory( memcachedNodesManager,
                         memcachedNodesManager.getSessionIdFormat(),
-                        statistics, operationTimeout ) : new BinaryConnectionFactory();
+                        statistics, operationTimeout, maxReconnectDelay ) : new BinaryConnectionFactory() {
+                    @Override
+                    public long getOperationTimeout() {
+                        return operationTimeout;
+                    }
+                    @Override
+                    public long getMaxReconnectDelay() {
+                        return maxReconnectDelay;
+                    }
+                };
             }
         }
         return memcachedNodesManager.isEncodeNodeIdInSessionId()
-                ? new SuffixLocatorConnectionFactory( memcachedNodesManager, memcachedNodesManager.getSessionIdFormat(), statistics, operationTimeout )
-                : new DefaultConnectionFactory();
+                ? new SuffixLocatorConnectionFactory( memcachedNodesManager, memcachedNodesManager.getSessionIdFormat(), statistics, operationTimeout, maxReconnectDelay )
+                : new DefaultConnectionFactory() {
+                    @Override
+                    public long getOperationTimeout() {
+                        return operationTimeout;
+                    }
+                    @Override
+                    public long getMaxReconnectDelay() {
+                        return maxReconnectDelay;
+                    }
+                };
     }
 
     static class ConnectionType {
diff --git a/core/src/main/java/de/javakaffee/web/msm/MemcachedNodesManager.java b/core/src/main/java/de/javakaffee/web/msm/MemcachedNodesManager.java
@@ -16,7 +16,7 @@
 package de.javakaffee.web.msm;
 
 import static de.javakaffee.web.msm.Configurations.NODE_AVAILABILITY_CACHE_TTL_KEY;
-import static de.javakaffee.web.msm.Configurations.getSystemPropertyInt;
+import static de.javakaffee.web.msm.Configurations.getSystemProperty;
 
 import java.net.InetSocketAddress;
 import java.net.URI;
@@ -73,7 +73,7 @@ public static interface MemcachedClientCallback {
     private static final String COUCHBASE_BUCKET_NODES_REGEX = COUCHBASE_BUCKET_NODE_REGEX + "(?:(?:\\s+|,)" + COUCHBASE_BUCKET_NODE_REGEX + ")*";
     private static final Pattern COUCHBASE_BUCKET_NODES_PATTERN = Pattern.compile( COUCHBASE_BUCKET_NODES_REGEX );
 
-    private static final int NODE_AVAILABILITY_CACHE_TTL = getSystemPropertyInt(NODE_AVAILABILITY_CACHE_TTL_KEY, 1000);
+    private static final int NODE_AVAILABILITY_CACHE_TTL = getSystemProperty(NODE_AVAILABILITY_CACHE_TTL_KEY, 1000);
 
 	private final String _memcachedNodes;
     private final NodeIdList _primaryNodeIds;
diff --git a/core/src/main/java/de/javakaffee/web/msm/MemcachedSessionService.java b/core/src/main/java/de/javakaffee/web/msm/MemcachedSessionService.java
@@ -17,6 +17,8 @@
 package de.javakaffee.web.msm;
 
 
+import static de.javakaffee.web.msm.Configurations.MAX_RECONNECT_DELAY_KEY;
+import static de.javakaffee.web.msm.Configurations.getSystemProperty;
 import static de.javakaffee.web.msm.Statistics.StatsType.DELETE_FROM_MEMCACHED;
 import static de.javakaffee.web.msm.Statistics.StatsType.LOAD_FROM_MEMCACHED;
 import static de.javakaffee.web.msm.Statistics.StatsType.SESSION_DESERIALIZATION;
@@ -34,6 +36,7 @@
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
+import net.spy.memcached.DefaultConnectionFactory;
 import net.spy.memcached.MemcachedClient;
 
 import org.apache.catalina.Container;
@@ -464,7 +467,10 @@ protected MemcachedClient createMemcachedClient( final MemcachedNodesManager mem
         if ( ! _enabled.get() ) {
             return null;
         }
-        return new MemcachedClientFactory().createMemcachedClient(memcachedNodesManager, _memcachedProtocol, _username, _password, _operationTimeout, statistics);
+
+        final long maxReconnectDelay = getSystemProperty(MAX_RECONNECT_DELAY_KEY, DefaultConnectionFactory.DEFAULT_MAX_RECONNECT_DELAY);
+        return new MemcachedClientFactory().createMemcachedClient(memcachedNodesManager, _memcachedProtocol, _username, _password, _operationTimeout,
+                maxReconnectDelay, statistics);
     }
 
     private TranscoderFactory createTranscoderFactory() throws InstantiationException, IllegalAccessException, ClassNotFoundException {
diff --git a/core/src/main/java/de/javakaffee/web/msm/SuffixLocatorBinaryConnectionFactory.java b/core/src/main/java/de/javakaffee/web/msm/SuffixLocatorBinaryConnectionFactory.java
@@ -46,6 +46,7 @@ public final class SuffixLocatorBinaryConnectionFactory extends DefaultConnectio
     private final SessionIdFormat _sessionIdFormat;
     private final Statistics _statistics;
     private final long _operationTimeout;
+    private final long _maxReconnectDelay;
     private final AuthDescriptor _authDescriptor;
 
     /**
@@ -56,11 +57,12 @@ public final class SuffixLocatorBinaryConnectionFactory extends DefaultConnectio
      *            the {@link SessionIdFormat}
      */
     public SuffixLocatorBinaryConnectionFactory( final MemcachedNodesManager memcachedNodesManager, final SessionIdFormat sessionIdFormat,
-            final Statistics statistics, final long operationTimeout, final AuthDescriptor authDescriptor) {
+            final Statistics statistics, final long operationTimeout, final long maxReconnectDelay, final AuthDescriptor authDescriptor) {
         _memcachedNodesManager = memcachedNodesManager;
         _sessionIdFormat = sessionIdFormat;
         _statistics = statistics;
         _operationTimeout = operationTimeout;
+        _maxReconnectDelay = maxReconnectDelay;
         _authDescriptor = authDescriptor;
     }
 
@@ -72,8 +74,8 @@ public SuffixLocatorBinaryConnectionFactory( final MemcachedNodesManager memcach
      *            the {@link SessionIdFormat}
      */
     public SuffixLocatorBinaryConnectionFactory( final MemcachedNodesManager memcachedNodesManager, final SessionIdFormat sessionIdFormat,
-            final Statistics statistics, final long operationTimeout ) {
-        this(memcachedNodesManager, sessionIdFormat, statistics, operationTimeout, null);
+            final Statistics statistics, final long operationTimeout, final long maxReconnectDelay ) {
+        this(memcachedNodesManager, sessionIdFormat, statistics, operationTimeout, maxReconnectDelay, null);
     }
 
     /**
@@ -129,6 +131,11 @@ public long getOperationTimeout() {
         return _operationTimeout;
     }
 
+    @Override
+    public long getMaxReconnectDelay() {
+        return _maxReconnectDelay;
+    }
+
     @Override
     public AuthDescriptor getAuthDescriptor() {
         return _authDescriptor;
diff --git a/core/src/main/java/de/javakaffee/web/msm/SuffixLocatorConnectionFactory.java b/core/src/main/java/de/javakaffee/web/msm/SuffixLocatorConnectionFactory.java
@@ -39,6 +39,7 @@ public final class SuffixLocatorConnectionFactory extends DefaultConnectionFacto
     private final SessionIdFormat _sessionIdFormat;
     private final Statistics _statistics;
     private final long _operationTimeout;
+    private final long _maxReconnectDelay;
 
     /**
      * Creates a new instance.
@@ -48,11 +49,12 @@ public final class SuffixLocatorConnectionFactory extends DefaultConnectionFacto
      *            the {@link SessionIdFormat}
      */
     public SuffixLocatorConnectionFactory( final MemcachedNodesManager memcachedNodesManager, final SessionIdFormat sessionIdFormat,
-            final Statistics statistics, long operationTimeout ) {
+            final Statistics statistics, final long operationTimeout, final long maxReconnectDelay ) {
         _memcachedNodesManager = memcachedNodesManager;
         _sessionIdFormat = sessionIdFormat;
         _statistics = statistics;
         _operationTimeout = operationTimeout;
+        _maxReconnectDelay = maxReconnectDelay;
     }
 
     /**
@@ -81,9 +83,14 @@ public Transcoder<Object> getDefaultTranscoder() {
         transcoder.setCompressionThreshold( SerializingTranscoder.DEFAULT_COMPRESSION_THRESHOLD );
         return new TranscoderWrapperStatisticsSupport( _statistics, transcoder );
     }
-    
+
     @Override
     public long getOperationTimeout() {
     	return _operationTimeout;
     }
+
+    @Override
+    public long getMaxReconnectDelay() {
+        return _maxReconnectDelay;
+    }
 }
diff --git a/core/src/test/java/de/javakaffee/web/msm/MemcachedSessionManagerIntegrationTest.java b/core/src/test/java/de/javakaffee/web/msm/MemcachedSessionManagerIntegrationTest.java
@@ -131,7 +131,7 @@ private TomcatBuilder tcBuilder() {
     private MemcachedClient createMemcachedClient( final String memcachedNodes, final InetSocketAddress address ) throws IOException, InterruptedException {
     	final MemcachedNodesManager nodesManager = MemcachedNodesManager.createFor(memcachedNodes, null, _memcachedClientCallback);
         final ConnectionFactory cf = nodesManager.isEncodeNodeIdInSessionId()
-            ? new SuffixLocatorConnectionFactory( nodesManager, nodesManager.getSessionIdFormat(), Statistics.create(), 1000 )
+            ? new SuffixLocatorConnectionFactory( nodesManager, nodesManager.getSessionIdFormat(), Statistics.create(), 1000, 1000 )
             : new DefaultConnectionFactory();
         final MemcachedClient result = new MemcachedClient( cf, Arrays.asList( address ) );
 
diff --git a/core/src/test/java/de/javakaffee/web/msm/integration/NonStickySessionsIntegrationTest.java b/core/src/test/java/de/javakaffee/web/msm/integration/NonStickySessionsIntegrationTest.java
@@ -131,7 +131,7 @@ public void setUp() throws Throwable {
 
         final MemcachedNodesManager nodesManager = MemcachedNodesManager.createFor(MEMCACHED_NODES, null, _memcachedClientCallback);
         _client =
-                new MemcachedClient( new SuffixLocatorConnectionFactory( nodesManager, nodesManager.getSessionIdFormat(), Statistics.create(), 1000 ),
+                new MemcachedClient( new SuffixLocatorConnectionFactory( nodesManager, nodesManager.getSessionIdFormat(), Statistics.create(), 1000, 1000 ),
                         Arrays.asList( address1, address2 ) );
 
         final SchemeRegistry schemeRegistry = new SchemeRegistry();
diff --git a/core/src/test/java/de/javakaffee/web/msm/integration/TomcatFailoverIntegrationTest.java b/core/src/test/java/de/javakaffee/web/msm/integration/TomcatFailoverIntegrationTest.java
@@ -116,7 +116,7 @@ public void setUp(final Method testMethod) throws Throwable {
 
         final MemcachedNodesManager nodesManager = MemcachedNodesManager.createFor(_memcachedNodes, null, _memcachedClientCallback);
         final ConnectionFactory cf = nodesManager.isEncodeNodeIdInSessionId()
-            ? new SuffixLocatorConnectionFactory( nodesManager, nodesManager.getSessionIdFormat(), Statistics.create(), 1000 )
+            ? new SuffixLocatorConnectionFactory( nodesManager, nodesManager.getSessionIdFormat(), Statistics.create(), 1000, 1000 )
             : new DefaultConnectionFactory();
         _client = new MemcachedClient( cf, Arrays.asList( address ) );
 
