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/05 13:36:38 UTC
[10/51] [partial] zookeeper git commit: ZOOKEEPER-3032: MAVEN
MIGRATION - branch-3.5 - move java server, client
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/test/org/apache/zookeeper/server/admin/JettyAdminServerTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/admin/JettyAdminServerTest.java b/src/java/test/org/apache/zookeeper/server/admin/JettyAdminServerTest.java
deleted file mode 100644
index bc8aab6..0000000
--- a/src/java/test/org/apache/zookeeper/server/admin/JettyAdminServerTest.java
+++ /dev/null
@@ -1,162 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server.admin;
-
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.net.MalformedURLException;
-import java.net.URL;
-
-import org.apache.zookeeper.PortAssignment;
-import org.apache.zookeeper.ZKTestCase;
-import org.apache.zookeeper.server.ZooKeeperServerMainTest;
-import org.apache.zookeeper.server.admin.AdminServer.AdminServerException;
-import org.apache.zookeeper.server.quorum.QuorumPeerTestBase;
-import org.apache.zookeeper.test.ClientBase;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class JettyAdminServerTest extends ZKTestCase{
- protected static final Logger LOG = LoggerFactory.getLogger(JettyAdminServerTest.class);
-
- private static final String URL_FORMAT = "http://localhost:%d/commands";
- private static final int jettyAdminPort = PortAssignment.unique();
-
- @Before
- public void enableServer() {
- // Override setting in ZKTestCase
- System.setProperty("zookeeper.admin.enableServer", "true");
- System.setProperty("zookeeper.admin.serverPort", "" + jettyAdminPort);
- }
-
- /**
- * Tests that we can start and query a JettyAdminServer.
- */
- @Test
- public void testJettyAdminServer() throws AdminServerException, IOException {
- JettyAdminServer server = new JettyAdminServer();;
- try {
- server.start();
- queryAdminServer(jettyAdminPort);
- } finally {
- server.shutdown();
- }
- }
-
- /**
- * Starts a standalone server and tests that we can query its AdminServer.
- */
- @Test
- public void testStandalone() throws Exception {
- ClientBase.setupTestEnv();
-
- final int CLIENT_PORT = PortAssignment.unique();
-
- ZooKeeperServerMainTest.MainThread main = new ZooKeeperServerMainTest.MainThread(CLIENT_PORT, false, null);
- main.start();
-
- Assert.assertTrue("waiting for server being up",
- ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT,
- ClientBase.CONNECTION_TIMEOUT));
-
- queryAdminServer(jettyAdminPort);
-
- main.shutdown();
-
- Assert.assertTrue("waiting for server down",
- ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT,
- ClientBase.CONNECTION_TIMEOUT));
- }
-
- /**
- * Starts a quorum of two servers and tests that we can query both AdminServers.
- */
- @Test
- public void testQuorum() throws Exception {
- ClientBase.setupTestEnv();
-
- final int CLIENT_PORT_QP1 = PortAssignment.unique();
- final int CLIENT_PORT_QP2 = PortAssignment.unique();
-
- final int ADMIN_SERVER_PORT1 = PortAssignment.unique();
- final int ADMIN_SERVER_PORT2 = PortAssignment.unique();
-
- String quorumCfgSection = String.format
- ("server.1=127.0.0.1:%d:%d;%d\nserver.2=127.0.0.1:%d:%d;%d",
- PortAssignment.unique(), PortAssignment.unique(), CLIENT_PORT_QP1,
- PortAssignment.unique(), PortAssignment.unique(), CLIENT_PORT_QP2
- );
- QuorumPeerTestBase.MainThread q1 = new QuorumPeerTestBase.MainThread(
- 1, CLIENT_PORT_QP1, ADMIN_SERVER_PORT1, quorumCfgSection, null);
- q1.start();
-
- // Since JettyAdminServer reads a system property to determine its port,
- // make sure it initializes itself before setting the system property
- // again with the second port number
- Thread.sleep(500);
-
- QuorumPeerTestBase.MainThread q2 = new QuorumPeerTestBase.MainThread(
- 2, CLIENT_PORT_QP2, ADMIN_SERVER_PORT2, quorumCfgSection, null);
- q2.start();
-
- Thread.sleep(500);
-
- Assert.assertTrue("waiting for server 1 being up",
- ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_QP1,
- ClientBase.CONNECTION_TIMEOUT));
- Assert.assertTrue("waiting for server 2 being up",
- ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_QP2,
- ClientBase.CONNECTION_TIMEOUT));
-
- queryAdminServer(ADMIN_SERVER_PORT1);
- queryAdminServer(ADMIN_SERVER_PORT2);
-
- q1.shutdown();
- q2.shutdown();
-
- Assert.assertTrue("waiting for server 1 down",
- ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT_QP1,
- ClientBase.CONNECTION_TIMEOUT));
- Assert.assertTrue("waiting for server 2 down",
- ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT_QP2,
- ClientBase.CONNECTION_TIMEOUT));
- }
-
- /**
- * Check that we can load the commands page of an AdminServer running at
- * localhost:port. (Note that this should work even if no zk server is set.)
- */
- private void queryAdminServer(int port) throws MalformedURLException, IOException {
- queryAdminServer(String.format(URL_FORMAT, port));
- }
-
- /**
- * Check that loading urlStr results in a non-zero length response.
- */
- private void queryAdminServer(String urlStr) throws MalformedURLException, IOException {
- URL url = new URL(urlStr);
- BufferedReader dis = new BufferedReader(new InputStreamReader((url.openStream())));
- String line = dis.readLine();
- Assert.assertTrue(line.length() > 0);
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/test/org/apache/zookeeper/server/persistence/FileTxnLogTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/persistence/FileTxnLogTest.java b/src/java/test/org/apache/zookeeper/server/persistence/FileTxnLogTest.java
deleted file mode 100644
index 77b7210..0000000
--- a/src/java/test/org/apache/zookeeper/server/persistence/FileTxnLogTest.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.zookeeper.server.persistence;
-
-import org.apache.zookeeper.ZKTestCase;
-import org.apache.zookeeper.ZooDefs;
-import org.apache.zookeeper.server.ServerStats;
-import org.apache.zookeeper.test.ClientBase;
-import org.apache.zookeeper.txn.CreateTxn;
-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.File;
-import java.io.IOException;
-import java.util.Arrays;
-
-import static org.hamcrest.core.Is.is;
-import static org.hamcrest.core.IsEqual.equalTo;
-import static org.mockito.Mockito.mock;
-
-public class FileTxnLogTest extends ZKTestCase {
- protected static final Logger LOG = LoggerFactory.getLogger(FileTxnLogTest.class);
-
- private static final int KB = 1024;
-
- @Test
- public void testInvalidPreallocSize() {
- Assert.assertEquals("file should not be padded",
- 10 * KB, FilePadding.calculateFileSizeWithPadding(7 * KB, 10 * KB, 0));
- Assert.assertEquals("file should not be padded",
- 10 * KB, FilePadding.calculateFileSizeWithPadding(7 * KB, 10 * KB, -1));
- }
-
- @Test
- public void testCalculateFileSizeWithPaddingWhenNotToCurrentSize() {
- Assert.assertEquals("file should not be padded",
- 10 * KB, FilePadding.calculateFileSizeWithPadding(5 * KB, 10 * KB, 10 * KB));
- }
-
- @Test
- public void testCalculateFileSizeWithPaddingWhenCloseToCurrentSize() {
- Assert.assertEquals("file should be padded an additional 10 KB",
- 20 * KB, FilePadding.calculateFileSizeWithPadding(7 * KB, 10 * KB, 10 * KB));
- }
-
- @Test
- public void testFileSizeGreaterThanPosition() {
- Assert.assertEquals("file should be padded to 40 KB",
- 40 * KB, FilePadding.calculateFileSizeWithPadding(31 * KB, 10 * KB, 10 * KB));
- }
-
- @Test
- public void testPreAllocSizeSmallerThanTxnData() throws IOException {
- File logDir = ClientBase.createTmpDir();
- FileTxnLog fileTxnLog = new FileTxnLog(logDir);
-
- // Set a small preAllocSize (.5 MB)
- final int preAllocSize = 500 * KB;
- FilePadding.setPreallocSize(preAllocSize);
-
- // Create dummy txn larger than preAllocSize
- // Since the file padding inserts a 0, we will fill the data with 0xff to ensure we corrupt the data if we put the 0 in the data
- byte[] data = new byte[2 * preAllocSize];
- Arrays.fill(data, (byte) 0xff);
-
- // Append and commit 2 transactions to the log
- // Prior to ZOOKEEPER-2249, attempting to pad in association with the second transaction will corrupt the first
- fileTxnLog.append(new TxnHeader(1, 1, 1, 1, ZooDefs.OpCode.create),
- new CreateTxn("/testPreAllocSizeSmallerThanTxnData1", data, ZooDefs.Ids.OPEN_ACL_UNSAFE, false, 0));
- fileTxnLog.commit();
- fileTxnLog.append(new TxnHeader(1, 1, 2, 2, ZooDefs.OpCode.create),
- new CreateTxn("/testPreAllocSizeSmallerThanTxnData2", new byte[]{}, ZooDefs.Ids.OPEN_ACL_UNSAFE, false, 0));
- fileTxnLog.commit();
- fileTxnLog.close();
-
- // Read the log back from disk, this will throw a java.io.IOException: CRC check failed prior to ZOOKEEPER-2249
- FileTxnLog.FileTxnIterator fileTxnIterator = new FileTxnLog.FileTxnIterator(logDir, 0);
-
- // Verify the data in the first transaction
- CreateTxn createTxn = (CreateTxn) fileTxnIterator.getTxn();
- Assert.assertTrue(Arrays.equals(createTxn.getData(), data));
-
- // Verify the data in the second transaction
- fileTxnIterator.next();
- createTxn = (CreateTxn) fileTxnIterator.getTxn();
- Assert.assertTrue(Arrays.equals(createTxn.getData(), new byte[]{}));
- }
-
- @Test
- public void testSetPreallocSize() {
- long customPreallocSize = 10101;
- FileTxnLog.setPreallocSize(customPreallocSize);
- Assert.assertThat(FilePadding.getPreAllocSize(), is(equalTo(customPreallocSize)));
- }
-
- public void testSyncThresholdExceedCount() throws IOException {
- // Given ...
-
- // Set threshold to -1, as after the first commit it takes 0ms to commit to disk.
- java.lang.System.setProperty(FileTxnLog.ZOOKEEPER_FSYNC_WARNING_THRESHOLD_MS_PROPERTY, "-1");
- ServerStats.Provider providerMock = mock(ServerStats.Provider.class);
- ServerStats serverStats = new ServerStats(providerMock);
-
- File logDir = ClientBase.createTmpDir();
- FileTxnLog fileTxnLog = new FileTxnLog(logDir);
- fileTxnLog.setServerStats(serverStats);
-
- // Verify serverStats is 0 before any commit
- Assert.assertEquals(0L, serverStats.getFsyncThresholdExceedCount());
-
- // When ...
- for (int i = 0; i < 50; i++) {
- fileTxnLog.append(new TxnHeader(1, 1, 1, 1, ZooDefs.OpCode.create),
- new CreateTxn("/testFsyncThresholdCountIncreased", new byte[]{}, ZooDefs.Ids.OPEN_ACL_UNSAFE, false, 0));
- fileTxnLog.commit(); // only 1 commit, otherwise it will be flaky
- // Then ... verify serverStats is updated to the number of commits (as threshold is set to 0)
- Assert.assertEquals((long) i + 1 , serverStats.getFsyncThresholdExceedCount());
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/test/org/apache/zookeeper/server/persistence/FileTxnSnapLogTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/persistence/FileTxnSnapLogTest.java b/src/java/test/org/apache/zookeeper/server/persistence/FileTxnSnapLogTest.java
deleted file mode 100644
index 9334d93..0000000
--- a/src/java/test/org/apache/zookeeper/server/persistence/FileTxnSnapLogTest.java
+++ /dev/null
@@ -1,224 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server.persistence;
-
-import org.apache.jute.Record;
-import org.apache.zookeeper.ZooDefs;
-import org.apache.zookeeper.server.Request;
-import org.apache.zookeeper.test.ClientBase;
-import org.apache.zookeeper.test.TestUtils;
-import org.apache.zookeeper.txn.SetDataTxn;
-import org.apache.zookeeper.txn.TxnHeader;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.File;
-import java.io.IOException;
-
-public class FileTxnSnapLogTest {
-
- private File tmpDir;
-
- private File logDir;
-
- private File snapDir;
-
- private File logVersionDir;
-
- private File snapVersionDir;
-
- @Before
- public void setUp() throws Exception {
- tmpDir = ClientBase.createEmptyTestDir();
- logDir = new File(tmpDir, "logdir");
- snapDir = new File(tmpDir, "snapdir");
- }
-
- @After
- public void tearDown() throws Exception {
- if(tmpDir != null){
- TestUtils.deleteFileRecursively(tmpDir);
- }
- this.tmpDir = null;
- this.logDir = null;
- this.snapDir = null;
- this.logVersionDir = null;
- this.snapVersionDir = null;
- }
-
- private File createVersionDir(File parentDir) {
- File versionDir = new File(parentDir, FileTxnSnapLog.version + FileTxnSnapLog.VERSION);
- versionDir.mkdirs();
- return versionDir;
- }
-
- private void createLogFile(File dir, long zxid) throws IOException {
- File file = new File(dir.getPath() + File.separator + Util.makeLogName(zxid));
- file.createNewFile();
- }
-
- private void createSnapshotFile(File dir, long zxid) throws IOException {
- File file = new File(dir.getPath() + File.separator + Util.makeSnapshotName(zxid));
- file.createNewFile();
- }
-
- private void twoDirSetupWithCorrectFiles() throws IOException {
- logVersionDir = createVersionDir(logDir);
- snapVersionDir = createVersionDir(snapDir);
-
- // transaction log files in log dir
- createLogFile(logVersionDir,1);
- createLogFile(logVersionDir,2);
-
- // snapshot files in snap dir
- createSnapshotFile(snapVersionDir,1);
- createSnapshotFile(snapVersionDir,2);
- }
-
- private void singleDirSetupWithCorrectFiles() throws IOException {
- logVersionDir = createVersionDir(logDir);
-
- // transaction log and snapshot files in the same dir
- createLogFile(logVersionDir,1);
- createLogFile(logVersionDir,2);
- createSnapshotFile(logVersionDir,1);
- createSnapshotFile(logVersionDir,2);
- }
-
- private FileTxnSnapLog createFileTxnSnapLogWithNoAutoCreateDataDir(File logDir, File snapDir) throws IOException {
- return createFileTxnSnapLogWithAutoCreateDataDir(logDir, snapDir, "false");
- }
-
- private FileTxnSnapLog createFileTxnSnapLogWithAutoCreateDataDir(File logDir, File snapDir, String autoCreateValue) throws IOException {
- String priorAutocreateDirValue = System.getProperty(FileTxnSnapLog.ZOOKEEPER_DATADIR_AUTOCREATE);
- System.setProperty(FileTxnSnapLog.ZOOKEEPER_DATADIR_AUTOCREATE, autoCreateValue);
- FileTxnSnapLog fileTxnSnapLog;
- try {
- fileTxnSnapLog = new FileTxnSnapLog(logDir, snapDir);
- } finally {
- if (priorAutocreateDirValue == null) {
- System.clearProperty(FileTxnSnapLog.ZOOKEEPER_DATADIR_AUTOCREATE);
- } else {
- System.setProperty(FileTxnSnapLog.ZOOKEEPER_DATADIR_AUTOCREATE, priorAutocreateDirValue);
- }
- }
- return fileTxnSnapLog;
- }
-
- /**
- * Test verifies the auto creation of log dir and snap dir.
- * Sets "zookeeper.datadir.autocreate" to true.
- */
- @Test
- public void testWithAutoCreateDataDir() throws IOException {
- Assert.assertFalse("log directory already exists", logDir.exists());
- Assert.assertFalse("snapshot directory already exists", snapDir.exists());
-
- FileTxnSnapLog fileTxnSnapLog = createFileTxnSnapLogWithAutoCreateDataDir(logDir, snapDir, "true");
-
- Assert.assertTrue(logDir.exists());
- Assert.assertTrue(snapDir.exists());
- Assert.assertTrue(fileTxnSnapLog.getDataDir().exists());
- Assert.assertTrue(fileTxnSnapLog.getSnapDir().exists());
- }
-
- /**
- * Test verifies server should fail when log dir or snap dir doesn't exist.
- * Sets "zookeeper.datadir.autocreate" to false.
- */
- @Test(expected = FileTxnSnapLog.DatadirException.class)
- public void testWithoutAutoCreateDataDir() throws Exception {
- Assert.assertFalse("log directory already exists", logDir.exists());
- Assert.assertFalse("snapshot directory already exists", snapDir.exists());
-
- try {
- createFileTxnSnapLogWithAutoCreateDataDir(logDir, snapDir, "false");
- } catch (FileTxnSnapLog.DatadirException e) {
- Assert.assertFalse(logDir.exists());
- Assert.assertFalse(snapDir.exists());
- // rethrow exception
- throw e;
- }
- Assert.fail("Expected exception from FileTxnSnapLog");
- }
-
- @Test
- public void testGetTxnLogSyncElapsedTime() throws IOException {
- FileTxnSnapLog fileTxnSnapLog = createFileTxnSnapLogWithAutoCreateDataDir(logDir, snapDir, "true");
-
- TxnHeader hdr = new TxnHeader(1, 1, 1, 1, ZooDefs.OpCode.setData);
- Record txn = new SetDataTxn("/foo", new byte[0], 1);
- Request req = new Request(0, 0, 0, hdr, txn, 0);
-
- try {
- fileTxnSnapLog.append(req);
- fileTxnSnapLog.commit();
- long syncElapsedTime = fileTxnSnapLog.getTxnLogElapsedSyncTime();
- Assert.assertNotEquals("Did not update syncElapsedTime!", -1L, syncElapsedTime);
- } finally {
- fileTxnSnapLog.close();
- }
- }
-
- @Test
- public void testDirCheckWithCorrectFiles() throws IOException {
- twoDirSetupWithCorrectFiles();
-
- try {
- createFileTxnSnapLogWithNoAutoCreateDataDir(logDir, snapDir);
- } catch (FileTxnSnapLog.LogDirContentCheckException | FileTxnSnapLog.SnapDirContentCheckException e) {
- Assert.fail("Should not throw ContentCheckException.");
- }
- }
-
- @Test
- public void testDirCheckWithSingleDirSetup() throws IOException {
- singleDirSetupWithCorrectFiles();
-
- try {
- createFileTxnSnapLogWithNoAutoCreateDataDir(logDir, logDir);
- } catch (FileTxnSnapLog.LogDirContentCheckException | FileTxnSnapLog.SnapDirContentCheckException e) {
- Assert.fail("Should not throw ContentCheckException.");
- }
- }
-
- @Test(expected = FileTxnSnapLog.LogDirContentCheckException.class)
- public void testDirCheckWithSnapFilesInLogDir() throws IOException {
- twoDirSetupWithCorrectFiles();
-
- // add snapshot files to the log version dir
- createSnapshotFile(logVersionDir,3);
- createSnapshotFile(logVersionDir,4);
-
- createFileTxnSnapLogWithNoAutoCreateDataDir(logDir, snapDir);
- }
-
- @Test(expected = FileTxnSnapLog.SnapDirContentCheckException.class)
- public void testDirCheckWithLogFilesInSnapDir() throws IOException {
- twoDirSetupWithCorrectFiles();
-
- // add transaction log files to the snap version dir
- createLogFile(snapVersionDir,3);
- createLogFile(snapVersionDir,4);
-
- createFileTxnSnapLogWithNoAutoCreateDataDir(logDir, snapDir);
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/test/org/apache/zookeeper/server/persistence/TxnLogToolkitTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/persistence/TxnLogToolkitTest.java b/src/java/test/org/apache/zookeeper/server/persistence/TxnLogToolkitTest.java
deleted file mode 100644
index 79a69c4..0000000
--- a/src/java/test/org/apache/zookeeper/server/persistence/TxnLogToolkitTest.java
+++ /dev/null
@@ -1,155 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server.persistence;
-
-import org.apache.commons.io.FileUtils;
-import org.apache.zookeeper.test.ClientBase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.io.ByteArrayOutputStream;
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.io.PrintStream;
-import java.util.Scanner;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-import static org.hamcrest.CoreMatchers.containsString;
-import static org.hamcrest.core.IsNot.not;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
-
-
-public class TxnLogToolkitTest {
- private static final File testData = new File(
- System.getProperty("test.data.dir", "build/test/data"));
-
- private final ByteArrayOutputStream outContent = new ByteArrayOutputStream();
- private final ByteArrayOutputStream errContent = new ByteArrayOutputStream();
- private File mySnapDir;
-
- @Before
- public void setUp() throws IOException {
- System.setOut(new PrintStream(outContent));
- System.setErr(new PrintStream(errContent));
- File snapDir = new File(testData, "invalidsnap");
- mySnapDir = ClientBase.createTmpDir();
- FileUtils.copyDirectory(snapDir, mySnapDir);
- }
-
- @After
- public void tearDown() throws IOException {
- System.setOut(System.out);
- System.setErr(System.err);
- mySnapDir.setWritable(true);
- FileUtils.deleteDirectory(mySnapDir);
- }
-
- @Test
- public void testDumpMode() throws Exception {
- // Arrange
- File logfile = new File(new File(mySnapDir, "version-2"), "log.274");
- TxnLogToolkit lt = new TxnLogToolkit(false, false, logfile.toString(), true);
-
- // Act
- lt.dump(null);
-
- // Assert
- // no exception thrown
- }
-
- @Test(expected = TxnLogToolkit.TxnLogToolkitException.class)
- public void testInitMissingFile() throws FileNotFoundException, TxnLogToolkit.TxnLogToolkitException {
- // Arrange & Act
- File logfile = new File("this_file_should_not_exists");
- TxnLogToolkit lt = new TxnLogToolkit(false, false, logfile.toString(), true);
- }
-
- @Test(expected = TxnLogToolkit.TxnLogToolkitException.class)
- public void testInitWithRecoveryFileExists() throws IOException, TxnLogToolkit.TxnLogToolkitException {
- // Arrange & Act
- File logfile = new File(new File(mySnapDir, "version-2"), "log.274");
- File recoveryFile = new File(new File(mySnapDir, "version-2"), "log.274.fixed");
- recoveryFile.createNewFile();
- TxnLogToolkit lt = new TxnLogToolkit(true, false, logfile.toString(), true);
- }
-
- @Test
- public void testDumpWithCrcError() throws Exception {
- // Arrange
- File logfile = new File(new File(mySnapDir, "version-2"), "log.42");
- TxnLogToolkit lt = new TxnLogToolkit(false, false, logfile.toString(), true);
-
- // Act
- lt.dump(null);
-
- // Assert
- String output = outContent.toString();
- Pattern p = Pattern.compile("^CRC ERROR.*session 0x8061fac5ddeb0000 cxid 0x0 zxid 0x8800000002 createSession 30000$", Pattern.MULTILINE);
- Matcher m = p.matcher(output);
- assertTrue("Output doesn't indicate CRC error for the broken session id: " + output, m.find());
- }
-
- @Test
- public void testRecoveryFixBrokenFile() throws Exception {
- // Arrange
- File logfile = new File(new File(mySnapDir, "version-2"), "log.42");
- TxnLogToolkit lt = new TxnLogToolkit(true, false, logfile.toString(), true);
-
- // Act
- lt.dump(null);
-
- // Assert
- String output = outContent.toString();
- assertThat(output, containsString("CRC FIXED"));
-
- // Should be able to dump the recovered logfile with no CRC error
- outContent.reset();
- logfile = new File(new File(mySnapDir, "version-2"), "log.42.fixed");
- lt = new TxnLogToolkit(false, false, logfile.toString(), true);
- lt.dump(null);
- output = outContent.toString();
- assertThat(output, not(containsString("CRC ERROR")));
- }
-
- @Test
- public void testRecoveryInteractiveMode() throws Exception {
- // Arrange
- File logfile = new File(new File(mySnapDir, "version-2"), "log.42");
- TxnLogToolkit lt = new TxnLogToolkit(true, false, logfile.toString(), false);
-
- // Act
- lt.dump(new Scanner("y\n"));
-
- // Assert
- String output = outContent.toString();
- assertThat(output, containsString("CRC ERROR"));
-
- // Should be able to dump the recovered logfile with no CRC error
- outContent.reset();
- logfile = new File(new File(mySnapDir, "version-2"), "log.42.fixed");
- lt = new TxnLogToolkit(false, false, logfile.toString(), true);
- lt.dump(null);
- output = outContent.toString();
- assertThat(output, not(containsString("CRC ERROR")));
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/test/org/apache/zookeeper/server/quorum/BufferStatsTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/quorum/BufferStatsTest.java b/src/java/test/org/apache/zookeeper/server/quorum/BufferStatsTest.java
deleted file mode 100644
index 8435e85..0000000
--- a/src/java/test/org/apache/zookeeper/server/quorum/BufferStatsTest.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server.quorum;
-
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-
-public class BufferStatsTest {
- @Test
- public void testSetProposalSizeSetMinMax() {
- BufferStats stats = new BufferStats();
- assertEquals(-1, stats.getLastBufferSize());
- assertEquals(-1, stats.getMinBufferSize());
- assertEquals(-1, stats.getMaxBufferSize());
- stats.setLastBufferSize(10);
- assertEquals(10, stats.getLastBufferSize());
- assertEquals(10, stats.getMinBufferSize());
- assertEquals(10, stats.getMaxBufferSize());
- stats.setLastBufferSize(20);
- assertEquals(20, stats.getLastBufferSize());
- assertEquals(10, stats.getMinBufferSize());
- assertEquals(20, stats.getMaxBufferSize());
- stats.setLastBufferSize(5);
- assertEquals(5, stats.getLastBufferSize());
- assertEquals(5, stats.getMinBufferSize());
- assertEquals(20, stats.getMaxBufferSize());
- }
-
- @Test
- public void testReset() {
- BufferStats stats = new BufferStats();
- stats.setLastBufferSize(10);
- assertEquals(10, stats.getLastBufferSize());
- assertEquals(10, stats.getMinBufferSize());
- assertEquals(10, stats.getMaxBufferSize());
- stats.reset();
- assertEquals(-1, stats.getLastBufferSize());
- assertEquals(-1, stats.getMinBufferSize());
- assertEquals(-1, stats.getMaxBufferSize());
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/test/org/apache/zookeeper/server/quorum/CommitProcessorConcurrencyTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/quorum/CommitProcessorConcurrencyTest.java b/src/java/test/org/apache/zookeeper/server/quorum/CommitProcessorConcurrencyTest.java
deleted file mode 100644
index 8cbed91..0000000
--- a/src/java/test/org/apache/zookeeper/server/quorum/CommitProcessorConcurrencyTest.java
+++ /dev/null
@@ -1,142 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server.quorum;
-
-import java.io.ByteArrayOutputStream;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-
-import org.apache.jute.BinaryOutputArchive;
-import org.apache.zookeeper.ZKTestCase;
-import org.apache.zookeeper.ZooDefs.Ids;
-import org.apache.zookeeper.ZooDefs.OpCode;
-import org.apache.zookeeper.data.Id;
-import org.apache.zookeeper.proto.CreateRequest;
-import org.apache.zookeeper.proto.GetDataRequest;
-import org.apache.zookeeper.proto.SyncRequest;
-import org.apache.zookeeper.server.Request;
-import org.apache.zookeeper.server.RequestProcessor;
-import org.apache.zookeeper.server.WorkerService;
-import org.apache.zookeeper.server.RequestProcessor.RequestProcessorException;
-import org.apache.zookeeper.server.ZooKeeperServerListener;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class CommitProcessorConcurrencyTest extends ZKTestCase {
- protected static final Logger LOG =
- LoggerFactory.getLogger(CommitProcessorConcurrencyTest.class);
-
- Boolean executedFlag = false;
- MockCommitProcessor processor;
-
- @Before
- public void setUp() throws Exception {
- processor = new MockCommitProcessor();
- }
-
- @After
- public void tearDown() throws Exception {
- processor.shutdown();
- }
-
- class MockCommitProcessor extends CommitProcessor {
-
- MockCommitProcessor() {
- super(
- new RequestProcessor() {
- public void processRequest(Request request)
- throws RequestProcessorException {
- executedFlag = true;
- }
- public void shutdown(){}
- },
- "0",
- false, new ZooKeeperServerListener(){
-
- @Override
- public void notifyStopping(String errMsg, int exitCode) {
-
- }});
- }
-
- public void testStart() {
- this.stopped = false;
- this.workerPool = new WorkerService(
- "CommitProcWork", 1, true);
- }
-
- public void addToCommittedRequests(Request req) {
- this.committedRequests.add(req);
- }
-
- public void addToNextPending(Request req) {
- this.nextPending.set(req);
- }
-
- public void addToQueuedRequests(Request req) {
- //this.numRequestsProcessing.incrementAndGet();
- this.queuedRequests.add(req);
- }
-
- public void testProcessCommitted() {
- this.processCommitted();
- }
-
- @Override
- public void shutdown() {
- this.workerPool.stop();
- }
- }
-
- /*
- * We populate the necessary data structures in the CommitProcessor
- * instance and run processCommitted
- */
- @Test
- public void raceTest()
- throws Exception {
-
- ByteArrayOutputStream boas = new ByteArrayOutputStream();
- BinaryOutputArchive boa = BinaryOutputArchive.getArchive(boas);
- GetDataRequest getReq = new GetDataRequest("/testrace", false);
- getReq.serialize(boa, "request");
- ByteBuffer bb = ByteBuffer.wrap(boas.toByteArray());
- Request readReq = new Request(null, 0x0, 0, OpCode.getData,
- bb, new ArrayList<Id>());
-
- boas.reset();
- SyncRequest syncReq = new SyncRequest("/testrace");
- syncReq.serialize(boa, "request");
- bb = ByteBuffer.wrap(boas.toByteArray());
- Request writeReq = new Request(null, 0x0, 0, OpCode.sync,
- bb, new ArrayList<Id>());
-
- processor.addToCommittedRequests(writeReq);
- processor.addToQueuedRequests(readReq);
- processor.addToQueuedRequests(writeReq);
-
- processor.testStart();
- processor.testProcessCommitted();
- Assert.assertFalse("Next request processor executed", executedFlag);
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/test/org/apache/zookeeper/server/quorum/CommitProcessorTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/quorum/CommitProcessorTest.java b/src/java/test/org/apache/zookeeper/server/quorum/CommitProcessorTest.java
deleted file mode 100644
index 1418158..0000000
--- a/src/java/test/org/apache/zookeeper/server/quorum/CommitProcessorTest.java
+++ /dev/null
@@ -1,425 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server.quorum;
-
-import java.io.ByteArrayOutputStream;
-import java.io.File;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Random;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingQueue;
-
-import org.apache.jute.BinaryOutputArchive;
-import org.apache.zookeeper.ZKTestCase;
-import org.apache.zookeeper.data.Id;
-import org.apache.zookeeper.ZooDefs.Ids;
-import org.apache.zookeeper.ZooDefs.OpCode;
-import org.apache.zookeeper.proto.CreateRequest;
-import org.apache.zookeeper.proto.GetDataRequest;
-import org.apache.zookeeper.server.FinalRequestProcessor;
-import org.apache.zookeeper.server.PrepRequestProcessor;
-import org.apache.zookeeper.server.Request;
-import org.apache.zookeeper.server.RequestProcessor;
-import org.apache.zookeeper.server.SessionTracker;
-import org.apache.zookeeper.server.ZooKeeperServer;
-import org.apache.zookeeper.server.quorum.CommitProcessor;
-import org.apache.zookeeper.test.ClientBase;
-
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * The following are invariant regardless of the particular implementation
- * of the CommitProcessor, and are tested for:
- *
- * 1. For each session, requests are processed and the client sees its
- * responses in order.
- * 2. Write requests are processed in zxid order across all sessions.
- *
- * The following are also tested for here, but are specific to this
- * particular implementation. The underlying issue is that watches can be
- * reset while reading the data. For reads/writes on two different sessions
- * on different nodes, or with reads that do not set watches, the reads can
- * happen in any order relative to the writes. For a read in one session that
- * resets a watch that is triggered by a write on another session, however,
- * we need to ensure that there is no race condition
- *
- * 3. The pipeline needs to be drained before a write request can enter.
- * 4. No in-flight write requests while processing a read request.
- */
-public class CommitProcessorTest extends ZKTestCase {
- protected static final Logger LOG =
- LoggerFactory.getLogger(CommitProcessorTest.class);
-
-
- private AtomicInteger processedReadRequests = new AtomicInteger(0);
- private AtomicInteger processedWriteRequests = new AtomicInteger(0);
-
- TestZooKeeperServer zks;
- File tmpDir;
- ArrayList<TestClientThread> testClients =
- new ArrayList<TestClientThread>();
-
- public void setUp(int numCommitThreads, int numClientThreads)
- throws Exception {
- System.setProperty(
- CommitProcessor.ZOOKEEPER_COMMIT_PROC_NUM_WORKER_THREADS,
- Integer.toString(numCommitThreads));
- System.setProperty("zookeeper.admin.enableServer", "false");
- tmpDir = ClientBase.createTmpDir();
- ClientBase.setupTestEnv();
- zks = new TestZooKeeperServer(tmpDir, tmpDir, 4000);
- zks.startup();
- for(int i=0; i<numClientThreads; ++i) {
- TestClientThread client = new TestClientThread();
- testClients.add(client);
- client.start();
- }
- }
-
- @After
- public void tearDown() throws Exception {
- LOG.info("tearDown starting");
- for(TestClientThread client : testClients) {
- client.interrupt();
- client.join();
- }
- zks.shutdown();
-
- if (tmpDir != null) {
- Assert.assertTrue("delete " + tmpDir.toString(),
- ClientBase.recursiveDelete(tmpDir));
- }
- }
-
- private class TestClientThread extends Thread {
- long sessionId;
- int cxid;
- int nodeId;
-
- public TestClientThread() {
- sessionId = zks.getSessionTracker().createSession(5000);
- }
-
- public void sendWriteRequest() throws Exception {
- ByteArrayOutputStream boas = new ByteArrayOutputStream();
- BinaryOutputArchive boa = BinaryOutputArchive.getArchive(boas);
- CreateRequest createReq = new CreateRequest(
- "/session" + Long.toHexString(sessionId) + "-" + (++nodeId),
- new byte[0], Ids.OPEN_ACL_UNSAFE, 1);
- createReq.serialize(boa, "request");
- ByteBuffer bb = ByteBuffer.wrap(boas.toByteArray());
- Request req = new Request(null, sessionId, ++cxid, OpCode.create,
- bb, new ArrayList<Id>());
- zks.firstProcessor.processRequest(req);
- }
-
- public void sendReadRequest() throws Exception {
- ByteArrayOutputStream boas = new ByteArrayOutputStream();
- BinaryOutputArchive boa = BinaryOutputArchive.getArchive(boas);
- GetDataRequest getDataRequest = new GetDataRequest(
- "/session" + Long.toHexString(sessionId) + "-" + nodeId, false);
- getDataRequest.serialize(boa, "request");
- ByteBuffer bb = ByteBuffer.wrap(boas.toByteArray());
- Request req = new Request(null, sessionId, ++cxid, OpCode.getData,
- bb, new ArrayList<Id>());
- zks.firstProcessor.processRequest(req);
- }
-
- public void run() {
- Random rand = new Random(Thread.currentThread().getId());
- try {
- sendWriteRequest();
- for(int i=0; i<1000; ++i) {
- // Do 25% write / 75% read request mix
- if (rand.nextInt(100) < 25) {
- sendWriteRequest();
- } else {
- sendReadRequest();
- }
- }
- } catch (Exception e) {
- LOG.error("Uncaught exception in test: ", e);
- }
- }
- }
-
- @Test
- public void testNoCommitWorkers() throws Exception {
- setUp(0, 10);
- synchronized(this) {
- wait(5000);
- }
- checkProcessedRequest();
- Assert.assertFalse(fail);
- }
-
- @Test
- public void testOneCommitWorker() throws Exception {
- setUp(1, 10);
- synchronized(this) {
- wait(5000);
- }
- checkProcessedRequest();
- Assert.assertFalse(fail);
- }
-
- @Test
- public void testManyCommitWorkers() throws Exception {
- setUp(10, 10);
- synchronized(this) {
- wait(5000);
- }
- checkProcessedRequest();
- Assert.assertFalse(fail);
-
- }
-
- private void checkProcessedRequest() {
- Assert.assertTrue("No read requests processed",
- processedReadRequests.get() > 0);
- Assert.assertTrue("No write requests processed",
- processedWriteRequests.get() > 0);
- }
-
- volatile boolean fail = false;
- synchronized private void failTest(String reason) {
- fail = true;
- notifyAll();
- Assert.fail(reason);
- }
-
- private class TestZooKeeperServer extends ZooKeeperServer {
- PrepRequestProcessor firstProcessor;
- CommitProcessor commitProcessor;
-
- public TestZooKeeperServer(File snapDir, File logDir, int tickTime)
- throws IOException {
- super(snapDir, logDir, tickTime);
- }
-
- public SessionTracker getSessionTracker() {
- return sessionTracker;
- }
-
- // Leader mock: Prep -> MockProposal -> Commit -> validate -> Final
- // Have side thread call commitProc.commit()
- @Override
- protected void setupRequestProcessors() {
- RequestProcessor finalProcessor = new FinalRequestProcessor(zks);
- // ValidateProcessor is set up in a similar fashion to ToBeApplied
- // processor, so it can do pre/post validating of requests
- ValidateProcessor validateProcessor =
- new ValidateProcessor(finalProcessor);
- commitProcessor = new CommitProcessor(validateProcessor, "1", true,
- getZooKeeperServerListener());
- validateProcessor.setCommitProcessor(commitProcessor);
- commitProcessor.start();
- MockProposalRequestProcessor proposalProcessor =
- new MockProposalRequestProcessor(commitProcessor);
- proposalProcessor.start();
- firstProcessor = new PrepRequestProcessor(zks, proposalProcessor);
- firstProcessor.start();
- }
- }
-
- private class MockProposalRequestProcessor extends Thread
- implements RequestProcessor {
- private final CommitProcessor commitProcessor;
- private final LinkedBlockingQueue<Request> proposals =
- new LinkedBlockingQueue<Request>();
-
- public MockProposalRequestProcessor(CommitProcessor commitProcessor) {
- this.commitProcessor = commitProcessor;
- }
-
- @Override
- public void run() {
- Random rand = new Random(Thread.currentThread().getId());
- try {
- while(true) {
- Request request = proposals.take();
- Thread.sleep(10 + rand.nextInt(190));
- commitProcessor.commit(request);
- }
- } catch (InterruptedException e) {
- // ignore
- }
- }
-
- @Override
- public void processRequest(Request request)
- throws RequestProcessorException {
- commitProcessor.processRequest(request);
- if (request.getHdr() != null) {
- // fake propose request
- proposals.add(request);
- }
- }
-
- @Override
- public void shutdown() {
- // TODO Auto-generated method stub
-
- }
- }
-
- private class ValidateProcessor implements RequestProcessor {
- Random rand = new Random(Thread.currentThread().getId());
- RequestProcessor nextProcessor;
- CommitProcessor commitProcessor;
- AtomicLong expectedZxid = new AtomicLong(1);
- ConcurrentHashMap<Long, AtomicInteger> cxidMap =
- new ConcurrentHashMap<Long, AtomicInteger>();
-
- AtomicInteger outstandingReadRequests = new AtomicInteger(0);
- AtomicInteger outstandingWriteRequests = new AtomicInteger(0);
-
- public ValidateProcessor(RequestProcessor nextProcessor) {
- this.nextProcessor = nextProcessor;
- }
-
- public void setCommitProcessor(CommitProcessor commitProcessor) {
- this.commitProcessor = commitProcessor;
- }
-
-
- @Override
- public void processRequest(Request request)
- throws RequestProcessorException {
- boolean isWriteRequest = commitProcessor.needCommit(request);
- if (isWriteRequest) {
- outstandingWriteRequests.incrementAndGet();
- validateWriteRequestVariant(request);
- LOG.debug("Starting write request zxid=" + request.zxid);
- } else {
- LOG.debug("Starting read request cxid="
- + request.cxid + " for session 0x"
- + Long.toHexString(request.sessionId));
- outstandingReadRequests.incrementAndGet();
- validateReadRequestVariant(request);
- }
-
- // Insert random delay to test thread race conditions
- try {
- Thread.sleep(10 + rand.nextInt(290));
- } catch(InterruptedException e) {
- // ignore
- }
- nextProcessor.processRequest(request);
-
- /*
- * The commit workers will have to execute this line before they
- * wake up the commit processor. So this value is up-to-date when
- * variant check is performed
- */
- if (isWriteRequest) {
- outstandingWriteRequests.decrementAndGet();
- LOG.debug("Done write request zxid=" + request.zxid);
- processedWriteRequests.incrementAndGet();
- } else {
- outstandingReadRequests.decrementAndGet();
- LOG.debug("Done read request cxid="
- + request.cxid + " for session 0x"
- + Long.toHexString(request.sessionId));
- processedReadRequests.incrementAndGet();
- }
- validateRequest(request);
- }
-
- /**
- * Validate that this is the only request in the pipeline
- */
- private void validateWriteRequestVariant(Request request) {
- long zxid = request.getHdr().getZxid();
- int readRequests = outstandingReadRequests.get();
- if (readRequests != 0) {
- failTest("There are " + readRequests + " outstanding"
- + " read requests while issuing a write request zxid="
- + zxid);
- }
- int writeRequests = outstandingWriteRequests.get();
- if (writeRequests > 1) {
- failTest("There are " + writeRequests + " outstanding"
- + " write requests while issuing a write request zxid="
- + zxid + " (expected one)");
- }
- }
-
- /**
- * Validate that no write request is in the pipeline while working
- * on a read request
- */
- private void validateReadRequestVariant(Request request) {
- int writeRequests = outstandingWriteRequests.get();
- if (writeRequests != 0) {
- failTest("There are " + writeRequests + " outstanding"
- + " write requests while issuing a read request cxid="
- + request.cxid + " for session 0x"
- + Long.toHexString(request.sessionId));
- }
- }
-
- private void validateRequest(Request request) {
- LOG.info("Got request " + request);
-
- // Zxids should always be in order for write requests
- if (request.getHdr() != null) {
- long zxid = request.getHdr().getZxid();
- if (!expectedZxid.compareAndSet(zxid, zxid + 1)) {
- failTest("Write request, expected_zxid="
- + expectedZxid.get() + "; req_zxid=" + zxid);
- }
- }
-
- // Each session should see its cxids in order
- AtomicInteger sessionCxid = cxidMap.get(request.sessionId);
- if (sessionCxid == null) {
- sessionCxid = new AtomicInteger(request.cxid + 1);
- AtomicInteger existingSessionCxid =
- cxidMap.putIfAbsent(request.sessionId, sessionCxid);
- if (existingSessionCxid != null) {
- failTest("Race condition adding cxid=" + request.cxid
- + " for session 0x"
- + Long.toHexString(request.sessionId)
- + " with other_cxid=" + existingSessionCxid.get());
- }
- } else {
- if (!sessionCxid.compareAndSet(
- request.cxid, request.cxid + 1)) {
- failTest("Expected_cxid=" + sessionCxid.get()
- + "; req_cxid=" + request.cxid);
- }
- }
- }
-
- @Override
- public void shutdown() {
- // TODO Auto-generated method stub
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/test/org/apache/zookeeper/server/quorum/EphemeralNodeDeletionTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/quorum/EphemeralNodeDeletionTest.java b/src/java/test/org/apache/zookeeper/server/quorum/EphemeralNodeDeletionTest.java
deleted file mode 100644
index 9546c25..0000000
--- a/src/java/test/org/apache/zookeeper/server/quorum/EphemeralNodeDeletionTest.java
+++ /dev/null
@@ -1,244 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.zookeeper.server.quorum;
-
-import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-
-import java.io.IOException;
-import java.net.SocketTimeoutException;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.zookeeper.AsyncCallback;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.PortAssignment;
-import org.apache.zookeeper.ZooDefs.Ids;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.data.Stat;
-import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
-import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
-import org.apache.zookeeper.test.ClientBase;
-import org.apache.zookeeper.test.ClientBase.CountdownWatcher;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
-
-import javax.security.sasl.SaslException;
-
-public class EphemeralNodeDeletionTest extends QuorumPeerTestBase {
- private static int SERVER_COUNT = 3;
- private MainThread[] mt = new MainThread[SERVER_COUNT];
-
- /**
- * Test case for https://issues.apache.org/jira/browse/ZOOKEEPER-2355.
- * ZooKeeper ephemeral node is never deleted if follower fail while reading
- * the proposal packet.
- */
-
- @Test(timeout = 120000)
- public void testEphemeralNodeDeletion() throws Exception {
- final int clientPorts[] = new int[SERVER_COUNT];
- StringBuilder sb = new StringBuilder();
- String server;
-
- for (int i = 0; i < SERVER_COUNT; i++) {
- clientPorts[i] = PortAssignment.unique();
- server = "server." + i + "=127.0.0.1:" + PortAssignment.unique()
- + ":" + PortAssignment.unique() + ":participant;127.0.0.1:"
- + clientPorts[i];
- sb.append(server + "\n");
- }
- String currentQuorumCfgSection = sb.toString();
- // start all the servers
- for (int i = 0; i < SERVER_COUNT; i++) {
- mt[i] = new MainThread(i, clientPorts[i], currentQuorumCfgSection,
- false) {
- @Override
- public TestQPMain getTestQPMain() {
- return new MockTestQPMain();
- }
- };
- mt[i].start();
- }
-
- // ensure all servers started
- for (int i = 0; i < SERVER_COUNT; i++) {
- Assert.assertTrue("waiting for server " + i + " being up",
- ClientBase.waitForServerUp("127.0.0.1:" + clientPorts[i],
- CONNECTION_TIMEOUT));
- }
-
- CountdownWatcher watch = new CountdownWatcher();
- ZooKeeper zk = new ZooKeeper("127.0.0.1:" + clientPorts[1],
- ClientBase.CONNECTION_TIMEOUT, watch);
- watch.waitForConnected(ClientBase.CONNECTION_TIMEOUT);
-
- /**
- * now the problem scenario starts
- */
-
- Stat firstEphemeralNode = new Stat();
-
- // 1: create ephemeral node
- String nodePath = "/e1";
- zk.create(nodePath, "1".getBytes(), Ids.OPEN_ACL_UNSAFE,
- CreateMode.EPHEMERAL, firstEphemeralNode);
- assertEquals("Current session and ephemeral owner should be same",
- zk.getSessionId(), firstEphemeralNode.getEphemeralOwner());
-
- // 2: inject network problem in one of the follower
- CustomQuorumPeer follower = (CustomQuorumPeer) getByServerState(mt,
- ServerState.FOLLOWING);
- follower.setInjectError(true);
-
- // 3: close the session so that ephemeral node is deleted
- zk.close();
-
- // remove the error
- follower.setInjectError(false);
-
- Assert.assertTrue("Faulted Follower should have joined quorum by now",
- ClientBase.waitForServerUp(
- "127.0.0.1:" + follower.getClientPort(),
- CONNECTION_TIMEOUT));
-
- QuorumPeer leader = getByServerState(mt, ServerState.LEADING);
- assertNotNull("Leader should not be null", leader);
- Assert.assertTrue("Leader must be running", ClientBase.waitForServerUp(
- "127.0.0.1:" + leader.getClientPort(), CONNECTION_TIMEOUT));
-
- watch = new CountdownWatcher();
- zk = new ZooKeeper("127.0.0.1:" + leader.getClientPort(),
- ClientBase.CONNECTION_TIMEOUT, watch);
- watch.waitForConnected(ClientBase.CONNECTION_TIMEOUT);
-
- Stat exists = zk.exists(nodePath, false);
- assertNull("Node must have been deleted from leader", exists);
-
- CountdownWatcher followerWatch = new CountdownWatcher();
- ZooKeeper followerZK = new ZooKeeper(
- "127.0.0.1:" + follower.getClientPort(),
- ClientBase.CONNECTION_TIMEOUT, followerWatch);
- followerWatch.waitForConnected(ClientBase.CONNECTION_TIMEOUT);
- Stat nodeAtFollower = followerZK.exists(nodePath, false);
-
- // Problem 1: Follower had one extra ephemeral node /e1
- assertNull("ephemeral node must not exist", nodeAtFollower);
-
- // Create the node with another session
- Stat currentEphemeralNode = new Stat();
- zk.create(nodePath, "2".getBytes(), Ids.OPEN_ACL_UNSAFE,
- CreateMode.EPHEMERAL, currentEphemeralNode);
-
- // close the session and newly created ephemeral node should be deleted
- zk.close();
-
- SyncCallback cb = new SyncCallback();
- followerZK.sync(nodePath, cb, null);
- cb.sync.await(CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS);
-
- nodeAtFollower = followerZK.exists(nodePath, false);
-
- // Problem 2: Before fix, after session close the ephemeral node
- // was not getting deleted. But now after the fix after session close
- // ephemeral node is getting deleted.
- assertNull("After session close ephemeral node must be deleted",
- nodeAtFollower);
- followerZK.close();
- }
-
- @After
- public void tearDown() {
- // stop all severs
- for (int i = 0; i < mt.length; i++) {
- try {
- mt[i].shutdown();
- } catch (InterruptedException e) {
- LOG.warn("Quorum Peer interrupted while shutting it down", e);
- }
- }
- }
-
- private QuorumPeer getByServerState(MainThread[] mt, ServerState state) {
- for (int i = mt.length - 1; i >= 0; i--) {
- QuorumPeer quorumPeer = mt[i].getQuorumPeer();
- if (null != quorumPeer && state == quorumPeer.getPeerState()) {
- return quorumPeer;
- }
- }
- return null;
- }
-
- static class CustomQuorumPeer extends QuorumPeer {
- private boolean injectError = false;
-
- public CustomQuorumPeer() throws SaslException {
-
- }
-
- @Override
- protected Follower makeFollower(FileTxnSnapLog logFactory)
- throws IOException {
- return new Follower(this, new FollowerZooKeeperServer(logFactory,
- this, this.getZkDb())) {
-
- @Override
- void readPacket(QuorumPacket pp) throws IOException {
- /**
- * In real scenario got SocketTimeoutException while reading
- * the packet from leader because of network problem, but
- * here throwing SocketTimeoutException based on whether
- * error is injected or not
- */
- super.readPacket(pp);
- if (injectError && pp.getType() == Leader.PROPOSAL) {
- String type = LearnerHandler.packetToString(pp);
- throw new SocketTimeoutException(
- "Socket timeout while reading the packet for operation "
- + type);
- }
- }
-
- };
- }
-
- public void setInjectError(boolean injectError) {
- this.injectError = injectError;
- }
-
- }
-
- static class MockTestQPMain extends TestQPMain {
- @Override
- protected QuorumPeer getQuorumPeer() throws SaslException {
- return new CustomQuorumPeer();
- }
- }
-
- private static class SyncCallback implements AsyncCallback.VoidCallback {
- private final CountDownLatch sync = new CountDownLatch(1);
-
- @Override
- public void processResult(int rc, String path, Object ctx) {
- sync.countDown();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/test/org/apache/zookeeper/server/quorum/FLEBackwardElectionRoundTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/quorum/FLEBackwardElectionRoundTest.java b/src/java/test/org/apache/zookeeper/server/quorum/FLEBackwardElectionRoundTest.java
deleted file mode 100644
index 6c88f0f..0000000
--- a/src/java/test/org/apache/zookeeper/server/quorum/FLEBackwardElectionRoundTest.java
+++ /dev/null
@@ -1,155 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server.quorum;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
-import java.util.HashMap;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.zookeeper.PortAssignment;
-import org.apache.zookeeper.ZKTestCase;
-import org.apache.zookeeper.server.quorum.QuorumCnxManager;
-import org.apache.zookeeper.server.quorum.QuorumPeer;
-import org.apache.zookeeper.server.quorum.Vote;
-import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
-import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
-import org.apache.zookeeper.test.ClientBase;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-public class FLEBackwardElectionRoundTest extends ZKTestCase {
- protected static final Logger LOG = LoggerFactory.getLogger(FLELostMessageTest.class);
-
- int count;
- HashMap<Long,QuorumServer> peers;
- File tmpdir[];
- int port[];
-
- QuorumCnxManager cnxManagers[];
-
- @Before
- public void setUp() throws Exception {
- count = 3;
-
- peers = new HashMap<Long,QuorumServer>(count);
- tmpdir = new File[count];
- port = new int[count];
- cnxManagers = new QuorumCnxManager[count - 1];
- }
-
- @After
- public void tearDown() throws Exception {
- for(int i = 0; i < (count - 1); i++){
- if(cnxManagers[i] != null){
- cnxManagers[i].halt();
- }
- }
- }
-
- /**
- * This test is checking the following case. A server S is
- * currently LOOKING and it receives notifications from
- * a quorum indicating they are following S. The election
- * round E of S is higher than the election round E' in the
- * notification messages, so S becomes the leader and sets
- * its epoch back to E'. In the meanwhile, one or more
- * followers turn to LOOKING and elect S in election round E.
- * Having leader and followers with different election rounds
- * might prevent other servers from electing a leader because
- * they can't get a consistent set of notifications from a
- * quorum.
- *
- * {@link https://issues.apache.org/jira/browse/ZOOKEEPER-1514}
- *
- *
- * @throws Exception
- */
-
- @Test
- public void testBackwardElectionRound() throws Exception {
- LOG.info("TestLE: {}, {}", getTestName(), count);
- for(int i = 0; i < count; i++) {
- int clientport = PortAssignment.unique();
- peers.put(Long.valueOf(i),
- new QuorumServer(i,
- new InetSocketAddress(clientport),
- new InetSocketAddress(PortAssignment.unique())));
- tmpdir[i] = ClientBase.createTmpDir();
- port[i] = clientport;
- }
-
- ByteBuffer initialMsg0 = getMsg();
- ByteBuffer initialMsg1 = getMsg();
-
- /*
- * Start server 0
- */
- QuorumPeer peer = new QuorumPeer(peers, tmpdir[0], tmpdir[0], port[0], 3, 0, 1000, 2, 2);
- peer.startLeaderElection();
- FLETestUtils.LEThread thread = new FLETestUtils.LEThread(peer, 0);
- thread.start();
-
- /*
- * Start mock server 1
- */
- QuorumPeer mockPeer = new QuorumPeer(peers, tmpdir[1], tmpdir[1], port[1], 3, 1, 1000, 2, 2);
- cnxManagers[0] = mockPeer.createCnxnManager();
- cnxManagers[0].listener.start();
-
- cnxManagers[0].toSend(0l, initialMsg0);
-
- /*
- * Start mock server 2
- */
- mockPeer = new QuorumPeer(peers, tmpdir[2], tmpdir[2], port[2], 3, 2, 1000, 2, 2);
- cnxManagers[1] = mockPeer.createCnxnManager();
- cnxManagers[1].listener.start();
-
- cnxManagers[1].toSend(0l, initialMsg1);
-
- /*
- * Run another instance of leader election.
- */
- thread.join(5000);
- thread = new FLETestUtils.LEThread(peer, 0);
- thread.start();
-
- /*
- * Send the same messages, this time should not make 0 the leader.
- */
- cnxManagers[0].toSend(0l, initialMsg0);
- cnxManagers[1].toSend(0l, initialMsg1);
-
- thread.join(5000);
-
- if (!thread.isAlive()) {
- Assert.fail("Should not have joined");
- }
-
- }
-
- private ByteBuffer getMsg() {
- return FLETestUtils.createMsg(ServerState.FOLLOWING.ordinal(), 0, 0, 1);
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/test/org/apache/zookeeper/server/quorum/FLELostMessageTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/quorum/FLELostMessageTest.java b/src/java/test/org/apache/zookeeper/server/quorum/FLELostMessageTest.java
deleted file mode 100644
index 6583f90..0000000
--- a/src/java/test/org/apache/zookeeper/server/quorum/FLELostMessageTest.java
+++ /dev/null
@@ -1,105 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server.quorum;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.util.HashMap;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.zookeeper.PortAssignment;
-import org.apache.zookeeper.ZKTestCase;
-import org.apache.zookeeper.server.quorum.FastLeaderElection;
-import org.apache.zookeeper.server.quorum.QuorumCnxManager;
-import org.apache.zookeeper.server.quorum.QuorumPeer;
-import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
-import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
-import org.apache.zookeeper.test.ClientBase;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-public class FLELostMessageTest extends ZKTestCase {
- protected static final Logger LOG = LoggerFactory.getLogger(FLELostMessageTest.class);
-
- int count;
- HashMap<Long,QuorumServer> peers;
- File tmpdir[];
- int port[];
-
- QuorumCnxManager cnxManager;
-
- @Before
- public void setUp() throws Exception {
- count = 3;
-
- peers = new HashMap<Long,QuorumServer>(count);
- tmpdir = new File[count];
- port = new int[count];
- }
-
- @After
- public void tearDown() throws Exception {
- cnxManager.halt();
- }
-
- @Test
- public void testLostMessage() throws Exception {
- LOG.info("TestLE: {}, {}", getTestName(), count);
- for(int i = 0; i < count; i++) {
- int clientport = PortAssignment.unique();
- peers.put(Long.valueOf(i),
- new QuorumServer(i,
- new InetSocketAddress(clientport),
- new InetSocketAddress(PortAssignment.unique())));
- tmpdir[i] = ClientBase.createTmpDir();
- port[i] = clientport;
- }
-
- /*
- * Start server 0
- */
- QuorumPeer peer = new QuorumPeer(peers, tmpdir[1], tmpdir[1], port[1], 3, 1, 1000, 2, 2);
- peer.startLeaderElection();
- FLETestUtils.LEThread thread = new FLETestUtils.LEThread(peer, 1);
- thread.start();
-
- /*
- * Start mock server 1
- */
- mockServer();
- thread.join(5000);
- if (thread.isAlive()) {
- Assert.fail("Threads didn't join");
- }
- }
-
- void mockServer() throws InterruptedException, IOException {
- QuorumPeer peer = new QuorumPeer(peers, tmpdir[0], tmpdir[0], port[0], 3, 0, 1000, 2, 2);
- cnxManager = peer.createCnxnManager();
- cnxManager.listener.start();
-
- cnxManager.toSend(1l, FLETestUtils.createMsg(ServerState.LOOKING.ordinal(), 0, 0, 0));
- cnxManager.recvQueue.take();
- cnxManager.toSend(1L, FLETestUtils.createMsg(ServerState.FOLLOWING.ordinal(), 1, 0, 0));
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/test/org/apache/zookeeper/server/quorum/FLETestUtils.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/quorum/FLETestUtils.java b/src/java/test/org/apache/zookeeper/server/quorum/FLETestUtils.java
deleted file mode 100644
index a907abd..0000000
--- a/src/java/test/org/apache/zookeeper/server/quorum/FLETestUtils.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server.quorum;
-
-import java.nio.ByteBuffer;
-
-import org.apache.zookeeper.ZKTestCase;
-import org.apache.zookeeper.server.quorum.FastLeaderElection;
-import org.apache.zookeeper.server.quorum.QuorumPeer;
-import org.apache.zookeeper.server.quorum.Vote;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.junit.Assert;
-
-import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
-
-public class FLETestUtils extends ZKTestCase {
- protected static final Logger LOG = LoggerFactory.getLogger(FLETestUtils.class);
-
- /*
- * Thread to run an instance of leader election for
- * a given quorum peer.
- */
- 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());
-
- Assert.assertTrue("State is not leading.", peer.getPeerState() == ServerState.LEADING);
- } catch (Exception e) {
- e.printStackTrace();
- }
- LOG.info("Joining");
- }
- }
-
- /*
- * Creates a leader election notification message.
- */
- static ByteBuffer createMsg(int state, long leader, long zxid, long epoch){
- return FastLeaderElection.buildMsg(state, leader, zxid, 1, epoch);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/test/org/apache/zookeeper/server/quorum/FuzzySnapshotRelatedTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/quorum/FuzzySnapshotRelatedTest.java b/src/java/test/org/apache/zookeeper/server/quorum/FuzzySnapshotRelatedTest.java
deleted file mode 100644
index b49a426..0000000
--- a/src/java/test/org/apache/zookeeper/server/quorum/FuzzySnapshotRelatedTest.java
+++ /dev/null
@@ -1,212 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server.quorum;
-
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import javax.security.sasl.SaslException;
-
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException.NoNodeException;
-import org.apache.zookeeper.KeeperException.NodeExistsException;
-import org.apache.zookeeper.Op;
-import org.apache.zookeeper.PortAssignment;
-import org.apache.zookeeper.ZooDefs.Ids;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.ZooKeeper.States;
-import org.apache.zookeeper.data.ACL;
-import org.apache.zookeeper.data.Stat;
-import org.apache.zookeeper.server.DataTree;
-import org.apache.zookeeper.server.ZKDatabase;
-import org.apache.zookeeper.server.ZooKeeperServer;
-import org.apache.zookeeper.test.ClientBase;
-
-import org.junit.Assert;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Test cases used to catch corner cases due to fuzzy snapshot.
- */
-public class FuzzySnapshotRelatedTest extends QuorumPeerTestBase {
-
- private static final Logger LOG = LoggerFactory.getLogger(FuzzySnapshotRelatedTest.class);
-
- MainThread[] mt = null;
- ZooKeeper[] zk = null;
- int leaderId;
- int followerA;
-
- @Before
- public void setup() throws Exception {
- LOG.info("Start up a 3 server quorum");
- final int ENSEMBLE_SERVERS = 3;
- final int clientPorts[] = new int[ENSEMBLE_SERVERS];
- StringBuilder sb = new StringBuilder();
- String server;
-
- for (int i = 0; i < ENSEMBLE_SERVERS; i++) {
- clientPorts[i] = PortAssignment.unique();
- server = "server." + i + "=127.0.0.1:" + PortAssignment.unique()
- + ":" + PortAssignment.unique() + ":participant;127.0.0.1:"
- + clientPorts[i];
- sb.append(server + "\n");
- }
- String currentQuorumCfgSection = sb.toString();
-
- // start servers
- mt = new MainThread[ENSEMBLE_SERVERS];
- zk = new ZooKeeper[ENSEMBLE_SERVERS];
- for (int i = 0; i < ENSEMBLE_SERVERS; i++) {
- mt[i] = new MainThread(i, clientPorts[i], currentQuorumCfgSection,
- false) {
- @Override
- public TestQPMain getTestQPMain() {
- return new CustomizedQPMain();
- }
- };
- mt[i].start();
- zk[i] = new ZooKeeper("127.0.0.1:" + clientPorts[i],
- ClientBase.CONNECTION_TIMEOUT, this);
- }
- QuorumPeerMainTest.waitForAll(zk, States.CONNECTED);
- LOG.info("all servers started");
-
- leaderId = -1;
- followerA = -1;
- for (int i = 0; i < ENSEMBLE_SERVERS; i++) {
- if (mt[i].main.quorumPeer.leader != null) {
- leaderId = i;
- } else if (followerA == -1) {
- followerA = i;
- }
- }
- }
-
- @After
- public void tearDown() throws Exception {
- if (mt != null) {
- for (MainThread t: mt) {
- t.shutdown();
- }
- }
-
- if (zk != null) {
- for (ZooKeeper z: zk) {
- z.close();
- }
- }
- }
-
- @Test
- public void testMultiOpConsistency() throws Exception {
- LOG.info("Create a parent node");
- final String path = "/testMultiOpConsistency";
- createEmptyNode(zk[followerA], path);
-
- LOG.info("Hook to catch the 2nd sub create node txn in multi-op");
- CustomDataTree dt =
- (CustomDataTree) mt[followerA].main.quorumPeer.getZkDb().getDataTree();
-
- final ZooKeeperServer zkServer = mt[followerA].main.quorumPeer.getActiveServer();
-
- String node1 = path + "/1";
- String node2 = path + "/2";
-
- dt.addNodeCreateListener(node2, new NodeCreateListener() {
- @Override
- public void process(String path) {
- LOG.info("Take a snapshot");
- zkServer.takeSnapshot();
- }
- });
-
- LOG.info("Issue a multi op to create 2 nodes");
- zk[followerA].multi(Arrays.asList(
- Op.create(node1, node1.getBytes(),
- Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT),
- Op.create(node2, node2.getBytes(),
- Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT))
- );
-
- LOG.info("Restart the server");
- mt[followerA].shutdown();
- QuorumPeerMainTest.waitForOne(zk[followerA], States.CONNECTING);
-
- mt[followerA].start();
- QuorumPeerMainTest.waitForOne(zk[followerA], States.CONNECTED);
-
- LOG.info("Make sure the node consistent with leader");
- Assert.assertEquals(new String(zk[leaderId].getData(node2, null, null)),
- new String(zk[followerA].getData(node2, null, null)));
- }
-
- private void createEmptyNode(ZooKeeper zk, String path) throws Exception {
- zk.create(path, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
- }
-
- static interface NodeCreateListener {
- public void process(String path);
-
- }
-
- static class CustomDataTree extends DataTree {
- Map<String, NodeCreateListener> nodeCreateListeners =
- new HashMap<String, NodeCreateListener>();
-
- @Override
- public void createNode(final String path, byte data[], List<ACL> acl,
- long ephemeralOwner, int parentCVersion, long zxid,
- long time, Stat outputStat)
- throws NoNodeException, NodeExistsException {
- NodeCreateListener listener = nodeCreateListeners.get(path);
- if (listener != null) {
- listener.process(path);
- }
- super.createNode(path, data, acl, ephemeralOwner, parentCVersion,
- zxid, time, outputStat);
- }
-
- public void addNodeCreateListener(String path, NodeCreateListener listener) {
- nodeCreateListeners.put(path, listener);
- }
- }
-
- static class CustomizedQPMain extends TestQPMain {
- @Override
- protected QuorumPeer getQuorumPeer() throws SaslException {
- return new QuorumPeer() {
- @Override
- public void setZKDatabase(ZKDatabase database) {
- super.setZKDatabase(new ZKDatabase(this.getTxnFactory()) {
- @Override
- public DataTree createDataTree() {
- return new CustomDataTree();
- }
- });
- }
- };
- }
- }
-}