finalized the changes and removed old classes
Project: http://git-wip-us.apache.org/repos/asf/curator/repo Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/36ddd58a Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/36ddd58a Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/36ddd58a Branch: refs/heads/CURATOR-265 Commit: 36ddd58a31045addfd9d984353956f7a99c09221 Parents: d42ef17 Author: randgalt <randg...@apache.org> Authored: Fri Sep 25 21:49:12 2015 -0500 Committer: randgalt <randg...@apache.org> Committed: Fri Sep 25 21:49:12 2015 -0500 ---------------------------------------------------------------------- .../framework/api/ACLBackgroundPathable.java | 25 -- .../api/ACLVersionBackgroundPathable.java | 25 -- .../api/BackgroundPathableQuietly.java | 23 - .../api/CreateModalPathAndBytesable.java | 25 -- .../api/IncrementalReconfigBuilder.java | 33 -- .../api/JoinAddStatConfigEnsembleable.java | 34 -- .../api/JoinLeaveStatConfigEnsembleable.java | 34 -- .../framework/api/JoinStatConfigurable.java | 30 -- .../api/LeaveAddStatConfigEnsembleable.java | 33 -- .../framework/api/SyncReconfigurable.java | 30 -- .../framework/imps/TestReconfiguration.java | 58 ++- .../framework/imps/TestReconfigurationX.java | 425 ------------------- .../org/apache/curator/test/TestingCluster.java | 16 +- 13 files changed, 49 insertions(+), 742 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/curator/blob/36ddd58a/curator-framework/src/main/java/org/apache/curator/framework/api/ACLBackgroundPathable.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/ACLBackgroundPathable.java b/curator-framework/src/main/java/org/apache/curator/framework/api/ACLBackgroundPathable.java deleted file mode 100644 index d63281d..0000000 --- a/curator-framework/src/main/java/org/apache/curator/framework/api/ACLBackgroundPathable.java +++ /dev/null @@ -1,25 +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.api; - -public interface ACLBackgroundPathable<T> extends - ACLable<BackgroundPathable<T>>, - BackgroundPathable<T> -{ -} http://git-wip-us.apache.org/repos/asf/curator/blob/36ddd58a/curator-framework/src/main/java/org/apache/curator/framework/api/ACLVersionBackgroundPathable.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/ACLVersionBackgroundPathable.java b/curator-framework/src/main/java/org/apache/curator/framework/api/ACLVersionBackgroundPathable.java deleted file mode 100644 index bc8e6bf..0000000 --- a/curator-framework/src/main/java/org/apache/curator/framework/api/ACLVersionBackgroundPathable.java +++ /dev/null @@ -1,25 +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.api; - -public interface ACLVersionBackgroundPathable<T> extends - ACLable<Versionable<BackgroundPathable<T>>>, - Versionable<BackgroundPathable<T>> -{ -} http://git-wip-us.apache.org/repos/asf/curator/blob/36ddd58a/curator-framework/src/main/java/org/apache/curator/framework/api/BackgroundPathableQuietly.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/BackgroundPathableQuietly.java b/curator-framework/src/main/java/org/apache/curator/framework/api/BackgroundPathableQuietly.java deleted file mode 100644 index 13202aa..0000000 --- a/curator-framework/src/main/java/org/apache/curator/framework/api/BackgroundPathableQuietly.java +++ /dev/null @@ -1,23 +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.api; - -public interface BackgroundPathableQuietly<T> extends BackgroundPathable<T>, Quietly<BackgroundPathable<T>> -{ -} http://git-wip-us.apache.org/repos/asf/curator/blob/36ddd58a/curator-framework/src/main/java/org/apache/curator/framework/api/CreateModalPathAndBytesable.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/CreateModalPathAndBytesable.java b/curator-framework/src/main/java/org/apache/curator/framework/api/CreateModalPathAndBytesable.java deleted file mode 100644 index 94bfe7e..0000000 --- a/curator-framework/src/main/java/org/apache/curator/framework/api/CreateModalPathAndBytesable.java +++ /dev/null @@ -1,25 +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.api; - -public interface CreateModalPathAndBytesable<T> extends - CreateModable<PathAndBytesable<T>>, - PathAndBytesable<T> -{ -} http://git-wip-us.apache.org/repos/asf/curator/blob/36ddd58a/curator-framework/src/main/java/org/apache/curator/framework/api/IncrementalReconfigBuilder.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/IncrementalReconfigBuilder.java b/curator-framework/src/main/java/org/apache/curator/framework/api/IncrementalReconfigBuilder.java deleted file mode 100644 index 0ad6426..0000000 --- a/curator-framework/src/main/java/org/apache/curator/framework/api/IncrementalReconfigBuilder.java +++ /dev/null @@ -1,33 +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.api; - -/** - * An incremental reconfiguration builder. - * This builder has access only to the incremental reconfiguration methods join and leave, so that we prevent - * mixing concepts that can't be used together. - * @param <T> - */ -public interface IncrementalReconfigBuilder<T> extends - Joinable<IncrementalReconfigBuilder<T>>, - Leaveable<IncrementalReconfigBuilder<T>>, - DataCallbackable<AsyncReconfigurable>, - Statable<SyncReconfigurable> { - -} http://git-wip-us.apache.org/repos/asf/curator/blob/36ddd58a/curator-framework/src/main/java/org/apache/curator/framework/api/JoinAddStatConfigEnsembleable.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/JoinAddStatConfigEnsembleable.java b/curator-framework/src/main/java/org/apache/curator/framework/api/JoinAddStatConfigEnsembleable.java deleted file mode 100644 index 4356ba7..0000000 --- a/curator-framework/src/main/java/org/apache/curator/framework/api/JoinAddStatConfigEnsembleable.java +++ /dev/null @@ -1,34 +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.api; - -/** - * An incremental reconfiguration builder. - * This builder has access only to the incremental reconfiguration methods joining and leaving, so that we prevent - * mixing concepts that can't be used together. - */ -public interface JoinAddStatConfigEnsembleable extends - Joinable<AddStatConfigEnsembleable>, - Addable<JoinStatConfigurable>, - ConfigureEnsembleable, - Statable<ConfigureEnsembleable> -{ - -} http://git-wip-us.apache.org/repos/asf/curator/blob/36ddd58a/curator-framework/src/main/java/org/apache/curator/framework/api/JoinLeaveStatConfigEnsembleable.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/JoinLeaveStatConfigEnsembleable.java b/curator-framework/src/main/java/org/apache/curator/framework/api/JoinLeaveStatConfigEnsembleable.java deleted file mode 100644 index fac16a9..0000000 --- a/curator-framework/src/main/java/org/apache/curator/framework/api/JoinLeaveStatConfigEnsembleable.java +++ /dev/null @@ -1,34 +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.api; - -/** - * An incremental reconfiguration builder. - * This builder has access only to the incremental reconfiguration methods joining and leaving, so that we prevent - * mixing concepts that can't be used together. - */ -public interface JoinLeaveStatConfigEnsembleable extends - Joinable<LeaveStatConfigEnsembleable>, - Leaveable<JoinStatConfigEnsembleable>, - ConfigureEnsembleable, - Statable<ConfigureEnsembleable> -{ - -} http://git-wip-us.apache.org/repos/asf/curator/blob/36ddd58a/curator-framework/src/main/java/org/apache/curator/framework/api/JoinStatConfigurable.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/JoinStatConfigurable.java b/curator-framework/src/main/java/org/apache/curator/framework/api/JoinStatConfigurable.java deleted file mode 100644 index 18713e4..0000000 --- a/curator-framework/src/main/java/org/apache/curator/framework/api/JoinStatConfigurable.java +++ /dev/null @@ -1,30 +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.api; - -/** - * An incremental reconfiguration builder. - * This builder has access only to the incremental reconfiguration methods joining and leaving, so that we prevent - * mixing concepts that can't be used together. - */ -public interface JoinStatConfigurable extends - Joinable<ConfigureEnsembleable> -{ -} http://git-wip-us.apache.org/repos/asf/curator/blob/36ddd58a/curator-framework/src/main/java/org/apache/curator/framework/api/LeaveAddStatConfigEnsembleable.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/LeaveAddStatConfigEnsembleable.java b/curator-framework/src/main/java/org/apache/curator/framework/api/LeaveAddStatConfigEnsembleable.java deleted file mode 100644 index b5125dc..0000000 --- a/curator-framework/src/main/java/org/apache/curator/framework/api/LeaveAddStatConfigEnsembleable.java +++ /dev/null @@ -1,33 +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.api; - -/** - * An non-incremental reconfiguration builder. - * This builder has access only to the non-incremental reconfiguration methods withMembers, so that we prevent - * mixing concepts that can't be used together. - */ -public interface LeaveAddStatConfigEnsembleable extends - Leaveable<AddStatConfigEnsembleable>, - Addable<LeaveStatConfigEnsembleable>, - ConfigureEnsembleable, - Statable<ConfigureEnsembleable> -{ -} http://git-wip-us.apache.org/repos/asf/curator/blob/36ddd58a/curator-framework/src/main/java/org/apache/curator/framework/api/SyncReconfigurable.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/SyncReconfigurable.java b/curator-framework/src/main/java/org/apache/curator/framework/api/SyncReconfigurable.java deleted file mode 100644 index bd7b96b..0000000 --- a/curator-framework/src/main/java/org/apache/curator/framework/api/SyncReconfigurable.java +++ /dev/null @@ -1,30 +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.api; - -public interface SyncReconfigurable { - - /** - * Sets the configuration version to use. - * @param config The version of the configuration. - * @return The configuration data. - * @throws Exception - */ - byte[] fromConfig(long config) throws Exception; -} http://git-wip-us.apache.org/repos/asf/curator/blob/36ddd58a/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 99e5a2e..37be4f1 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 @@ -25,10 +25,11 @@ import org.apache.curator.framework.CuratorFrameworkFactory; 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.retry.RetryOneTime; +import org.apache.curator.retry.ExponentialBackoffRetry; import org.apache.curator.test.BaseClassForTests; import org.apache.curator.test.InstanceSpec; import org.apache.curator.test.TestingCluster; +import org.apache.curator.test.TestingZooKeeperServer; import org.apache.curator.test.Timing; import org.apache.curator.utils.CloseableUtils; import org.apache.zookeeper.WatchedEvent; @@ -51,6 +52,7 @@ import java.util.concurrent.CountDownLatch; public class TestReconfiguration extends BaseClassForTests { + private final Timing timing = new Timing(); private TestingCluster cluster; @BeforeMethod @@ -146,7 +148,7 @@ public class TestReconfiguration extends BaseClassForTests @Test public void testBasicGetConfig() throws Exception { - try ( CuratorFramework client = CuratorFrameworkFactory.newClient(cluster.getConnectString(), new RetryOneTime(1)) ) + try ( CuratorFramework client = newClient()) { client.start(); QuorumVerifier quorumVerifier = toQuorumVerifier(client.getConfig().forEnsemble()); @@ -158,8 +160,7 @@ public class TestReconfiguration extends BaseClassForTests @Test public void testAdd() throws Exception { - Timing timing = new Timing(); - try ( CuratorFramework client = CuratorFrameworkFactory.newClient(cluster.getConnectString(), new RetryOneTime(1)) ) + try ( CuratorFramework client = newClient()) { client.start(); @@ -167,7 +168,7 @@ public class TestReconfiguration extends BaseClassForTests assertConfig(oldConfig, cluster.getInstances()); CountDownLatch latch = setChangeWaiter(client); - try ( TestingCluster newCluster = new TestingCluster(1, false) ) + try ( TestingCluster newCluster = new TestingCluster(TestingCluster.makeSpecs(1, false)) ) { newCluster.start(); @@ -186,8 +187,7 @@ public class TestReconfiguration extends BaseClassForTests @Test public void testAddAsync() throws Exception { - Timing timing = new Timing(); - try ( CuratorFramework client = CuratorFrameworkFactory.newClient(cluster.getConnectString(), new RetryOneTime(1)) ) + try ( CuratorFramework client = newClient()) { client.start(); @@ -195,7 +195,7 @@ public class TestReconfiguration extends BaseClassForTests assertConfig(oldConfig, cluster.getInstances()); CountDownLatch latch = setChangeWaiter(client); - try ( TestingCluster newCluster = new TestingCluster(1, false) ) + try ( TestingCluster newCluster = new TestingCluster(TestingCluster.makeSpecs(1, false)) ) { newCluster.start(); @@ -227,8 +227,7 @@ public class TestReconfiguration extends BaseClassForTests @Test public void testAddAndRemove() throws Exception { - Timing timing = new Timing(); - try ( CuratorFramework client = CuratorFrameworkFactory.newClient(cluster.getConnectString(), new RetryOneTime(1)) ) + try ( CuratorFramework client = newClient()) { client.start(); @@ -237,7 +236,7 @@ public class TestReconfiguration extends BaseClassForTests CountDownLatch latch = setChangeWaiter(client); - try ( TestingCluster newCluster = new TestingCluster(1, false) ) + try ( TestingCluster newCluster = new TestingCluster(TestingCluster.makeSpecs(1, false)) ) { newCluster.start(); @@ -264,6 +263,43 @@ public class TestReconfiguration extends BaseClassForTests } } + @Test + public void testNewMembers() throws Exception + { + cluster.close(); + cluster = new TestingCluster(5); + List<TestingZooKeeperServer> servers = cluster.getServers(); + List<InstanceSpec> smallCluster = Lists.newArrayList(); + for ( int i = 0; i < 3; ++i ) // only start 3 of the 5 + { + TestingZooKeeperServer server = servers.get(i); + server.start(); + smallCluster.add(server.getInstanceSpec()); + } + + try ( CuratorFramework client = newClient()) + { + client.start(); + + QuorumVerifier oldConfig = toQuorumVerifier(client.getConfig().forEnsemble()); + Assert.assertEquals(cluster.getInstances().size(), 5); + assertConfig(oldConfig, cluster.getInstances()); + + CountDownLatch latch = setChangeWaiter(client); + + client.reconfig().withNewMembers(toReconfigSpec(smallCluster)).forEnsemble(); + + Assert.assertTrue(timing.awaitLatch(latch)); + QuorumVerifier newConfig = toQuorumVerifier(client.getConfig().forEnsemble()); + assertConfig(newConfig, smallCluster); + } + } + + private CuratorFramework newClient() + { + return CuratorFrameworkFactory.newClient(cluster.getConnectString(), timing.session(), timing.connection(), new ExponentialBackoffRetry(timing.forSleepingABit().milliseconds(), 3)); + } + private CountDownLatch setChangeWaiter(CuratorFramework client) throws Exception { final CountDownLatch latch = new CountDownLatch(1); http://git-wip-us.apache.org/repos/asf/curator/blob/36ddd58a/curator-framework/src/test/java/org/apache/curator/framework/imps/TestReconfigurationX.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/test/java/org/apache/curator/framework/imps/TestReconfigurationX.java b/curator-framework/src/test/java/org/apache/curator/framework/imps/TestReconfigurationX.java deleted file mode 100644 index 7554ddd..0000000 --- a/curator-framework/src/test/java/org/apache/curator/framework/imps/TestReconfigurationX.java +++ /dev/null @@ -1,425 +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.imps; - -import com.google.common.collect.ImmutableList; -import org.apache.curator.ensemble.EnsembleListener; -import org.apache.curator.ensemble.dynamic.DynamicEnsembleProvider; -import org.apache.curator.framework.CuratorFramework; -import org.apache.curator.framework.CuratorFrameworkFactory; -import org.apache.curator.framework.api.BackgroundCallback; -import org.apache.curator.framework.api.CuratorEvent; -import org.apache.curator.framework.ensemble.EnsembleTracker; -import org.apache.curator.retry.RetryOneTime; -import org.apache.curator.test.InstanceSpec; -import org.apache.curator.test.TestingCluster; -import org.apache.curator.test.Timing; -import org.apache.curator.utils.CloseableUtils; -import org.apache.zookeeper.data.Stat; -import org.apache.zookeeper.server.quorum.flexible.QuorumMaj; -import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier; -import org.testng.Assert; -import org.testng.annotations.AfterMethod; -import org.testng.annotations.BeforeMethod; -import org.testng.annotations.Test; -import java.io.IOException; -import java.io.StringReader; -import java.util.HashMap; -import java.util.Map; -import java.util.Properties; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeoutException; -import java.util.concurrent.atomic.AtomicReference; - -public class TestReconfigurationX -{ - private static final Timing timing = new Timing(); - private TestingCluster cluster; - private DynamicEnsembleProvider dynamicEnsembleProvider; - private WaitOnDelegateListener waitOnDelegateListener; - private EnsembleTracker ensembleTracker; - private CuratorFramework client; - - private String connectionString1to5; - private String connectionString2to5; - private String connectionString3to5; - - @BeforeMethod - public void setup() throws Exception - { - ImmutableList.Builder<InstanceSpec> builder = ImmutableList.builder(); - for ( int i = 1; i <= 5; ++i ) - { - builder.add(new InstanceSpec(null, -1, -1, -1, true, i, -1, -1)); - } - - cluster = new TestingCluster(builder.build()); - cluster.start(); - - connectionString1to5 = cluster.getConnectString(); - connectionString2to5 = getConnectionString(cluster, 2, 3, 4, 5); - connectionString3to5 = getConnectionString(cluster, 3, 4, 5); - - dynamicEnsembleProvider = new DynamicEnsembleProvider(connectionString1to5); - client = CuratorFrameworkFactory.builder() - .ensembleProvider(dynamicEnsembleProvider) - .retryPolicy(new RetryOneTime(1)) - .build(); - client.start(); - client.blockUntilConnected(); - - //Wrap around the dynamic ensemble provider, so that we can wait until it has received the event. - waitOnDelegateListener = new WaitOnDelegateListener(dynamicEnsembleProvider); - ensembleTracker = new EnsembleTracker(client); - ensembleTracker.getListenable().addListener(waitOnDelegateListener); - ensembleTracker.start(); - //Wait for the initial event. - waitOnDelegateListener.waitForEvent(); - } - - @AfterMethod - public void tearDown() throws IOException - { - CloseableUtils.closeQuietly(ensembleTracker); - CloseableUtils.closeQuietly(client); - CloseableUtils.closeQuietly(cluster); - } - - @Test - public void testSyncIncremental() throws Exception - { - Stat stat = new Stat(); - byte[] bytes = client.getConfig().storingStatIn(stat).forEnsemble(); - Assert.assertNotNull(bytes); - QuorumVerifier qv = getQuorumVerifier(bytes); - Assert.assertEquals(qv.getAllMembers().size(), 5); - String server1 = getServerString(qv, cluster, 1L); - String server2 = getServerString(qv, cluster, 2L); - - //Remove Servers - bytes = client.reconfig().leaving("1").storingStatIn(stat).fromConfig(qv.getVersion()).forEnsemble(); - qv = getQuorumVerifier(bytes); - Assert.assertEquals(qv.getAllMembers().size(), 4); - - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString2to5); - - bytes = client.reconfig().leaving("2").storingStatIn(stat).fromConfig(qv.getVersion()).forEnsemble(); - qv = getQuorumVerifier(bytes); - Assert.assertEquals(qv.getAllMembers().size(), 3); - - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString3to5); - - //Add Servers - bytes = client.reconfig().joining("server.2=" + server2).storingStatIn(stat).fromConfig(qv.getVersion()).forEnsemble(); - qv = getQuorumVerifier(bytes); - Assert.assertEquals(qv.getAllMembers().size(), 4); - - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString2to5); - - bytes = client.reconfig().joining("server.1=" + server1).storingStatIn(stat).fromConfig(qv.getVersion()).forEnsemble(); - qv = getQuorumVerifier(bytes); - Assert.assertEquals(qv.getAllMembers().size(), 5); - - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString1to5); - } - - @Test - public void testAsyncIncremental() throws Exception - { - final AtomicReference<byte[]> bytes = new AtomicReference<>(); - final BackgroundCallback callback = new BackgroundCallback() - { - @Override - public void processResult(CuratorFramework client, CuratorEvent event) throws Exception - { - bytes.set(event.getData()); - //We only need the latch on getConfig. - if ( event.getContext() != null ) - { - ((CountDownLatch)event.getContext()).countDown(); - } - } - - }; - - CountDownLatch latch = new CountDownLatch(1); - client.getConfig().inBackground(callback, latch).forEnsemble(); - Assert.assertTrue(timing.awaitLatch(latch)); - Assert.assertNotNull(bytes.get()); - QuorumVerifier qv = getQuorumVerifier(bytes.get()); - Assert.assertEquals(qv.getAllMembers().size(), 5); - String server1 = getServerString(qv, cluster, 1L); - String server2 = getServerString(qv, cluster, 2L); - - //Remove Servers - client.reconfig().inBackground(callback).leaving("1").fromConfig(qv.getVersion()).forEnsemble(); - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString2to5); - qv = getQuorumVerifier(bytes.get()); - Assert.assertEquals(qv.getAllMembers().size(), 4); - - client.reconfig().inBackground(callback, latch).leaving("2").fromConfig(qv.getVersion()).forEnsemble(); - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString3to5); - qv = getQuorumVerifier(bytes.get()); - Assert.assertEquals(qv.getAllMembers().size(), 3); - - //Add Servers - client.reconfig().inBackground(callback, latch).joining("server.2=" + server2).fromConfig(qv.getVersion()).forEnsemble(); - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString2to5); - qv = getQuorumVerifier(bytes.get()); - Assert.assertEquals(qv.getAllMembers().size(), 4); - - client.reconfig().inBackground(callback, latch).joining("server.1=" + server1).fromConfig(qv.getVersion()).forEnsemble(); - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString1to5); - qv = getQuorumVerifier(bytes.get()); - Assert.assertEquals(qv.getAllMembers().size(), 5); - } - - @Test - public void testSyncNonIncremental() throws Exception - { - Stat stat = new Stat(); - byte[] bytes = client.getConfig().storingStatIn(stat).forEnsemble(); - Assert.assertNotNull(bytes); - QuorumVerifier qv = getQuorumVerifier(bytes); - Assert.assertEquals(qv.getAllMembers().size(), 5); - String server1 = getServerString(qv, cluster, 1L); - String server2 = getServerString(qv, cluster, 2L); - String server3 = getServerString(qv, cluster, 3L); - String server4 = getServerString(qv, cluster, 4L); - String server5 = getServerString(qv, cluster, 5L); - - //Remove Servers - bytes = client.reconfig() - .withNewMembers("server.2=" + server2, - "server.3=" + server3, - "server.4=" + server4, - "server.5=" + server5) - .storingStatIn(stat).fromConfig(qv.getVersion()).forEnsemble(); - qv = getQuorumVerifier(bytes); - Assert.assertEquals(qv.getAllMembers().size(), 4); - - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString2to5); - - bytes = client.reconfig() - .withNewMembers("server.3=" + server3, - "server.4=" + server4, - "server.5=" + server5) - .storingStatIn(stat).fromConfig(qv.getVersion()).forEnsemble(); - - qv = getQuorumVerifier(bytes); - Assert.assertEquals(qv.getAllMembers().size(), 3); - - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString3to5); - - //Add Servers - bytes = client.reconfig() - .withNewMembers("server.2=" + server2, - "server.3=" + server3, - "server.4=" + server4, - "server.5=" + server5) - .storingStatIn(stat).fromConfig(qv.getVersion()).forEnsemble(); - qv = getQuorumVerifier(bytes); - Assert.assertEquals(qv.getAllMembers().size(), 4); - - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString2to5); - - bytes = client.reconfig() - .withNewMembers("server.1=" + server1, - "server.2=" + server2, - "server.3=" + server3, - "server.4=" + server4, - "server.5=" + server5) - .storingStatIn(stat).fromConfig(qv.getVersion()).forEnsemble(); - qv = getQuorumVerifier(bytes); - Assert.assertEquals(qv.getAllMembers().size(), 5); - - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString1to5); - } - - @Test - public void testAsyncNonIncremental() throws Exception - { - final AtomicReference<byte[]> bytes = new AtomicReference<>(); - final BackgroundCallback callback = new BackgroundCallback() - { - @Override - public void processResult(CuratorFramework client, CuratorEvent event) throws Exception - { - bytes.set(event.getData()); - ((CountDownLatch)event.getContext()).countDown(); - } - - }; - - CountDownLatch latch = new CountDownLatch(1); - client.getConfig().inBackground(callback, latch).forEnsemble(); - Assert.assertTrue(timing.awaitLatch(latch)); - Assert.assertNotNull(bytes.get()); - QuorumVerifier qv = getQuorumVerifier(bytes.get()); - Assert.assertEquals(qv.getAllMembers().size(), 5); - String server1 = getServerString(qv, cluster, 1L); - String server2 = getServerString(qv, cluster, 2L); - String server3 = getServerString(qv, cluster, 3L); - String server4 = getServerString(qv, cluster, 4L); - String server5 = getServerString(qv, cluster, 5L); - - //Remove Servers - client.reconfig().inBackground(callback, latch) - .withNewMembers("server.2=" + server2, - "server.3=" + server3, - "server.4=" + server4, - "server.5=" + server5) - .fromConfig(qv.getVersion()).forEnsemble(); - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString2to5); - qv = getQuorumVerifier(bytes.get()); - Assert.assertEquals(qv.getAllMembers().size(), 4); - - client.reconfig().inBackground(callback, latch) - .withNewMembers("server.3=" + server3, - "server.4=" + server4, - "server.5=" + server5) - .fromConfig(qv.getVersion()).forEnsemble(); - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString3to5); - qv = getQuorumVerifier(bytes.get()); - Assert.assertEquals(qv.getAllMembers().size(), 3); - - //Add Servers - client.reconfig().inBackground(callback, latch) - .withNewMembers("server.2=" + server2, - "server.3=" + server3, - "server.4=" + server4, - "server.5=" + server5) - .fromConfig(qv.getVersion()).forEnsemble(); - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString2to5); - qv = getQuorumVerifier(bytes.get()); - Assert.assertEquals(qv.getAllMembers().size(), 4); - - client.reconfig().inBackground(callback, latch) - .withNewMembers("server.1=" + server1, - "server.2=" + server2, - "server.3=" + server3, - "server.4=" + server4, - "server.5=" + server5) - .fromConfig(qv.getVersion()).forEnsemble(); - waitOnDelegateListener.waitForEvent(); - Assert.assertEquals(dynamicEnsembleProvider.getConnectionString(), connectionString1to5); - qv = getQuorumVerifier(bytes.get()); - Assert.assertEquals(qv.getAllMembers().size(), 5); - } - - private static QuorumVerifier getQuorumVerifier(byte[] bytes) throws Exception - { - Properties properties = new Properties(); - properties.load(new StringReader(new String(bytes))); - return new QuorumMaj(properties); - } - - private static InstanceSpec getInstance(TestingCluster cluster, int id) - { - for ( InstanceSpec spec : cluster.getInstances() ) - { - if ( spec.getServerId() == id ) - { - return spec; - } - } - throw new IllegalStateException("InstanceSpec with id:" + id + " not found"); - } - - private static String getServerString(QuorumVerifier qv, TestingCluster cluster, long id) throws Exception - { - String str = qv.getAllMembers().get(id).toString(); - //check if connection string is already there. - if ( str.contains(";") ) - { - return str; - } - else - { - return str + ";" + getInstance(cluster, (int)id).getConnectString(); - } - } - - private static String getConnectionString(TestingCluster cluster, long... ids) throws Exception - { - StringBuilder sb = new StringBuilder(); - Map<Long, InstanceSpec> specs = new HashMap<>(); - for ( InstanceSpec spec : cluster.getInstances() ) - { - specs.put((long)spec.getServerId(), spec); - } - for ( long id : ids ) - { - if ( sb.length() != 0 ) - { - sb.append(","); - } - sb.append(specs.get(id).getConnectString()); - } - return sb.toString(); - } - - //Simple EnsembleListener that can wait until the delegate handles the event. - private static class WaitOnDelegateListener implements EnsembleListener - { - private CountDownLatch latch = new CountDownLatch(1); - - private final EnsembleListener delegate; - - private WaitOnDelegateListener(EnsembleListener delegate) - { - this.delegate = delegate; - } - - @Override - public void connectionStringUpdated(String connectionString) - { - delegate.connectionStringUpdated(connectionString); - latch.countDown(); - } - - public void waitForEvent() throws InterruptedException, TimeoutException - { - if ( timing.awaitLatch(latch) ) - { - latch = new CountDownLatch(1); - } - else - { - throw new TimeoutException("Failed to receive event in time."); - } - } - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/curator/blob/36ddd58a/curator-test/src/main/java/org/apache/curator/test/TestingCluster.java ---------------------------------------------------------------------- diff --git a/curator-test/src/main/java/org/apache/curator/test/TestingCluster.java b/curator-test/src/main/java/org/apache/curator/test/TestingCluster.java index b8dada8..3d38fe1 100644 --- a/curator-test/src/main/java/org/apache/curator/test/TestingCluster.java +++ b/curator-test/src/main/java/org/apache/curator/test/TestingCluster.java @@ -52,18 +52,6 @@ public class TestingCluster implements Closeable } /** - * Creates an ensemble comprised of <code>n</code> servers. Each server will use - * a temp directory and random ports - * - * @param instanceQty number of servers to create in the ensemble - * @param resetServerIds if true, server Ids are reset first - */ - public TestingCluster(int instanceQty, boolean resetServerIds) - { - this(makeSpecs(instanceQty, resetServerIds)); - } - - /** * Creates an ensemble using the given server specs * * @param specs the server specs @@ -254,12 +242,12 @@ public class TestingCluster implements Closeable return null; } - private static Map<InstanceSpec, Collection<InstanceSpec>> makeSpecs(int instanceQty) + public static Map<InstanceSpec, Collection<InstanceSpec>> makeSpecs(int instanceQty) { return makeSpecs(instanceQty, true); } - private static Map<InstanceSpec, Collection<InstanceSpec>> makeSpecs(int instanceQty, boolean resetServerIds) + public static Map<InstanceSpec, Collection<InstanceSpec>> makeSpecs(int instanceQty, boolean resetServerIds) { if ( resetServerIds ) {
