QPID-5409: Make replication node monitor interval/timeout configurable from the vhost level. Eliminate thread pool

per remote replication node.

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/java-broker-bdb-ha@1559084 13f79535-47bb-0310-9956-ffa450edef68

8 files changed, 147 insertions, 47 deletions

diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacade.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacade.java
index 379efef613..e017bf13ad 100644
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacade.java
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacade.java
@@ -38,13 +38,16 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.log4j.Logger;
@@ -53,6 +56,7 @@ import org.apache.qpid.server.replication.ReplicationGroupListener;
 import org.apache.qpid.server.store.StoreFuture;
 import org.apache.qpid.server.store.berkeleydb.replication.RemoteReplicationNode;
 import org.apache.qpid.server.store.berkeleydb.replication.RemoteReplicationNodeFactory;
+import org.apache.qpid.server.util.DaemonThreadFactory;
 
 import com.sleepycat.je.Database;
 import com.sleepycat.je.DatabaseConfig;
@@ -80,7 +84,11 @@ import com.sleepycat.je.rep.util.ReplicationGroupAdmin;
 
 public class ReplicatedEnvironmentFacade implements EnvironmentFacade, StateChangeListener
 {
+    public static final String GROUP_CHECK_INTERVAL_PROPERTY_NAME = "qpid.bdb.ha.group_check_interval";
+
     private static final Logger LOGGER = Logger.getLogger(ReplicatedEnvironmentFacade.class);
+    private static final long DEFAULT_GROUP_CHECK_INTERVAL = 1000l;
+    private static final long GROUP_CHECK_INTERVAL = Long.getLong(GROUP_CHECK_INTERVAL_PROPERTY_NAME, DEFAULT_GROUP_CHECK_INTERVAL);
 
     @SuppressWarnings("serial")
     private static final Map<String, String> REPCONFIG_DEFAULTS = Collections.unmodifiableMap(new HashMap<String, String>()
@@ -136,11 +144,11 @@ public class ReplicatedEnvironmentFacade implements EnvironmentFacade, StateChan
     private final Map<String, String> _environmentParameters;
     private final Map<String, String> _replicationEnvironmentParameters;
     private final String _name;
-    private final ExecutorService _restartEnvironmentExecutor = Executors.newFixedThreadPool(1);
+    private final ExecutorService _restartEnvironmentExecutor;
     private final ScheduledExecutorService _groupChangeExecutor;
     private final AtomicReference<State> _state = new AtomicReference<State>(State.INITIAL);
     private final ConcurrentMap<String, Database> _databases = new ConcurrentHashMap<String, Database>();
-    private final ConcurrentMap<String, org.apache.qpid.server.model.ReplicationNode> _remoteReplicationNodes = new ConcurrentHashMap<String, org.apache.qpid.server.model.ReplicationNode>();
+    private final ConcurrentMap<String, RemoteReplicationNode> _remoteReplicationNodes = new ConcurrentHashMap<String, RemoteReplicationNode>();
     private final RemoteReplicationNodeFactory _remoteReplicationNodeFactory;
 
     private volatile CommitThreadWrapper _commitThreadWrapper;
@@ -167,19 +175,13 @@ public class ReplicatedEnvironmentFacade implements EnvironmentFacade, StateChan
         _environmentParameters = (Map<String, String>)replicationNode.getAttribute(PARAMETERS);
         _replicationEnvironmentParameters = (Map<String, String>)replicationNode.getAttribute(REPLICATION_PARAMETERS);
 
-        _groupChangeExecutor = Executors.newScheduledThreadPool(1, new ThreadFactory()
-        {
-            @Override
-            public Thread newThread(Runnable r)
-            {
-                return new Thread(r, "GroupChangeLearner_" + _groupName);
-            }
-        });
+        _restartEnvironmentExecutor = Executors.newFixedThreadPool(1, new DaemonThreadFactory("Environment-Restarter:" + _groupName));
+        _groupChangeExecutor = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors() + 1, new DaemonThreadFactory("Group-Change-Learner:" + _groupName));
+
         _remoteReplicationNodeFactory = remoteReplicationNodeFactory;
         _state.set(State.OPENING);
-        //TODO: add ability to alter the execution period
-        _groupChangeExecutor.scheduleWithFixedDelay(new GroupChangeLearner(), 1, 1, TimeUnit.SECONDS);
-
+        _groupChangeExecutor.scheduleWithFixedDelay(new GroupChangeLearner(), 0, GROUP_CHECK_INTERVAL, TimeUnit.MILLISECONDS);
+        _groupChangeExecutor.schedule(new RemoteNodeStateLearner(), _remoteReplicationNodeFactory.getRemoteNodeMonitorInterval(), TimeUnit.MILLISECONDS);
         _environment = createEnvironment();
         populateExistingRemoteReplicationNodes();
         _commitThreadWrapper = startCommitThread(_name, _environment);
@@ -809,6 +811,60 @@ public class ReplicatedEnvironmentFacade implements EnvironmentFacade, StateChan
         }
     }
 
+    //TODO: move the class into external class
+    private class RemoteNodeStateLearner implements Callable<Void>
+    {
+        @Override
+        public Void call()
+        {
+            long remoteNodeMonitorInterval = _remoteReplicationNodeFactory.getRemoteNodeMonitorInterval();
+            try
+            {
+                Set<Future<Void>> futures = new HashSet<Future<Void>>();
+                for (Map.Entry<String, RemoteReplicationNode> entry : _remoteReplicationNodes.entrySet())
+                {
+                    final RemoteReplicationNode  node = entry.getValue();
+                    Future<Void> future = _groupChangeExecutor.submit(new Callable<Void>()
+                    {
+                        @Override
+                        public Void call()
+                        {
+                            node.updateNodeState();
+                            return null;
+                        }
+                    });
+                    futures.add(future);
+                }
+
+                for (Future<Void> future : futures)
+                {
+                    try
+                    {
+                        future.get(remoteNodeMonitorInterval, TimeUnit.MILLISECONDS);
+                    }
+                    catch (InterruptedException e)
+                    {
+                        Thread.currentThread().interrupt();
+                    }
+                    catch (ExecutionException e)
+                    {
+                        LOGGER.warn("Cannot update node state for group " + _groupName, e.getCause());
+                    }
+                    catch (TimeoutException e)
+                    {
+                        LOGGER.warn("Timeout whilst updating node state for group " + _groupName);
+                        future.cancel(true);
+                    }
+                }
+            }
+            finally
+            {
+                _groupChangeExecutor.schedule(this, remoteNodeMonitorInterval, TimeUnit.MILLISECONDS);
+            }
+            return null;
+        }
+    }
+
     private class LoggingAsyncExceptionListener implements ExceptionListener
     {
         @Override
diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeFactory.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeFactory.java
index 5eeda17373..44f809a70b 100644
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeFactory.java
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeFactory.java
@@ -83,5 +83,11 @@ public class ReplicatedEnvironmentFacadeFactory implements EnvironmentFacadeFact
             attributes.put(ReplicationNode.HOST_PORT, replicationNode.getHostName() + ":" + replicationNode.getPort());
             return new RemoteReplicationNode(replicationNode, groupName, _virtualHost, _virtualHost.getTaskExecutor());
         }
+
+        @Override
+        public long getRemoteNodeMonitorInterval()
+        {
+            return (Long)_virtualHost.getAttribute(VirtualHost.REMOTE_REPLICATION_NODE_MONITOR_INTERVAL);
+        }
     }
 }
diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNode.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNode.java
index df2eddad89..bff20fb306 100644
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNode.java
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNode.java
@@ -24,10 +24,6 @@ import java.io.IOException;
 import java.security.AccessControlException;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
 
 import org.apache.log4j.Logger;
 import org.apache.qpid.server.configuration.updater.TaskExecutor;
@@ -48,17 +44,12 @@ import com.sleepycat.je.rep.utilint.ServiceDispatcher.ServiceConnectFailedExcept
 /**
  * Represents a remote replication node in a BDB group.
  */
-public class RemoteReplicationNode extends AbstractAdapter implements ReplicationNode, Runnable
+public class RemoteReplicationNode extends AbstractAdapter implements ReplicationNode
 {
     private static final Logger LOGGER = Logger.getLogger(RemoteReplicationNode.class);
 
-    //TODO: add attributes for setting the intervals below
-    private static final int DEFAULT_SOCKET_TIMEOUT = 10000;
-    private static final long DEFAULT_STATE_UPDATE_INTERVAL = 1000; //TODO: set it to bigger value
-
-    // TODO: needs to be shared between all remote nodes
-    private final ScheduledExecutorService _updateStateExecutor;
     private final com.sleepycat.je.rep.ReplicationNode _replicationNode;
+    private final VirtualHost _virtualHost;
     private final String _hostPort;
     private final String _groupName;
 
@@ -73,18 +64,7 @@ public class RemoteReplicationNode extends AbstractAdapter implements Replicatio
         _groupName = groupName;
         _hostPort = replicationNode.getHostName() + ":" + replicationNode.getPort();
         _replicationNode = replicationNode;
-        _updateStateExecutor = Executors.newScheduledThreadPool(1, new ThreadFactory()
-        {
-            @Override
-            public Thread newThread(Runnable r)
-            {
-                return new Thread(r, "Remote node state updater " + getName() + "-" + getAttribute(GROUP_NAME));
-            }
-        });
-
-        //TODO: add attribute for update interval
-        long stateUpdateInterval = DEFAULT_STATE_UPDATE_INTERVAL;
-        _updateStateExecutor.schedule(this, stateUpdateInterval, TimeUnit.MILLISECONDS);
+        _virtualHost = virtualHost;
     }
 
     @Override
@@ -166,7 +146,6 @@ public class RemoteReplicationNode extends AbstractAdapter implements Replicatio
     {
         if (desiredState == State.STOPPED)
         {
-            _updateStateExecutor.shutdown();
             return true;
         }
         else
@@ -206,9 +185,10 @@ public class RemoteReplicationNode extends AbstractAdapter implements Replicatio
         return super.getAttribute(name);
     }
 
-    private void updateNodeState()
+    public void updateNodeState()
     {
-        DbPing ping = new DbPing(_replicationNode, _groupName, DEFAULT_SOCKET_TIMEOUT);
+        Long monitorTimeout = (Long)_virtualHost.getAttribute(VirtualHost.REMOTE_REPLICATION_NODE_MONITOR_TIMEOUT);
+        DbPing ping = new DbPing(_replicationNode, _groupName, monitorTimeout.intValue());
         String oldRole = _role;
         long oldJoinTime = _joinTime;
         long oldTransactionId = _lastTransactionId;
@@ -247,9 +227,4 @@ public class RemoteReplicationNode extends AbstractAdapter implements Replicatio
         }
     }
 
-    @Override
-    public void run()
-    {
-        updateNodeState();
-    }
 }
diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNodeFactory.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNodeFactory.java
index 31b55d6574..e02c40009b 100644
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNodeFactory.java
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNodeFactory.java
@@ -23,4 +23,6 @@ package org.apache.qpid.server.store.berkeleydb.replication;
 public interface RemoteReplicationNodeFactory
 {
     RemoteReplicationNode create(com.sleepycat.je.rep.ReplicationNode jeNode, String groupName);
+
+    long getRemoteNodeMonitorInterval();
 }
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeTest.java
index 050d0c7747..ca06db5400 100644
--- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeTest.java
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeTest.java
@@ -109,6 +109,9 @@ public class ReplicatedEnvironmentFacadeTest extends EnvironmentFacadeTestCase
     public void setUp() throws Exception
     {
         super.setUp();
+
+        when(_virtualHost.getAttribute(VirtualHost.REMOTE_REPLICATION_NODE_MONITOR_INTERVAL)).thenReturn(100L);
+        when(_virtualHost.getAttribute(VirtualHost.REMOTE_REPLICATION_NODE_MONITOR_TIMEOUT)).thenReturn(100L);
     }
 
     @Override
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
index 48c35797a4..35235daf94 100644
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
@@ -91,6 +91,8 @@ public interface VirtualHost extends ConfiguredObject
     String CONFIG_PATH                          = "configPath";
 
     String QUIESCE_ON_MASTER_CHANGE             = "quiesceOnMasterChange";
+    String REMOTE_REPLICATION_NODE_MONITOR_INTERVAL = "remoteReplicationNodeMonitorInterval";
+    String REMOTE_REPLICATION_NODE_MONITOR_TIMEOUT = "remoteReplicationNodeMonitorTimeout";
 
     // Attributes
     public static final Collection<String> AVAILABLE_ATTRIBUTES =
@@ -126,7 +128,9 @@ public interface VirtualHost extends ConfiguredObject
                             QUEUE_ALERT_THRESHOLD_QUEUE_DEPTH_BYTES,
                             QUEUE_ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES,
                             CONFIG_PATH,
-                            QUIESCE_ON_MASTER_CHANGE));
+                            QUIESCE_ON_MASTER_CHANGE,
+                            REMOTE_REPLICATION_NODE_MONITOR_INTERVAL,
+                            REMOTE_REPLICATION_NODE_MONITOR_TIMEOUT));
 
     int CURRENT_CONFIG_VERSION = 3;
 
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java
index 57bf80a883..9fd1815b67 100644
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java
@@ -97,8 +97,22 @@ public final class VirtualHostAdapter extends AbstractAdapter implements Virtual
         put(STORE_TYPE, String.class);
         put(CONFIG_PATH, String.class);
         put(STATE, State.class);
+        put(REMOTE_REPLICATION_NODE_MONITOR_INTERVAL, Long.class);
+        put(REMOTE_REPLICATION_NODE_MONITOR_TIMEOUT, Long.class);
+        put(QUIESCE_ON_MASTER_CHANGE, Boolean.class);
     }});
 
+    private static final long DEFAULT_REMOTE_REPLICATION_NODE_MONITOR_INTERVAL = 10000L;
+    private static final long DEFAULT_REMOTE_REPLICATION_NODE_MONITOR_TIMEOUT = 1000L;
+
+    @SuppressWarnings("serial")
+    static final Map<String, Object> DEFAULTS = new HashMap<String, Object>()
+    {{
+        put(REMOTE_REPLICATION_NODE_MONITOR_INTERVAL, DEFAULT_REMOTE_REPLICATION_NODE_MONITOR_INTERVAL);
+        put(REMOTE_REPLICATION_NODE_MONITOR_TIMEOUT, DEFAULT_REMOTE_REPLICATION_NODE_MONITOR_TIMEOUT);
+        put(QUIESCE_ON_MASTER_CHANGE, false);
+    }};
+
     private org.apache.qpid.server.virtualhost.VirtualHost _virtualHost;
 
     private final Map<AMQConnectionModel, ConnectionAdapter> _connectionAdapters =
@@ -120,7 +134,7 @@ public final class VirtualHostAdapter extends AbstractAdapter implements Virtual
 
     public VirtualHostAdapter(UUID id, Map<String, Object> attributes, Broker broker, StatisticsGatherer brokerStatisticsGatherer, TaskExecutor taskExecutor)
     {
-        super(id, null, MapValueConverter.convert(attributes, ATTRIBUTE_TYPES, false), taskExecutor, false);
+        super(id, DEFAULTS, MapValueConverter.convert(attributes, ATTRIBUTE_TYPES, false), taskExecutor, false);
         _taskExecutor = taskExecutor;
         _broker = broker;
         _brokerStatisticsGatherer = brokerStatisticsGatherer;
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/util/DaemonThreadFactory.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/util/DaemonThreadFactory.java
new file mode 100644
index 0000000000..4f1f830fd0
--- /dev/null
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/util/DaemonThreadFactory.java
@@ -0,0 +1,40 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.util;
+
+import java.util.concurrent.ThreadFactory;
+
+public final class DaemonThreadFactory implements ThreadFactory
+{
+    private String _threadName;
+    public DaemonThreadFactory(String threadName)
+    {
+        _threadName = threadName;
+    }
+
+    @Override
+    public Thread newThread(Runnable r)
+    {
+        Thread thread = new Thread(r, _threadName);
+        thread.setDaemon(true);
+        return thread;
+    }
+}
\ No newline at end of file