You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by iv...@apache.org on 2012/11/23 11:49:55 UTC
svn commit: r1412818 [1/2] - in /zookeeper/bookkeeper/trunk: ./
bookkeeper-benchmark/
bookkeeper-benchmark/src/test/java/org/apache/bookkeeper/benchmark/
bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/
bookkeeper-server/src/test/java/org/...
Author: ivank
Date: Fri Nov 23 10:49:51 2012
New Revision: 1412818
URL: http://svn.apache.org/viewvc?rev=1412818&view=rev
Log:
BOOKKEEPER-467: Allocate ports for testing dynamically (ivank)
Added:
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/PortManager.java
Modified:
zookeeper/bookkeeper/trunk/CHANGES.txt
zookeeper/bookkeeper/trunk/bookkeeper-benchmark/pom.xml
zookeeper/bookkeeper/trunk/bookkeeper-benchmark/src/test/java/org/apache/bookkeeper/benchmark/TestBenchmark.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CookieTest.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/UpgradeTest.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerManager.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookKeeperClusterTestCase.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieZKExpireTest.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/NIOServerFactoryTest.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/TestBackwardCompat.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/ZooKeeperUtil.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/client/TestPubSubClient.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/client/netty/TestMultiplexing.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/HedwigHubTestBase.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/HedwigRegionTestBase.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/PubSubServerStandAloneTestBase.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/TestBackwardCompat.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/TestPubSubServerStartup.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/TestThrottlingDelivery.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/filter/TestMessageFilter.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigHub.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigRegion.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/netty/TestPubSubServer.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/BookKeeperTestBase.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/MessageBoundedPersistenceTest.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/TestDeadlock.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/subscriptions/TestUpdateSubscriptionState.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/topics/TestConcurrentTopicAcquisition.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/zookeeper/ZooKeeperTestBase.java
Modified: zookeeper/bookkeeper/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/CHANGES.txt?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/CHANGES.txt (original)
+++ zookeeper/bookkeeper/trunk/CHANGES.txt Fri Nov 23 10:49:51 2012
@@ -146,6 +146,8 @@ Trunk (unreleased changes)
IMPROVEMENTS:
+ BOOKKEEPER-467: Allocate ports for testing dynamically (ivank)
+
bookkeeper-server:
BOOKKEEPER-328: Bookie DeathWatcher is missing thread name (Rakesh via sijie)
Modified: zookeeper/bookkeeper/trunk/bookkeeper-benchmark/pom.xml
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-benchmark/pom.xml?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-benchmark/pom.xml (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-benchmark/pom.xml Fri Nov 23 10:49:51 2012
@@ -118,6 +118,13 @@
<type>jar</type>
<scope>compile</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.bookkeeper</groupId>
+ <artifactId>bookkeeper-server</artifactId>
+ <version>${project.parent.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
<!--
Annoying dependency we need to include because
zookeeper uses log4j and so we transatively do, but
Modified: zookeeper/bookkeeper/trunk/bookkeeper-benchmark/src/test/java/org/apache/bookkeeper/benchmark/TestBenchmark.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-benchmark/src/test/java/org/apache/bookkeeper/benchmark/TestBenchmark.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-benchmark/src/test/java/org/apache/bookkeeper/benchmark/TestBenchmark.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-benchmark/src/test/java/org/apache/bookkeeper/benchmark/TestBenchmark.java Fri Nov 23 10:49:51 2012
@@ -24,9 +24,11 @@ import org.junit.AfterClass;
import org.junit.Test;
import org.junit.Assert;
+import java.net.InetSocketAddress;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.util.LocalBookKeeper;
+import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,78 +46,18 @@ import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
-public class TestBenchmark {
+public class TestBenchmark extends BookKeeperClusterTestCase {
protected static final Logger LOG = LoggerFactory.getLogger(TestBenchmark.class);
- private static Thread ensembleThread = null;
- private final static String zkString = "localhost:2181";
- private static List<String> bookies = null;
-
- @BeforeClass
- public static void startEnsemble() throws Exception {
- final int numBookies = 5;
-
- ensembleThread = new Thread() {
- public void run() {
- try {
- LocalBookKeeper.main(new String[]{String.valueOf(numBookies)});
- } catch (InterruptedException ie) {
- LOG.info("Shutting down ensemble thread");
- } catch (Exception e) {
- LOG.error("Error running bookkeeper ensemble", e);
- }
- }
- };
- ensembleThread.start();
-
- if (!LocalBookKeeper.waitForServerUp(zkString, 5000)) {
- throw new Exception("Failed to start zookeeper");
- }
- ZooKeeper zk = null;
- try {
- final CountDownLatch connectLatch = new CountDownLatch(1);
-
- zk = new ZooKeeper(zkString, 15000, new Watcher() {
- @Override
- public void process(WatchedEvent event) {
- if (event.getState() == KeeperState.SyncConnected) {
- connectLatch.countDown();
- }
- }});
- if (!connectLatch.await(10, TimeUnit.SECONDS)) {
- LOG.error("Couldn't connect to zookeeper at " + zkString);
- } else {
- for (int i = 0; i < 10; i++) {
- try {
- bookies = zk.getChildren("/ledgers/available", false);
- if (zk.getChildren("/ledgers/available", false).size()
- == numBookies) {
- return;
- }
- } catch (Exception e) {
- // do nothing
- }
- Thread.sleep(1000);
- }
- throw new Exception("Not enough bookies started");
- }
- } finally {
- zk.close();
- }
- }
-
- @AfterClass
- public static void stopEnsemble() throws Exception {
- if (ensembleThread != null) {
- ensembleThread.interrupt();
- ensembleThread.join();
- }
+ public TestBenchmark() {
+ super(5);
}
@Test
public void testThroughputLatency() throws Exception {
String latencyFile = System.getProperty("test.latency.file", "latencyDump.dat");
BenchThroughputLatency.main(new String[] {
+ "--zookeeper", zkUtil.getZooKeeperConnectString(),
"--time", "10",
"--skipwarmup",
"--throttle", "1",
@@ -126,25 +68,26 @@ public class TestBenchmark {
@Test
public void testBookie() throws Exception {
- String bookie = bookies.get(0);
- String[] parts = bookie.split(":");
+ InetSocketAddress bookie = getBookie(0);
BenchBookie.main(new String[] {
- "--host", parts[0],
- "--port", parts[1],
- "--zookeeper", zkString
+ "--host", bookie.getHostName(),
+ "--port", String.valueOf(bookie.getPort()),
+ "--zookeeper", zkUtil.getZooKeeperConnectString()
});
}
@Test
public void testReadThroughputLatency() throws Exception {
- AtomicBoolean threwException = new AtomicBoolean(false);
+ final AtomicBoolean threwException = new AtomicBoolean(false);
Thread t = new Thread() {
public void run() {
try {
BenchReadThroughputLatency.main(new String[] {
+ "--zookeeper", zkUtil.getZooKeeperConnectString(),
"--listen", "10"});
} catch (Throwable t) {
LOG.error("Error reading", t);
+ threwException.set(true);
}
}
};
@@ -157,7 +100,7 @@ public class TestBenchmark {
long lastLedgerId = 0;
Assert.assertTrue("Thread should be running", t.isAlive());
for (int i = 0; i < 10; i++) {
- BookKeeper bk = new BookKeeper(zkString);
+ BookKeeper bk = new BookKeeper(zkUtil.getZooKeeperConnectString());
LedgerHandle lh = bk.createLedger(BookKeeper.DigestType.CRC32, "benchPasswd".getBytes());
lastLedgerId = lh.getId();
try {
@@ -179,6 +122,7 @@ public class TestBenchmark {
Assert.assertFalse("Thread should be finished", t.isAlive());
BenchReadThroughputLatency.main(new String[] {
+ "--zookeeper", zkUtil.getZooKeeperConnectString(),
"--ledger", String.valueOf(lastLedgerId)});
final long nextLedgerId = lastLedgerId+1;
@@ -186,16 +130,18 @@ public class TestBenchmark {
public void run() {
try {
BenchReadThroughputLatency.main(new String[] {
+ "--zookeeper", zkUtil.getZooKeeperConnectString(),
"--ledger", String.valueOf(nextLedgerId)});
} catch (Throwable t) {
LOG.error("Error reading", t);
+ threwException.set(true);
}
}
};
t.start();
Assert.assertTrue("Thread should be running", t.isAlive());
- BookKeeper bk = new BookKeeper(zkString);
+ BookKeeper bk = new BookKeeper(zkUtil.getZooKeeperConnectString());
LedgerHandle lh = bk.createLedger(BookKeeper.DigestType.CRC32, "benchPasswd".getBytes());
try {
for (int j = 0; j < 100; j++) {
@@ -212,5 +158,6 @@ public class TestBenchmark {
Thread.sleep(1000); // wait for 10 seconds for reading to finish
}
Assert.assertFalse("Thread should be finished", t.isAlive());
+ Assert.assertFalse("A thread has thrown an exception, check logs", threwException.get());
}
}
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CookieTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CookieTest.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CookieTest.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CookieTest.java Fri Nov 23 10:49:51 2012
@@ -26,6 +26,7 @@ import org.apache.bookkeeper.client.Book
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.test.ZooKeeperUtil;
+import org.apache.bookkeeper.test.PortManager;
import org.apache.zookeeper.ZooKeeper;
import java.io.File;
@@ -41,6 +42,7 @@ import static org.apache.bookkeeper.book
public class CookieTest {
ZooKeeperUtil zkutil;
ZooKeeper zkc = null;
+ final int bookiePort = PortManager.nextFreePort();
@Before
public void setupZooKeeper() throws Exception {
@@ -77,7 +79,7 @@ public class CookieTest {
.setZkServers(zkutil.getZooKeeperConnectString())
.setJournalDirName(newDirectory(false))
.setLedgerDirNames(new String[] { newDirectory(false) })
- .setBookiePort(3181);
+ .setBookiePort(bookiePort);
try {
Bookie b = new Bookie(conf);
} catch (Exception e) {
@@ -95,7 +97,7 @@ public class CookieTest {
ServerConfiguration conf1 = new ServerConfiguration()
.setJournalDirName(newDirectory())
.setLedgerDirNames(new String[] { newDirectory() })
- .setBookiePort(3181);
+ .setBookiePort(bookiePort);
Cookie c = Cookie.generateCookie(conf1);
c.writeToZooKeeper(zkc, conf1);
@@ -105,7 +107,7 @@ public class CookieTest {
.setZkServers(zkutil.getZooKeeperConnectString())
.setJournalDirName(journalDir)
.setLedgerDirNames(new String[] { ledgerDir })
- .setBookiePort(3181);
+ .setBookiePort(bookiePort);
Cookie c2 = Cookie.generateCookie(conf2);
c2.writeToDirectory(new File(journalDir, "current"));
c2.writeToDirectory(new File(ledgerDir, "current"));
@@ -132,7 +134,7 @@ public class CookieTest {
.setZkServers(zkutil.getZooKeeperConnectString())
.setJournalDirName(journalDir)
.setLedgerDirNames(ledgerDirs)
- .setBookiePort(3181);
+ .setBookiePort(bookiePort);
Bookie b = new Bookie(conf); // should work fine
b.start();
@@ -173,7 +175,7 @@ public class CookieTest {
.setZkServers(zkutil.getZooKeeperConnectString())
.setJournalDirName(journalDir)
.setLedgerDirNames(new String[] { ledgerDir0 })
- .setBookiePort(3181);
+ .setBookiePort(bookiePort);
Bookie b = new Bookie(conf); // should work fine
b.start();
@@ -205,7 +207,7 @@ public class CookieTest {
.setZkServers(zkutil.getZooKeeperConnectString())
.setJournalDirName(journalDir)
.setLedgerDirNames(new String[] { ledgerDir0 , newDirectory() })
- .setBookiePort(3181);
+ .setBookiePort(bookiePort);
Bookie b = new Bookie(conf); // should work fine
b.start();
@@ -230,7 +232,7 @@ public class CookieTest {
.setZkServers(zkutil.getZooKeeperConnectString())
.setJournalDirName(newDirectory())
.setLedgerDirNames(new String[] { newDirectory() , newDirectory() })
- .setBookiePort(3181);
+ .setBookiePort(bookiePort);
Bookie b = new Bookie(conf); // should work fine
b.start();
b.shutdown();
@@ -256,7 +258,7 @@ public class CookieTest {
.setZkServers(zkutil.getZooKeeperConnectString())
.setJournalDirName(newDirectory())
.setLedgerDirNames(new String[] { newDirectory() , newDirectory() })
- .setBookiePort(3181);
+ .setBookiePort(bookiePort);
Bookie b = new Bookie(conf); // should work fine
b.start();
b.shutdown();
@@ -265,7 +267,7 @@ public class CookieTest {
.setZkServers(zkutil.getZooKeeperConnectString())
.setJournalDirName(newDirectory())
.setLedgerDirNames(new String[] { newDirectory() , newDirectory() })
- .setBookiePort(3181);
+ .setBookiePort(bookiePort);
try {
b = new Bookie(conf);
fail("Shouldn't have been able to start");
@@ -279,7 +281,9 @@ public class CookieTest {
*/
@Test
public void testVerifyCookieWithFormat() throws Exception {
- ClientConfiguration adminConf = new ClientConfiguration();
+ ClientConfiguration adminConf = new ClientConfiguration()
+ .setZkServers(zkutil.getZooKeeperConnectString());
+
adminConf.setProperty("bookkeeper.format", true);
// Format the BK Metadata and generate INSTANCEID
BookKeeperAdmin.format(adminConf, false, true);
@@ -288,7 +292,7 @@ public class CookieTest {
.setZkServers(zkutil.getZooKeeperConnectString())
.setJournalDirName(newDirectory(false))
.setLedgerDirNames(new String[] { newDirectory(false) })
- .setBookiePort(3181);
+ .setBookiePort(bookiePort);
// Bookie should start successfully for fresh env.
new Bookie(bookieConf);
@@ -319,7 +323,7 @@ public class CookieTest {
.setZkServers(zkutil.getZooKeeperConnectString())
.setJournalDirName(newV2JournalDirectory())
.setLedgerDirNames(new String[] { newV2LedgerDirectory() })
- .setBookiePort(3181);
+ .setBookiePort(bookiePort);
try {
Bookie b = new Bookie(conf);
fail("Shouldn't have been able to start");
@@ -339,7 +343,7 @@ public class CookieTest {
.setZkServers(zkutil.getZooKeeperConnectString())
.setJournalDirName(newV1JournalDirectory())
.setLedgerDirNames(new String[] { newV1LedgerDirectory() })
- .setBookiePort(3181);
+ .setBookiePort(bookiePort);
try {
Bookie b = new Bookie(conf);
fail("Shouldn't have been able to start");
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/UpgradeTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/UpgradeTest.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/UpgradeTest.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/UpgradeTest.java Fri Nov 23 10:49:51 2012
@@ -46,6 +46,7 @@ import org.apache.bookkeeper.conf.Server
import org.apache.zookeeper.ZooKeeper;
import org.apache.bookkeeper.test.ZooKeeperUtil;
+import org.apache.bookkeeper.test.PortManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -55,6 +56,7 @@ public class UpgradeTest {
ZooKeeperUtil zkutil;
ZooKeeper zkc = null;
+ final static int bookiePort = PortManager.nextFreePort();
@Before
public void setupZooKeeper() throws Exception {
@@ -166,7 +168,7 @@ public class UpgradeTest {
.setZkServers(zkServers)
.setJournalDirName(journalDir)
.setLedgerDirNames(new String[] { ledgerDir })
- .setBookiePort(3181);
+ .setBookiePort(bookiePort);
Bookie b = null;
try {
b = new Bookie(conf);
@@ -223,7 +225,7 @@ public class UpgradeTest {
.setZkServers(zkutil.getZooKeeperConnectString())
.setJournalDirName(journalDir)
.setLedgerDirNames(new String[] { ledgerDir })
- .setBookiePort(3181);
+ .setBookiePort(bookiePort);
FileSystemUpgrade.upgrade(conf); // should work fine with current directory
Bookie b = new Bookie(conf);
b.start();
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java Fri Nov 23 10:49:51 2012
@@ -245,6 +245,7 @@ public class BookieRecoveryTest extends
// Shutdown the first bookie server
LOG.info("Finished writing all ledger entries so shutdown one of the bookies.");
+ int initialPort = bsConfs.get(0).getBookiePort();
bs.get(0).shutdown();
bs.remove(0);
@@ -298,6 +299,7 @@ public class BookieRecoveryTest extends
// Shutdown the first bookie server
LOG.info("Finished writing all ledger entries so shutdown one of the bookies.");
+ int initialPort = bsConfs.get(0).getBookiePort();
bs.get(0).shutdown();
bs.remove(0);
@@ -352,6 +354,7 @@ public class BookieRecoveryTest extends
// Shutdown the first bookie server
LOG.info("Finished writing all ledger entries so shutdown one of the bookies.");
+ int initialPort = bsConfs.get(0).getBookiePort();
bs.get(0).shutdown();
bs.remove(0);
@@ -395,6 +398,7 @@ public class BookieRecoveryTest extends
// Shutdown the first bookie server
LOG.info("Finished writing all ledger entries so shutdown one of the bookies.");
+ int initialPort = bsConfs.get(0).getBookiePort();
bs.get(0).shutdown();
bs.remove(0);
@@ -694,6 +698,7 @@ public class BookieRecoveryTest extends
// Shutdown the first bookie server
LOG.info("Finished writing all ledger entries so shutdown one of the bookies.");
+ int initialPort = bsConfs.get(0).getBookiePort();
bs.get(0).shutdown();
bs.remove(0);
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerManager.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerManager.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerManager.java Fri Nov 23 10:49:51 2012
@@ -196,12 +196,13 @@ public class TestLedgerManager extends B
private final CyclicBarrier barrier;
private ZooKeeper zkc;
- CreateLMThread(String root, String factoryCls, CyclicBarrier barrier) throws Exception {
+ CreateLMThread(String zkConnectString, String root,
+ String factoryCls, CyclicBarrier barrier) throws Exception {
this.factoryCls = factoryCls;
this.barrier = barrier;
this.root = root;
final CountDownLatch latch = new CountDownLatch(1);
- zkc = new ZooKeeper("127.0.0.1", 10000, new Watcher() {
+ zkc = new ZooKeeper(zkConnectString, 10000, new Watcher() {
public void process(WatchedEvent event) {
latch.countDown();
}
@@ -248,7 +249,8 @@ public class TestLedgerManager extends B
CyclicBarrier barrier = new CyclicBarrier(numThreads+1);
List<CreateLMThread> threads = new ArrayList<CreateLMThread>(numThreads);
for (int i = 0; i < numThreads; i++) {
- CreateLMThread t = new CreateLMThread(root0, FlatLedgerManagerFactory.class.getName(), barrier);
+ CreateLMThread t = new CreateLMThread(zkUtil.getZooKeeperConnectString(),
+ root0, FlatLedgerManagerFactory.class.getName(), barrier);
t.start();
threads.add(t);
}
@@ -277,14 +279,15 @@ public class TestLedgerManager extends B
CyclicBarrier barrier = new CyclicBarrier(numThreadsEach*2+1);
List<CreateLMThread> threadsA = new ArrayList<CreateLMThread>(numThreadsEach);
for (int i = 0; i < numThreadsEach; i++) {
- CreateLMThread t = new CreateLMThread(root0, FlatLedgerManagerFactory.class.getName(), barrier);
+ CreateLMThread t = new CreateLMThread(zkUtil.getZooKeeperConnectString(),
+ root0, FlatLedgerManagerFactory.class.getName(), barrier);
t.start();
threadsA.add(t);
}
List<CreateLMThread> threadsB = new ArrayList<CreateLMThread>(numThreadsEach);
for (int i = 0; i < numThreadsEach; i++) {
- CreateLMThread t = new CreateLMThread(root0,
- HierarchicalLedgerManagerFactory.class.getName(), barrier);
+ CreateLMThread t = new CreateLMThread(zkUtil.getZooKeeperConnectString(),
+ root0, HierarchicalLedgerManagerFactory.class.getName(), barrier);
t.start();
threadsB.add(t);
}
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookKeeperClusterTestCase.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookKeeperClusterTestCase.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookKeeperClusterTestCase.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookKeeperClusterTestCase.java Fri Nov 23 10:49:51 2012
@@ -60,8 +60,6 @@ public abstract class BookKeeperClusterT
protected List<File> tmpDirs = new LinkedList<File>();
protected List<BookieServer> bs = new LinkedList<BookieServer>();
protected List<ServerConfiguration> bsConfs = new LinkedList<ServerConfiguration>();
- protected Integer initialPort = 5000;
- private Integer nextPort = initialPort;
protected int numBookies;
protected BookKeeperTestClient bkc;
@@ -351,7 +349,7 @@ public abstract class BookKeeperClusterT
f.delete();
f.mkdir();
- int port = nextPort++;
+ int port = PortManager.nextFreePort();
ServerConfiguration conf = newServerConfiguration(port, zkUtil.getZooKeeperConnectString(),
f, new File[] { f });
bsConfs.add(conf);
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieZKExpireTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieZKExpireTest.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieZKExpireTest.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieZKExpireTest.java Fri Nov 23 10:49:51 2012
@@ -57,7 +57,7 @@ public class BookieZKExpireTest extends
}
}
- ServerConfiguration conf = newServerConfiguration(initialPort + 1,
+ ServerConfiguration conf = newServerConfiguration(PortManager.nextFreePort(),
zkUtil.getZooKeeperConnectString(), f, new File[] { f });
server = new BookieServer(conf);
server.start();
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/NIOServerFactoryTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/NIOServerFactoryTest.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/NIOServerFactoryTest.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/NIOServerFactoryTest.java Fri Nov 23 10:49:51 2012
@@ -47,10 +47,11 @@ public class NIOServerFactoryTest extend
@Test
public void testProblemProcessor() throws Exception {
ServerConfiguration conf = new ServerConfiguration();
- conf.setBookiePort(22334);
+ int port = PortManager.nextFreePort();
+ conf.setBookiePort(port);
NIOServerFactory factory = new NIOServerFactory(conf, problemProcessor);
factory.start();
- Socket s = new Socket("127.0.0.1", 22334);
+ Socket s = new Socket("127.0.0.1", port);
s.setSoTimeout(5000);
try {
s.getOutputStream().write("\0\0\0\4\0\0\0\1".getBytes());
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/PortManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/PortManager.java?rev=1412818&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/PortManager.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/PortManager.java Fri Nov 23 10:49:51 2012
@@ -0,0 +1,55 @@
+/*
+ *
+ * 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.bookkeeper.test;
+
+import java.net.ServerSocket;
+import java.io.IOException;
+/**
+ * Port manager allows a base port to be specified on the commandline.
+ * Tests will then use ports, counting up from this base port.
+ * This allows multiple instances of the bookkeeper tests to run at once.
+ */
+public class PortManager {
+ private static int nextPort = getBasePort();
+
+ public synchronized static int nextFreePort() {
+ while (true) {
+ ServerSocket ss = null;
+ try {
+ int port = nextPort++;
+ ss = new ServerSocket(port);
+ ss.setReuseAddress(true);
+ return port;
+ } catch (IOException ioe) {
+ } finally {
+ if (ss != null) {
+ try {
+ ss.close();
+ } catch (IOException ioe) {}
+ }
+ }
+ }
+ }
+
+ private static int getBasePort() {
+ return Integer.valueOf(System.getProperty("test.basePort", "15000"));
+ }
+}
\ No newline at end of file
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/TestBackwardCompat.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/TestBackwardCompat.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/TestBackwardCompat.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/TestBackwardCompat.java Fri Nov 23 10:49:51 2012
@@ -44,7 +44,6 @@ public class TestBackwardCompat {
static Logger LOG = LoggerFactory.getLogger(TestBackwardCompat.class);
private static ZooKeeperUtil zkUtil = new ZooKeeperUtil();;
- private static int nextPort = 3181;
private static byte[] ENTRY_DATA = "ThisIsAnEntry".getBytes();
static void waitUp(int port) throws Exception {
@@ -367,7 +366,7 @@ public class TestBackwardCompat {
ledgerDir.delete();
ledgerDir.mkdir();
- int port = nextPort++;
+ int port = PortManager.nextFreePort();
// start old server
Server410 s410 = new Server410(journalDir, ledgerDir, port);
@@ -425,7 +424,7 @@ public class TestBackwardCompat {
ledgerDir.delete();
ledgerDir.mkdir();
- int port = nextPort++;
+ int port = PortManager.nextFreePort();
// start server, upgrade
Server400 s400 = new Server400(journalDir, ledgerDir, port);
s400.start();
@@ -513,7 +512,7 @@ public class TestBackwardCompat {
ledgerDir.delete();
ledgerDir.mkdir();
- int port = nextPort++;
+ int port = PortManager.nextFreePort();
// start server, upgrade
Server410 s410 = new Server410(journalDir, ledgerDir, port);
s410.start();
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/ZooKeeperUtil.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/ZooKeeperUtil.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/ZooKeeperUtil.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/ZooKeeperUtil.java Fri Nov 23 10:49:51 2012
@@ -49,7 +49,7 @@ public class ZooKeeperUtil {
static final Logger LOG = LoggerFactory.getLogger(ZooKeeperUtil.class);
// ZooKeeper related variables
- protected static Integer ZooKeeperDefaultPort = 2181;
+ protected final static Integer zooKeeperPort = PortManager.nextFreePort();
private final InetSocketAddress zkaddr;
protected ZooKeeperServer zks;
@@ -59,8 +59,8 @@ public class ZooKeeperUtil {
private final String connectString;
public ZooKeeperUtil() {
- zkaddr = new InetSocketAddress(ZooKeeperDefaultPort);
- connectString= "localhost:" + ZooKeeperDefaultPort;
+ zkaddr = new InetSocketAddress(zooKeeperPort);
+ connectString= "localhost:" + zooKeeperPort;
}
public ZooKeeper getZooKeeperClient() {
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/client/TestPubSubClient.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/client/TestPubSubClient.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/client/TestPubSubClient.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/client/TestPubSubClient.java Fri Nov 23 10:49:51 2012
@@ -51,6 +51,7 @@ import org.apache.hedwig.server.PubSubSe
import org.apache.hedwig.util.Callback;
import org.apache.hedwig.util.ConcurrencyUtils;
import org.apache.hedwig.util.SubscriptionListener;
+import org.apache.hedwig.util.HedwigSocketAddress;
@RunWith(Parameterized.class)
public class TestPubSubClient extends PubSubServerStandAloneTestBase {
@@ -173,6 +174,11 @@ public class TestPubSubClient extends Pu
super.setUp();
client = new HedwigClient(new ClientConfiguration() {
@Override
+ public HedwigSocketAddress getDefaultServerHedwigSocketAddress() {
+ return getDefaultHedwigAddress();
+ }
+
+ @Override
public boolean isSubscriptionChannelSharingEnabled() {
return TestPubSubClient.this.isSubscriptionChannelSharingEnabled;
}
@@ -516,7 +522,12 @@ public class TestPubSubClient extends Pu
subscriber.startDelivery(topic, subscriberId, new TestMessageHandler());
// new a client
- HedwigClient client2 = new HedwigClient(new ClientConfiguration());
+ HedwigClient client2 = new HedwigClient(new ClientConfiguration() {
+ @Override
+ public HedwigSocketAddress getDefaultServerHedwigSocketAddress() {
+ return getDefaultHedwigAddress();
+ }
+ });
Subscriber subscriber2 = client2.getSubscriber();
Publisher publisher2 = client2.getPublisher();
SynchronousQueue<SubscriptionEvent> eventQueue2 =
@@ -553,8 +564,24 @@ public class TestPubSubClient extends Pu
@Test
public void testSyncSubscribeWithListenerWhenReleasingTopic() throws Exception {
+ client.close();
+
tearDownHubServer();
startHubServer(new RetentionServerConfiguration());
+ client = new HedwigClient(new ClientConfiguration() {
+ @Override
+ public HedwigSocketAddress getDefaultServerHedwigSocketAddress() {
+ return getDefaultHedwigAddress();
+ }
+
+ @Override
+ public boolean isSubscriptionChannelSharingEnabled() {
+ return TestPubSubClient.this.isSubscriptionChannelSharingEnabled;
+ }
+ });
+ publisher = client.getPublisher();
+ subscriber = client.getSubscriber();
+
ByteString topic = ByteString.copyFromUtf8("mySyncSubscribeWithListenerWhenReleasingTopic");
ByteString subscriberId = ByteString.copyFromUtf8("mysub");
subscriber.addSubscriptionListener(new TestSubscriptionListener());
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/client/netty/TestMultiplexing.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/client/netty/TestMultiplexing.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/client/netty/TestMultiplexing.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/client/netty/TestMultiplexing.java Fri Nov 23 10:49:51 2012
@@ -225,9 +225,8 @@ public class TestMultiplexing extends He
@Override
@Before
public void setUp() throws Exception {
- numServers = 1;
super.setUp();
- client = new HedwigClient(new ClientConfiguration() {
+ client = new HedwigClient(new HubClientConfiguration() {
@Override
public boolean isSubscriptionChannelSharingEnabled() {
return true;
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/HedwigHubTestBase.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/HedwigHubTestBase.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/HedwigHubTestBase.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/HedwigHubTestBase.java Fri Nov 23 10:49:51 2012
@@ -27,9 +27,13 @@ import org.slf4j.LoggerFactory;
import org.junit.After;
import org.junit.Before;
+import org.apache.hedwig.client.conf.ClientConfiguration;
import org.apache.hedwig.server.common.ServerConfiguration;
import org.apache.hedwig.server.netty.PubSubServer;
import org.apache.hedwig.server.persistence.BookKeeperTestBase;
+import org.apache.hedwig.util.HedwigSocketAddress;
+
+import org.apache.bookkeeper.test.PortManager;
/**
* This is a base class for any tests that need a Hedwig Hub(s) setup with an
@@ -50,10 +54,9 @@ public abstract class HedwigHubTestBase
// PubSubServer variables
// Default number of PubSubServer hubs to setup. Extending classes can
// override this.
- protected int numServers = 1;
- protected int initialServerPort = 4080;
- protected int initialSSLServerPort = 9876;
+ protected final int numServers;
protected List<PubSubServer> serversList;
+ protected List<HedwigSocketAddress> serverAddresses;
public HedwigHubTestBase() {
this(1);
@@ -61,6 +64,12 @@ public abstract class HedwigHubTestBase
protected HedwigHubTestBase(int numServers) {
this.numServers = numServers;
+
+ serverAddresses = new LinkedList<HedwigSocketAddress>();
+ for (int i = 0; i < numServers; i++) {
+ serverAddresses.add(new HedwigSocketAddress("localhost",
+ PortManager.nextFreePort(), PortManager.nextFreePort()));
+ }
}
// Default child class of the ServerConfiguration to be used here.
@@ -106,6 +115,13 @@ public abstract class HedwigHubTestBase
}
}
+ public class HubClientConfiguration extends ClientConfiguration {
+ @Override
+ public HedwigSocketAddress getDefaultServerHedwigSocketAddress() {
+ return serverAddresses.get(0);
+ }
+ }
+
// Method to get a ServerConfiguration for the PubSubServers created using
// the specified ports. Extending child classes can override this. This
// default implementation will return the HubServerConfiguration object
@@ -117,9 +133,11 @@ public abstract class HedwigHubTestBase
protected void startHubServers() throws Exception {
// Now create the PubSubServer Hubs
serversList = new LinkedList<PubSubServer>();
+
for (int i = 0; i < numServers; i++) {
- PubSubServer s = new PubSubServer(
- getServerConfiguration(initialServerPort + i, initialSSLServerPort + i));
+ ServerConfiguration conf = getServerConfiguration(serverAddresses.get(i).getPort(),
+ serverAddresses.get(i).getSSLPort());
+ PubSubServer s = new PubSubServer(conf);
serversList.add(s);
s.start();
}
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/HedwigRegionTestBase.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/HedwigRegionTestBase.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/HedwigRegionTestBase.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/HedwigRegionTestBase.java Fri Nov 23 10:49:51 2012
@@ -36,6 +36,8 @@ import org.apache.hedwig.server.netty.Pu
import org.apache.hedwig.server.persistence.BookKeeperTestBase;
import org.apache.hedwig.util.HedwigSocketAddress;
+import org.apache.bookkeeper.test.PortManager;
+
/**
* This is a base class for any tests that need a Hedwig Region(s) setup with a
* number of Hedwig hubs per region, an associated HedwigClient per region and
@@ -58,8 +60,7 @@ public abstract class HedwigRegionTestBa
// override this.
protected int numRegions = 2;
protected int numServersPerRegion = 1;
- protected int initialServerPort = 4080;
- protected int initialSSLServerPort = 9876;
+
// Map with keys being Region names and values being the list of Hedwig
// Hubs (PubSubServers) for that particular region.
protected Map<String, List<PubSubServer>> regionServersMap;
@@ -67,6 +68,9 @@ public abstract class HedwigRegionTestBa
// instance.
protected Map<String, HedwigClient> regionClientsMap;
+ protected Map<String, Integer> regionNameToIndexMap;
+ protected Map<Integer, List<HedwigSocketAddress>> regionHubAddresses;
+
// String constant used as the prefix for the region names.
protected static final String REGION_PREFIX = "region";
@@ -89,14 +93,15 @@ public abstract class HedwigRegionTestBa
protected void setRegionList() {
List<String> myRegionList = new LinkedList<String>();
for (int i = 0; i < numRegions; i++) {
- int curDefaultServerPort = initialServerPort + (i * numServersPerRegion);
- int curDefaultSSLServerPort = initialSSLServerPort + (i * numServersPerRegion);
+ int curDefaultServerPort = regionHubAddresses.get(i).get(0).getPort();
+ int curDefaultSSLServerPort = regionHubAddresses.get(i).get(0).getSSLPort();
// Add this region default server port if it is for a region
// other than its own.
- if (curDefaultServerPort > serverPort
- || Math.abs(serverPort - curDefaultServerPort) >= numServersPerRegion)
+ if (regionNameToIndexMap.get(regionName) != i) {
myRegionList.add("localhost:" + curDefaultServerPort + ":" + curDefaultSSLServerPort);
+ }
}
+
regionList = myRegionList;
}
@@ -168,7 +173,12 @@ public abstract class HedwigRegionTestBa
// Method to get a ClientConfiguration for the Cross Region Hedwig Client.
protected ClientConfiguration getRegionClientConfiguration() {
- return new ClientConfiguration();
+ return new ClientConfiguration() {
+ @Override
+ public HedwigSocketAddress getDefaultServerHedwigSocketAddress() {
+ return regionHubAddresses.get(0).get(0);
+ }
+ };
}
@Override
@@ -181,6 +191,19 @@ public abstract class HedwigRegionTestBa
// Create the Hedwig PubSubServer Hubs for all of the regions
regionServersMap = new HashMap<String, List<PubSubServer>>(numRegions, 1.0f);
regionClientsMap = new HashMap<String, HedwigClient>(numRegions, 1.0f);
+
+ regionHubAddresses = new HashMap<Integer, List<HedwigSocketAddress>>(numRegions, 1.0f);
+ for (int i = 0; i < numRegions; i++) {
+ List<HedwigSocketAddress> addresses = new LinkedList<HedwigSocketAddress>();
+ for (int j = 0; j < numServersPerRegion; j++) {
+ HedwigSocketAddress a = new HedwigSocketAddress("localhost",
+ PortManager.nextFreePort(), PortManager.nextFreePort());
+ addresses.add(a);
+ }
+ regionHubAddresses.put(i, addresses);
+ }
+ regionNameToIndexMap = new HashMap<String, Integer>();
+
for (int i = 0; i < numRegions; i++) {
startRegion(i);
}
@@ -230,18 +253,21 @@ public abstract class HedwigRegionTestBa
protected void startRegion(int i) throws Exception {
String regionName = REGION_PREFIX + i;
+ regionNameToIndexMap.put(regionName, i);
+
if (logger.isDebugEnabled()) {
logger.debug("Start region : " + regionName);
}
+
List<PubSubServer> serversList = new LinkedList<PubSubServer>();
// For the current region, create the necessary amount of hub
// servers. We will basically increment through the port numbers
// starting from the initial ones defined.
for (int j = 0; j < numServersPerRegion; j++) {
+ HedwigSocketAddress a = regionHubAddresses.get(i).get(j);
PubSubServer s = new PubSubServer(
- getServerConfiguration(initialServerPort
- + (j + i * numServersPerRegion),
- initialSSLServerPort + (j + i * numServersPerRegion),
+ getServerConfiguration(a.getPort(),
+ a.getSSLPort(),
regionName),
getRegionClientConfiguration());
serversList.add(s);
@@ -251,8 +277,9 @@ public abstract class HedwigRegionTestBa
regionServersMap.put(regionName, serversList);
// Create a Hedwig Client that points to the first Hub server
// created in the loop above for the current region.
- HedwigClient regionClient = new HedwigClient(getClientConfiguration(initialServerPort
- + (i * numServersPerRegion), initialSSLServerPort + (i * numServersPerRegion)));
+ HedwigClient regionClient = new HedwigClient(
+ getClientConfiguration(regionHubAddresses.get(i).get(0).getPort(),
+ regionHubAddresses.get(i).get(0).getSSLPort()));
regionClientsMap.put(regionName, regionClient);
}
}
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/PubSubServerStandAloneTestBase.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/PubSubServerStandAloneTestBase.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/PubSubServerStandAloneTestBase.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/PubSubServerStandAloneTestBase.java Fri Nov 23 10:49:51 2012
@@ -24,8 +24,11 @@ import org.slf4j.LoggerFactory;
import org.junit.After;
import org.junit.Before;
+import org.apache.bookkeeper.test.PortManager;
+
import org.apache.hedwig.server.common.ServerConfiguration;
import org.apache.hedwig.server.netty.PubSubServer;
+import org.apache.hedwig.util.HedwigSocketAddress;
/**
* This is a base class for any tests that need a StandAlone PubSubServer setup.
@@ -35,10 +38,23 @@ public abstract class PubSubServerStandA
protected static Logger logger = LoggerFactory.getLogger(PubSubServerStandAloneTestBase.class);
protected class StandAloneServerConfiguration extends ServerConfiguration {
+ final int port = PortManager.nextFreePort();
+ final int sslPort = PortManager.nextFreePort();
+
@Override
public boolean isStandalone() {
return true;
}
+
+ @Override
+ public int getServerPort() {
+ return port;
+ }
+
+ @Override
+ public int getSSLServerPort() {
+ return sslPort;
+ }
}
public ServerConfiguration getStandAloneServerConfiguration() {
@@ -46,6 +62,7 @@ public abstract class PubSubServerStandA
}
protected PubSubServer server;
+ protected HedwigSocketAddress defaultAddress;
@Override
@Before
@@ -65,10 +82,19 @@ public abstract class PubSubServerStandA
}
protected void startHubServer() throws Exception {
- startHubServer(getStandAloneServerConfiguration());
+ ServerConfiguration conf = getStandAloneServerConfiguration();
+ defaultAddress = new HedwigSocketAddress("localhost", conf.getServerPort(),
+ conf.getSSLServerPort());
+ startHubServer(conf);
+ }
+
+ protected HedwigSocketAddress getDefaultHedwigAddress() {
+ return defaultAddress;
}
protected void startHubServer(ServerConfiguration conf) throws Exception {
+ defaultAddress = new HedwigSocketAddress("localhost", conf.getServerPort(),
+ conf.getSSLServerPort());
server = new PubSubServer(conf);
server.start();
}
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/TestBackwardCompat.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/TestBackwardCompat.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/TestBackwardCompat.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/TestBackwardCompat.java Fri Nov 23 10:49:51 2012
@@ -34,6 +34,10 @@ import org.junit.Test;
import static org.junit.Assert.*;
import org.apache.bookkeeper.test.ZooKeeperUtil;
+import org.apache.bookkeeper.test.PortManager;
+
+import org.apache.hedwig.util.HedwigSocketAddress;
+
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
import org.slf4j.Logger;
@@ -55,9 +59,6 @@ public class TestBackwardCompat extends
List<org.apache.hw_v4_0_0.bookkeeper.conf.ServerConfiguration> bkConfs;
List<org.apache.hw_v4_0_0.bookkeeper.proto.BookieServer> bks;
- private int initialPort = 5000;
- private int nextPort = initialPort;
-
BookKeeperCluster400(int numBookies) {
this.numBookies = numBookies;
}
@@ -83,10 +84,11 @@ public class TestBackwardCompat extends
}
protected void startBookieServer() throws Exception {
+ int port = PortManager.nextFreePort();
File tmpDir = org.apache.hw_v4_0_0.hedwig.util.FileUtils.createTempDirectory(
- getClass().getName() + (nextPort - initialPort), "test");
+ getClass().getName() + port, "test");
org.apache.hw_v4_0_0.bookkeeper.conf.ServerConfiguration conf = newServerConfiguration(
- nextPort++, zkUtil.getZooKeeperConnectString(), tmpDir, new File[] { tmpDir });
+ port, zkUtil.getZooKeeperConnectString(), tmpDir, new File[] { tmpDir });
bks.add(startBookie(conf));
bkConfs.add(conf);
}
@@ -129,12 +131,22 @@ public class TestBackwardCompat extends
org.apache.hw_v4_0_0.hedwig.server.common.ServerConfiguration conf;
org.apache.hw_v4_0_0.hedwig.server.netty.PubSubServer server;
- Server400(final String zkHosts) {
+ Server400(final String zkHosts, final int port, final int sslPort) {
conf = new org.apache.hw_v4_0_0.hedwig.server.common.ServerConfiguration() {
@Override
public String getZkHost() {
return zkHosts;
}
+
+ @Override
+ public int getServerPort() {
+ return port;
+ }
+
+ @Override
+ public int getSSLServerPort() {
+ return sslPort;
+ }
};
}
@@ -155,8 +167,14 @@ public class TestBackwardCompat extends
org.apache.hw_v4_0_0.hedwig.client.api.Publisher publisher;
org.apache.hw_v4_0_0.hedwig.client.api.Subscriber subscriber;
- Client400() {
- conf = new org.apache.hw_v4_0_0.hedwig.client.conf.ClientConfiguration();
+ Client400(final String connectString) {
+ conf = new org.apache.hw_v4_0_0.hedwig.client.conf.ClientConfiguration() {
+ @Override
+ protected org.apache.hw_v4_0_0.hedwig.util.HedwigSocketAddress
+ getDefaultServerHedwigSocketAddress() {
+ return new org.apache.hw_v4_0_0.hedwig.util.HedwigSocketAddress(connectString);
+ }
+ };
client = new org.apache.hw_v4_0_0.hedwig.client.HedwigClient(conf);
publisher = client.getPublisher();
subscriber = client.getSubscriber();
@@ -184,9 +202,6 @@ public class TestBackwardCompat extends
List<org.apache.hw_v4_1_0.bookkeeper.conf.ServerConfiguration> bkConfs;
List<org.apache.hw_v4_1_0.bookkeeper.proto.BookieServer> bks;
- private int initialPort = 5000;
- private int nextPort = initialPort;
-
BookKeeperCluster410(int numBookies) {
this.numBookies = numBookies;
}
@@ -212,10 +227,11 @@ public class TestBackwardCompat extends
}
protected void startBookieServer() throws Exception {
+ int port = PortManager.nextFreePort();
File tmpDir = org.apache.hw_v4_1_0.hedwig.util.FileUtils.createTempDirectory(
- getClass().getName() + (nextPort - initialPort), "test");
+ getClass().getName() + port, "test");
org.apache.hw_v4_1_0.bookkeeper.conf.ServerConfiguration conf = newServerConfiguration(
- nextPort++, zkUtil.getZooKeeperConnectString(), tmpDir, new File[] { tmpDir });
+ port, zkUtil.getZooKeeperConnectString(), tmpDir, new File[] { tmpDir });
bks.add(startBookie(conf));
bkConfs.add(conf);
}
@@ -258,7 +274,7 @@ public class TestBackwardCompat extends
org.apache.hw_v4_1_0.hedwig.server.common.ServerConfiguration conf;
org.apache.hw_v4_1_0.hedwig.server.netty.PubSubServer server;
- Server410(final String zkHosts) {
+ Server410(final String zkHosts, final int port, final int sslPort) {
conf = new org.apache.hw_v4_1_0.hedwig.server.common.ServerConfiguration() {
@Override
public int getConsumeInterval() {
@@ -268,6 +284,16 @@ public class TestBackwardCompat extends
public String getZkHost() {
return zkHosts;
}
+
+ @Override
+ public int getServerPort() {
+ return port;
+ }
+
+ @Override
+ public int getSSLServerPort() {
+ return sslPort;
+ }
};
}
@@ -324,7 +350,7 @@ public class TestBackwardCompat extends
}
}
- Client410() {
+ Client410(final String connectString) {
conf = new org.apache.hw_v4_1_0.hedwig.client.conf.ClientConfiguration() {
@Override
public boolean isAutoSendConsumeMessageEnabled() {
@@ -334,6 +360,11 @@ public class TestBackwardCompat extends
public int getConsumedMessagesBufferSize() {
return 1;
}
+ @Override
+ protected org.apache.hw_v4_1_0.hedwig.util.HedwigSocketAddress
+ getDefaultServerHedwigSocketAddress() {
+ return new org.apache.hw_v4_1_0.hedwig.util.HedwigSocketAddress(connectString);
+ }
};
client = new org.apache.hw_v4_1_0.hedwig.client.HedwigClient(conf);
publisher = client.getPublisher();
@@ -438,8 +469,6 @@ public class TestBackwardCompat extends
List<org.apache.bookkeeper.conf.ServerConfiguration> bkConfs;
List<org.apache.bookkeeper.proto.BookieServer> bks;
- private int initialPort = 5000;
- private int nextPort = initialPort;
BookKeeperClusterCurrent(int numBookies) {
this.numBookies = numBookies;
@@ -466,10 +495,11 @@ public class TestBackwardCompat extends
}
protected void startBookieServer() throws Exception {
+ int port = PortManager.nextFreePort();
File tmpDir = org.apache.hedwig.util.FileUtils.createTempDirectory(
- getClass().getName() + (nextPort - initialPort), "test");
+ getClass().getName() + port, "test");
org.apache.bookkeeper.conf.ServerConfiguration conf = newServerConfiguration(
- nextPort++, zkUtil.getZooKeeperConnectString(), tmpDir, new File[] { tmpDir });
+ port, zkUtil.getZooKeeperConnectString(), tmpDir, new File[] { tmpDir });
bks.add(startBookie(conf));
bkConfs.add(conf);
}
@@ -509,7 +539,7 @@ public class TestBackwardCompat extends
org.apache.hedwig.server.common.ServerConfiguration conf;
org.apache.hedwig.server.netty.PubSubServer server;
- ServerCurrent(final String zkHosts) {
+ ServerCurrent(final String zkHosts, final int port, final int sslPort) {
conf = new org.apache.hedwig.server.common.ServerConfiguration() {
@Override
public int getConsumeInterval() {
@@ -520,6 +550,16 @@ public class TestBackwardCompat extends
public String getZkHost() {
return zkHosts;
}
+
+ @Override
+ public int getServerPort() {
+ return port;
+ }
+
+ @Override
+ public int getSSLServerPort() {
+ return sslPort;
+ }
};
}
@@ -576,11 +616,11 @@ public class TestBackwardCompat extends
}
}
- ClientCurrent() {
- this(true);
+ ClientCurrent(final String connectString) {
+ this(true, connectString);
}
- ClientCurrent(final boolean autoConsumeEnabled) {
+ ClientCurrent(final boolean autoConsumeEnabled, final String connectString) {
conf = new org.apache.hedwig.client.conf.ClientConfiguration() {
@Override
public boolean isAutoSendConsumeMessageEnabled() {
@@ -590,6 +630,10 @@ public class TestBackwardCompat extends
public int getConsumedMessagesBufferSize() {
return 1;
}
+ @Override
+ protected HedwigSocketAddress getDefaultServerHedwigSocketAddress() {
+ return new HedwigSocketAddress(connectString);
+ }
};
client = new org.apache.hedwig.client.HedwigClient(conf);
publisher = client.getPublisher();
@@ -833,12 +877,15 @@ public class TestBackwardCompat extends
ByteString topic = ByteString.copyFromUtf8("testMessageBoundCompat");
ByteString subid = ByteString.copyFromUtf8("mysub");
+ int port = PortManager.nextFreePort();
+ int sslPort = PortManager.nextFreePort();
+
// start bookkeeper 400
BookKeeperCluster400 bkc400 = new BookKeeperCluster400(3);
bkc400.start();
// start 400 server
- Server400 s400 = new Server400(zkUtil.getZooKeeperConnectString());
+ Server400 s400 = new Server400(zkUtil.getZooKeeperConnectString(), port, sslPort);
s400.start();
org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions options5cur =
@@ -846,7 +893,7 @@ public class TestBackwardCompat extends
.setCreateOrAttach(org.apache.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.CREATE_OR_ATTACH)
.setMessageBound(5).build();
- ClientCurrent ccur = new ClientCurrent();
+ ClientCurrent ccur = new ClientCurrent("localhost:" + port + ":" + sslPort);
ccur.subscribe(topic, subid, options5cur);
ccur.closeSubscription(topic, subid);
ccur.sendXExpectLastY(topic, subid, 50, 50);
@@ -860,7 +907,7 @@ public class TestBackwardCompat extends
bkc410.start();
// start 410 server
- Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString());
+ Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString(), port, sslPort);
s410.start();
ccur.subscribe(topic, subid, options5cur);
@@ -876,7 +923,7 @@ public class TestBackwardCompat extends
bkccur.start();
// start current server
- ServerCurrent scur = new ServerCurrent(zkUtil.getZooKeeperConnectString());
+ ServerCurrent scur = new ServerCurrent(zkUtil.getZooKeeperConnectString(), port, sslPort);
scur.start();
ccur.subscribe(topic, subid, options5cur);
@@ -907,12 +954,15 @@ public class TestBackwardCompat extends
BookKeeperCluster410 bkc410 = new BookKeeperCluster410(3);
bkc410.start();
+ int port = PortManager.nextFreePort();
+ int sslPort = PortManager.nextFreePort();
+
// start 410 server
- Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString());
+ Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString(), port, sslPort);
s410.start();
- ClientCurrent ccur = new ClientCurrent();
- Client410 c410 = new Client410();
+ ClientCurrent ccur = new ClientCurrent("localhost:"+port+":"+sslPort);
+ Client410 c410 = new Client410("localhost:"+port+":"+sslPort);
// client c410 could publish message to 410 server
assertNull(c410.publish(topic, data));
@@ -924,7 +974,7 @@ public class TestBackwardCompat extends
s410.stop();
// start current server
- ServerCurrent scur = new ServerCurrent(zkUtil.getZooKeeperConnectString());
+ ServerCurrent scur = new ServerCurrent(zkUtil.getZooKeeperConnectString(), port, sslPort);
scur.start();
// client c410 could publish message to 410 server
@@ -956,15 +1006,18 @@ public class TestBackwardCompat extends
BookKeeperCluster410 bkc410 = new BookKeeperCluster410(3);
bkc410.start();
+ int port = PortManager.nextFreePort();
+ int sslPort = PortManager.nextFreePort();
+
// start 410 server
- Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString());
+ Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString(), port, sslPort);
s410.start();
- Client410 c410 = new Client410();
+ Client410 c410 = new Client410("localhost:"+port+":"+sslPort);
c410.subscribe(topic, sub410);
c410.closeSubscription(topic, sub410);
- ClientCurrent ccur = new ClientCurrent();
+ ClientCurrent ccur = new ClientCurrent("localhost:"+port+":"+sslPort);
ccur.subscribe(topic, subcur);
ccur.closeSubscription(topic, subcur);
@@ -974,7 +1027,8 @@ public class TestBackwardCompat extends
s410.stop();
// start current server
- ServerCurrent scur = new ServerCurrent(zkUtil.getZooKeeperConnectString());
+ ServerCurrent scur = new ServerCurrent(zkUtil.getZooKeeperConnectString(),
+ port, sslPort);
scur.start();
c410.subscribe(topic, sub410);
@@ -1013,8 +1067,12 @@ public class TestBackwardCompat extends
BookKeeperClusterCurrent bkccur= new BookKeeperClusterCurrent(3);
bkccur.start();
+ int port = PortManager.nextFreePort();
+ int sslPort = PortManager.nextFreePort();
+
// start hub server
- ServerCurrent scur = new ServerCurrent(zkUtil.getZooKeeperConnectString());
+ ServerCurrent scur = new ServerCurrent(zkUtil.getZooKeeperConnectString(),
+ port, sslPort);
scur.start();
org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions options5cur =
@@ -1030,7 +1088,7 @@ public class TestBackwardCompat extends
.setCreateOrAttach(org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.CREATE_OR_ATTACH)
.setMessageBound(20).build();
- Client410 c410 = new Client410();
+ Client410 c410 = new Client410("localhost:"+port+":"+sslPort);
c410.subscribe(topic, subid, options20v410);
c410.closeSubscription(topic, subid);
c410.sendXExpectLastY(topic, subid, 50, 20);
@@ -1040,7 +1098,7 @@ public class TestBackwardCompat extends
// the message bound isn't updated.
c410.sendXExpectLastY(topic, subid, 50, 20);
- ClientCurrent ccur = new ClientCurrent();
+ ClientCurrent ccur = new ClientCurrent("localhost:"+port+":"+sslPort);
ccur.subscribe(topic, subid, options5cur);
ccur.closeSubscription(topic, subid);
// the message bound should be updated.
@@ -1070,11 +1128,14 @@ public class TestBackwardCompat extends
BookKeeperCluster410 bkc410 = new BookKeeperCluster410(3);
bkc410.start();
+ int port = PortManager.nextFreePort();
+ int sslPort = PortManager.nextFreePort();
+
// start hub server 410
- Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString());
+ Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString(), port, sslPort);
s410.start();
- ClientCurrent ccur = new ClientCurrent();
+ ClientCurrent ccur = new ClientCurrent("localhost:"+port+":"+sslPort);
ccur.subscribe(topic, subid);
ccur.closeSubscription(topic, subid);
@@ -1108,11 +1169,14 @@ public class TestBackwardCompat extends
BookKeeperCluster410 bkc410 = new BookKeeperCluster410(3);
bkc410.start();
+ int port = PortManager.nextFreePort();
+ int sslPort = PortManager.nextFreePort();
+
// start hub server 410
- Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString());
+ Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString(), port, sslPort);
s410.start();
- ClientCurrent ccur = new ClientCurrent(false);
+ ClientCurrent ccur = new ClientCurrent(false, "localhost:"+port+":"+sslPort);
ccur.throttleX41(topic, subid, 10);
ccur.close();
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/TestPubSubServerStartup.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/TestPubSubServerStartup.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/TestPubSubServerStartup.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/TestPubSubServerStartup.java Fri Nov 23 10:49:51 2012
@@ -25,6 +25,7 @@ import java.net.MalformedURLException;
import junit.framework.Assert;
+import org.apache.bookkeeper.test.PortManager;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.hedwig.server.common.ServerConfiguration;
import org.apache.hedwig.server.netty.PubSubServer;
@@ -62,8 +63,14 @@ public class TestPubSubServerStartup {
private void instantiateAndDestroyPubSubServer() throws IOException, InterruptedException, ConfigurationException,
MalformedURLException, Exception {
- String hedwigParams = "default_server_host=localhost:4080\n" + "zookeeper_connection_string=localhost:2181\n"
- + "zk_timeout=2000\n";
+ int zkPort = PortManager.nextFreePort();
+ int hwPort = PortManager.nextFreePort();
+ int hwSSLPort = PortManager.nextFreePort();
+ String hedwigParams = "default_server_host=localhost:" + hwPort + "\n"
+ + "zk_host=localhost:" + zkPort + "\n"
+ + "server_port=" + hwPort + "\n"
+ + "ssl_server_port=" + hwSSLPort + "\n"
+ + "zk_timeout=2000\n";
File hedwigConfigFile = new File(System.getProperty("java.io.tmpdir") + "/hedwig.cfg");
writeStringToFile(hedwigParams, hedwigConfigFile);
@@ -73,19 +80,19 @@ public class TestPubSubServerStartup {
zkTmpDir.delete();
zkTmpDir.mkdir();
- ZooKeeperServer zks = new ZooKeeperServer(zkTmpDir, zkTmpDir, 2181);
+ ZooKeeperServer zks = new ZooKeeperServer(zkTmpDir, zkTmpDir, zkPort);
NIOServerCnxnFactory serverFactory = new NIOServerCnxnFactory();
- serverFactory.configure(new InetSocketAddress(2181), 100);
+ serverFactory.configure(new InetSocketAddress(zkPort), 100);
serverFactory.startup(zks);
- boolean b = ClientBase.waitForServerUp("127.0.0.1:2181", 5000);
+ boolean b = ClientBase.waitForServerUp("127.0.0.1:" + zkPort, 5000);
ServerConfiguration serverConf = new ServerConfiguration();
serverConf.loadConf(hedwigConfigFile.toURI().toURL());
logger.info("Zookeeper server up and running!");
- ZooKeeper zkc = new ZooKeeper("127.0.0.1", 2181, null);
+ ZooKeeper zkc = new ZooKeeper("127.0.0.1:" + zkPort, 5000, null);
// initialize the zk client with (fake) values
zkc.create("/ledgers", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
@@ -109,7 +116,7 @@ public class TestPubSubServerStartup {
zkTmpDir.delete();
- ClientBase.waitForServerDown("localhost:2181", 10000);
+ ClientBase.waitForServerDown("localhost:" + zkPort, 10000);
}
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/TestThrottlingDelivery.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/TestThrottlingDelivery.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/TestThrottlingDelivery.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/TestThrottlingDelivery.java Fri Nov 23 10:49:51 2012
@@ -66,7 +66,7 @@ public class TestThrottlingDelivery exte
}
}
- protected class ThrottleDeliveryClientConfiguration extends ClientConfiguration {
+ protected class ThrottleDeliveryClientConfiguration extends HubClientConfiguration {
int messageWindowSize;
@@ -173,13 +173,13 @@ public class TestThrottlingDelivery exte
protected boolean isSubscriptionChannelSharingEnabled;
public TestThrottlingDelivery(boolean isSubscriptionChannelSharingEnabled) {
+ super(1);
this.isSubscriptionChannelSharingEnabled = isSubscriptionChannelSharingEnabled;
}
@Override
@Before
public void setUp() throws Exception {
- numServers = 1;
super.setUp();
}
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/filter/TestMessageFilter.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/filter/TestMessageFilter.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/filter/TestMessageFilter.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/filter/TestMessageFilter.java Fri Nov 23 10:49:51 2012
@@ -146,13 +146,16 @@ public class TestMessageFilter extends H
}
}
+ public TestMessageFilter() {
+ super(1);
+ }
+
@Override
@Before
public void setUp() throws Exception {
- numServers = 1;
super.setUp();
- conf = new ClientConfiguration() {
+ conf = new HubClientConfiguration() {
@Override
public boolean isAutoSendConsumeMessageEnabled() {
return false;
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigHub.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigHub.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigHub.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigHub.java Fri Nov 23 10:49:51 2012
@@ -60,6 +60,8 @@ import org.apache.hedwig.server.proxy.Pr
import org.apache.hedwig.server.regions.HedwigHubClient;
import org.apache.hedwig.util.Callback;
import org.apache.hedwig.util.ConcurrencyUtils;
+import org.apache.hedwig.util.HedwigSocketAddress;
+import org.apache.bookkeeper.test.PortManager;
@RunWith(Parameterized.class)
public class TestHedwigHub extends HedwigHubTestBase {
@@ -94,12 +96,25 @@ public class TestHedwigHub extends Hedwi
protected boolean isSubscriptionChannelSharingEnabled;
public TestHedwigHub(Mode mode, boolean isSubscriptionChannelSharingEnabled) {
+ super(3);
this.mode = mode;
this.isSubscriptionChannelSharingEnabled = isSubscriptionChannelSharingEnabled;
}
protected HedwigProxy proxy;
- protected ProxyConfiguration proxyConf = new ProxyConfiguration();
+ protected ProxyConfiguration proxyConf = new ProxyConfiguration() {
+ final int proxyPort = PortManager.nextFreePort();
+
+ @Override
+ public HedwigSocketAddress getDefaultServerHedwigSocketAddress() {
+ return serverAddresses.get(0);
+ }
+
+ @Override
+ public int getProxyPort() {
+ return proxyPort;
+ }
+ };
// SynchronousQueues to verify async calls
private final SynchronousQueue<Boolean> queue = new SynchronousQueue<Boolean>();
@@ -190,7 +205,7 @@ public class TestHedwigHub extends Hedwi
}
}
- class TestClientConfiguration extends ClientConfiguration {
+ class TestClientConfiguration extends HubClientConfiguration {
@Override
public InetSocketAddress getDefaultServerHost() {
@@ -223,7 +238,6 @@ public class TestHedwigHub extends Hedwi
@Override
@Before
public void setUp() throws Exception {
- numServers = 3;
super.setUp();
if (mode == Mode.PROXY) {
proxy = new HedwigProxy(proxyConf);
@@ -676,7 +690,7 @@ public class TestHedwigHub extends Hedwi
// subscriberId to be in the "hub" specific format.
@Test
public void testSyncHubSubscribeWithInvalidSubscriberId() throws Exception {
- Client hubClient = new HedwigHubClient(new ClientConfiguration());
+ Client hubClient = new HedwigHubClient(new HubClientConfiguration());
Subscriber hubSubscriber = hubClient.getSubscriber();
boolean subscribeSuccess = false;
try {
@@ -692,7 +706,7 @@ public class TestHedwigHub extends Hedwi
@Test
public void testAsyncHubSubscribeWithInvalidSubscriberId() throws Exception {
- Client hubClient = new HedwigHubClient(new ClientConfiguration());
+ Client hubClient = new HedwigHubClient(new HubClientConfiguration());
Subscriber hubSubscriber = hubClient.getSubscriber();
hubSubscriber.asyncSubscribe(getTopic(0), localSubscriberId, CreateOrAttach.CREATE_OR_ATTACH, new TestCallback(
queue), null);
@@ -702,7 +716,7 @@ public class TestHedwigHub extends Hedwi
@Test
public void testSyncHubUnsubscribeWithInvalidSubscriberId() throws Exception {
- Client hubClient = new HedwigHubClient(new ClientConfiguration());
+ Client hubClient = new HedwigHubClient(new HubClientConfiguration());
Subscriber hubSubscriber = hubClient.getSubscriber();
boolean unsubscribeSuccess = false;
try {
@@ -718,7 +732,7 @@ public class TestHedwigHub extends Hedwi
@Test
public void testAsyncHubUnsubscribeWithInvalidSubscriberId() throws Exception {
- Client hubClient = new HedwigHubClient(new ClientConfiguration());
+ Client hubClient = new HedwigHubClient(new HubClientConfiguration());
Subscriber hubSubscriber = hubClient.getSubscriber();
hubSubscriber.asyncUnsubscribe(getTopic(0), localSubscriberId, new TestCallback(queue), null);
assertFalse(queue.take());
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigRegion.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigRegion.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigRegion.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigRegion.java Fri Nov 23 10:49:51 2012
@@ -40,6 +40,7 @@ import org.apache.hedwig.server.HedwigRe
import org.apache.hedwig.server.common.ServerConfiguration;
import org.apache.hedwig.server.integration.TestHedwigHub.TestCallback;
import org.apache.hedwig.server.integration.TestHedwigHub.TestMessageHandler;
+import org.apache.hedwig.util.HedwigSocketAddress;
@RunWith(Parameterized.class)
public class TestHedwigRegion extends HedwigRegionTestBase {
@@ -69,6 +70,10 @@ public class TestHedwigRegion extends He
public boolean isSubscriptionChannelSharingEnabled() {
return isSubscriptionChannelSharingEnabled;
}
+ @Override
+ public HedwigSocketAddress getDefaultServerHedwigSocketAddress() {
+ return regionHubAddresses.get(0).get(0);
+ }
}
protected ServerConfiguration getServerConfiguration(int serverPort, int sslServerPort, String regionName) {
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/netty/TestPubSubServer.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/netty/TestPubSubServer.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/netty/TestPubSubServer.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/netty/TestPubSubServer.java Fri Nov 23 10:49:51 2012
@@ -31,6 +31,7 @@ import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.junit.Test;
+import org.apache.bookkeeper.test.PortManager;
import com.google.protobuf.ByteString;
import org.apache.hedwig.client.conf.ClientConfiguration;
import org.apache.hedwig.client.HedwigClient;
@@ -125,7 +126,7 @@ public class TestPubSubServer extends Pu
SynchronousQueue<Throwable> queue = new SynchronousQueue<Throwable>();
RecordingUncaughtExceptionHandler uncaughtExceptionHandler = new RecordingUncaughtExceptionHandler(queue);
- final int port = 9876;
+ final int port = PortManager.nextFreePort();
PubSubServer server = startServer(uncaughtExceptionHandler, port, new TopicManagerInstantiator() {
@@ -155,8 +156,8 @@ public class TestPubSubServer extends Pu
SynchronousQueue<Throwable> queue = new SynchronousQueue<Throwable>();
RecordingUncaughtExceptionHandler uncaughtExceptionHandler = new RecordingUncaughtExceptionHandler(queue);
- final int port = 9876;
- final String hostPort = "127.0.0.1:33221";
+ final int port = PortManager.nextFreePort();
+ final String hostPort = "127.0.0.1:" + PortManager.nextFreePort();
PubSubServer server = startServer(uncaughtExceptionHandler, port, new TopicManagerInstantiator() {
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/BookKeeperTestBase.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/BookKeeperTestBase.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/BookKeeperTestBase.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/BookKeeperTestBase.java Fri Nov 23 10:49:51 2012
@@ -25,6 +25,8 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Random;
+import org.apache.bookkeeper.test.PortManager;
+
import org.apache.bookkeeper.bookie.Bookie;
import org.apache.bookkeeper.bookie.BookieException;
import org.apache.bookkeeper.conf.ClientConfiguration;
@@ -89,8 +91,6 @@ public class BookKeeperTestBase extends
// BookKeeper Server variables
private List<BookieServer> bookiesList;
private List<ServerConfiguration> bkConfsList;
- private int initialPort = 5000;
- private int nextPort = initialPort;
// String constants used for creating the bookie server files.
private static final String PREFIX = "bookie";
@@ -198,10 +198,11 @@ public class BookKeeperTestBase extends
}
public void startUpNewBookieServer() throws Exception {
+ int port = PortManager.nextFreePort();
File tmpDir = FileUtils.createTempDirectory(
- PREFIX + (nextPort - initialPort), SUFFIX);
+ PREFIX + port, SUFFIX);
ServerConfiguration conf = newServerConfiguration(
- nextPort++, hostPort, tmpDir, new File[] { tmpDir });
+ port, hostPort, tmpDir, new File[] { tmpDir });
bookiesList.add(startBookie(conf));
bkConfsList.add(conf);
}
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/MessageBoundedPersistenceTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/MessageBoundedPersistenceTest.java?rev=1412818&r1=1412817&r2=1412818&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/MessageBoundedPersistenceTest.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/MessageBoundedPersistenceTest.java Fri Nov 23 10:49:51 2012
@@ -71,7 +71,7 @@ public class MessageBoundedPersistenceTe
return new SmallReadAheadServerConfiguration(serverPort, sslServerPort);
}
- private class MessageBoundClientConfiguration extends ClientConfiguration {
+ private class MessageBoundClientConfiguration extends HubClientConfiguration {
final int messageBound;
public MessageBoundClientConfiguration(int bound) {
@@ -152,7 +152,7 @@ public class MessageBoundedPersistenceTe
public void testMultipleSubscribers() throws Exception {
ByteString topic = ByteString.copyFromUtf8("multiSubTopic");
- Client client = new HedwigClient(new ClientConfiguration());
+ Client client = new HedwigClient(new HubClientConfiguration());
Publisher pub = client.getPublisher();
Subscriber sub = client.getSubscriber();
@@ -194,7 +194,7 @@ public class MessageBoundedPersistenceTe
public void testUpdateMessageBound() throws Exception {
ByteString topic = ByteString.copyFromUtf8("UpdateMessageBound");
- Client client = new HedwigClient(new ClientConfiguration());
+ Client client = new HedwigClient(new HubClientConfiguration());
Publisher pub = client.getPublisher();
Subscriber sub = client.getSubscriber();