curator-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From randg...@apache.org
Subject [1/7] curator git commit: 1. EnsembleTracker should always be on, it now is 2. Removed DynamicEnsembleProvider. This should not be optional. EnsembleTracker now always publishes config changes which will end up calling ZooKeeper.updateServerList() 3. Tes
Date Fri, 09 Oct 2015 00:12:18 GMT
Repository: curator
Updated Branches:
  refs/heads/CURATOR-3.0 0fe4d969f -> 779ff5ea4


1. EnsembleTracker should always be on, it now is
2. Removed DynamicEnsembleProvider. This should not be optional. EnsembleTracker now always
publishes config changes which will end up calling ZooKeeper.updateServerList()
3. Testing


Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/26364c61
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/26364c61
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/26364c61

Branch: refs/heads/CURATOR-3.0
Commit: 26364c6186fc7c09a9462557b1ca791e9aa70006
Parents: a7076bc
Author: randgalt <randgalt@apache.org>
Authored: Sat Sep 26 13:13:02 2015 -0500
Committer: randgalt <randgalt@apache.org>
Committed: Sat Sep 26 13:13:02 2015 -0500

----------------------------------------------------------------------
 .../org/apache/curator/ConnectionState.java     |  32 +++-
 .../java/org/apache/curator/HandleHolder.java   |   4 +-
 .../ClassicConnectionHandlingPolicy.java        |   4 +-
 .../connection/ConnectionHandlingPolicy.java    |   6 +-
 .../StandardConnectionHandlingPolicy.java       |   4 +-
 .../curator/ensemble/EnsembleListener.java      |  24 ---
 .../curator/ensemble/EnsembleProvider.java      |   2 +
 .../dynamic/DynamicEnsembleProvider.java        |  61 ------
 .../exhibitor/ExhibitorEnsembleProvider.java    |   6 +
 .../ensemble/fixed/FixedEnsembleProvider.java   |  13 +-
 .../framework/ensemble/EnsembleTracker.java     | 191 -------------------
 .../framework/imps/CuratorFrameworkImpl.java    |   7 +
 .../curator/framework/imps/EnsembleTracker.java | 150 +++++++++++++++
 .../framework/imps/TestReconfiguration.java     |  57 +++++-
 .../locks/TestInterProcessSemaphoreCluster.java |   5 +
 15 files changed, 263 insertions(+), 303 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/curator/blob/26364c61/curator-client/src/main/java/org/apache/curator/ConnectionState.java
----------------------------------------------------------------------
diff --git a/curator-client/src/main/java/org/apache/curator/ConnectionState.java b/curator-client/src/main/java/org/apache/curator/ConnectionState.java
index eea2ce0..4c1e6ad 100644
--- a/curator-client/src/main/java/org/apache/curator/ConnectionState.java
+++ b/curator-client/src/main/java/org/apache/curator/ConnectionState.java
@@ -199,12 +199,14 @@ class ConnectionState implements Watcher, Closeable
 
     private synchronized void checkTimeouts() throws Exception
     {
-        Callable<Boolean> hasNewConnectionString  = new Callable<Boolean>()
+        final AtomicReference<String> newConnectionString = new AtomicReference<>();
+        Callable<String> hasNewConnectionString = new Callable<String>()
         {
             @Override
-            public Boolean call()
+            public String call()
             {
-                return zooKeeper.hasNewConnectionString();
+                newConnectionString.set(zooKeeper.getNewConnectionString());
+                return newConnectionString.get();
             }
         };
         int lastNegotiatedSessionTimeoutMs = getLastNegotiatedSessionTimeoutMs();
@@ -220,7 +222,7 @@ class ConnectionState implements Watcher, Closeable
 
             case NEW_CONNECTION_STRING:
             {
-                handleNewConnectionString();
+                handleNewConnectionString(newConnectionString.get());
                 break;
             }
 
@@ -298,22 +300,34 @@ class ConnectionState implements Watcher, Closeable
         }
         }
 
-        if ( checkNewConnectionString && zooKeeper.hasNewConnectionString() )
+        if ( checkNewConnectionString )
         {
-            handleNewConnectionString();
+            String newConnectionString = zooKeeper.getNewConnectionString();
+            if ( newConnectionString != null )
+            {
+                handleNewConnectionString(newConnectionString);
+            }
         }
 
         return isConnected;
     }
 
-    private void handleNewConnectionString()
+    private void handleNewConnectionString(String newConnectionString)
     {
-        log.info("Connection string changed");
+        log.info("Connection string changed to: " + newConnectionString);
         tracer.get().addCount("connection-string-changed", 1);
 
         try
         {
-            reset();
+            ZooKeeper zooKeeper = this.zooKeeper.getZooKeeper();
+            if ( zooKeeper == null )
+            {
+                log.warn("Could not update the connection string because getZooKeeper() returned
null.");
+            }
+            else
+            {
+                zooKeeper.updateServerList(newConnectionString);
+            }
         }
         catch ( Exception e )
         {

http://git-wip-us.apache.org/repos/asf/curator/blob/26364c61/curator-client/src/main/java/org/apache/curator/HandleHolder.java
----------------------------------------------------------------------
diff --git a/curator-client/src/main/java/org/apache/curator/HandleHolder.java b/curator-client/src/main/java/org/apache/curator/HandleHolder.java
index 8652f0c..98b39ce 100644
--- a/curator-client/src/main/java/org/apache/curator/HandleHolder.java
+++ b/curator-client/src/main/java/org/apache/curator/HandleHolder.java
@@ -67,10 +67,10 @@ class HandleHolder
         return (helper != null) ? helper.getConnectionString() : null;
     }
 
-    boolean hasNewConnectionString() 
+    String getNewConnectionString()
     {
         String helperConnectionString = (helper != null) ? helper.getConnectionString() :
null;
-        return (helperConnectionString != null) && !ensembleProvider.getConnectionString().equals(helperConnectionString);
+        return ((helperConnectionString != null) && !ensembleProvider.getConnectionString().equals(helperConnectionString))
? helperConnectionString : null;
     }
 
     void closeAndClear() throws Exception

http://git-wip-us.apache.org/repos/asf/curator/blob/26364c61/curator-client/src/main/java/org/apache/curator/connection/ClassicConnectionHandlingPolicy.java
----------------------------------------------------------------------
diff --git a/curator-client/src/main/java/org/apache/curator/connection/ClassicConnectionHandlingPolicy.java
b/curator-client/src/main/java/org/apache/curator/connection/ClassicConnectionHandlingPolicy.java
index 8116308..f620ffb 100644
--- a/curator-client/src/main/java/org/apache/curator/connection/ClassicConnectionHandlingPolicy.java
+++ b/curator-client/src/main/java/org/apache/curator/connection/ClassicConnectionHandlingPolicy.java
@@ -56,14 +56,14 @@ public class ClassicConnectionHandlingPolicy implements ConnectionHandlingPolicy
     }
 
     @Override
-    public CheckTimeoutsResult checkTimeouts(Callable<Boolean> hasNewConnectionString,
long connectionStartMs, int sessionTimeoutMs, int connectionTimeoutMs) throws Exception
+    public CheckTimeoutsResult checkTimeouts(Callable<String> hasNewConnectionString,
long connectionStartMs, int sessionTimeoutMs, int connectionTimeoutMs) throws Exception
     {
         CheckTimeoutsResult result = CheckTimeoutsResult.NOP;
         int minTimeout = Math.min(sessionTimeoutMs, connectionTimeoutMs);
         long elapsed = System.currentTimeMillis() - connectionStartMs;
         if ( elapsed >= minTimeout )
         {
-            if ( hasNewConnectionString.call() )
+            if ( hasNewConnectionString.call() != null )
             {
                 result = CheckTimeoutsResult.NEW_CONNECTION_STRING;
             }

http://git-wip-us.apache.org/repos/asf/curator/blob/26364c61/curator-client/src/main/java/org/apache/curator/connection/ConnectionHandlingPolicy.java
----------------------------------------------------------------------
diff --git a/curator-client/src/main/java/org/apache/curator/connection/ConnectionHandlingPolicy.java
b/curator-client/src/main/java/org/apache/curator/connection/ConnectionHandlingPolicy.java
index c47577d..8f6a147 100644
--- a/curator-client/src/main/java/org/apache/curator/connection/ConnectionHandlingPolicy.java
+++ b/curator-client/src/main/java/org/apache/curator/connection/ConnectionHandlingPolicy.java
@@ -99,13 +99,13 @@ public interface ConnectionHandlingPolicy
      * Check timeouts. NOTE: this method is only called when an attempt to access to the
ZooKeeper instances
      * is made and the connection has not completed.
      *
-     * @param hasNewConnectionString proc to call to check if there is a new connection string.
Important: the internal state is cleared after
-     *                               this is called so you MUST handle the new connection
string if <tt>true</tt> is returned
+     * @param getNewConnectionString proc to call to check if there is a new connection string.
Important: the internal state is cleared after
+     *                               this is called so you MUST handle the new connection
string if non null is returned
      * @param connectionStartMs the epoch/ms time that the connection was first initiated
      * @param sessionTimeoutMs the configured/negotiated session timeout in milliseconds
      * @param connectionTimeoutMs the configured connection timeout in milliseconds
      * @return result
      * @throws Exception errors
      */
-    CheckTimeoutsResult checkTimeouts(Callable<Boolean> hasNewConnectionString, long
connectionStartMs, int sessionTimeoutMs, int connectionTimeoutMs) throws Exception;
+    CheckTimeoutsResult checkTimeouts(Callable<String> getNewConnectionString, long
connectionStartMs, int sessionTimeoutMs, int connectionTimeoutMs) throws Exception;
 }

http://git-wip-us.apache.org/repos/asf/curator/blob/26364c61/curator-client/src/main/java/org/apache/curator/connection/StandardConnectionHandlingPolicy.java
----------------------------------------------------------------------
diff --git a/curator-client/src/main/java/org/apache/curator/connection/StandardConnectionHandlingPolicy.java
b/curator-client/src/main/java/org/apache/curator/connection/StandardConnectionHandlingPolicy.java
index 9f311de..6995815 100644
--- a/curator-client/src/main/java/org/apache/curator/connection/StandardConnectionHandlingPolicy.java
+++ b/curator-client/src/main/java/org/apache/curator/connection/StandardConnectionHandlingPolicy.java
@@ -76,9 +76,9 @@ public class StandardConnectionHandlingPolicy implements ConnectionHandlingPolic
     }
 
     @Override
-    public CheckTimeoutsResult checkTimeouts(Callable<Boolean> hasNewConnectionString,
long connectionStartMs, int sessionTimeoutMs, int connectionTimeoutMs) throws Exception
+    public CheckTimeoutsResult checkTimeouts(Callable<String> hasNewConnectionString,
long connectionStartMs, int sessionTimeoutMs, int connectionTimeoutMs) throws Exception
     {
-        if ( hasNewConnectionString.call() )
+        if ( hasNewConnectionString.call() != null )
         {
             return CheckTimeoutsResult.NEW_CONNECTION_STRING;
         }

http://git-wip-us.apache.org/repos/asf/curator/blob/26364c61/curator-client/src/main/java/org/apache/curator/ensemble/EnsembleListener.java
----------------------------------------------------------------------
diff --git a/curator-client/src/main/java/org/apache/curator/ensemble/EnsembleListener.java
b/curator-client/src/main/java/org/apache/curator/ensemble/EnsembleListener.java
deleted file mode 100644
index 8f963cd..0000000
--- a/curator-client/src/main/java/org/apache/curator/ensemble/EnsembleListener.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/**
- * 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.curator.ensemble;
-
-public interface EnsembleListener {
-
-    void connectionStringUpdated(String connectionString);
-}

http://git-wip-us.apache.org/repos/asf/curator/blob/26364c61/curator-client/src/main/java/org/apache/curator/ensemble/EnsembleProvider.java
----------------------------------------------------------------------
diff --git a/curator-client/src/main/java/org/apache/curator/ensemble/EnsembleProvider.java
b/curator-client/src/main/java/org/apache/curator/ensemble/EnsembleProvider.java
index b118294..c03726f 100644
--- a/curator-client/src/main/java/org/apache/curator/ensemble/EnsembleProvider.java
+++ b/curator-client/src/main/java/org/apache/curator/ensemble/EnsembleProvider.java
@@ -51,4 +51,6 @@ public interface EnsembleProvider extends Closeable
      * @throws IOException errors
      */
     public void         close() throws IOException;
+
+    public void setConnectionString(String connectionString);
 }

http://git-wip-us.apache.org/repos/asf/curator/blob/26364c61/curator-client/src/main/java/org/apache/curator/ensemble/dynamic/DynamicEnsembleProvider.java
----------------------------------------------------------------------
diff --git a/curator-client/src/main/java/org/apache/curator/ensemble/dynamic/DynamicEnsembleProvider.java
b/curator-client/src/main/java/org/apache/curator/ensemble/dynamic/DynamicEnsembleProvider.java
deleted file mode 100644
index 70b755f..0000000
--- a/curator-client/src/main/java/org/apache/curator/ensemble/dynamic/DynamicEnsembleProvider.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/**
- * 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.curator.ensemble.dynamic;
-
-import com.google.common.base.Preconditions;
-import org.apache.curator.ensemble.EnsembleListener;
-import org.apache.curator.ensemble.EnsembleProvider;
-
-import java.io.IOException;
-import java.util.concurrent.atomic.AtomicReference;
-
-public class DynamicEnsembleProvider implements EnsembleProvider, EnsembleListener {
-
-    private final AtomicReference<String> connectionString = new AtomicReference<String>();
-
-    /**
-     * The connection string to use
-     *
-     * @param connectionString connection string
-     */
-    public DynamicEnsembleProvider(String connectionString)
-    {
-        this.connectionString.set(Preconditions.checkNotNull(connectionString, "connectionString
cannot be null"));
-    }
-
-    @Override
-    public void start() throws Exception {
-        // NOP
-    }
-
-    @Override
-    public String getConnectionString() {
-        return connectionString.get();
-    }
-
-    @Override
-    public void close() throws IOException {
-        // NOP
-    }
-
-    @Override
-    public void connectionStringUpdated(String connectionString) {
-        this.connectionString.set(connectionString);
-    }
-}

http://git-wip-us.apache.org/repos/asf/curator/blob/26364c61/curator-client/src/main/java/org/apache/curator/ensemble/exhibitor/ExhibitorEnsembleProvider.java
----------------------------------------------------------------------
diff --git a/curator-client/src/main/java/org/apache/curator/ensemble/exhibitor/ExhibitorEnsembleProvider.java
b/curator-client/src/main/java/org/apache/curator/ensemble/exhibitor/ExhibitorEnsembleProvider.java
index 02a01e5..4cbf5ee 100644
--- a/curator-client/src/main/java/org/apache/curator/ensemble/exhibitor/ExhibitorEnsembleProvider.java
+++ b/curator-client/src/main/java/org/apache/curator/ensemble/exhibitor/ExhibitorEnsembleProvider.java
@@ -145,6 +145,12 @@ public class ExhibitorEnsembleProvider implements EnsembleProvider
         return connectionString.get();
     }
 
+    @Override
+    public void setConnectionString(String connectionString)
+    {
+        log.info("setConnectionString received. Ignoring. " + connectionString);
+    }
+
     @VisibleForTesting
     protected void poll()
     {

http://git-wip-us.apache.org/repos/asf/curator/blob/26364c61/curator-client/src/main/java/org/apache/curator/ensemble/fixed/FixedEnsembleProvider.java
----------------------------------------------------------------------
diff --git a/curator-client/src/main/java/org/apache/curator/ensemble/fixed/FixedEnsembleProvider.java
b/curator-client/src/main/java/org/apache/curator/ensemble/fixed/FixedEnsembleProvider.java
index 411c712..159497d 100644
--- a/curator-client/src/main/java/org/apache/curator/ensemble/fixed/FixedEnsembleProvider.java
+++ b/curator-client/src/main/java/org/apache/curator/ensemble/fixed/FixedEnsembleProvider.java
@@ -21,13 +21,14 @@ package org.apache.curator.ensemble.fixed;
 import com.google.common.base.Preconditions;
 import org.apache.curator.ensemble.EnsembleProvider;
 import java.io.IOException;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * Standard ensemble provider that wraps a fixed connection string
  */
 public class FixedEnsembleProvider implements EnsembleProvider
 {
-    private final String connectionString;
+    private final AtomicReference<String> connectionString = new AtomicReference<>();
 
     /**
      * The connection string to use
@@ -36,7 +37,7 @@ public class FixedEnsembleProvider implements EnsembleProvider
      */
     public FixedEnsembleProvider(String connectionString)
     {
-        this.connectionString = Preconditions.checkNotNull(connectionString, "connectionString
cannot be null");
+        this.connectionString.set(Preconditions.checkNotNull(connectionString, "connectionString
cannot be null"));
     }
 
     @Override
@@ -52,8 +53,14 @@ public class FixedEnsembleProvider implements EnsembleProvider
     }
 
     @Override
+    public void setConnectionString(String connectionString)
+    {
+        this.connectionString.set(connectionString);
+    }
+
+    @Override
     public String getConnectionString()
     {
-        return connectionString;
+        return connectionString.get();
     }
 }

http://git-wip-us.apache.org/repos/asf/curator/blob/26364c61/curator-framework/src/main/java/org/apache/curator/framework/ensemble/EnsembleTracker.java
----------------------------------------------------------------------
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/ensemble/EnsembleTracker.java
b/curator-framework/src/main/java/org/apache/curator/framework/ensemble/EnsembleTracker.java
deleted file mode 100644
index 375e1f0..0000000
--- a/curator-framework/src/main/java/org/apache/curator/framework/ensemble/EnsembleTracker.java
+++ /dev/null
@@ -1,191 +0,0 @@
-/**
- * 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.curator.framework.ensemble;
-
-import com.google.common.base.Function;
-import com.google.common.base.Preconditions;
-import org.apache.curator.ensemble.EnsembleListener;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.api.BackgroundCallback;
-import org.apache.curator.framework.api.CuratorEvent;
-import org.apache.curator.framework.api.CuratorWatcher;
-import org.apache.curator.framework.listen.ListenerContainer;
-import org.apache.curator.framework.state.ConnectionState;
-import org.apache.curator.framework.state.ConnectionStateListener;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.server.quorum.QuorumPeer;
-import org.apache.zookeeper.server.quorum.flexible.QuorumMaj;
-import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.io.ByteArrayInputStream;
-import java.io.Closeable;
-import java.io.IOException;
-import java.util.Properties;
-import java.util.concurrent.atomic.AtomicReference;
-
-/**
- * Tracks changes to the ensemble and notifies registered {@link org.apache.curator.ensemble.EnsembleListener}
instances.
- */
-public class EnsembleTracker implements Closeable
-{
-    private final Logger log = LoggerFactory.getLogger(getClass());
-    private final CuratorFramework client;
-    private final AtomicReference<State> state = new AtomicReference<>(State.LATENT);
-    private final ListenerContainer<EnsembleListener> listeners = new ListenerContainer<>();
-    private final ConnectionStateListener connectionStateListener = new ConnectionStateListener()
-    {
-        @Override
-        public void stateChanged(CuratorFramework client, ConnectionState newState)
-        {
-            if ( (newState == ConnectionState.CONNECTED) || (newState == ConnectionState.RECONNECTED)
)
-            {
-                try
-                {
-                    reset();
-                }
-                catch ( Exception e )
-                {
-                    log.error("Trying to reset after reconnection", e);
-                }
-            }
-        }
-    };
-
-    private final CuratorWatcher watcher = new CuratorWatcher()
-    {
-        @Override
-        public void process(WatchedEvent event) throws Exception
-        {
-            if ( event.getType() == Watcher.Event.EventType.NodeDataChanged )
-            {
-                reset();
-            }
-        }
-    };
-
-    private enum State
-    {
-        LATENT,
-        STARTED,
-        CLOSED
-    }
-
-    private final BackgroundCallback backgroundCallback = new BackgroundCallback()
-    {
-        @Override
-        public void processResult(CuratorFramework client, CuratorEvent event) throws Exception
-        {
-            processBackgroundResult(event);
-        }
-    };
-
-    public EnsembleTracker(CuratorFramework client)
-    {
-        this.client = client;
-    }
-
-    public void start() throws Exception
-    {
-        Preconditions.checkState(state.compareAndSet(State.LATENT, State.STARTED), "Cannot
be started more than once");
-        client.getConnectionStateListenable().addListener(connectionStateListener);
-        reset();
-    }
-
-    @Override
-    public void close() throws IOException
-    {
-        if ( state.compareAndSet(State.STARTED, State.CLOSED) )
-        {
-            listeners.clear();
-        }
-        client.getConnectionStateListenable().removeListener(connectionStateListener);
-    }
-
-    /**
-     * Return the ensemble listenable
-     *
-     * @return listenable
-     */
-    public ListenerContainer<EnsembleListener> getListenable()
-    {
-        Preconditions.checkState(state.get() != State.CLOSED, "Closed");
-
-        return listeners;
-    }
-
-    private void reset() throws Exception
-    {
-        client.getConfig().usingWatcher(watcher).inBackground(backgroundCallback).forEnsemble();
-    }
-
-    private void processBackgroundResult(CuratorEvent event) throws Exception
-    {
-        switch ( event.getType() )
-        {
-            case GET_CONFIG:
-            {
-                if ( event.getResultCode() == KeeperException.Code.OK.intValue() )
-                {
-                    processConfigData(event.getData());
-                }
-            }
-        }
-    }
-
-    private void processConfigData(byte[] data) throws Exception
-    {
-        Properties properties = new Properties();
-        properties.load(new ByteArrayInputStream(data));
-        QuorumVerifier qv = new QuorumMaj(properties);
-        StringBuilder sb = new StringBuilder();
-        for ( QuorumPeer.QuorumServer server : qv.getAllMembers().values() )
-        {
-            if ( sb.length() != 0 )
-            {
-                sb.append(",");
-            }
-            sb.append(server.clientAddr.getAddress().getHostAddress()).append(":").append(server.clientAddr.getPort());
-        }
-
-        final String connectionString = sb.toString();
-        listeners.forEach
-            (
-                new Function<EnsembleListener, Void>()
-                {
-                    @Override
-                    public Void apply(EnsembleListener listener)
-                    {
-                        try
-                        {
-                            listener.connectionStringUpdated(connectionString);
-                        }
-                        catch ( Exception e )
-                        {
-                            log.error("Calling listener", e);
-                        }
-                        return null;
-                    }
-                }
-            );
-    }
-}

http://git-wip-us.apache.org/repos/asf/curator/blob/26364c61/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java
----------------------------------------------------------------------
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java
b/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java
index da9067d..f2f578c 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java
@@ -88,6 +88,7 @@ public class CuratorFrameworkImpl implements CuratorFramework
     private final ConnectionStateErrorPolicy connectionStateErrorPolicy;
     private final AtomicLong currentInstanceIndex = new AtomicLong(-1);
     private final InternalConnectionHandler internalConnectionHandler;
+    private final EnsembleTracker ensembleTracker;
 
     private volatile ExecutorService executorService;
     private final AtomicBoolean logAsErrorConnectionErrors = new AtomicBoolean(false);
@@ -150,6 +151,8 @@ public class CuratorFrameworkImpl implements CuratorFramework
         failedDeleteManager = new FailedDeleteManager(this);
         failedRemoveWatcherManager = new FailedRemoveWatchManager(this);
         namespaceFacadeCache = new NamespaceFacadeCache(this);
+
+        ensembleTracker = new EnsembleTracker(this, builder.getEnsembleProvider());
     }
 
     private List<AuthInfo> buildAuths(CuratorFrameworkFactory.Builder builder)
@@ -217,6 +220,7 @@ public class CuratorFrameworkImpl implements CuratorFramework
         useContainerParentsIfAvailable = parent.useContainerParentsIfAvailable;
         connectionStateErrorPolicy = parent.connectionStateErrorPolicy;
         internalConnectionHandler = parent.internalConnectionHandler;
+        ensembleTracker = null;
     }
 
     @Override
@@ -306,6 +310,8 @@ public class CuratorFrameworkImpl implements CuratorFramework
                     return null;
                 }
             });
+
+            ensembleTracker.start();
         }
         catch ( Exception e )
         {
@@ -351,6 +357,7 @@ public class CuratorFrameworkImpl implements CuratorFramework
                 }
             }
 
+            ensembleTracker.close();
             listeners.clear();
             unhandledErrorListeners.clear();
             connectionStateManager.close();

http://git-wip-us.apache.org/repos/asf/curator/blob/26364c61/curator-framework/src/main/java/org/apache/curator/framework/imps/EnsembleTracker.java
----------------------------------------------------------------------
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/EnsembleTracker.java
b/curator-framework/src/main/java/org/apache/curator/framework/imps/EnsembleTracker.java
new file mode 100644
index 0000000..d8092fe
--- /dev/null
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/EnsembleTracker.java
@@ -0,0 +1,150 @@
+/**
+ * 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.curator.framework.imps;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import org.apache.curator.ensemble.EnsembleProvider;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.api.BackgroundCallback;
+import org.apache.curator.framework.api.CuratorEvent;
+import org.apache.curator.framework.api.CuratorEventType;
+import org.apache.curator.framework.api.CuratorWatcher;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.curator.framework.state.ConnectionStateListener;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.server.quorum.QuorumPeer;
+import org.apache.zookeeper.server.quorum.flexible.QuorumMaj;
+import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.ByteArrayInputStream;
+import java.io.Closeable;
+import java.util.Arrays;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicReference;
+
+@VisibleForTesting
+public class EnsembleTracker implements Closeable
+{
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    private final CuratorFramework client;
+    private final EnsembleProvider ensembleProvider;
+    private final AtomicReference<State> state = new AtomicReference<>(State.LATENT);
+    private final ConnectionStateListener connectionStateListener = new ConnectionStateListener()
+    {
+        @Override
+        public void stateChanged(CuratorFramework client, ConnectionState newState)
+        {
+            if ( (newState == ConnectionState.CONNECTED) || (newState == ConnectionState.RECONNECTED)
)
+            {
+                try
+                {
+                    reset();
+                }
+                catch ( Exception e )
+                {
+                    log.error("Trying to reset after reconnection", e);
+                }
+            }
+        }
+    };
+
+    private final CuratorWatcher watcher = new CuratorWatcher()
+    {
+        @Override
+        public void process(WatchedEvent event) throws Exception
+        {
+            if ( event.getType() == Watcher.Event.EventType.NodeDataChanged )
+            {
+                reset();
+            }
+        }
+    };
+
+    private enum State
+    {
+        LATENT,
+        STARTED,
+        CLOSED
+    }
+
+    EnsembleTracker(CuratorFramework client, EnsembleProvider ensembleProvider)
+    {
+        this.client = client;
+        this.ensembleProvider = ensembleProvider;
+    }
+
+    public void start() throws Exception
+    {
+        Preconditions.checkState(state.compareAndSet(State.LATENT, State.STARTED), "Cannot
be started more than once");
+        client.getConnectionStateListenable().addListener(connectionStateListener);
+        reset();
+    }
+
+    @Override
+    public void close()
+    {
+        client.getConnectionStateListenable().removeListener(connectionStateListener);
+    }
+
+    private void reset() throws Exception
+    {
+        BackgroundCallback backgroundCallback = new BackgroundCallback()
+        {
+            @Override
+            public void processResult(CuratorFramework client, CuratorEvent event) throws
Exception
+            {
+                if ( event.getType() == CuratorEventType.GET_CONFIG )
+                {
+                    processConfigData(event.getData());
+                }
+            }
+        };
+        client.getConfig().usingWatcher(watcher).inBackground(backgroundCallback).forEnsemble();
+    }
+
+    @VisibleForTesting
+    public static String configToConnectionString(byte[] data) throws Exception
+    {
+        Properties properties = new Properties();
+        properties.load(new ByteArrayInputStream(data));
+        QuorumVerifier qv = new QuorumMaj(properties);
+        StringBuilder sb = new StringBuilder();
+        for ( QuorumPeer.QuorumServer server : qv.getAllMembers().values() )
+        {
+            if ( sb.length() != 0 )
+            {
+                sb.append(",");
+            }
+            sb.append(server.clientAddr.getAddress().getHostAddress()).append(":").append(server.clientAddr.getPort());
+        }
+
+        return sb.toString();
+    }
+
+    private void processConfigData(byte[] data) throws Exception
+    {
+        log.info("New config event received: " + Arrays.toString(data));
+        String connectionString = configToConnectionString(data);
+        ensembleProvider.setConnectionString(connectionString);
+    }
+}

http://git-wip-us.apache.org/repos/asf/curator/blob/26364c61/curator-framework/src/test/java/org/apache/curator/framework/imps/TestReconfiguration.java
----------------------------------------------------------------------
diff --git a/curator-framework/src/test/java/org/apache/curator/framework/imps/TestReconfiguration.java
b/curator-framework/src/test/java/org/apache/curator/framework/imps/TestReconfiguration.java
index e7d2229..101360a 100644
--- a/curator-framework/src/test/java/org/apache/curator/framework/imps/TestReconfiguration.java
+++ b/curator-framework/src/test/java/org/apache/curator/framework/imps/TestReconfiguration.java
@@ -20,6 +20,7 @@
 package org.apache.curator.framework.imps;
 
 import com.google.common.collect.Lists;
+import org.apache.curator.ensemble.EnsembleProvider;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.api.BackgroundCallback;
@@ -43,17 +44,20 @@ import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 import java.io.ByteArrayInputStream;
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
 
 public class TestReconfiguration extends BaseClassForTests
 {
     private final Timing timing = new Timing();
     private TestingCluster cluster;
+    private EnsembleProvider ensembleProvider;
 
     @BeforeMethod
     @Override
@@ -72,6 +76,7 @@ public class TestReconfiguration extends BaseClassForTests
     public void teardown() throws Exception
     {
         CloseableUtils.closeQuietly(cluster);
+        ensembleProvider = null;
 
         super.teardown();
     }
@@ -151,9 +156,11 @@ public class TestReconfiguration extends BaseClassForTests
         try ( CuratorFramework client = newClient())
         {
             client.start();
-            QuorumVerifier quorumVerifier = toQuorumVerifier(client.getConfig().forEnsemble());
+            byte[] configData = client.getConfig().forEnsemble();
+            QuorumVerifier quorumVerifier = toQuorumVerifier(configData);
             System.out.println(quorumVerifier);
             assertConfig(quorumVerifier, cluster.getInstances());
+            Assert.assertEquals(EnsembleTracker.configToConnectionString(configData), ensembleProvider.getConnectionString());
         }
     }
 
@@ -176,10 +183,12 @@ public class TestReconfiguration extends BaseClassForTests
 
                 Assert.assertTrue(timing.awaitLatch(latch));
 
-                QuorumVerifier newConfig = toQuorumVerifier(client.getConfig().forEnsemble());
+                byte[] newConfigData = client.getConfig().forEnsemble();
+                QuorumVerifier newConfig = toQuorumVerifier(newConfigData);
                 List<InstanceSpec> newInstances = Lists.newArrayList(cluster.getInstances());
                 newInstances.addAll(newCluster.getInstances());
                 assertConfig(newConfig, newInstances);
+                Assert.assertEquals(EnsembleTracker.configToConnectionString(newConfigData),
ensembleProvider.getConnectionString());
             }
         }
     }
@@ -216,10 +225,12 @@ public class TestReconfiguration extends BaseClassForTests
                 Assert.assertTrue(timing.awaitLatch(callbackLatch));
                 Assert.assertTrue(timing.awaitLatch(latch));
 
-                QuorumVerifier newConfig = toQuorumVerifier(client.getConfig().forEnsemble());
+                byte[] newConfigData = client.getConfig().forEnsemble();
+                QuorumVerifier newConfig = toQuorumVerifier(newConfigData);
                 List<InstanceSpec> newInstances = Lists.newArrayList(cluster.getInstances());
                 newInstances.addAll(newCluster.getInstances());
                 assertConfig(newConfig, newInstances);
+                Assert.assertEquals(EnsembleTracker.configToConnectionString(newConfigData),
ensembleProvider.getConnectionString());
             }
         }
     }
@@ -254,11 +265,13 @@ public class TestReconfiguration extends BaseClassForTests
 
                 Assert.assertTrue(timing.awaitLatch(latch));
 
-                QuorumVerifier newConfig = toQuorumVerifier(client.getConfig().forEnsemble());
+                byte[] newConfigData = client.getConfig().forEnsemble();
+                QuorumVerifier newConfig = toQuorumVerifier(newConfigData);
                 ArrayList<InstanceSpec> newInstances = Lists.newArrayList(oldInstances);
                 newInstances.addAll(instances);
                 newInstances.remove(removeSpec);
                 assertConfig(newConfig, newInstances);
+                Assert.assertEquals(EnsembleTracker.configToConnectionString(newConfigData),
ensembleProvider.getConnectionString());
             }
         }
     }
@@ -290,15 +303,47 @@ public class TestReconfiguration extends BaseClassForTests
             client.reconfig().withNewMembers(toReconfigSpec(smallCluster)).forEnsemble();
 
             Assert.assertTrue(timing.awaitLatch(latch));
-            QuorumVerifier newConfig = toQuorumVerifier(client.getConfig().forEnsemble());
+            byte[] newConfigData = client.getConfig().forEnsemble();
+            QuorumVerifier newConfig = toQuorumVerifier(newConfigData);
             Assert.assertEquals(newConfig.getAllMembers().size(), 3);
             assertConfig(newConfig, smallCluster);
+            Assert.assertEquals(EnsembleTracker.configToConnectionString(newConfigData),
ensembleProvider.getConnectionString());
         }
     }
 
     private CuratorFramework newClient()
     {
-        return CuratorFrameworkFactory.newClient(cluster.getConnectString(), timing.session(),
timing.connection(), new ExponentialBackoffRetry(timing.forSleepingABit().milliseconds(),
3));
+        final AtomicReference<String> connectString = new AtomicReference<>(cluster.getConnectString());
+        ensembleProvider = new EnsembleProvider()
+        {
+            @Override
+            public void start() throws Exception
+            {
+            }
+
+            @Override
+            public String getConnectionString()
+            {
+                return connectString.get();
+            }
+
+            @Override
+            public void close() throws IOException
+            {
+            }
+
+            @Override
+            public void setConnectionString(String connectionString)
+            {
+                connectString.set(connectionString);
+            }
+        };
+        return CuratorFrameworkFactory.builder()
+            .ensembleProvider(ensembleProvider)
+            .sessionTimeoutMs(timing.session())
+            .connectionTimeoutMs(timing.connection())
+            .retryPolicy(new ExponentialBackoffRetry(timing.forSleepingABit().milliseconds(),
3))
+            .build();
     }
 
     private CountDownLatch setChangeWaiter(CuratorFramework client) throws Exception

http://git-wip-us.apache.org/repos/asf/curator/blob/26364c61/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessSemaphoreCluster.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessSemaphoreCluster.java
b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessSemaphoreCluster.java
index f4cb7bb..ee49288 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessSemaphoreCluster.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessSemaphoreCluster.java
@@ -65,6 +65,11 @@ public class TestInterProcessSemaphoreCluster
             final EnsembleProvider          provider = new EnsembleProvider()
             {
                 @Override
+                public void setConnectionString(String connectionString)
+                {
+                }
+
+                @Override
                 public void start() throws Exception
                 {
                 }


Mime
View raw message