helix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ka...@apache.org
Subject [3/3] git commit: [HELIX-392] Write a test to ensure that ZK connection loss is silent
Date Fri, 14 Mar 2014 17:21:16 GMT
[HELIX-392] Write a test to ensure that ZK connection loss is silent


Project: http://git-wip-us.apache.org/repos/asf/helix/repo
Commit: http://git-wip-us.apache.org/repos/asf/helix/commit/6e587b2d
Tree: http://git-wip-us.apache.org/repos/asf/helix/tree/6e587b2d
Diff: http://git-wip-us.apache.org/repos/asf/helix/diff/6e587b2d

Branch: refs/heads/helix-0.6.2-release
Commit: 6e587b2d1252fed54b42f6daefeee12242f22b1c
Parents: 5019d96
Author: Kanak Biscuitwala <kanak@apache.org>
Authored: Wed Feb 26 17:55:08 2014 -0800
Committer: Kanak Biscuitwala <kanak@apache.org>
Committed: Fri Mar 14 10:02:36 2014 -0700

----------------------------------------------------------------------
 .../TestCorrectnessOnConnectivityLoss.java      | 203 +++++++++++++++++++
 1 file changed, 203 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/helix/blob/6e587b2d/helix-core/src/test/java/org/apache/helix/integration/TestCorrectnessOnConnectivityLoss.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/TestCorrectnessOnConnectivityLoss.java
b/helix-core/src/test/java/org/apache/helix/integration/TestCorrectnessOnConnectivityLoss.java
new file mode 100644
index 0000000..3b44f2c
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/integration/TestCorrectnessOnConnectivityLoss.java
@@ -0,0 +1,203 @@
+package org.apache.helix.integration;
+
+/*
+ * 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.
+ */
+
+import java.util.Map;
+
+import org.I0Itec.zkclient.ZkServer;
+import org.apache.helix.HelixManager;
+import org.apache.helix.HelixManagerFactory;
+import org.apache.helix.InstanceType;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.TestHelper;
+import org.apache.helix.integration.manager.ClusterControllerManager;
+import org.apache.helix.model.IdealState.RebalanceMode;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelFactory;
+import org.apache.helix.participant.statemachine.StateModelInfo;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+import org.apache.helix.participant.statemachine.Transition;
+import org.apache.helix.spectator.RoutingTableProvider;
+import org.apache.helix.tools.ClusterStateVerifier;
+import org.apache.helix.tools.ClusterStateVerifier.BestPossAndExtViewZkVerifier;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.google.common.collect.Maps;
+
+public class TestCorrectnessOnConnectivityLoss {
+  private static final String ZK_ADDR = "localhost:2189";
+  private ZkServer _zkServer;
+  private String _clusterName;
+  private ClusterControllerManager _controller;
+
+  @BeforeMethod
+  public void beforeMethod() throws Exception {
+    _zkServer = TestHelper.startZkServer(ZK_ADDR);
+
+    String className = TestHelper.getTestClassName();
+    String methodName = TestHelper.getTestMethodName();
+    _clusterName = className + "_" + methodName;
+    TestHelper.setupCluster(_clusterName, ZK_ADDR, 12918, // participant start port
+        "localhost", // participant host
+        "resource", // resource name prefix
+        1, // number of resources
+        1, // number of partitions
+        1, // number of participants
+        1, // number of replicas
+        "OnlineOffline", // state model
+        RebalanceMode.FULL_AUTO, // automatic assignment
+        true); // rebalance
+
+    _controller = new ClusterControllerManager(ZK_ADDR, _clusterName, "controller0");
+    _controller.connect();
+  }
+
+  @Test
+  public void testParticipant() throws Exception {
+    Map<String, Integer> stateReachedCounts = Maps.newHashMap();
+    HelixManager participant =
+        HelixManagerFactory.getZKHelixManager(_clusterName, "localhost_12918",
+            InstanceType.PARTICIPANT, ZK_ADDR);
+    participant.getStateMachineEngine().registerStateModelFactory("OnlineOffline",
+        new MyStateModelFactory(stateReachedCounts));
+    participant.connect();
+
+    Thread.sleep(1000);
+
+    // Ensure that the external view coalesces
+    boolean result =
+        ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,
+            _clusterName));
+    Assert.assertTrue(result);
+
+    // Ensure that there was only one state transition
+    Assert.assertEquals(stateReachedCounts.size(), 1);
+    Assert.assertTrue(stateReachedCounts.containsKey("ONLINE"));
+    Assert.assertEquals(stateReachedCounts.get("ONLINE").intValue(), 1);
+
+    // Now let's stop the ZK server; this should do nothing
+    TestHelper.stopZkServer(_zkServer);
+    Thread.sleep(1000);
+
+    // Verify no change
+    Assert.assertEquals(stateReachedCounts.size(), 1);
+    Assert.assertTrue(stateReachedCounts.containsKey("ONLINE"));
+    Assert.assertEquals(stateReachedCounts.get("ONLINE").intValue(), 1);
+  }
+
+  @SuppressWarnings("deprecation")
+  @Test
+  public void testSpectator() throws Exception {
+    Map<String, Integer> stateReachedCounts = Maps.newHashMap();
+    HelixManager participant =
+        HelixManagerFactory.getZKHelixManager(_clusterName, "localhost_12918",
+            InstanceType.PARTICIPANT, ZK_ADDR);
+    participant.getStateMachineEngine().registerStateModelFactory("OnlineOffline",
+        new MyStateModelFactory(stateReachedCounts));
+    participant.connect();
+
+    RoutingTableProvider routingTableProvider = new RoutingTableProvider();
+    HelixManager spectator =
+        HelixManagerFactory.getZKHelixManager(_clusterName, "spectator", InstanceType.SPECTATOR,
+            ZK_ADDR);
+    spectator.connect();
+    spectator.addConfigChangeListener(routingTableProvider);
+    spectator.addExternalViewChangeListener(routingTableProvider);
+    Thread.sleep(1000);
+
+    // Now let's stop the ZK server; this should do nothing
+    TestHelper.stopZkServer(_zkServer);
+    Thread.sleep(1000);
+
+    // Verify routing table still works
+    Assert.assertEquals(routingTableProvider.getInstances("resource0", "ONLINE").size(),
1);
+    Assert.assertEquals(routingTableProvider.getInstances("resource0", "OFFLINE").size(),
0);
+  }
+
+  @AfterMethod
+  public void afterMethod() throws Exception {
+    TestHelper.stopZkServer(_zkServer);
+  }
+
+  @StateModelInfo(initialState = "OFFLINE", states = {
+      "MASTER", "SLAVE", "OFFLINE", "ERROR"
+  })
+  public static class MyStateModel extends StateModel {
+    private final Map<String, Integer> _counts;
+
+    public MyStateModel(Map<String, Integer> counts) {
+      _counts = counts;
+    }
+
+    @Transition(to = "ONLINE", from = "OFFLINE")
+    public void onBecomeOnlineFromOffline(Message message, NotificationContext context) {
+      incrementCount(message.getToState());
+    }
+
+    @Transition(to = "OFFLINE", from = "ONLINE")
+    public void onBecomeOfflineFromOnline(Message message, NotificationContext context) {
+      incrementCount(message.getToState());
+    }
+
+    @Transition(to = "DROPPED", from = "OFFLINE")
+    public void onBecomeDroppedFromOffline(Message message, NotificationContext context)
{
+      incrementCount(message.getToState());
+    }
+
+    @Transition(to = "OFFLINE", from = "ERROR")
+    public void onBecomeOfflineFromError(Message message, NotificationContext context) {
+      incrementCount(message.getToState());
+    }
+
+    @Transition(to = "DROPPED", from = "ERROR")
+    public void onBecomeDroppedFromError(Message message, NotificationContext context) {
+      incrementCount(message.getToState());
+    }
+
+    @Override
+    public void rollbackOnError(Message message, NotificationContext context,
+        StateTransitionError error) {
+      incrementCount("rollback");
+    }
+
+    private synchronized void incrementCount(String toState) {
+      int current = (_counts.containsKey(toState)) ? _counts.get(toState) : 0;
+      _counts.put(toState, current + 1);
+    }
+  }
+
+  public static class MyStateModelFactory extends StateModelFactory<MyStateModel> {
+
+    private final Map<String, Integer> _counts;
+
+    public MyStateModelFactory(Map<String, Integer> counts) {
+      _counts = counts;
+    }
+
+    @Override
+    public MyStateModel createNewStateModel(String partitionId) {
+      return new MyStateModel(_counts);
+    }
+  }
+}


Mime
View raw message