You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by an...@apache.org on 2018/10/24 09:31:00 UTC
[05/51] [partial] zookeeper git commit: ZOOKEEPER-3032: MAVEN
MIGRATION - branch-3.5 - zookeeper-server
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/LENonTerminateTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/LENonTerminateTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/LENonTerminateTest.java
new file mode 100644
index 0000000..2bbf7b5
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/LENonTerminateTest.java
@@ -0,0 +1,378 @@
+/**
+ * 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.zookeeper.test;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.InetSocketAddress;
+import java.net.SocketException;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.server.ServerCnxnFactory;
+import org.apache.zookeeper.server.quorum.Election;
+import org.apache.zookeeper.server.quorum.FLELostMessageTest;
+import org.apache.zookeeper.server.quorum.LeaderElection;
+import org.apache.zookeeper.server.quorum.QuorumPeer;
+import org.apache.zookeeper.server.quorum.Vote;
+import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
+import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
+import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
+import org.apache.zookeeper.server.quorum.flexible.QuorumMaj;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+@SuppressWarnings("deprecation")
+public class LENonTerminateTest extends ZKTestCase {
+ public static class MockLeaderElection extends LeaderElection {
+ public MockLeaderElection(QuorumPeer self) {
+ super(self);
+ }
+
+ /**
+ * Temporary for 3.3.0 - we want to ensure that a round of voting happens
+ * before any of the peers update their votes. The easiest way to do that
+ * is to add a latch that all wait on after counting their votes.
+ *
+ * In 3.4.0 we intend to make this class more testable, and therefore
+ * there should be much less duplicated code.
+ *
+ * JMX bean method calls are removed to reduce noise.
+ */
+ public Vote lookForLeader() throws InterruptedException {
+ self.setCurrentVote(new Vote(self.getId(),
+ self.getLastLoggedZxid()));
+ // We are going to look for a leader by casting a vote for ourself
+ byte requestBytes[] = new byte[4];
+ ByteBuffer requestBuffer = ByteBuffer.wrap(requestBytes);
+ byte responseBytes[] = new byte[28];
+ ByteBuffer responseBuffer = ByteBuffer.wrap(responseBytes);
+ /* The current vote for the leader. Initially me! */
+ DatagramSocket s = null;
+ try {
+ s = new DatagramSocket();
+ s.setSoTimeout(200);
+ } catch (SocketException e1) {
+ LOG.error("Socket exception when creating socket for leader election", e1);
+ System.exit(4);
+ }
+ DatagramPacket requestPacket = new DatagramPacket(requestBytes,
+ requestBytes.length);
+ DatagramPacket responsePacket = new DatagramPacket(responseBytes,
+ responseBytes.length);
+ int xid = epochGen.nextInt();
+ while (self.isRunning()) {
+ HashMap<InetSocketAddress, Vote> votes =
+ new HashMap<InetSocketAddress, Vote>(self.getVotingView().size());
+
+ requestBuffer.clear();
+ requestBuffer.putInt(xid);
+ requestPacket.setLength(4);
+ HashSet<Long> heardFrom = new HashSet<Long>();
+ for (QuorumServer server :
+ self.getVotingView().values())
+ {
+ LOG.info("Server address: " + server.addr);
+ try {
+ requestPacket.setSocketAddress(server.addr);
+ } catch (IllegalArgumentException e) {
+ // Sun doesn't include the address that causes this
+ // exception to be thrown, so we wrap the exception
+ // in order to capture this critical detail.
+ throw new IllegalArgumentException(
+ "Unable to set socket address on packet, msg:"
+ + e.getMessage() + " with addr:" + server.addr,
+ e);
+ }
+
+ try {
+ s.send(requestPacket);
+ responsePacket.setLength(responseBytes.length);
+ s.receive(responsePacket);
+ if (responsePacket.getLength() != responseBytes.length) {
+ LOG.error("Got a short response: "
+ + responsePacket.getLength());
+ continue;
+ }
+ responseBuffer.clear();
+ int recvedXid = responseBuffer.getInt();
+ if (recvedXid != xid) {
+ LOG.error("Got bad xid: expected " + xid
+ + " got " + recvedXid);
+ continue;
+ }
+ long peerId = responseBuffer.getLong();
+ heardFrom.add(peerId);
+ //if(server.id != peerId){
+ Vote vote = new Vote(responseBuffer.getLong(),
+ responseBuffer.getLong());
+ InetSocketAddress addr =
+ (InetSocketAddress) responsePacket
+ .getSocketAddress();
+ votes.put(addr, vote);
+ //}
+ } catch (IOException e) {
+ LOG.warn("Ignoring exception while looking for leader",
+ e);
+ // Errors are okay, since hosts may be
+ // down
+ }
+ }
+
+ ElectionResult result = countVotes(votes, heardFrom);
+
+ /**
+ * This is the only difference from LeaderElection - wait for
+ * this latch on the first time through this method. This ensures
+ * that the first round of voting happens before setCurrentVote
+ * is called below.
+ */
+ LOG.info("Waiting for first round of voting to complete");
+ latch.countDown();
+ Assert.assertTrue("Thread timed out waiting for latch",
+ latch.await(10000, TimeUnit.MILLISECONDS));
+
+ // ZOOKEEPER-569:
+ // If no votes are received for live peers, reset to voting
+ // for ourselves as otherwise we may hang on to a vote
+ // for a dead peer
+ if (result.numValidVotes == 0) {
+ self.setCurrentVote(new Vote(self.getId(),
+ self.getLastLoggedZxid()));
+ } else {
+ if (result.winner.getId() >= 0) {
+ self.setCurrentVote(result.vote);
+ // To do: this doesn't use a quorum verifier
+ if (result.winningCount > (self.getVotingView().size() / 2)) {
+ self.setCurrentVote(result.winner);
+ s.close();
+ Vote current = self.getCurrentVote();
+ LOG.info("Found leader: my type is: " + self.getLearnerType());
+ /*
+ * We want to make sure we implement the state machine
+ * correctly. If we are a PARTICIPANT, once a leader
+ * is elected we can move either to LEADING or
+ * FOLLOWING. However if we are an OBSERVER, it is an
+ * error to be elected as a Leader.
+ */
+ if (self.getLearnerType() == LearnerType.OBSERVER) {
+ if (current.getId() == self.getId()) {
+ // This should never happen!
+ LOG.error("OBSERVER elected as leader!");
+ Thread.sleep(100);
+ }
+ else {
+ self.setPeerState(ServerState.OBSERVING);
+ Thread.sleep(100);
+ return current;
+ }
+ } else {
+ self.setPeerState((current.getId() == self.getId())
+ ? ServerState.LEADING: ServerState.FOLLOWING);
+ if (self.getPeerState() == ServerState.FOLLOWING) {
+ Thread.sleep(100);
+ }
+ return current;
+ }
+ }
+ }
+ }
+ Thread.sleep(1000);
+ }
+ return null;
+ }
+ }
+
+ public static class MockQuorumPeer extends QuorumPeer {
+ public MockQuorumPeer(Map<Long,QuorumServer> quorumPeers, File snapDir,
+ File logDir, int clientPort, int electionAlg,
+ long myid, int tickTime, int initLimit, int syncLimit)
+ throws IOException
+ {
+ super(quorumPeers, snapDir, logDir, electionAlg,
+ myid,tickTime, initLimit,syncLimit, false,
+ ServerCnxnFactory.createFactory(clientPort, -1),
+ new QuorumMaj(quorumPeers));
+ }
+
+ protected Election createElectionAlgorithm(int electionAlgorithm){
+ LOG.info("Returning mocked leader election");
+ return new MockLeaderElection(this);
+ }
+ }
+
+
+ protected static final Logger LOG = LoggerFactory.getLogger(FLELostMessageTest.class);
+
+ int count;
+ HashMap<Long,QuorumServer> peers;
+ File tmpdir[];
+ int port[];
+
+ @Before
+ public void setUp() throws Exception {
+ count = 3;
+
+ peers = new HashMap<Long,QuorumServer>(count);
+ tmpdir = new File[count];
+ port = new int[count];
+ }
+
+ static final CountDownLatch latch = new CountDownLatch(2);
+ static final CountDownLatch mockLatch = new CountDownLatch(1);
+
+ private static class LEThread extends Thread {
+ private int i;
+ private QuorumPeer peer;
+
+ LEThread(QuorumPeer peer, int i) {
+ this.i = i;
+ this.peer = peer;
+ LOG.info("Constructor: " + getName());
+
+ }
+
+ public void run(){
+ try{
+ Vote v = null;
+ peer.setPeerState(ServerState.LOOKING);
+ LOG.info("Going to call leader election: " + i);
+ v = peer.getElectionAlg().lookForLeader();
+
+ if (v == null){
+ Assert.fail("Thread " + i + " got a null vote");
+ }
+
+ /*
+ * A real zookeeper would take care of setting the current vote. Here
+ * we do it manually.
+ */
+ peer.setCurrentVote(v);
+
+ LOG.info("Finished election: " + i + ", " + v.getId());
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ LOG.info("Joining");
+ }
+ }
+
+ /**
+ * This tests ZK-569.
+ * With three peers A, B and C, the following could happen:
+ * 1. Round 1, A,B and C all vote for themselves
+ * 2. Round 2, C dies, A and B vote for C
+ * 3. Because C has died, votes for it are ignored, but A and B never
+ * reset their votes. Hence LE never terminates. ZK-569 fixes this by
+ * resetting votes to themselves if the set of votes for live peers is null.
+ */
+ @Test
+ public void testNonTermination() throws Exception {
+ LOG.info("TestNonTermination: " + getTestName()+ ", " + count);
+ for(int i = 0; i < count; i++) {
+ int clientport = PortAssignment.unique();
+ peers.put(Long.valueOf(i),
+ new QuorumServer(i,
+ new InetSocketAddress("127.0.0.1", clientport),
+ new InetSocketAddress("127.0.0.1", PortAssignment.unique())));
+ tmpdir[i] = ClientBase.createTmpDir();
+ port[i] = clientport;
+ }
+
+ /*
+ * peer1 and peer2 are A and B in the above example.
+ */
+ QuorumPeer peer1 = new MockQuorumPeer(peers, tmpdir[0], tmpdir[0], port[0], 0, 0, 2, 2, 2);
+ peer1.startLeaderElection();
+ LEThread thread1 = new LEThread(peer1, 0);
+
+ QuorumPeer peer2 = new MockQuorumPeer(peers, tmpdir[1], tmpdir[1], port[1], 0, 1, 2, 2, 2);
+ peer2.startLeaderElection();
+ LEThread thread2 = new LEThread(peer2, 1);
+
+ /*
+ * Start mock server.
+ */
+ Thread thread3 = new Thread() {
+ public void run() {
+ try {
+ mockServer();
+ } catch (Exception e) {
+ LOG.error("exception", e);
+ Assert.fail("Exception when running mocked server " + e);
+ }
+ }
+ };
+
+ thread3.start();
+ Assert.assertTrue("mockServer did not start in 5s",
+ mockLatch.await(5000, TimeUnit.MILLISECONDS));
+ thread1.start();
+ thread2.start();
+ /*
+ * Occasionally seen false negatives with a 5s timeout.
+ */
+ thread1.join(15000);
+ thread2.join(15000);
+ thread3.join(15000);
+ if (thread1.isAlive() || thread2.isAlive() || thread3.isAlive()) {
+ Assert.fail("Threads didn't join");
+ }
+ }
+
+ /**
+ * MockServer plays the role of peer C. Respond to two requests for votes
+ * with vote for self and then Assert.fail.
+ */
+ void mockServer() throws InterruptedException, IOException {
+ byte b[] = new byte[36];
+ ByteBuffer responseBuffer = ByteBuffer.wrap(b);
+ DatagramPacket packet = new DatagramPacket(b, b.length);
+ QuorumServer server = peers.get(Long.valueOf(2));
+ DatagramSocket udpSocket = new DatagramSocket(server.addr.getPort());
+ LOG.info("In MockServer");
+ mockLatch.countDown();
+ Vote current = new Vote(2, 1);
+ for (int i=0;i<2;++i) {
+ udpSocket.receive(packet);
+ responseBuffer.rewind();
+ LOG.info("Received " + responseBuffer.getInt() + " " + responseBuffer.getLong() + " " + responseBuffer.getLong());
+ LOG.info("From " + packet.getSocketAddress());
+ responseBuffer.clear();
+ responseBuffer.getInt(); // Skip the xid
+ responseBuffer.putLong(2);
+
+ responseBuffer.putLong(current.getId());
+ responseBuffer.putLong(current.getZxid());
+ packet.setData(b);
+ udpSocket.send(packet);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/LETest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/LETest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/LETest.java
new file mode 100644
index 0000000..c6b1833
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/LETest.java
@@ -0,0 +1,142 @@
+/**
+ * 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.zookeeper.test;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Random;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.server.quorum.LeaderElection;
+import org.apache.zookeeper.server.quorum.QuorumPeer;
+import org.apache.zookeeper.server.quorum.Vote;
+import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
+import org.junit.Assert;
+import org.junit.Ignore;
+import org.junit.Test;
+
+@SuppressWarnings("deprecation")
+public class LETest extends ZKTestCase {
+ private static final Logger LOG = LoggerFactory.getLogger(LETest.class);
+ volatile Vote votes[];
+ volatile boolean leaderDies;
+ volatile long leader = -1;
+ Random rand = new Random();
+ class LEThread extends Thread {
+ LeaderElection le;
+ int i;
+ QuorumPeer peer;
+ LEThread(LeaderElection le, QuorumPeer peer, int i) {
+ this.le = le;
+ this.i = i;
+ this.peer = peer;
+ }
+ public void run() {
+ try {
+ Vote v = null;
+ while(true) {
+ v = le.lookForLeader();
+ votes[i] = v;
+ if (v.getId() == i) {
+ synchronized(LETest.this) {
+ if (leaderDies) {
+ leaderDies = false;
+ peer.stopLeaderElection();
+ LOG.info("Leader " + i + " dying");
+ leader = -2;
+ } else {
+ leader = i;
+ }
+ LETest.this.notifyAll();
+ }
+ break;
+ }
+ synchronized(LETest.this) {
+ if (leader == -1) {
+ LETest.this.wait();
+ }
+ if (leader == v.getId()) {
+ break;
+ }
+ }
+ Thread.sleep(rand.nextInt(1000));
+ peer.setCurrentVote(new Vote(peer.getId(), 0));
+ }
+ LOG.info("Thread " + i + " votes " + v);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ @Test
+ @Ignore("ZOOKEEPER-1932, this test is flaky and already removed in master")
+ public void testLE() throws Exception {
+ int count = 30;
+ HashMap<Long,QuorumServer> peers = new HashMap<Long,QuorumServer>(count);
+ ArrayList<LEThread> threads = new ArrayList<LEThread>(count);
+ File tmpdir[] = new File[count];
+ int port[] = new int[count];
+ votes = new Vote[count];
+ for(int i = 0; i < count; i++) {
+ peers.put(Long.valueOf(i),
+ new QuorumServer(i,
+ new InetSocketAddress("127.0.0.1",
+ PortAssignment.unique())));
+ tmpdir[i] = ClientBase.createTmpDir();
+ port[i] = PortAssignment.unique();
+ }
+ LeaderElection le[] = new LeaderElection[count];
+ leaderDies = true;
+ boolean allowOneBadLeader = leaderDies;
+ for(int i = 0; i < le.length; i++) {
+ QuorumPeer peer = new QuorumPeer(peers, tmpdir[i], tmpdir[i],
+ port[i], 0, i, 1000, 2, 2);
+ peer.startLeaderElection();
+ le[i] = new LeaderElection(peer);
+ LEThread thread = new LEThread(le[i], peer, i);
+ thread.start();
+ threads.add(thread);
+ }
+ for(int i = 0; i < threads.size(); i++) {
+ threads.get(i).join(15000);
+ if (threads.get(i).isAlive()) {
+ Assert.fail("Threads didn't join");
+ }
+ }
+ long id = votes[0].getId();
+ for(int i = 1; i < votes.length; i++) {
+ if (votes[i] == null) {
+ Assert.fail("Thread " + i + " had a null vote");
+ }
+ if (votes[i].getId() != id) {
+ if (allowOneBadLeader && votes[i].getId() == i) {
+ allowOneBadLeader = false;
+ } else {
+ Assert.fail("Thread " + i + " got " + votes[i].getId() + " expected " + id);
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/LeaderSessionTrackerTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/LeaderSessionTrackerTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/LeaderSessionTrackerTest.java
new file mode 100644
index 0000000..a45b9cb
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/LeaderSessionTrackerTest.java
@@ -0,0 +1,163 @@
+/**
+ * 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.zookeeper.test;
+
+
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+
+import org.apache.jute.BinaryOutputArchive;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooDefs.OpCode;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.proto.CreateRequest;
+import org.apache.zookeeper.server.Request;
+import org.apache.zookeeper.server.quorum.QuorumPeer;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Due to race condition or bad client code, the leader may get request from
+ * expired session. We need to make sure that we never allow ephmeral node
+ * to be created in those case, but we do allow normal node to be created.
+ */
+public class LeaderSessionTrackerTest extends ZKTestCase {
+
+ protected static final Logger LOG = LoggerFactory
+ .getLogger(LeaderSessionTrackerTest.class);
+
+ QuorumUtil qu;
+
+ @Before
+ public void setUp() throws Exception {
+ qu = new QuorumUtil(1);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ qu.shutdownAll();
+ }
+
+ @Test
+ public void testExpiredSessionWithLocalSession() throws Exception {
+ testCreateEphemeral(true);
+ }
+
+ @Test
+ public void testExpiredSessionWithoutLocalSession() throws Exception {
+ testCreateEphemeral(false);
+ }
+
+ /**
+ * When we create ephemeral node, we need to check against global
+ * session, so the leader never accept request from an expired session
+ * (that we no longer track)
+ *
+ * This is not the same as SessionInvalidationTest since session
+ * is not in closing state
+ */
+ public void testCreateEphemeral(boolean localSessionEnabled) throws Exception {
+ if (localSessionEnabled) {
+ qu.enableLocalSession(true);
+ }
+ qu.startAll();
+ QuorumPeer leader = qu.getLeaderQuorumPeer();
+
+ ZooKeeper zk = ClientBase.createZKClient(qu.getConnectString(leader));
+
+ CreateRequest createRequest = new CreateRequest("/impossible",
+ new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL.toFlag());
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
+ createRequest.serialize(boa, "request");
+ ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
+
+ // Mimic sessionId generated by follower's local session tracker
+ long sid = qu.getFollowerQuorumPeers().get(0).getActiveServer()
+ .getServerId();
+ long fakeSessionId = (sid << 56) + 1;
+
+ LOG.info("Fake session Id: " + Long.toHexString(fakeSessionId));
+
+ Request request = new Request(null, fakeSessionId, 0, OpCode.create,
+ bb, new ArrayList<Id>());
+
+ // Submit request directly to leader
+ leader.getActiveServer().submitRequest(request);
+
+ // Make sure that previous request is finished
+ zk.create("/ok", new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+
+ Stat stat = zk.exists("/impossible", null);
+ Assert.assertEquals("Node from fake session get created", null, stat);
+
+ }
+
+ /**
+ * When local session is enabled, leader will allow persistent node
+ * to be create for unknown session
+ */
+ @Test
+ public void testCreatePersistent() throws Exception {
+ qu.enableLocalSession(true);
+ qu.startAll();
+
+ QuorumPeer leader = qu.getLeaderQuorumPeer();
+
+ ZooKeeper zk = ClientBase.createZKClient(qu.getConnectString(leader));
+
+ CreateRequest createRequest = new CreateRequest("/success",
+ new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT.toFlag());
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
+ createRequest.serialize(boa, "request");
+ ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
+
+ // Mimic sessionId generated by follower's local session tracker
+ long sid = qu.getFollowerQuorumPeers().get(0).getActiveServer()
+ .getServerId();
+ long locallSession = (sid << 56) + 1;
+
+ LOG.info("Local session Id: " + Long.toHexString(locallSession));
+
+ Request request = new Request(null, locallSession, 0, OpCode.create,
+ bb, new ArrayList<Id>());
+
+ // Submit request directly to leader
+ leader.getActiveServer().submitRequest(request);
+
+ // Make sure that previous request is finished
+ zk.create("/ok", new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+
+ Stat stat = zk.exists("/success", null);
+ Assert.assertTrue("Request from local sesson failed", stat != null);
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/LoadFromLogNoServerTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/LoadFromLogNoServerTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/LoadFromLogNoServerTest.java
new file mode 100644
index 0000000..4d56f60
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/LoadFromLogNoServerTest.java
@@ -0,0 +1,176 @@
+/**
+ * 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.zookeeper.test;
+
+import org.apache.jute.BinaryInputArchive;
+import org.apache.jute.BinaryOutputArchive;
+import org.apache.jute.Record;
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.common.Time;
+import org.apache.zookeeper.server.DataNode;
+import org.apache.zookeeper.server.DataTree;
+import org.apache.zookeeper.server.persistence.FileHeader;
+import org.apache.zookeeper.server.persistence.FileTxnLog;
+import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
+import org.apache.zookeeper.txn.CreateTxn;
+import org.apache.zookeeper.txn.DeleteTxn;
+import org.apache.zookeeper.txn.MultiTxn;
+import org.apache.zookeeper.txn.Txn;
+import org.apache.zookeeper.txn.TxnHeader;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+public class LoadFromLogNoServerTest extends ZKTestCase {
+ protected static final Logger LOG = LoggerFactory.getLogger(LoadFromLogNoServerTest.class);
+
+ /**
+ * For ZOOKEEPER-1046. Verify if cversion and pzxid if incremented
+ * after create/delete failure during restore.
+ */
+ @Test
+ public void testTxnFailure() throws Exception {
+ long count = 1;
+ File tmpDir = ClientBase.createTmpDir();
+ FileTxnSnapLog logFile = new FileTxnSnapLog(tmpDir, tmpDir);
+ DataTree dt = new DataTree();
+ dt.createNode("/test", new byte[0], null, 0, -1, 1, 1);
+ for (count = 1; count <= 3; count++) {
+ dt.createNode("/test/" + count, new byte[0], null, 0, -1, count,
+ Time.currentElapsedTime());
+ }
+ DataNode zk = dt.getNode("/test");
+
+ // Make create to fail, then verify cversion.
+ LOG.info("Attempting to create " + "/test/" + (count - 1));
+ doOp(logFile, ZooDefs.OpCode.create, "/test/" + (count - 1), dt, zk, -1);
+
+ LOG.info("Attempting to create " + "/test/" + (count - 1));
+ doOp(logFile, ZooDefs.OpCode.create, "/test/" + (count - 1), dt, zk,
+ zk.stat.getCversion() + 1);
+
+ LOG.info("Attempting to create " + "/test/" + (count - 1));
+ doOp(logFile, ZooDefs.OpCode.multi, "/test/" + (count - 1), dt, zk,
+ zk.stat.getCversion() + 1);
+
+ LOG.info("Attempting to create " + "/test/" + (count - 1));
+ doOp(logFile, ZooDefs.OpCode.multi, "/test/" + (count - 1), dt, zk,
+ -1);
+
+ // Make delete fo fail, then verify cversion.
+ // this doesn't happen anymore, we only set the cversion on create
+ // LOG.info("Attempting to delete " + "/test/" + (count + 1));
+ // doOp(logFile, OpCode.delete, "/test/" + (count + 1), dt, zk);
+ }
+
+ /*
+ * Does create/delete depending on the type and verifies
+ * if cversion before the operation is 1 less than cversion afer.
+ */
+ private void doOp(FileTxnSnapLog logFile, int type, String path,
+ DataTree dt, DataNode parent, int cversion) throws Exception {
+ int lastSlash = path.lastIndexOf('/');
+ String parentName = path.substring(0, lastSlash);
+
+ int prevCversion = parent.stat.getCversion();
+ long prevPzxid = parent.stat.getPzxid();
+ List<String> child = dt.getChildren(parentName, null, null);
+ StringBuilder childStr = new StringBuilder();
+ for (String s : child) {
+ childStr.append(s).append(" ");
+ }
+ LOG.info("Children: " + childStr + " for " + parentName);
+ LOG.info("(cverions, pzxid): " + prevCversion + ", " + prevPzxid);
+
+ Record txn = null;
+ TxnHeader txnHeader = null;
+ if (type == ZooDefs.OpCode.delete) {
+ txn = new DeleteTxn(path);
+ txnHeader = new TxnHeader(0xabcd, 0x123, prevPzxid + 1,
+ Time.currentElapsedTime(), ZooDefs.OpCode.delete);
+ } else if (type == ZooDefs.OpCode.create) {
+ txnHeader = new TxnHeader(0xabcd, 0x123, prevPzxid + 1,
+ Time.currentElapsedTime(), ZooDefs.OpCode.create);
+ txn = new CreateTxn(path, new byte[0], null, false, cversion);
+ }
+ else if (type == ZooDefs.OpCode.multi) {
+ txnHeader = new TxnHeader(0xabcd, 0x123, prevPzxid + 1,
+ Time.currentElapsedTime(), ZooDefs.OpCode.create);
+ txn = new CreateTxn(path, new byte[0], null, false, cversion);
+ List<Txn> txnList = new ArrayList<Txn>();
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
+ txn.serialize(boa, "request") ;
+ ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
+ Txn txact = new Txn(ZooDefs.OpCode.create, bb.array());
+ txnList.add(txact);
+ txn = new MultiTxn(txnList);
+ txnHeader = new TxnHeader(0xabcd, 0x123, prevPzxid + 1,
+ Time.currentElapsedTime(), ZooDefs.OpCode.multi);
+ }
+ logFile.processTransaction(txnHeader, dt, null, txn);
+
+ int newCversion = parent.stat.getCversion();
+ long newPzxid = parent.stat.getPzxid();
+ child = dt.getChildren(parentName, null, null);
+ childStr = new StringBuilder();
+ for (String s : child) {
+ childStr.append(s).append(" ");
+ }
+ LOG.info("Children: " + childStr + " for " + parentName);
+ LOG.info("(cverions, pzxid): " +newCversion + ", " + newPzxid);
+ Assert.assertTrue(type + " <cversion, pzxid> verification failed. Expected: <" +
+ (prevCversion + 1) + ", " + (prevPzxid + 1) + ">, found: <" +
+ newCversion + ", " + newPzxid + ">",
+ (newCversion == prevCversion + 1 && newPzxid == prevPzxid + 1));
+ }
+
+ /**
+ * Simulates ZOOKEEPER-1069 and verifies that flush() before padLogFile
+ * fixes it.
+ */
+ @Test
+ public void testPad() throws Exception {
+ File tmpDir = ClientBase.createTmpDir();
+ FileTxnLog txnLog = new FileTxnLog(tmpDir);
+ TxnHeader txnHeader = new TxnHeader(0xabcd, 0x123, 0x123,
+ Time.currentElapsedTime(), ZooDefs.OpCode.create);
+ Record txn = new CreateTxn("/Test", new byte[0], null, false, 1);
+ txnLog.append(txnHeader, txn);
+ FileInputStream in = new FileInputStream(tmpDir.getPath() + "/log." +
+ Long.toHexString(txnHeader.getZxid()));
+ BinaryInputArchive ia = BinaryInputArchive.getArchive(in);
+ FileHeader header = new FileHeader();
+ header.deserialize(ia, "fileheader");
+ LOG.info("Received magic : " + header.getMagic() +
+ " Expected : " + FileTxnLog.TXNLOG_MAGIC);
+ Assert.assertTrue("Missing magic number ",
+ header.getMagic() == FileTxnLog.TXNLOG_MAGIC);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/LoadFromLogTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/LoadFromLogTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/LoadFromLogTest.java
new file mode 100644
index 0000000..90de755
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/LoadFromLogTest.java
@@ -0,0 +1,310 @@
+/**
+ * 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.zookeeper.test;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException.NoNodeException;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.server.SyncRequestProcessor;
+import org.apache.zookeeper.server.ZooKeeperServer;
+import org.apache.zookeeper.server.persistence.FileTxnLog;
+import org.apache.zookeeper.server.persistence.FileTxnLog.FileTxnIterator;
+import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
+import org.apache.zookeeper.server.persistence.TxnLog.TxnIterator;
+import org.apache.zookeeper.server.persistence.Util;
+import org.apache.zookeeper.txn.TxnHeader;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+
+public class LoadFromLogTest extends ClientBase {
+ private static final int NUM_MESSAGES = 300;
+ protected static final Logger LOG = LoggerFactory.getLogger(LoadFromLogTest.class);
+
+ // setting up the quorum has a transaction overhead for creating and closing the session
+ private static final int TRANSACTION_OVERHEAD = 2;
+ private static final int TOTAL_TRANSACTIONS = NUM_MESSAGES + TRANSACTION_OVERHEAD;
+
+ @Before
+ public void setUp() throws Exception {
+ SyncRequestProcessor.setSnapCount(50);
+ super.setUp();
+ }
+
+ /**
+ * test that all transactions from the Log are loaded, and only once
+ * @throws Exception an exception might be thrown here
+ */
+ @Test
+ public void testLoad() throws Exception {
+ // generate some transactions that will get logged
+ ZooKeeper zk = createZKClient(hostPort);
+ try {
+ for (int i = 0; i< NUM_MESSAGES; i++) {
+ zk.create("/invalidsnap-" + i, new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ }
+ } finally {
+ zk.close();
+ }
+ stopServer();
+
+ // now verify that the FileTxnLog reads every transaction only once
+ File logDir = new File(tmpDir, FileTxnSnapLog.version + FileTxnSnapLog.VERSION);
+ FileTxnLog txnLog = new FileTxnLog(logDir);
+ TxnIterator itr = txnLog.read(0);
+
+ // Check that storage space return some value
+ FileTxnIterator fileItr = (FileTxnIterator) itr;
+ long storageSize = fileItr.getStorageSize();
+ LOG.info("Txnlog size: " + storageSize + " bytes");
+ Assert.assertTrue("Storage size is greater than zero ",
+ (storageSize > 0));
+
+ long expectedZxid = 0;
+ long lastZxid = 0;
+ TxnHeader hdr;
+ do {
+ hdr = itr.getHeader();
+ expectedZxid++;
+ Assert.assertTrue("not the same transaction. lastZxid=" + lastZxid + ", zxid=" + hdr.getZxid(), lastZxid != hdr.getZxid());
+ Assert.assertTrue("excepting next transaction. expected=" + expectedZxid + ", retreived=" + hdr.getZxid(), (hdr.getZxid() == expectedZxid));
+ lastZxid = hdr.getZxid();
+ }while(itr.next());
+
+ Assert.assertTrue("processed all transactions. " + expectedZxid + " == " + TOTAL_TRANSACTIONS, (expectedZxid == TOTAL_TRANSACTIONS));
+ }
+
+ /**
+ * test that we fail to load txnlog of a request zxid that is older
+ * than what exist on disk
+ * @throws Exception an exception might be thrown here
+ */
+ @Test
+ public void testLoadFailure() throws Exception {
+ // generate some transactions that will get logged
+ ZooKeeper zk = createZKClient(hostPort);
+ try {
+ for (int i = 0; i< NUM_MESSAGES; i++) {
+ zk.create("/data-", new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT_SEQUENTIAL);
+ }
+ } finally {
+ zk.close();
+ }
+ stopServer();
+
+ File logDir = new File(tmpDir, FileTxnSnapLog.version + FileTxnSnapLog.VERSION);
+ File[] logFiles = FileTxnLog.getLogFiles(logDir.listFiles(), 0);
+ // Verify that we have at least NUM_MESSAGES / SNAPCOUNT txnlog
+ Assert.assertTrue(logFiles.length > NUM_MESSAGES / 100);
+ // Delete the first log file, so we will fail to read it back from disk
+ Assert.assertTrue("delete the first log file", logFiles[0].delete());
+
+ // Find zxid for the second log
+ long secondStartZxid = Util.getZxidFromName(logFiles[1].getName(), "log");
+
+ FileTxnLog txnLog = new FileTxnLog(logDir);
+ TxnIterator itr = txnLog.read(1, false);
+
+ // Oldest log is already remove, so this should point to the start of
+ // of zxid on the second log
+ Assert.assertEquals(secondStartZxid, itr.getHeader().getZxid());
+
+ itr = txnLog.read(secondStartZxid, false);
+ Assert.assertEquals(secondStartZxid, itr.getHeader().getZxid());
+ Assert.assertTrue(itr.next());
+
+ // Trying to get a second txn on second txnlog give us the
+ // the start of second log, since the first one is removed
+ long nextZxid = itr.getHeader().getZxid();
+
+ itr = txnLog.read(nextZxid, false);
+ Assert.assertEquals(secondStartZxid, itr.getHeader().getZxid());
+
+ // Trying to get a first txn on the third give us the
+ // the start of second log, since the first one is removed
+ long thirdStartZxid = Util.getZxidFromName(logFiles[2].getName(), "log");
+ itr = txnLog.read(thirdStartZxid, false);
+ Assert.assertEquals(secondStartZxid, itr.getHeader().getZxid());
+ Assert.assertTrue(itr.next());
+
+ nextZxid = itr.getHeader().getZxid();
+ itr = txnLog.read(nextZxid, false);
+ Assert.assertEquals(secondStartZxid, itr.getHeader().getZxid());
+ }
+
+ /**
+ * Test we can restore the snapshot that has data ahead of the zxid
+ * of the snapshot file.
+ */
+ @Test
+ public void testRestore() throws Exception {
+ // generate some transactions
+ ZooKeeper zk = createZKClient(hostPort);
+ String lastPath = null;
+ try {
+ zk.create("/invalidsnap", new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ for (int i = 0; i < NUM_MESSAGES; i++) {
+ lastPath = zk.create("/invalidsnap/test-", new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
+ }
+ } finally {
+ zk.close();
+ }
+ String[] tokens = lastPath.split("-");
+ String expectedPath = "/invalidsnap/test-"
+ + String.format("%010d",
+ (new Integer(tokens[1])).intValue() + 1);
+ ZooKeeperServer zks = getServer(serverFactory);
+ long eZxid = zks.getZKDatabase().getDataTreeLastProcessedZxid();
+ // force the zxid to be behind the content
+ zks.getZKDatabase().setlastProcessedZxid(
+ zks.getZKDatabase().getDataTreeLastProcessedZxid() - 10);
+ LOG.info("Set lastProcessedZxid to "
+ + zks.getZKDatabase().getDataTreeLastProcessedZxid());
+ // Force snapshot and restore
+ zks.takeSnapshot();
+ zks.shutdown();
+ stopServer();
+
+ startServer();
+ zks = getServer(serverFactory);
+ long fZxid = zks.getZKDatabase().getDataTreeLastProcessedZxid();
+
+ // Verify lastProcessedZxid is set correctly
+ Assert.assertTrue("Restore failed expected zxid=" + eZxid + " found="
+ + fZxid, fZxid == eZxid);
+ zk = createZKClient(hostPort);
+
+ // Verify correctness of data and whether sequential znode creation
+ // proceeds correctly after this point
+ String[] children;
+ String path;
+ try {
+ children = zk.getChildren("/invalidsnap", false).toArray(
+ new String[0]);
+ path = zk.create("/invalidsnap/test-", new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
+ } finally {
+ zk.close();
+ }
+ LOG.info("Expected " + expectedPath + " found " + path);
+ Assert.assertTrue("Error in sequential znode creation expected "
+ + expectedPath + " found " + path, path.equals(expectedPath));
+ Assert.assertTrue("Unexpected number of children " + children.length
+ + " expected " + NUM_MESSAGES,
+ (children.length == NUM_MESSAGES));
+ }
+
+ /**
+ * Test we can restore a snapshot that has errors and data ahead of the zxid
+ * of the snapshot file.
+ */
+ @Test
+ public void testRestoreWithTransactionErrors() throws Exception {
+ // generate some transactions
+ ZooKeeper zk = createZKClient(hostPort);
+ try {
+ for (int i = 0; i < NUM_MESSAGES; i++) {
+ try {
+ zk.create("/invaliddir/test-", new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
+ } catch(NoNodeException e) {
+ //Expected
+ }
+ }
+ } finally {
+ zk.close();
+ }
+
+ // force the zxid to be behind the content
+ ZooKeeperServer zks = getServer(serverFactory);
+ zks.getZKDatabase().setlastProcessedZxid(
+ zks.getZKDatabase().getDataTreeLastProcessedZxid() - 10);
+ LOG.info("Set lastProcessedZxid to "
+ + zks.getZKDatabase().getDataTreeLastProcessedZxid());
+
+ // Force snapshot and restore
+ zks.takeSnapshot();
+ zks.shutdown();
+ stopServer();
+
+ zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
+ startServer();
+ }
+
+ /**
+ * Verify snap/log dir create with/without autocreate enabled.
+ */
+ @Test
+ public void testDatadirAutocreate() throws Exception {
+ stopServer();
+
+ try {
+ // now verify autocreate off works
+ System.setProperty(FileTxnSnapLog.ZOOKEEPER_DATADIR_AUTOCREATE, "false");
+ tmpDir = createTmpDir();
+ startServer();
+ Assert.fail("Server should not have started without datadir");
+ } catch (IOException e) {
+ LOG.info("Server failed to start - correct behavior " + e);
+ } finally {
+ System.setProperty(FileTxnSnapLog.ZOOKEEPER_DATADIR_AUTOCREATE,
+ FileTxnSnapLog.ZOOKEEPER_DATADIR_AUTOCREATE_DEFAULT);
+ }
+ }
+
+ /**
+ * ZOOKEEPER-1573: test restoring a snapshot with deleted txns ahead of the
+ * snapshot file's zxid.
+ */
+ @Test
+ public void testReloadSnapshotWithMissingParent() throws Exception {
+ // create transactions to create the snapshot with create/delete pattern
+ ZooKeeper zk = createZKClient(hostPort);
+ zk.create("/a", "".getBytes(), Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ Stat stat = zk.exists("/a", false);
+ long createZxId = stat.getMzxid();
+ zk.create("/a/b", "".getBytes(), Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ zk.delete("/a/b", -1);
+ zk.delete("/a", -1);
+ // force the zxid to be behind the content
+ ZooKeeperServer zks = getServer(serverFactory);
+ zks.getZKDatabase().setlastProcessedZxid(createZxId);
+ LOG.info("Set lastProcessedZxid to {}", zks.getZKDatabase()
+ .getDataTreeLastProcessedZxid());
+ // Force snapshot and restore
+ zks.takeSnapshot();
+ zks.shutdown();
+ stopServer();
+
+ startServer();
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/LocalSessionRequestTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/LocalSessionRequestTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/LocalSessionRequestTest.java
new file mode 100644
index 0000000..603cd15
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/LocalSessionRequestTest.java
@@ -0,0 +1,148 @@
+/**
+ * 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.zookeeper.test;
+
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.server.TraceFormatter;
+import org.apache.zookeeper.server.ZKDatabase;
+import org.apache.zookeeper.server.quorum.Leader.Proposal;
+import org.apache.zookeeper.server.quorum.QuorumPeer;
+import org.apache.zookeeper.test.ClientBase.CountdownWatcher;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Validate that open/close session request of a local session to not propagate
+ * to other machines in the quorum. We verify this by checking that
+ * these request doesn't show up in committedLog on other machines.
+ */
+public class LocalSessionRequestTest extends ZKTestCase {
+ protected static final Logger LOG = LoggerFactory
+ .getLogger(LocalSessionRequestTest.class);
+ // Need to be short since we need to wait for session to expire
+ public static final int CONNECTION_TIMEOUT = 4000;
+
+ private final QuorumBase qb = new QuorumBase();
+
+ @Before
+ public void setUp() throws Exception {
+ LOG.info("STARTING quorum " + getClass().getName());
+ qb.localSessionsEnabled = true;
+ qb.localSessionsUpgradingEnabled = true;
+ qb.setUp();
+ ClientBase.waitForServerUp(qb.hostPort, 10000);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ LOG.info("STOPPING quorum " + getClass().getName());
+ qb.tearDown();
+ }
+
+ @Test
+ public void testLocalSessionsOnFollower() throws Exception {
+ testOpenCloseSession(false);
+ }
+
+ @Test
+ public void testLocalSessionsOnLeader() throws Exception {
+ testOpenCloseSession(true);
+ }
+
+ /**
+ * Walk through the target peer commmittedLog.
+ * @param sessionId
+ * @param peerId
+ */
+ private void validateRequestLog(long sessionId, int peerId) {
+ String session = Long.toHexString(sessionId);
+ LOG.info("Searching for txn of session 0x " + session +
+ " on peer " + peerId);
+ String peerType = peerId == qb.getLeaderIndex() ? "leader" : "follower";
+ QuorumPeer peer = qb.getPeerList().get(peerId);
+ ZKDatabase db = peer.getActiveServer().getZKDatabase();
+ for (Proposal p : db.getCommittedLog()) {
+ Assert.assertFalse("Should not see " +
+ TraceFormatter.op2String(p.request.type) +
+ " request from local session 0x" + session +
+ " on the " + peerType,
+ p.request.sessionId == sessionId);
+ }
+ }
+
+ /**
+ * Test that a CloseSession request generated by both the server (client
+ * disconnect) or by the client (client explicitly issue close()) doesn't
+ * get committed by the ensemble
+ */
+ public void testOpenCloseSession(boolean onLeader) throws Exception {
+ int leaderIdx = qb.getLeaderIndex();
+ Assert.assertFalse("No leader in quorum?", leaderIdx == -1);
+ int followerIdx = (leaderIdx + 1) % 5;
+ int testPeerIdx = onLeader ? leaderIdx : followerIdx;
+ int verifyPeerIdx = onLeader ? followerIdx : leaderIdx;
+
+ String hostPorts[] = qb.hostPort.split(",");
+
+ CountdownWatcher watcher = new CountdownWatcher();
+ DisconnectableZooKeeper client = new DisconnectableZooKeeper(
+ hostPorts[testPeerIdx], CONNECTION_TIMEOUT, watcher);
+ watcher.waitForConnected(CONNECTION_TIMEOUT);
+
+ long localSessionId1 = client.getSessionId();
+
+ // Cut the connection, so the server will create closeSession as part
+ // of expiring the session.
+ client.dontReconnect();
+ client.disconnect();
+ watcher.reset();
+
+ // We don't validate right away, will do another session create first
+
+ ZooKeeper zk = qb.createClient(watcher, hostPorts[testPeerIdx],
+ CONNECTION_TIMEOUT);
+ watcher.waitForConnected(CONNECTION_TIMEOUT);
+
+ long localSessionId2 = zk.getSessionId();
+
+ // Send closeSession request.
+ zk.close();
+ watcher.reset();
+
+ // This should be enough time for the first session to expire and for
+ // the closeSession request to propagate to other machines (if there is a bug)
+ // Since it is time sensitive, we have false negative when test
+ // machine is under load
+ Thread.sleep(CONNECTION_TIMEOUT * 2);
+
+ // Validate that we don't see any txn from the first session
+ validateRequestLog(localSessionId1, verifyPeerIdx);
+
+ // Validate that we don't see any txn from the second session
+ validateRequestLog(localSessionId2, verifyPeerIdx);
+
+ qb.shutdownServers();
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/LocalSessionsOnlyTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/LocalSessionsOnlyTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/LocalSessionsOnlyTest.java
new file mode 100644
index 0000000..3e240e0
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/LocalSessionsOnlyTest.java
@@ -0,0 +1,137 @@
+/**
+ * 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.zookeeper.test;
+
+import java.util.HashMap;
+import java.util.Map.Entry;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.test.ClientBase.CountdownWatcher;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests learners configured to use local sessions only. Expected
+ * behavior is that sessions created on the learner will never be
+ * made global. Operations requiring a global session (e.g.
+ * creation of ephemeral nodes) will fail with an error.
+ */
+public class LocalSessionsOnlyTest extends ZKTestCase {
+ protected static final Logger LOG = LoggerFactory
+ .getLogger(LocalSessionsOnlyTest.class);
+ public static final int CONNECTION_TIMEOUT = ClientBase.CONNECTION_TIMEOUT;
+
+ private final QuorumBase qb = new QuorumBase();
+
+ @Before
+ public void setUp() throws Exception {
+ LOG.info("STARTING quorum " + getClass().getName());
+ qb.localSessionsEnabled = true;
+ qb.localSessionsUpgradingEnabled = false;
+ qb.setUp();
+ ClientBase.waitForServerUp(qb.hostPort, 10000);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ LOG.info("STOPPING quorum " + getClass().getName());
+ qb.tearDown();
+ }
+
+ @Test
+ public void testLocalSessionsOnFollower() throws Exception {
+ testLocalSessions(false);
+ }
+
+ @Test
+ public void testLocalSessionsOnLeader() throws Exception {
+ testLocalSessions(true);
+ }
+
+ private void testLocalSessions(boolean testLeader) throws Exception {
+ String nodePrefix = "/testLocalSessions-"
+ + (testLeader ? "leaderTest-" : "followerTest-");
+ int leaderIdx = qb.getLeaderIndex();
+ Assert.assertFalse("No leader in quorum?", leaderIdx == -1);
+ int followerIdx = (leaderIdx + 1) % 5;
+ int testPeerIdx = testLeader ? leaderIdx : followerIdx;
+ String hostPorts[] = qb.hostPort.split(",");
+
+ CountdownWatcher watcher = new CountdownWatcher();
+ ZooKeeper zk = qb.createClient(watcher, hostPorts[testPeerIdx],
+ CONNECTION_TIMEOUT);
+ watcher.waitForConnected(CONNECTION_TIMEOUT);
+
+ long localSessionId = zk.getSessionId();
+
+ // Try creating some data.
+ for (int i = 0; i < 5; i++) {
+ zk.create(nodePrefix + i, new byte[0],
+ ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ }
+
+ // Now, try an ephemeral node. This should fail since we
+ // cannot create ephemeral nodes on a local session.
+ try {
+ zk.create(nodePrefix + "ephemeral", new byte[0],
+ ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
+ Assert.fail("Ephemeral node creation should fail.");
+ } catch (KeeperException.EphemeralOnLocalSessionException e) {
+ }
+
+ // Close the session.
+ zk.close();
+
+ // Validate data on both follower and leader
+ HashMap<String, Integer> peers = new HashMap<String, Integer>();
+ peers.put("leader", leaderIdx);
+ peers.put("follower", followerIdx);
+ for (Entry<String, Integer> entry: peers.entrySet()) {
+ watcher.reset();
+ // Try reconnecting with a new session.
+ // The data should be persisted, even though the session was not.
+ zk = qb.createClient(watcher, hostPorts[entry.getValue()],
+ CONNECTION_TIMEOUT);
+ watcher.waitForConnected(CONNECTION_TIMEOUT);
+
+ long newSessionId = zk.getSessionId();
+ Assert.assertFalse(newSessionId == localSessionId);
+
+ for (int i = 0; i < 5; i++) {
+ Assert.assertNotNull("Data not exists in " + entry.getKey(),
+ zk.exists(nodePrefix + i, null));
+ }
+
+ // We may get the correct exception but the txn may go through
+ Assert.assertNull("Data exists in " + entry.getKey(),
+ zk.exists(nodePrefix + "ephemeral", null));
+
+ zk.close();
+ }
+ qb.shutdownServers();
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/MaxCnxnsTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/MaxCnxnsTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/MaxCnxnsTest.java
new file mode 100644
index 0000000..a96e5a8
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/MaxCnxnsTest.java
@@ -0,0 +1,136 @@
+/**
+ * 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.zookeeper.test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.jute.BinaryOutputArchive;
+import org.apache.zookeeper.proto.ConnectRequest;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class MaxCnxnsTest extends ClientBase {
+ final private static int numCnxns = 30;
+ AtomicInteger numConnected = new AtomicInteger(0);
+ String host;
+ int port;
+
+ @Override
+ public void setUp() throws Exception {
+ maxCnxns = numCnxns;
+ super.setUp();
+ }
+
+ class CnxnThread extends Thread {
+
+ public CnxnThread(int i) {
+ super("CnxnThread-"+i);
+ }
+
+ public void run() {
+ SocketChannel sChannel = null;
+ try {
+ /*
+ * For future unwary socket programmers: although connect 'blocks' it
+ * does not require an accept on the server side to return. Therefore
+ * you can not assume that all the sockets are connected at the end of
+ * this for loop.
+ */
+ sChannel = SocketChannel.open();
+ sChannel.connect(new InetSocketAddress(host,port));
+ // Construct a connection request
+ ConnectRequest conReq = new ConnectRequest(0, 0,
+ 10000, 0, "password".getBytes());
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
+ boa.writeInt(-1, "len");
+ conReq.serialize(boa, "connect");
+ baos.close();
+ ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
+ bb.putInt(bb.capacity() - 4);
+ bb.rewind();
+
+ /* Send a connect request. Any socket that has been closed (or at least
+ * not added to the cnxn list on the server) will not have any bytes to
+ * read and get an eof.
+ *
+ * The trick here was finding a call that caused the server to put
+ * bytes in the input stream without closing the cnxn. None of
+ * the four letter commands do that, so we actually try to create
+ * a session which should send us something back, while maintaining
+ * the connection.
+ */
+
+ int eof = sChannel.write(bb);
+ // If the socket times out, we count that as Assert.failed -
+ // the server should respond within 10s
+ sChannel.socket().setSoTimeout(10000);
+ if (!sChannel.socket().isClosed()){
+ eof = sChannel.socket().getInputStream().read();
+ if (eof != -1) {
+ numConnected.incrementAndGet();
+ }
+ }
+ }
+ catch (IOException io) {
+ // "Connection reset by peer"
+ } finally {
+ if (sChannel != null) {
+ try {
+ sChannel.close();
+ } catch (Exception e) {
+ // Do nothing
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * Verify the ability to limit the number of concurrent connections.
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ @Test
+ public void testMaxCnxns() throws IOException, InterruptedException{
+ String split[] = hostPort.split(":");
+ host = split[0];
+ port = Integer.parseInt(split[1]);
+ int numThreads = numCnxns + 5;
+ CnxnThread[] threads = new CnxnThread[numThreads];
+
+ for (int i=0;i<numCnxns;++i) {
+ threads[i] = new CnxnThread(i);
+ }
+
+ for (int i=0;i<numCnxns;++i) {
+ threads[i].start();
+ }
+
+ for (int i=0;i<numCnxns;++i) {
+ threads[i].join();
+ }
+ Assert.assertSame(numCnxns,numConnected.get());
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/MultiAsyncTransactionTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/MultiAsyncTransactionTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/MultiAsyncTransactionTest.java
new file mode 100644
index 0000000..822b9c5
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/MultiAsyncTransactionTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.zookeeper.test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.zookeeper.AsyncCallback.MultiCallback;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Op;
+import org.apache.zookeeper.OpResult;
+import org.apache.zookeeper.OpResult.CreateResult;
+import org.apache.zookeeper.OpResult.ErrorResult;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooKeeper;
+import org.junit.Before;
+import org.junit.Test;
+
+public class MultiAsyncTransactionTest extends ClientBase {
+ private ZooKeeper zk;
+ private final AtomicInteger pendingOps = new AtomicInteger(0);
+
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ zk = createClient();
+ pendingOps.set(0);
+ }
+
+ private static class MultiResult {
+ int rc;
+ List<OpResult> results;
+ }
+
+ private void finishPendingOps() {
+ if (pendingOps.decrementAndGet() == 0) {
+ synchronized (pendingOps) {
+ pendingOps.notifyAll();
+ }
+ }
+ }
+
+ private void waitForPendingOps(int timeout) throws Exception {
+ synchronized(pendingOps) {
+ while(pendingOps.get() > 0) {
+ pendingOps.wait(timeout);
+ }
+ }
+ }
+
+ /**
+ * ZOOKEEPER-1624: PendingChanges of create sequential node request didn't
+ * get rollbacked correctly when multi-op failed. This cause
+ * create sequential node request in subsequent multi-op to failed because
+ * sequential node name generation is incorrect.
+ *
+ * The check is to make sure that each request in multi-op failed with
+ * the correct reason.
+ */
+ @Test
+ public void testSequentialNodeCreateInAsyncMulti() throws Exception {
+ final int iteration = 4;
+ final List<MultiResult> results = new ArrayList<MultiResult>();
+
+ pendingOps.set(iteration);
+
+ List<Op> ops = Arrays.asList(
+ Op.create("/node-", new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT_SEQUENTIAL),
+ Op.create("/dup", new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT));
+
+
+ for (int i = 0; i < iteration; ++i) {
+ zk.multi(ops, new MultiCallback() {
+ @Override
+ public void processResult(int rc, String path, Object ctx,
+ List<OpResult> opResults) {
+ MultiResult result = new MultiResult();
+ result.results = opResults;
+ result.rc = rc;
+ results.add(result);
+ finishPendingOps();
+ }
+ }, null);
+ }
+
+ waitForPendingOps(CONNECTION_TIMEOUT);
+
+ // Check that return code of all request are correct
+ assertEquals(KeeperException.Code.OK.intValue(), results.get(0).rc);
+ assertEquals(KeeperException.Code.NODEEXISTS.intValue(), results.get(1).rc);
+ assertEquals(KeeperException.Code.NODEEXISTS.intValue(), results.get(2).rc);
+ assertEquals(KeeperException.Code.NODEEXISTS.intValue(), results.get(3).rc);
+
+ // Check that the first operation is successful in all request
+ assertTrue(results.get(0).results.get(0) instanceof CreateResult);
+ assertEquals(KeeperException.Code.OK.intValue(),
+ ((ErrorResult) results.get(1).results.get(0)).getErr());
+ assertEquals(KeeperException.Code.OK.intValue(),
+ ((ErrorResult) results.get(2).results.get(0)).getErr());
+ assertEquals(KeeperException.Code.OK.intValue(),
+ ((ErrorResult) results.get(3).results.get(0)).getErr());
+
+ // Check that the second operation failed after the first request
+ assertEquals(KeeperException.Code.NODEEXISTS.intValue(),
+ ((ErrorResult) results.get(1).results.get(1)).getErr());
+ assertEquals(KeeperException.Code.NODEEXISTS.intValue(),
+ ((ErrorResult) results.get(2).results.get(1)).getErr());
+ assertEquals(KeeperException.Code.NODEEXISTS.intValue(),
+ ((ErrorResult) results.get(3).results.get(1)).getErr());
+
+ }
+}
\ No newline at end of file