You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by to...@apache.org on 2012/05/24 02:56:23 UTC
svn commit: r1342112 [2/2] - in
/hadoop/common/trunk/hadoop-common-project/hadoop-common: ./ dev-support/
src/main/bin/ src/main/docs/ src/main/java/
src/main/java/org/apache/hadoop/fs/ src/main/java/org/apache/hadoop/ha/
src/main/java/org/apache/hadoo...
Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/protocolPB/HAServiceProtocolServerSideTranslatorPB.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/protocolPB/HAServiceProtocolServerSideTranslatorPB.java?rev=1342112&r1=1342111&r2=1342112&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/protocolPB/HAServiceProtocolServerSideTranslatorPB.java (original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/protocolPB/HAServiceProtocolServerSideTranslatorPB.java Thu May 24 00:56:20 2012
@@ -19,12 +19,17 @@ package org.apache.hadoop.ha.protocolPB;
import java.io.IOException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.ha.HAServiceProtocol;
+import org.apache.hadoop.ha.HAServiceProtocol.StateChangeRequestInfo;
+import org.apache.hadoop.ha.HAServiceProtocol.RequestSource;
import org.apache.hadoop.ha.HAServiceStatus;
import org.apache.hadoop.ha.proto.HAServiceProtocolProtos.GetServiceStatusRequestProto;
import org.apache.hadoop.ha.proto.HAServiceProtocolProtos.GetServiceStatusResponseProto;
+import org.apache.hadoop.ha.proto.HAServiceProtocolProtos.HAStateChangeRequestInfoProto;
import org.apache.hadoop.ha.proto.HAServiceProtocolProtos.HAServiceStateProto;
import org.apache.hadoop.ha.proto.HAServiceProtocolProtos.MonitorHealthRequestProto;
import org.apache.hadoop.ha.proto.HAServiceProtocolProtos.MonitorHealthResponseProto;
@@ -56,6 +61,8 @@ public class HAServiceProtocolServerSide
TransitionToActiveResponseProto.newBuilder().build();
private static final TransitionToStandbyResponseProto TRANSITION_TO_STANDBY_RESP =
TransitionToStandbyResponseProto.newBuilder().build();
+ private static final Log LOG = LogFactory.getLog(
+ HAServiceProtocolServerSideTranslatorPB.class);
public HAServiceProtocolServerSideTranslatorPB(HAServiceProtocol server) {
this.server = server;
@@ -71,13 +78,33 @@ public class HAServiceProtocolServerSide
throw new ServiceException(e);
}
}
+
+ private StateChangeRequestInfo convert(HAStateChangeRequestInfoProto proto) {
+ RequestSource src;
+ switch (proto.getReqSource()) {
+ case REQUEST_BY_USER:
+ src = RequestSource.REQUEST_BY_USER;
+ break;
+ case REQUEST_BY_USER_FORCED:
+ src = RequestSource.REQUEST_BY_USER_FORCED;
+ break;
+ case REQUEST_BY_ZKFC:
+ src = RequestSource.REQUEST_BY_ZKFC;
+ break;
+ default:
+ LOG.warn("Unknown request source: " + proto.getReqSource());
+ src = null;
+ }
+
+ return new StateChangeRequestInfo(src);
+ }
@Override
public TransitionToActiveResponseProto transitionToActive(
RpcController controller, TransitionToActiveRequestProto request)
throws ServiceException {
try {
- server.transitionToActive();
+ server.transitionToActive(convert(request.getReqInfo()));
return TRANSITION_TO_ACTIVE_RESP;
} catch(IOException e) {
throw new ServiceException(e);
@@ -89,7 +116,7 @@ public class HAServiceProtocolServerSide
RpcController controller, TransitionToStandbyRequestProto request)
throws ServiceException {
try {
- server.transitionToStandby();
+ server.transitionToStandby(convert(request.getReqInfo()));
return TRANSITION_TO_STANDBY_RESP;
} catch(IOException e) {
throw new ServiceException(e);
Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/packages/templates/conf/hadoop-env.sh
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/packages/templates/conf/hadoop-env.sh?rev=1342112&r1=1342111&r2=1342112&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/packages/templates/conf/hadoop-env.sh (original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/packages/templates/conf/hadoop-env.sh Thu May 24 00:56:20 2012
@@ -53,6 +53,10 @@ export HADOOP_DATANODE_OPTS="-Dhadoop.se
export HADOOP_SECONDARYNAMENODE_OPTS="-Dhadoop.security.logger=${HADOOP_SECURITY_LOGGER:-INFO,RFAS} -Dhdfs.audit.logger=${HDFS_AUDIT_LOGGER:-INFO,NullAppender} $HADOOP_SECONDARYNAMENODE_OPTS"
+# The ZKFC does not need a large heap, and keeping it small avoids
+# any potential for long GC pauses
+export HADOOP_ZKFC_OPTS="-Xmx256m $HADOOP_ZKFC_OPTS"
+
# The following applies to multiple commands (fs, dfs, fsck, distcp etc)
export HADOOP_CLIENT_OPTS="-Xmx128m $HADOOP_CLIENT_OPTS"
#HADOOP_JAVA_PLATFORM_OPTS="-XX:-UsePerfData $HADOOP_JAVA_PLATFORM_OPTS"
Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/packages/templates/conf/hadoop-policy.xml
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/packages/templates/conf/hadoop-policy.xml?rev=1342112&r1=1342111&r2=1342112&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/packages/templates/conf/hadoop-policy.xml (original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/packages/templates/conf/hadoop-policy.xml Thu May 24 00:56:20 2012
@@ -223,6 +223,12 @@
<description>ACL for HAService protocol used by HAAdmin to manage the
active and stand-by states of namenode.</description>
</property>
+ <property>
+ <name>security.zkfc.protocol.acl</name>
+ <value>*</value>
+ <description>ACL for access to the ZK Failover Controller
+ </description>
+ </property>
<property>
<name>security.mrhs.client.protocol.acl</name>
Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/proto/HAServiceProtocol.proto
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/proto/HAServiceProtocol.proto?rev=1342112&r1=1342111&r2=1342112&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/proto/HAServiceProtocol.proto (original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/proto/HAServiceProtocol.proto Thu May 24 00:56:20 2012
@@ -27,6 +27,16 @@ enum HAServiceStateProto {
STANDBY = 2;
}
+enum HARequestSource {
+ REQUEST_BY_USER = 0;
+ REQUEST_BY_USER_FORCED = 1;
+ REQUEST_BY_ZKFC = 2;
+}
+
+message HAStateChangeRequestInfoProto {
+ required HARequestSource reqSource = 1;
+}
+
/**
* void request
*/
@@ -43,6 +53,7 @@ message MonitorHealthResponseProto {
* void request
*/
message TransitionToActiveRequestProto {
+ required HAStateChangeRequestInfoProto reqInfo = 1;
}
/**
@@ -55,6 +66,7 @@ message TransitionToActiveResponseProto
* void request
*/
message TransitionToStandbyRequestProto {
+ required HAStateChangeRequestInfoProto reqInfo = 1;
}
/**
Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml?rev=1342112&r1=1342111&r2=1342112&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml (original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml Thu May 24 00:56:20 2012
@@ -944,4 +944,63 @@
</description>
</property>
+<property>
+ <name>ha.zookeeper.quorum</name>
+ <description>
+ A list of ZooKeeper server addresses, separated by commas, that are
+ to be used by the ZKFailoverController in automatic failover.
+ </description>
+</property>
+
+<property>
+ <name>ha.zookeeper.session-timeout.ms</name>
+ <value>5000</value>
+ <description>
+ The session timeout to use when the ZKFC connects to ZooKeeper.
+ Setting this value to a lower value implies that server crashes
+ will be detected more quickly, but risks triggering failover too
+ aggressively in the case of a transient error or network blip.
+ </description>
+</property>
+
+<property>
+ <name>ha.zookeeper.parent-znode</name>
+ <value>/hadoop-ha</value>
+ <description>
+ The ZooKeeper znode under which the ZK failover controller stores
+ its information. Note that the nameservice ID is automatically
+ appended to this znode, so it is not normally necessary to
+ configure this, even in a federated environment.
+ </description>
+</property>
+
+<property>
+ <name>ha.zookeeper.acl</name>
+ <value>world:anyone:rwcda</value>
+ <description>
+ A comma-separated list of ZooKeeper ACLs to apply to the znodes
+ used by automatic failover. These ACLs are specified in the same
+ format as used by the ZooKeeper CLI.
+
+ If the ACL itself contains secrets, you may instead specify a
+ path to a file, prefixed with the '@' symbol, and the value of
+ this configuration will be loaded from within.
+ </description>
+</property>
+
+<property>
+ <name>ha.zookeeper.auth</name>
+ <value></value>
+ <description>
+ A comma-separated list of ZooKeeper authentications to add when
+ connecting to ZooKeeper. These are specified in the same format
+ as used by the "addauth" command in the ZK CLI. It is
+ important that the authentications specified here are sufficient
+ to access znodes with the ACL specified in ha.zookeeper.acl.
+
+ If the auths contain secrets, you may instead specify a
+ path to a file, prefixed with the '@' symbol, and the value of
+ this configuration will be loaded from within.
+ </description>
+</property>
</configuration>
Propchange: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/core/
------------------------------------------------------------------------------
Merged /hadoop/common/branches/HDFS-3042/hadoop-common-project/hadoop-common/src/test/core:r1306184-1342109
Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/ActiveStandbyElectorTestUtil.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/ActiveStandbyElectorTestUtil.java?rev=1342112&r1=1342111&r2=1342112&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/ActiveStandbyElectorTestUtil.java (original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/ActiveStandbyElectorTestUtil.java Thu May 24 00:56:20 2012
@@ -19,16 +19,25 @@ package org.apache.hadoop.ha;
import java.util.Arrays;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.test.MultithreadedTestUtil.TestContext;
+import org.apache.hadoop.util.StringUtils;
import org.apache.zookeeper.KeeperException.NoNodeException;
import org.apache.zookeeper.data.Stat;
import org.apache.zookeeper.server.ZooKeeperServer;
public abstract class ActiveStandbyElectorTestUtil {
+
+ private static final Log LOG = LogFactory.getLog(
+ ActiveStandbyElectorTestUtil.class);
+ private static final long LOG_INTERVAL_MS = 500;
public static void waitForActiveLockData(TestContext ctx,
ZooKeeperServer zks, String parentDir, byte[] activeData)
throws Exception {
+ long st = System.currentTimeMillis();
+ long lastPrint = st;
while (true) {
if (ctx != null) {
ctx.checkException();
@@ -42,10 +51,18 @@ public abstract class ActiveStandbyElect
Arrays.equals(activeData, data)) {
return;
}
+ if (System.currentTimeMillis() > lastPrint + LOG_INTERVAL_MS) {
+ LOG.info("Cur data: " + StringUtils.byteToHexString(data));
+ lastPrint = System.currentTimeMillis();
+ }
} catch (NoNodeException nne) {
if (activeData == null) {
return;
}
+ if (System.currentTimeMillis() > lastPrint + LOG_INTERVAL_MS) {
+ LOG.info("Cur data: no node");
+ lastPrint = System.currentTimeMillis();
+ }
}
Thread.sleep(50);
}
Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/DummyHAService.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/DummyHAService.java?rev=1342112&r1=1342111&r2=1342112&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/DummyHAService.java (original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/DummyHAService.java Thu May 24 00:56:20 2012
@@ -22,6 +22,8 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
import org.apache.hadoop.security.AccessControlException;
@@ -34,13 +36,19 @@ import com.google.common.collect.Lists;
* a mock implementation.
*/
class DummyHAService extends HAServiceTarget {
+ public static final Log LOG = LogFactory.getLog(DummyHAService.class);
+ private static final String DUMMY_FENCE_KEY = "dummy.fence.key";
volatile HAServiceState state;
HAServiceProtocol proxy;
+ ZKFCProtocol zkfcProxy = null;
NodeFencer fencer;
InetSocketAddress address;
boolean isHealthy = true;
boolean actUnreachable = false;
- boolean failToBecomeActive;
+ boolean failToBecomeActive, failToBecomeStandby, failToFence;
+
+ DummySharedResource sharedResource;
+ public int fenceCount = 0;
static ArrayList<DummyHAService> instances = Lists.newArrayList();
int index;
@@ -48,7 +56,14 @@ class DummyHAService extends HAServiceTa
DummyHAService(HAServiceState state, InetSocketAddress address) {
this.state = state;
this.proxy = makeMock();
- this.fencer = Mockito.mock(NodeFencer.class);
+ try {
+ Configuration conf = new Configuration();
+ conf.set(DUMMY_FENCE_KEY, DummyFencer.class.getName());
+ this.fencer = Mockito.spy(
+ NodeFencer.create(conf, DUMMY_FENCE_KEY));
+ } catch (BadFencingConfigurationException e) {
+ throw new RuntimeException(e);
+ }
this.address = address;
synchronized (instances) {
instances.add(this);
@@ -56,6 +71,10 @@ class DummyHAService extends HAServiceTa
}
}
+ public void setSharedResource(DummySharedResource rsrc) {
+ this.sharedResource = rsrc;
+ }
+
private HAServiceProtocol makeMock() {
return Mockito.spy(new MockHAProtocolImpl());
}
@@ -66,12 +85,24 @@ class DummyHAService extends HAServiceTa
}
@Override
+ public InetSocketAddress getZKFCAddress() {
+ return null;
+ }
+
+ @Override
public HAServiceProtocol getProxy(Configuration conf, int timeout)
throws IOException {
return proxy;
}
@Override
+ public ZKFCProtocol getZKFCProxy(Configuration conf, int timeout)
+ throws IOException {
+ assert zkfcProxy != null;
+ return zkfcProxy;
+ }
+
+ @Override
public NodeFencer getFencer() {
return fencer;
}
@@ -81,6 +112,11 @@ class DummyHAService extends HAServiceTa
}
@Override
+ public boolean isAutoFailoverEnabled() {
+ return true;
+ }
+
+ @Override
public String toString() {
return "DummyHAService #" + index;
}
@@ -101,20 +137,28 @@ class DummyHAService extends HAServiceTa
}
@Override
- public void transitionToActive() throws ServiceFailedException,
+ public void transitionToActive(StateChangeRequestInfo req) throws ServiceFailedException,
AccessControlException, IOException {
checkUnreachable();
if (failToBecomeActive) {
throw new ServiceFailedException("injected failure");
}
-
+ if (sharedResource != null) {
+ sharedResource.take(DummyHAService.this);
+ }
state = HAServiceState.ACTIVE;
}
@Override
- public void transitionToStandby() throws ServiceFailedException,
+ public void transitionToStandby(StateChangeRequestInfo req) throws ServiceFailedException,
AccessControlException, IOException {
checkUnreachable();
+ if (failToBecomeStandby) {
+ throw new ServiceFailedException("injected failure");
+ }
+ if (sharedResource != null) {
+ sharedResource.release(DummyHAService.this);
+ }
state = HAServiceState.STANDBY;
}
@@ -138,4 +182,26 @@ class DummyHAService extends HAServiceTa
public void close() throws IOException {
}
}
+
+ public static class DummyFencer implements FenceMethod {
+ public void checkArgs(String args) throws BadFencingConfigurationException {
+ }
+
+ @Override
+ public boolean tryFence(HAServiceTarget target, String args)
+ throws BadFencingConfigurationException {
+ LOG.info("tryFence(" + target + ")");
+ DummyHAService svc = (DummyHAService)target;
+ synchronized (svc) {
+ svc.fenceCount++;
+ }
+ if (svc.failToFence) {
+ LOG.info("Injected failure to fence");
+ return false;
+ }
+ svc.sharedResource.release(svc);
+ return true;
+ }
+ }
+
}
Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElector.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElector.java?rev=1342112&r1=1342111&r2=1342112&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElector.java (original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElector.java Thu May 24 00:56:20 2012
@@ -19,6 +19,7 @@
package org.apache.hadoop.ha;
import java.io.IOException;
+import java.util.Collections;
import java.util.List;
import org.apache.zookeeper.AsyncCallback;
@@ -40,6 +41,7 @@ import org.mockito.Mockito;
import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.ha.ActiveStandbyElector.ActiveStandbyElectorCallback;
import org.apache.hadoop.ha.ActiveStandbyElector.ActiveNotFoundException;
+import org.apache.hadoop.ha.HAZKUtil.ZKAuthInfo;
public class TestActiveStandbyElector {
@@ -51,9 +53,12 @@ public class TestActiveStandbyElector {
private ActiveStandbyElectorTester elector;
class ActiveStandbyElectorTester extends ActiveStandbyElector {
+ private int sleptFor = 0;
+
ActiveStandbyElectorTester(String hostPort, int timeout, String parent,
List<ACL> acl, ActiveStandbyElectorCallback app) throws IOException {
- super(hostPort, timeout, parent, acl, app);
+ super(hostPort, timeout, parent, acl,
+ Collections.<ZKAuthInfo>emptyList(), app);
}
@Override
@@ -61,6 +66,14 @@ public class TestActiveStandbyElector {
++count;
return mockZK;
}
+
+ @Override
+ protected void sleepFor(int ms) {
+ // don't sleep in unit tests! Instead, just record the amount of
+ // time slept
+ LOG.info("Would have slept for " + ms + "ms");
+ sleptFor += ms;
+ }
}
private static final String ZK_PARENT_NAME = "/parent/node";
@@ -147,6 +160,68 @@ public class TestActiveStandbyElector {
}
/**
+ * Verify that, when the callback fails to enter active state,
+ * the elector rejoins the election after sleeping for a short period.
+ */
+ @Test
+ public void testFailToBecomeActive() throws Exception {
+ mockNoPriorActive();
+ elector.joinElection(data);
+ Assert.assertEquals(0, elector.sleptFor);
+
+ Mockito.doThrow(new ServiceFailedException("failed to become active"))
+ .when(mockApp).becomeActive();
+ elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK,
+ ZK_LOCK_NAME);
+ // Should have tried to become active
+ Mockito.verify(mockApp).becomeActive();
+
+ // should re-join
+ Mockito.verify(mockZK, Mockito.times(2)).create(ZK_LOCK_NAME, data,
+ Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
+ Assert.assertEquals(2, count);
+ Assert.assertTrue(elector.sleptFor > 0);
+ }
+
+ /**
+ * Verify that, when the callback fails to enter active state, after
+ * a ZK disconnect (i.e from the StatCallback), that the elector rejoins
+ * the election after sleeping for a short period.
+ */
+ @Test
+ public void testFailToBecomeActiveAfterZKDisconnect() throws Exception {
+ mockNoPriorActive();
+ elector.joinElection(data);
+ Assert.assertEquals(0, elector.sleptFor);
+
+ elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
+ ZK_LOCK_NAME);
+ Mockito.verify(mockZK, Mockito.times(2)).create(ZK_LOCK_NAME, data,
+ Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
+
+ elector.processResult(Code.NODEEXISTS.intValue(), ZK_LOCK_NAME, mockZK,
+ ZK_LOCK_NAME);
+ verifyExistCall(1);
+
+ Stat stat = new Stat();
+ stat.setEphemeralOwner(1L);
+ Mockito.when(mockZK.getSessionId()).thenReturn(1L);
+
+ // Fake failure to become active from within the stat callback
+ Mockito.doThrow(new ServiceFailedException("fail to become active"))
+ .when(mockApp).becomeActive();
+ elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK, stat);
+ Mockito.verify(mockApp, Mockito.times(1)).becomeActive();
+
+ // should re-join
+ Mockito.verify(mockZK, Mockito.times(3)).create(ZK_LOCK_NAME, data,
+ Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
+ Assert.assertEquals(2, count);
+ Assert.assertTrue(elector.sleptFor > 0);
+ }
+
+
+ /**
* Verify that, if there is a record of a prior active node, the
* elector asks the application to fence it before becoming active.
*/
@@ -314,6 +389,7 @@ public class TestActiveStandbyElector {
*/
@Test
public void testStatNodeRetry() {
+ elector.joinElection(data);
elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
(Stat) null);
elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
@@ -334,6 +410,7 @@ public class TestActiveStandbyElector {
*/
@Test
public void testStatNodeError() {
+ elector.joinElection(data);
elector.processResult(Code.RUNTIMEINCONSISTENCY.intValue(), ZK_LOCK_NAME,
mockZK, (Stat) null);
Mockito.verify(mockApp, Mockito.times(0)).enterNeutralMode();
@@ -517,6 +594,8 @@ public class TestActiveStandbyElector {
*/
@Test
public void testQuitElection() throws Exception {
+ elector.joinElection(data);
+ Mockito.verify(mockZK, Mockito.times(0)).close();
elector.quitElection(true);
Mockito.verify(mockZK, Mockito.times(1)).close();
// no watches added
Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElectorRealZK.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElectorRealZK.java?rev=1342112&r1=1342111&r2=1342112&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElectorRealZK.java (original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElectorRealZK.java Thu May 24 00:56:20 2012
@@ -21,15 +21,16 @@ package org.apache.hadoop.ha;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
-import java.io.File;
+import java.util.Collections;
import java.util.UUID;
import org.apache.commons.logging.impl.Log4JLogger;
import org.apache.hadoop.ha.ActiveStandbyElector.ActiveStandbyElectorCallback;
+import org.apache.hadoop.ha.ActiveStandbyElector.State;
+import org.apache.hadoop.ha.HAZKUtil.ZKAuthInfo;
import org.apache.log4j.Level;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.server.ZooKeeperServer;
-import org.apache.zookeeper.test.ClientBase;
import org.junit.Test;
import org.mockito.AdditionalMatchers;
import org.mockito.Mockito;
@@ -39,7 +40,7 @@ import com.google.common.primitives.Ints
/**
* Test for {@link ActiveStandbyElector} using real zookeeper.
*/
-public class TestActiveStandbyElectorRealZK extends ClientBase {
+public class TestActiveStandbyElectorRealZK extends ClientBaseWithFixes {
static final int NUM_ELECTORS = 2;
static {
@@ -58,8 +59,6 @@ public class TestActiveStandbyElectorRea
@Override
public void setUp() throws Exception {
- // build.test.dir is used by zookeeper
- new File(System.getProperty("build.test.dir", "build")).mkdirs();
super.setUp();
zkServer = getServer(serverFactory);
@@ -68,7 +67,8 @@ public class TestActiveStandbyElectorRea
cbs[i] = Mockito.mock(ActiveStandbyElectorCallback.class);
appDatas[i] = Ints.toByteArray(i);
electors[i] = new ActiveStandbyElector(
- hostPort, 5000, PARENT_DIR, Ids.OPEN_ACL_UNSAFE, cbs[i]);
+ hostPort, 5000, PARENT_DIR, Ids.OPEN_ACL_UNSAFE,
+ Collections.<ZKAuthInfo>emptyList(), cbs[i]);
}
}
@@ -196,4 +196,63 @@ public class TestActiveStandbyElectorRea
checkFatalsAndReset();
}
+
+ @Test(timeout=15000)
+ public void testHandleSessionExpirationOfStandby() throws Exception {
+ // Let elector 0 be active
+ electors[0].ensureParentZNode();
+ electors[0].joinElection(appDatas[0]);
+ ZooKeeperServer zks = getServer(serverFactory);
+ ActiveStandbyElectorTestUtil.waitForActiveLockData(null,
+ zks, PARENT_DIR, appDatas[0]);
+ Mockito.verify(cbs[0], Mockito.timeout(1000)).becomeActive();
+ checkFatalsAndReset();
+
+ // Let elector 1 be standby
+ electors[1].joinElection(appDatas[1]);
+ ActiveStandbyElectorTestUtil.waitForElectorState(null, electors[1],
+ State.STANDBY);
+
+ LOG.info("========================== Expiring standby's session");
+ zks.closeSession(electors[1].getZKSessionIdForTests());
+
+ // Should enter neutral mode when disconnected
+ Mockito.verify(cbs[1], Mockito.timeout(1000)).enterNeutralMode();
+
+ // Should re-join the election and go back to STANDBY
+ ActiveStandbyElectorTestUtil.waitForElectorState(null, electors[1],
+ State.STANDBY);
+ checkFatalsAndReset();
+
+ LOG.info("========================== Quitting election");
+ electors[1].quitElection(false);
+
+ // Double check that we don't accidentally re-join the election
+ // by quitting elector 0 and ensuring elector 1 doesn't become active
+ electors[0].quitElection(false);
+
+ // due to receiving the "expired" event.
+ Thread.sleep(1000);
+ Mockito.verify(cbs[1], Mockito.never()).becomeActive();
+ ActiveStandbyElectorTestUtil.waitForActiveLockData(null,
+ zks, PARENT_DIR, null);
+
+ checkFatalsAndReset();
+ }
+
+ @Test(timeout=15000)
+ public void testDontJoinElectionOnDisconnectAndReconnect() throws Exception {
+ electors[0].ensureParentZNode();
+
+ stopServer();
+ ActiveStandbyElectorTestUtil.waitForElectorState(
+ null, electors[0], State.NEUTRAL);
+ startServer();
+ waitForServerUp(hostPort, CONNECTION_TIMEOUT);
+ // Have to sleep to allow time for the clients to reconnect.
+ Thread.sleep(2000);
+ Mockito.verify(cbs[0], Mockito.never()).becomeActive();
+ Mockito.verify(cbs[1], Mockito.never()).becomeActive();
+ checkFatalsAndReset();
+ }
}
Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestFailoverController.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestFailoverController.java?rev=1342112&r1=1342111&r2=1342112&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestFailoverController.java (original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestFailoverController.java Thu May 24 00:56:20 2012
@@ -27,11 +27,12 @@ import static org.mockito.Mockito.verify
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
+import org.apache.hadoop.ha.HAServiceProtocol.StateChangeRequestInfo;
+import org.apache.hadoop.ha.HAServiceProtocol.RequestSource;
import org.apache.hadoop.ha.TestNodeFencer.AlwaysSucceedFencer;
import org.apache.hadoop.ha.TestNodeFencer.AlwaysFailFencer;
import static org.apache.hadoop.ha.TestNodeFencer.setupFencer;
import org.apache.hadoop.security.AccessControlException;
-import org.apache.hadoop.test.MockitoUtil;
import org.junit.Test;
import org.mockito.Mockito;
@@ -118,7 +119,8 @@ public class TestFailoverController {
public void testFailoverToUnreadyService() throws Exception {
DummyHAService svc1 = new DummyHAService(HAServiceState.ACTIVE, svc1Addr);
DummyHAService svc2 = new DummyHAService(HAServiceState.STANDBY, svc2Addr);
- Mockito.doReturn(STATE_NOT_READY).when(svc2.proxy).getServiceStatus();
+ Mockito.doReturn(STATE_NOT_READY).when(svc2.proxy)
+ .getServiceStatus();
svc1.fencer = svc2.fencer = setupFencer(AlwaysSucceedFencer.class.getName());
try {
@@ -162,7 +164,7 @@ public class TestFailoverController {
public void testFailoverFromFaultyServiceSucceeds() throws Exception {
DummyHAService svc1 = new DummyHAService(HAServiceState.ACTIVE, svc1Addr);
Mockito.doThrow(new ServiceFailedException("Failed!"))
- .when(svc1.proxy).transitionToStandby();
+ .when(svc1.proxy).transitionToStandby(anyReqInfo());
DummyHAService svc2 = new DummyHAService(HAServiceState.STANDBY, svc2Addr);
svc1.fencer = svc2.fencer = setupFencer(AlwaysSucceedFencer.class.getName());
@@ -185,7 +187,7 @@ public class TestFailoverController {
public void testFailoverFromFaultyServiceFencingFailure() throws Exception {
DummyHAService svc1 = new DummyHAService(HAServiceState.ACTIVE, svc1Addr);
Mockito.doThrow(new ServiceFailedException("Failed!"))
- .when(svc1.proxy).transitionToStandby();
+ .when(svc1.proxy).transitionToStandby(anyReqInfo());
DummyHAService svc2 = new DummyHAService(HAServiceState.STANDBY, svc2Addr);
svc1.fencer = svc2.fencer = setupFencer(AlwaysFailFencer.class.getName());
@@ -284,7 +286,7 @@ public class TestFailoverController {
DummyHAService svc1 = spy(new DummyHAService(HAServiceState.ACTIVE, svc1Addr));
DummyHAService svc2 = new DummyHAService(HAServiceState.STANDBY, svc2Addr);
Mockito.doThrow(new ServiceFailedException("Failed!"))
- .when(svc2.proxy).transitionToActive();
+ .when(svc2.proxy).transitionToActive(anyReqInfo());
svc1.fencer = svc2.fencer = setupFencer(AlwaysSucceedFencer.class.getName());
try {
@@ -295,8 +297,8 @@ public class TestFailoverController {
}
// svc1 went standby then back to active
- verify(svc1.proxy).transitionToStandby();
- verify(svc1.proxy).transitionToActive();
+ verify(svc1.proxy).transitionToStandby(anyReqInfo());
+ verify(svc1.proxy).transitionToActive(anyReqInfo());
assertEquals(HAServiceState.ACTIVE, svc1.state);
assertEquals(HAServiceState.STANDBY, svc2.state);
}
@@ -306,7 +308,7 @@ public class TestFailoverController {
DummyHAService svc1 = new DummyHAService(HAServiceState.ACTIVE, svc1Addr);
DummyHAService svc2 = new DummyHAService(HAServiceState.STANDBY, svc2Addr);
Mockito.doThrow(new ServiceFailedException("Failed!"))
- .when(svc2.proxy).transitionToActive();
+ .when(svc2.proxy).transitionToActive(anyReqInfo());
svc1.fencer = svc2.fencer = setupFencer(AlwaysSucceedFencer.class.getName());
try {
@@ -327,7 +329,7 @@ public class TestFailoverController {
DummyHAService svc1 = new DummyHAService(HAServiceState.ACTIVE, svc1Addr);
DummyHAService svc2 = new DummyHAService(HAServiceState.STANDBY, svc2Addr);
Mockito.doThrow(new ServiceFailedException("Failed!"))
- .when(svc2.proxy).transitionToActive();
+ .when(svc2.proxy).transitionToActive(anyReqInfo());
svc1.fencer = svc2.fencer = setupFencer(AlwaysSucceedFencer.class.getName());
AlwaysSucceedFencer.fenceCalled = 0;
@@ -346,12 +348,16 @@ public class TestFailoverController {
assertSame(svc2, AlwaysSucceedFencer.fencedSvc);
}
+ private StateChangeRequestInfo anyReqInfo() {
+ return Mockito.<StateChangeRequestInfo>any();
+ }
+
@Test
public void testFailureToFenceOnFailbackFailsTheFailback() throws Exception {
DummyHAService svc1 = new DummyHAService(HAServiceState.ACTIVE, svc1Addr);
DummyHAService svc2 = new DummyHAService(HAServiceState.STANDBY, svc2Addr);
Mockito.doThrow(new IOException("Failed!"))
- .when(svc2.proxy).transitionToActive();
+ .when(svc2.proxy).transitionToActive(anyReqInfo());
svc1.fencer = svc2.fencer = setupFencer(AlwaysFailFencer.class.getName());
AlwaysFailFencer.fenceCalled = 0;
@@ -374,10 +380,10 @@ public class TestFailoverController {
public void testFailbackToFaultyServiceFails() throws Exception {
DummyHAService svc1 = new DummyHAService(HAServiceState.ACTIVE, svc1Addr);
Mockito.doThrow(new ServiceFailedException("Failed!"))
- .when(svc1.proxy).transitionToActive();
+ .when(svc1.proxy).transitionToActive(anyReqInfo());
DummyHAService svc2 = new DummyHAService(HAServiceState.STANDBY, svc2Addr);
Mockito.doThrow(new ServiceFailedException("Failed!"))
- .when(svc2.proxy).transitionToActive();
+ .when(svc2.proxy).transitionToActive(anyReqInfo());
svc1.fencer = svc2.fencer = setupFencer(AlwaysSucceedFencer.class.getName());
@@ -420,7 +426,8 @@ public class TestFailoverController {
private void doFailover(HAServiceTarget tgt1, HAServiceTarget tgt2,
boolean forceFence, boolean forceActive) throws FailoverFailedException {
- FailoverController fc = new FailoverController(conf);
+ FailoverController fc = new FailoverController(conf,
+ RequestSource.REQUEST_BY_USER);
fc.failover(tgt1, tgt2, forceFence, forceActive);
}
Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestZKFailoverController.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestZKFailoverController.java?rev=1342112&r1=1342111&r2=1342112&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestZKFailoverController.java (original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestZKFailoverController.java Thu May 24 00:56:20 2012
@@ -19,93 +19,58 @@ package org.apache.hadoop.ha;
import static org.junit.Assert.*;
-import java.io.File;
-import java.net.InetSocketAddress;
+import java.security.NoSuchAlgorithmException;
import org.apache.commons.logging.impl.Log4JLogger;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
+import org.apache.hadoop.ha.HAServiceProtocol.StateChangeRequestInfo;
import org.apache.hadoop.ha.HealthMonitor.State;
-import org.apache.hadoop.test.MultithreadedTestUtil;
-import org.apache.hadoop.test.MultithreadedTestUtil.TestContext;
-import org.apache.hadoop.test.MultithreadedTestUtil.TestingThread;
+import org.apache.hadoop.ha.MiniZKFCCluster.DummyZKFC;
+import org.apache.hadoop.test.GenericTestUtils;
import org.apache.log4j.Level;
-import org.apache.zookeeper.KeeperException.NoNodeException;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
-import org.apache.zookeeper.server.ZooKeeperServer;
-import org.apache.zookeeper.test.ClientBase;
+import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
-import com.google.common.primitives.Ints;
-
-public class TestZKFailoverController extends ClientBase {
+public class TestZKFailoverController extends ClientBaseWithFixes {
private Configuration conf;
- private DummyHAService svc1;
- private DummyHAService svc2;
- private TestContext ctx;
- private DummyZKFCThread thr1, thr2;
+ private MiniZKFCCluster cluster;
+ // Set up ZK digest-based credentials for the purposes of the tests,
+ // to make sure all of our functionality works with auth and ACLs
+ // present.
+ private static final String DIGEST_USER_PASS="test-user:test-password";
+ private static final String TEST_AUTH_GOOD =
+ "digest:" + DIGEST_USER_PASS;
+ private static final String DIGEST_USER_HASH;
static {
- ((Log4JLogger)ActiveStandbyElector.LOG).getLogger().setLevel(Level.ALL);
+ try {
+ DIGEST_USER_HASH = DigestAuthenticationProvider.generateDigest(
+ DIGEST_USER_PASS);
+ } catch (NoSuchAlgorithmException e) {
+ throw new RuntimeException(e);
+ }
}
+ private static final String TEST_ACL =
+ "digest:" + DIGEST_USER_HASH + ":rwcda";
- @Override
- public void setUp() throws Exception {
- // build.test.dir is used by zookeeper
- new File(System.getProperty("build.test.dir", "build")).mkdirs();
- super.setUp();
+ static {
+ ((Log4JLogger)ActiveStandbyElector.LOG).getLogger().setLevel(Level.ALL);
}
@Before
public void setupConfAndServices() {
conf = new Configuration();
- conf.set(ZKFailoverController.ZK_QUORUM_KEY, hostPort);
- // Fast check interval so tests run faster
- conf.setInt(CommonConfigurationKeys.HA_HM_CHECK_INTERVAL_KEY, 50);
- conf.setInt(CommonConfigurationKeys.HA_HM_CONNECT_RETRY_INTERVAL_KEY, 50);
- conf.setInt(CommonConfigurationKeys.HA_HM_SLEEP_AFTER_DISCONNECT_KEY, 50);
- svc1 = new DummyHAService(HAServiceState.INITIALIZING,
- new InetSocketAddress("svc1", 1234));
- svc2 = new DummyHAService(HAServiceState.INITIALIZING,
- new InetSocketAddress("svc2", 1234));
- }
-
- /**
- * Set up two services and their failover controllers. svc1 is started
- * first, so that it enters ACTIVE state, and then svc2 is started,
- * which enters STANDBY
- */
- private void setupFCs() throws Exception {
- // Format the base dir, should succeed
- assertEquals(0, runFC(svc1, "-formatZK"));
+ conf.set(ZKFailoverController.ZK_ACL_KEY, TEST_ACL);
+ conf.set(ZKFailoverController.ZK_AUTH_KEY, TEST_AUTH_GOOD);
- ctx = new MultithreadedTestUtil.TestContext();
- thr1 = new DummyZKFCThread(ctx, svc1);
- ctx.addThread(thr1);
- thr1.start();
-
- LOG.info("Waiting for svc1 to enter active state");
- waitForHAState(svc1, HAServiceState.ACTIVE);
-
- LOG.info("Adding svc2");
- thr2 = new DummyZKFCThread(ctx, svc2);
- thr2.start();
- waitForHAState(svc2, HAServiceState.STANDBY);
- }
-
- private void stopFCs() throws Exception {
- if (thr1 != null) {
- thr1.interrupt();
- }
- if (thr2 != null) {
- thr2.interrupt();
- }
- if (ctx != null) {
- ctx.stop();
- }
+ conf.set(ZKFailoverController.ZK_QUORUM_KEY, hostPort);
+ this.cluster = new MiniZKFCCluster(conf, getServer(serverFactory));
}
/**
@@ -114,20 +79,104 @@ public class TestZKFailoverController ex
*/
@Test(timeout=15000)
public void testFormatZK() throws Exception {
+ DummyHAService svc = cluster.getService(1);
// Run without formatting the base dir,
// should barf
assertEquals(ZKFailoverController.ERR_CODE_NO_PARENT_ZNODE,
- runFC(svc1));
+ runFC(svc));
// Format the base dir, should succeed
- assertEquals(0, runFC(svc1, "-formatZK"));
+ assertEquals(0, runFC(svc, "-formatZK"));
// Should fail to format if already formatted
assertEquals(ZKFailoverController.ERR_CODE_FORMAT_DENIED,
- runFC(svc1, "-formatZK", "-nonInteractive"));
+ runFC(svc, "-formatZK", "-nonInteractive"));
// Unless '-force' is on
- assertEquals(0, runFC(svc1, "-formatZK", "-force"));
+ assertEquals(0, runFC(svc, "-formatZK", "-force"));
+ }
+
+ /**
+ * Test that if ZooKeeper is not running, the correct error
+ * code is returned.
+ */
+ @Test(timeout=15000)
+ public void testNoZK() throws Exception {
+ stopServer();
+ DummyHAService svc = cluster.getService(1);
+ assertEquals(ZKFailoverController.ERR_CODE_NO_ZK,
+ runFC(svc));
+ }
+
+ @Test
+ public void testFormatOneClusterLeavesOtherClustersAlone() throws Exception {
+ DummyHAService svc = cluster.getService(1);
+
+ DummyZKFC zkfcInOtherCluster = new DummyZKFC(conf, cluster.getService(1)) {
+ @Override
+ protected String getScopeInsideParentNode() {
+ return "other-scope";
+ }
+ };
+
+ // Run without formatting the base dir,
+ // should barf
+ assertEquals(ZKFailoverController.ERR_CODE_NO_PARENT_ZNODE,
+ runFC(svc));
+
+ // Format the base dir, should succeed
+ assertEquals(0, runFC(svc, "-formatZK"));
+
+ // Run the other cluster without formatting, should barf because
+ // it uses a different parent znode
+ assertEquals(ZKFailoverController.ERR_CODE_NO_PARENT_ZNODE,
+ zkfcInOtherCluster.run(new String[]{}));
+
+ // Should succeed in formatting the second cluster
+ assertEquals(0, zkfcInOtherCluster.run(new String[]{"-formatZK"}));
+
+ // But should not have deleted the original base node from the first
+ // cluster
+ assertEquals(ZKFailoverController.ERR_CODE_FORMAT_DENIED,
+ runFC(svc, "-formatZK", "-nonInteractive"));
+ }
+
+ /**
+ * Test that automatic failover won't run against a target that hasn't
+ * explicitly enabled the feature.
+ */
+ @Test(timeout=10000)
+ public void testWontRunWhenAutoFailoverDisabled() throws Exception {
+ DummyHAService svc = cluster.getService(1);
+ svc = Mockito.spy(svc);
+ Mockito.doReturn(false).when(svc).isAutoFailoverEnabled();
+
+ assertEquals(ZKFailoverController.ERR_CODE_AUTO_FAILOVER_NOT_ENABLED,
+ runFC(svc, "-formatZK"));
+ assertEquals(ZKFailoverController.ERR_CODE_AUTO_FAILOVER_NOT_ENABLED,
+ runFC(svc));
+ }
+
+ /**
+ * Test that, if ACLs are specified in the configuration, that
+ * it sets the ACLs when formatting the parent node.
+ */
+ @Test(timeout=15000)
+ public void testFormatSetsAcls() throws Exception {
+ // Format the base dir, should succeed
+ DummyHAService svc = cluster.getService(1);
+ assertEquals(0, runFC(svc, "-formatZK"));
+
+ ZooKeeper otherClient = createClient();
+ try {
+ // client without auth should not be able to read it
+ Stat stat = new Stat();
+ otherClient.getData(ZKFailoverController.ZK_PARENT_ZNODE_DEFAULT,
+ false, stat);
+ fail("Was able to read data without authenticating!");
+ } catch (KeeperException.NoAuthException nae) {
+ // expected
+ }
}
/**
@@ -136,14 +185,14 @@ public class TestZKFailoverController ex
*/
@Test(timeout=15000)
public void testFencingMustBeConfigured() throws Exception {
- svc1 = Mockito.spy(svc1);
+ DummyHAService svc = Mockito.spy(cluster.getService(0));
Mockito.doThrow(new BadFencingConfigurationException("no fencing"))
- .when(svc1).checkFencingConfigured();
+ .when(svc).checkFencingConfigured();
// Format the base dir, should succeed
- assertEquals(0, runFC(svc1, "-formatZK"));
+ assertEquals(0, runFC(svc, "-formatZK"));
// Try to run the actual FC, should fail without a fencer
assertEquals(ZKFailoverController.ERR_CODE_NO_FENCER,
- runFC(svc1));
+ runFC(svc));
}
/**
@@ -155,66 +204,50 @@ public class TestZKFailoverController ex
@Test(timeout=15000)
public void testAutoFailoverOnBadHealth() throws Exception {
try {
- setupFCs();
+ cluster.start();
+ DummyHAService svc1 = cluster.getService(1);
+
+ LOG.info("Faking svc0 unhealthy, should failover to svc1");
+ cluster.setHealthy(0, false);
- LOG.info("Faking svc1 unhealthy, should failover to svc2");
- svc1.isHealthy = false;
- LOG.info("Waiting for svc1 to enter standby state");
- waitForHAState(svc1, HAServiceState.STANDBY);
- waitForHAState(svc2, HAServiceState.ACTIVE);
+ LOG.info("Waiting for svc0 to enter standby state");
+ cluster.waitForHAState(0, HAServiceState.STANDBY);
+ cluster.waitForHAState(1, HAServiceState.ACTIVE);
- LOG.info("Allowing svc1 to be healthy again, making svc2 unreachable " +
+ LOG.info("Allowing svc0 to be healthy again, making svc1 unreachable " +
"and fail to gracefully go to standby");
- svc1.isHealthy = true;
- svc2.actUnreachable = true;
-
- // Allow fencing to succeed
- Mockito.doReturn(true).when(svc2.fencer).fence(Mockito.same(svc2));
- // Should fail back to svc1 at this point
- waitForHAState(svc1, HAServiceState.ACTIVE);
- // and fence svc2
- Mockito.verify(svc2.fencer).fence(Mockito.same(svc2));
+ cluster.setUnreachable(1, true);
+ cluster.setHealthy(0, true);
+
+ // Should fail back to svc0 at this point
+ cluster.waitForHAState(0, HAServiceState.ACTIVE);
+ // and fence svc1
+ Mockito.verify(svc1.fencer).fence(Mockito.same(svc1));
} finally {
- stopFCs();
+ cluster.stop();
}
}
@Test(timeout=15000)
public void testAutoFailoverOnLostZKSession() throws Exception {
try {
- setupFCs();
+ cluster.start();
- // Expire svc1, it should fail over to svc2
- expireAndVerifyFailover(thr1, thr2);
+ // Expire svc0, it should fail over to svc1
+ cluster.expireAndVerifyFailover(0, 1);
- // Expire svc2, it should fail back to svc1
- expireAndVerifyFailover(thr2, thr1);
+ // Expire svc1, it should fail back to svc0
+ cluster.expireAndVerifyFailover(1, 0);
LOG.info("======= Running test cases second time to test " +
"re-establishment =========");
- // Expire svc1, it should fail over to svc2
- expireAndVerifyFailover(thr1, thr2);
+ // Expire svc0, it should fail over to svc1
+ cluster.expireAndVerifyFailover(0, 1);
- // Expire svc2, it should fail back to svc1
- expireAndVerifyFailover(thr2, thr1);
+ // Expire svc1, it should fail back to svc0
+ cluster.expireAndVerifyFailover(1, 0);
} finally {
- stopFCs();
- }
- }
-
- private void expireAndVerifyFailover(DummyZKFCThread fromThr,
- DummyZKFCThread toThr) throws Exception {
- DummyHAService fromSvc = fromThr.zkfc.localTarget;
- DummyHAService toSvc = toThr.zkfc.localTarget;
-
- fromThr.zkfc.getElectorForTests().preventSessionReestablishmentForTests();
- try {
- expireActiveLockHolder(fromSvc);
-
- waitForHAState(fromSvc, HAServiceState.STANDBY);
- waitForHAState(toSvc, HAServiceState.ACTIVE);
- } finally {
- fromThr.zkfc.getElectorForTests().allowSessionReestablishmentForTests();
+ cluster.stop();
}
}
@@ -225,33 +258,32 @@ public class TestZKFailoverController ex
@Test(timeout=15000)
public void testDontFailoverToUnhealthyNode() throws Exception {
try {
- setupFCs();
+ cluster.start();
- // Make svc2 unhealthy, and wait for its FC to notice the bad health.
- svc2.isHealthy = false;
- waitForHealthState(thr2.zkfc,
- HealthMonitor.State.SERVICE_UNHEALTHY);
+ // Make svc1 unhealthy, and wait for its FC to notice the bad health.
+ cluster.setHealthy(1, false);
+ cluster.waitForHealthState(1, HealthMonitor.State.SERVICE_UNHEALTHY);
- // Expire svc1
- thr1.zkfc.getElectorForTests().preventSessionReestablishmentForTests();
+ // Expire svc0
+ cluster.getElector(0).preventSessionReestablishmentForTests();
try {
- expireActiveLockHolder(svc1);
+ cluster.expireActiveLockHolder(0);
- LOG.info("Expired svc1's ZK session. Waiting a second to give svc2" +
+ LOG.info("Expired svc0's ZK session. Waiting a second to give svc1" +
" a chance to take the lock, if it is ever going to.");
Thread.sleep(1000);
// Ensure that no one holds the lock.
- waitForActiveLockHolder(null);
+ cluster.waitForActiveLockHolder(null);
} finally {
- LOG.info("Allowing svc1's elector to re-establish its connection");
- thr1.zkfc.getElectorForTests().allowSessionReestablishmentForTests();
+ LOG.info("Allowing svc0's elector to re-establish its connection");
+ cluster.getElector(0).allowSessionReestablishmentForTests();
}
- // svc1 should get the lock again
- waitForActiveLockHolder(svc1);
+ // svc0 should get the lock again
+ cluster.waitForActiveLockHolder(0);
} finally {
- stopFCs();
+ cluster.stop();
}
}
@@ -262,29 +294,38 @@ public class TestZKFailoverController ex
@Test(timeout=15000)
public void testBecomingActiveFails() throws Exception {
try {
- setupFCs();
-
- LOG.info("Making svc2 fail to become active");
- svc2.failToBecomeActive = true;
+ cluster.start();
+ DummyHAService svc1 = cluster.getService(1);
- LOG.info("Faking svc1 unhealthy, should NOT successfully " +
- "failover to svc2");
- svc1.isHealthy = false;
- waitForHealthState(thr1.zkfc, State.SERVICE_UNHEALTHY);
- waitForActiveLockHolder(null);
-
- Mockito.verify(svc2.proxy).transitionToActive();
-
- waitForHAState(svc1, HAServiceState.STANDBY);
- waitForHAState(svc2, HAServiceState.STANDBY);
+ LOG.info("Making svc1 fail to become active");
+ cluster.setFailToBecomeActive(1, true);
- LOG.info("Faking svc1 healthy again, should go back to svc1");
- svc1.isHealthy = true;
- waitForHAState(svc1, HAServiceState.ACTIVE);
- waitForHAState(svc2, HAServiceState.STANDBY);
- waitForActiveLockHolder(svc1);
+ LOG.info("Faking svc0 unhealthy, should NOT successfully " +
+ "failover to svc1");
+ cluster.setHealthy(0, false);
+ cluster.waitForHealthState(0, State.SERVICE_UNHEALTHY);
+ cluster.waitForActiveLockHolder(null);
+
+
+ Mockito.verify(svc1.proxy, Mockito.timeout(2000).atLeastOnce())
+ .transitionToActive(Mockito.<StateChangeRequestInfo>any());
+
+ cluster.waitForHAState(0, HAServiceState.STANDBY);
+ cluster.waitForHAState(1, HAServiceState.STANDBY);
+
+ LOG.info("Faking svc0 healthy again, should go back to svc0");
+ cluster.setHealthy(0, true);
+ cluster.waitForHAState(0, HAServiceState.ACTIVE);
+ cluster.waitForHAState(1, HAServiceState.STANDBY);
+ cluster.waitForActiveLockHolder(0);
+
+ // Ensure that we can fail back to svc1 once it it is able
+ // to become active (e.g the admin has restarted it)
+ LOG.info("Allowing svc1 to become active, expiring svc0");
+ svc1.failToBecomeActive = false;
+ cluster.expireAndVerifyFailover(0, 1);
} finally {
- stopFCs();
+ cluster.stop();
}
}
@@ -296,27 +337,25 @@ public class TestZKFailoverController ex
@Test(timeout=15000)
public void testZooKeeperFailure() throws Exception {
try {
- setupFCs();
+ cluster.start();
// Record initial ZK sessions
- long session1 = thr1.zkfc.getElectorForTests().getZKSessionIdForTests();
- long session2 = thr2.zkfc.getElectorForTests().getZKSessionIdForTests();
+ long session0 = cluster.getElector(0).getZKSessionIdForTests();
+ long session1 = cluster.getElector(1).getZKSessionIdForTests();
LOG.info("====== Stopping ZK server");
stopServer();
waitForServerDown(hostPort, CONNECTION_TIMEOUT);
LOG.info("====== Waiting for services to enter NEUTRAL mode");
- ActiveStandbyElectorTestUtil.waitForElectorState(ctx,
- thr1.zkfc.getElectorForTests(),
+ cluster.waitForElectorState(0,
ActiveStandbyElector.State.NEUTRAL);
- ActiveStandbyElectorTestUtil.waitForElectorState(ctx,
- thr2.zkfc.getElectorForTests(),
+ cluster.waitForElectorState(1,
ActiveStandbyElector.State.NEUTRAL);
LOG.info("====== Checking that the services didn't change HA state");
- assertEquals(HAServiceState.ACTIVE, svc1.state);
- assertEquals(HAServiceState.STANDBY, svc2.state);
+ assertEquals(HAServiceState.ACTIVE, cluster.getService(0).state);
+ assertEquals(HAServiceState.STANDBY, cluster.getService(1).state);
LOG.info("====== Restarting server");
startServer();
@@ -324,134 +363,224 @@ public class TestZKFailoverController ex
// Nodes should go back to their original states, since they re-obtain
// the same sessions.
- ActiveStandbyElectorTestUtil.waitForElectorState(ctx,
- thr1.zkfc.getElectorForTests(),
- ActiveStandbyElector.State.ACTIVE);
- ActiveStandbyElectorTestUtil.waitForElectorState(ctx,
- thr2.zkfc.getElectorForTests(),
- ActiveStandbyElector.State.STANDBY);
+ cluster.waitForElectorState(0, ActiveStandbyElector.State.ACTIVE);
+ cluster.waitForElectorState(1, ActiveStandbyElector.State.STANDBY);
// Check HA states didn't change.
- ActiveStandbyElectorTestUtil.waitForElectorState(ctx,
- thr1.zkfc.getElectorForTests(),
- ActiveStandbyElector.State.ACTIVE);
- ActiveStandbyElectorTestUtil.waitForElectorState(ctx,
- thr2.zkfc.getElectorForTests(),
- ActiveStandbyElector.State.STANDBY);
+ cluster.waitForHAState(0, HAServiceState.ACTIVE);
+ cluster.waitForHAState(1, HAServiceState.STANDBY);
+
// Check they re-used the same sessions and didn't spuriously reconnect
+ assertEquals(session0,
+ cluster.getElector(0).getZKSessionIdForTests());
assertEquals(session1,
- thr1.zkfc.getElectorForTests().getZKSessionIdForTests());
- assertEquals(session2,
- thr2.zkfc.getElectorForTests().getZKSessionIdForTests());
+ cluster.getElector(1).getZKSessionIdForTests());
} finally {
- stopFCs();
+ cluster.stop();
}
}
-
- /**
- * Expire the ZK session of the given service. This requires
- * (and asserts) that the given service be the current active.
- * @throws NoNodeException if no service holds the lock
- */
- private void expireActiveLockHolder(DummyHAService expectedActive)
- throws NoNodeException {
- ZooKeeperServer zks = getServer(serverFactory);
- Stat stat = new Stat();
- byte[] data = zks.getZKDatabase().getData(
- ZKFailoverController.ZK_PARENT_ZNODE_DEFAULT + "/" +
- ActiveStandbyElector.LOCK_FILENAME, stat, null);
-
- assertArrayEquals(Ints.toByteArray(expectedActive.index), data);
- long session = stat.getEphemeralOwner();
- LOG.info("Expiring svc " + expectedActive + "'s zookeeper session " + session);
- zks.closeSession(session);
- }
/**
- * Wait for the given HA service to enter the given HA state.
+ * Test that the ZKFC can gracefully cede its active status.
*/
- private void waitForHAState(DummyHAService svc, HAServiceState state)
- throws Exception {
- while (svc.state != state) {
- ctx.checkException();
- Thread.sleep(50);
+ @Test(timeout=15000)
+ public void testCedeActive() throws Exception {
+ try {
+ cluster.start();
+ DummyZKFC zkfc = cluster.getZkfc(0);
+ // It should be in active to start.
+ assertEquals(ActiveStandbyElector.State.ACTIVE,
+ zkfc.getElectorForTests().getStateForTests());
+
+ // Ask it to cede active for 3 seconds. It should respond promptly
+ // (i.e. the RPC itself should not take 3 seconds!)
+ ZKFCProtocol proxy = zkfc.getLocalTarget().getZKFCProxy(conf, 5000);
+ long st = System.currentTimeMillis();
+ proxy.cedeActive(3000);
+ long et = System.currentTimeMillis();
+ assertTrue("RPC to cedeActive took " + (et - st) + " ms",
+ et - st < 1000);
+
+ // Should be in "INIT" state since it's not in the election
+ // at this point.
+ assertEquals(ActiveStandbyElector.State.INIT,
+ zkfc.getElectorForTests().getStateForTests());
+
+ // After the prescribed 3 seconds, should go into STANDBY state,
+ // since the other node in the cluster would have taken ACTIVE.
+ cluster.waitForElectorState(0, ActiveStandbyElector.State.STANDBY);
+ long et2 = System.currentTimeMillis();
+ assertTrue("Should take ~3 seconds to rejoin. Only took " + (et2 - et) +
+ "ms before rejoining.",
+ et2 - et > 2800);
+ } finally {
+ cluster.stop();
}
}
- /**
- * Wait for the ZKFC to be notified of a change in health state.
- */
- private void waitForHealthState(DummyZKFC zkfc, State state)
- throws Exception {
- while (zkfc.getLastHealthState() != state) {
- ctx.checkException();
- Thread.sleep(50);
+ @Test(timeout=15000)
+ public void testGracefulFailover() throws Exception {
+ try {
+ cluster.start();
+
+ cluster.waitForActiveLockHolder(0);
+ cluster.getService(1).getZKFCProxy(conf, 5000).gracefulFailover();
+ cluster.waitForActiveLockHolder(1);
+ cluster.getService(0).getZKFCProxy(conf, 5000).gracefulFailover();
+ cluster.waitForActiveLockHolder(0);
+
+ assertEquals(0, cluster.getService(0).fenceCount);
+ assertEquals(0, cluster.getService(1).fenceCount);
+ } finally {
+ cluster.stop();
}
}
+
+ @Test(timeout=15000)
+ public void testGracefulFailoverToUnhealthy() throws Exception {
+ try {
+ cluster.start();
- /**
- * Wait for the given HA service to become the active lock holder.
- * If the passed svc is null, waits for there to be no active
- * lock holder.
- */
- private void waitForActiveLockHolder(DummyHAService svc)
- throws Exception {
- ZooKeeperServer zks = getServer(serverFactory);
- ActiveStandbyElectorTestUtil.waitForActiveLockData(ctx, zks,
- ZKFailoverController.ZK_PARENT_ZNODE_DEFAULT,
- (svc == null) ? null : Ints.toByteArray(svc.index));
+ cluster.waitForActiveLockHolder(0);
+
+ // Mark it unhealthy, wait for it to exit election
+ cluster.setHealthy(1, false);
+ cluster.waitForElectorState(1, ActiveStandbyElector.State.INIT);
+
+ // Ask for failover, it should fail, because it's unhealthy
+ try {
+ cluster.getService(1).getZKFCProxy(conf, 5000).gracefulFailover();
+ fail("Did not fail to graceful failover to unhealthy service!");
+ } catch (ServiceFailedException sfe) {
+ GenericTestUtils.assertExceptionContains(
+ cluster.getService(1).toString() +
+ " is not currently healthy.", sfe);
+ }
+ } finally {
+ cluster.stop();
+ }
}
+
+ @Test(timeout=15000)
+ public void testGracefulFailoverFailBecomingActive() throws Exception {
+ try {
+ cluster.start();
+ cluster.waitForActiveLockHolder(0);
+ cluster.setFailToBecomeActive(1, true);
+
+ // Ask for failover, it should fail and report back to user.
+ try {
+ cluster.getService(1).getZKFCProxy(conf, 5000).gracefulFailover();
+ fail("Did not fail to graceful failover when target failed " +
+ "to become active!");
+ } catch (ServiceFailedException sfe) {
+ GenericTestUtils.assertExceptionContains(
+ "Couldn't make " + cluster.getService(1) + " active", sfe);
+ GenericTestUtils.assertExceptionContains(
+ "injected failure", sfe);
+ }
+
+ // No fencing
+ assertEquals(0, cluster.getService(0).fenceCount);
+ assertEquals(0, cluster.getService(1).fenceCount);
- private int runFC(DummyHAService target, String ... args) throws Exception {
- DummyZKFC zkfc = new DummyZKFC(target);
- zkfc.setConf(conf);
- return zkfc.run(args);
+ // Service 0 should go back to being active after the failed failover
+ cluster.waitForActiveLockHolder(0);
+ } finally {
+ cluster.stop();
+ }
}
- /**
- * Test-thread which runs a ZK Failover Controller corresponding
- * to a given dummy service.
- */
- private class DummyZKFCThread extends TestingThread {
- private final DummyZKFC zkfc;
+ @Test(timeout=15000)
+ public void testGracefulFailoverFailBecomingStandby() throws Exception {
+ try {
+ cluster.start();
- public DummyZKFCThread(TestContext ctx, DummyHAService svc) {
- super(ctx);
- this.zkfc = new DummyZKFC(svc);
- zkfc.setConf(conf);
+ cluster.waitForActiveLockHolder(0);
+
+ // Ask for failover when old node fails to transition to standby.
+ // This should trigger fencing, since the cedeActive() command
+ // still works, but leaves the breadcrumb in place.
+ cluster.setFailToBecomeStandby(0, true);
+ cluster.getService(1).getZKFCProxy(conf, 5000).gracefulFailover();
+
+ // Check that the old node was fenced
+ assertEquals(1, cluster.getService(0).fenceCount);
+ } finally {
+ cluster.stop();
}
+ }
+
+ @Test(timeout=15000)
+ public void testGracefulFailoverFailBecomingStandbyAndFailFence()
+ throws Exception {
+ try {
+ cluster.start();
+
+ cluster.waitForActiveLockHolder(0);
+
+ // Ask for failover when old node fails to transition to standby.
+ // This should trigger fencing, since the cedeActive() command
+ // still works, but leaves the breadcrumb in place.
+ cluster.setFailToBecomeStandby(0, true);
+ cluster.setFailToFence(0, true);
- @Override
- public void doWork() throws Exception {
try {
- assertEquals(0, zkfc.run(new String[0]));
- } catch (InterruptedException ie) {
- // Interrupted by main thread, that's OK.
+ cluster.getService(1).getZKFCProxy(conf, 5000).gracefulFailover();
+ fail("Failover should have failed when old node wont fence");
+ } catch (ServiceFailedException sfe) {
+ GenericTestUtils.assertExceptionContains(
+ "Unable to fence " + cluster.getService(0), sfe);
}
+ } finally {
+ cluster.stop();
}
}
-
- private static class DummyZKFC extends ZKFailoverController {
- private final DummyHAService localTarget;
-
- public DummyZKFC(DummyHAService localTarget) {
- this.localTarget = localTarget;
- }
- @Override
- protected byte[] targetToData(HAServiceTarget target) {
- return Ints.toByteArray(((DummyHAService)target).index);
- }
-
- @Override
- protected HAServiceTarget dataToTarget(byte[] data) {
- int index = Ints.fromByteArray(data);
- return DummyHAService.getInstance(index);
- }
+ /**
+ * Test which exercises all of the inputs into ZKFC. This is particularly
+ * useful for running under jcarder to check for lock order violations.
+ */
+ @Test(timeout=30000)
+ public void testOneOfEverything() throws Exception {
+ try {
+ cluster.start();
+
+ // Failover by session expiration
+ LOG.info("====== Failing over by session expiration");
+ cluster.expireAndVerifyFailover(0, 1);
+ cluster.expireAndVerifyFailover(1, 0);
+
+ // Restart ZK
+ LOG.info("====== Restarting server");
+ stopServer();
+ waitForServerDown(hostPort, CONNECTION_TIMEOUT);
+ startServer();
+ waitForServerUp(hostPort, CONNECTION_TIMEOUT);
- @Override
- protected HAServiceTarget getLocalTarget() {
- return localTarget;
+ // Failover by bad health
+ cluster.setHealthy(0, false);
+ cluster.waitForHAState(0, HAServiceState.STANDBY);
+ cluster.waitForHAState(1, HAServiceState.ACTIVE);
+ cluster.setHealthy(1, true);
+ cluster.setHealthy(0, false);
+ cluster.waitForHAState(1, HAServiceState.ACTIVE);
+ cluster.waitForHAState(0, HAServiceState.STANDBY);
+ cluster.setHealthy(0, true);
+
+ cluster.waitForHealthState(0, State.SERVICE_HEALTHY);
+
+ // Graceful failovers
+ cluster.getZkfc(1).gracefulFailoverToYou();
+ cluster.getZkfc(0).gracefulFailoverToYou();
+ } finally {
+ cluster.stop();
}
}
+
+ private int runFC(DummyHAService target, String ... args) throws Exception {
+ DummyZKFC zkfc = new DummyZKFC(conf, target);
+ return zkfc.run(args);
+ }
+
}