You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bookkeeper.apache.org by si...@apache.org on 2018/08/21 21:08:22 UTC
[bookkeeper] branch master updated: Issue #1610: netty leak
detector logs 'LEAK: HashedWheelTimer.release() was not called before it's
garbage-collected.' in unit tests
This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
The following commit(s) were added to refs/heads/master by this push:
new 5f2d45c Issue #1610: netty leak detector logs 'LEAK: HashedWheelTimer.release() was not called before it's garbage-collected.' in unit tests
5f2d45c is described below
commit 5f2d45c8fa0619d50423f1672303fea2d1a59238
Author: Andrey Yegorov <ay...@salesforce.com>
AuthorDate: Tue Aug 21 14:08:14 2018 -0700
Issue #1610: netty leak detector logs 'LEAK: HashedWheelTimer.release() was not called before it's garbage-collected.' in unit tests
Descriptions of the changes in this PR:
Enabled netty's paranoid leak detection for tests.
Fixed most of the 'LEAK: HashedWheelTimer.release() was not called before it's garbage-collected.' issues. Did not fix ones produced by AuditorElector. In long term we should fix these, ByteBuf leaks in BufferedChannel and whatever else we find and make tests fail if netty logs a leak.
### Motivation
I used to have random test timeouts/failures caused by delayed bookie shutdown when I built repo with OpenJDK 8 but ran tests after switching to Oracle JDK 8 (or vice versa). After these changes tests did not fail in these runs.
### Changes
Made bk/bkadmin/auditor close resources properly in the tests and shutdown calls.
Changed order of initialization in Bookkeeper so it creates HashedWheelTimer after bookie client, in case bookie client creation fails due to misconfiguration.
Master Issue: #1610
Author: Andrey Yegorov <ay...@salesforce.com>
Reviewers: Enrico Olivelli <eo...@gmail.com>, Sijie Guo <si...@apache.org>
This closes #1613 from dlg99/fix/issue_1610_hashedwheeltimer, closes #1610
---
.../org/apache/bookkeeper/client/BookKeeper.java | 8 +-
.../bookkeeper/proto/BookieRequestProcessor.java | 10 +-
.../org/apache/bookkeeper/proto/BookieServer.java | 1 +
.../org/apache/bookkeeper/replication/Auditor.java | 7 +-
.../java/org/apache/bookkeeper/auth/TestAuth.java | 37 +--
.../bookkeeper/client/TestBookieWatcher.java | 57 ++--
.../bookkeeper/client/UpdateLedgerOpTest.java | 316 +++++++++++----------
.../replication/AuditorPeriodicCheckTest.java | 24 +-
pom.xml | 2 +-
9 files changed, 240 insertions(+), 222 deletions(-)
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
index 4fc7728..6544e44 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
@@ -440,6 +440,10 @@ public class BookKeeper implements org.apache.bookkeeper.client.api.BookKeeper {
this.ownEventLoopGroup = false;
}
+ // initialize bookie client
+ this.bookieClient = new BookieClientImpl(conf, this.eventLoopGroup, this.mainWorkerPool,
+ scheduler, rootStatsLogger);
+
if (null == requestTimer) {
this.requestTimer = new HashedWheelTimer(
new ThreadFactoryBuilder().setNameFormat("BookieClientTimer-%d").build(),
@@ -455,9 +459,7 @@ public class BookKeeper implements org.apache.bookkeeper.client.api.BookKeeper {
this.placementPolicy = initializeEnsemblePlacementPolicy(conf,
dnsResolver, this.requestTimer, this.featureProvider, this.statsLogger);
- // initialize bookie client
- this.bookieClient = new BookieClientImpl(conf, this.eventLoopGroup, this.mainWorkerPool,
- scheduler, rootStatsLogger);
+
this.bookieWatcher = new BookieWatcherImpl(
conf, this.placementPolicy, metadataDriver.getRegistrationClient(),
this.statsLogger.scope(WATCHER_SCOPE));
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieRequestProcessor.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieRequestProcessor.java
index aa9e048..8ee363f 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieRequestProcessor.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieRequestProcessor.java
@@ -214,15 +214,16 @@ public class BookieRequestProcessor implements RequestProcessor {
this.serverCfg.getNumHighPriorityWorkerThreads(),
"BookieHighPriorityThread-" + serverCfg.getBookiePort(),
OrderedExecutor.NO_TASK_LIMIT, statsLogger);
- this.requestTimer = new HashedWheelTimer(
- new ThreadFactoryBuilder().setNameFormat("BookieRequestTimer-%d").build(),
- this.serverCfg.getRequestTimerTickDurationMs(),
- TimeUnit.MILLISECONDS, this.serverCfg.getRequestTimerNumTicks());
this.shFactory = shFactory;
if (shFactory != null) {
shFactory.init(NodeType.Server, serverCfg);
}
+ this.requestTimer = new HashedWheelTimer(
+ new ThreadFactoryBuilder().setNameFormat("BookieRequestTimer-%d").build(),
+ this.serverCfg.getRequestTimerTickDurationMs(),
+ TimeUnit.MILLISECONDS, this.serverCfg.getRequestTimerNumTicks());
+
if (waitTimeoutOnBackpressureMillis > 0) {
blacklistedChannels = Optional.of(CacheBuilder.newBuilder()
.expireAfterWrite(waitTimeoutOnBackpressureMillis, TimeUnit.MILLISECONDS)
@@ -396,6 +397,7 @@ public class BookieRequestProcessor implements RequestProcessor {
shutdownExecutor(longPollThreadPool);
}
shutdownExecutor(highPriorityThreadPool);
+ requestTimer.stop();
}
private OrderedExecutor createExecutor(
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieServer.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieServer.java
index 559873e..27e8ab2 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieServer.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieServer.java
@@ -138,6 +138,7 @@ public class BookieServer {
// fail fast, when bookie startup is not successful
if (!this.bookie.isRunning()) {
exitCode = bookie.getExitCode();
+ this.requestProcessor.close();
return;
}
this.nettyServer.start();
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java
index 8578a5b..920ac57 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java
@@ -77,7 +77,7 @@ import org.slf4j.LoggerFactory;
*
* <p>TODO: eliminate the direct usage of zookeeper here {@link https://github.com/apache/bookkeeper/issues/1332}
*/
-public class Auditor {
+public class Auditor implements AutoCloseable {
private static final Logger LOG = LoggerFactory.getLogger(Auditor.class);
private final ServerConfiguration conf;
private BookKeeper bkc;
@@ -735,6 +735,11 @@ public class Auditor {
}
}
+ @Override
+ public void close() {
+ shutdown();
+ }
+
/**
* Return true if auditor is running otherwise return false.
*
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/auth/TestAuth.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/auth/TestAuth.java
index 5d86036..cca97e8 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/auth/TestAuth.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/auth/TestAuth.java
@@ -69,13 +69,13 @@ public class TestAuth extends BookKeeperClusterTestCase {
private void connectAndWriteToBookie(ClientConfiguration conf, AtomicLong ledgerWritten)
throws Exception {
LOG.info("Connecting to bookie");
- BookKeeper bkc = new BookKeeper(conf, zkc);
- LedgerHandle l = bkc.createLedger(1, 1, DigestType.CRC32,
- PASSWD);
- ledgerWritten.set(l.getId());
- l.addEntry(ENTRY);
- l.close();
- bkc.close();
+ try (BookKeeper bkc = new BookKeeper(conf, zkc);
+ LedgerHandle l = bkc.createLedger(1, 1, DigestType.CRC32,
+ PASSWD)) {
+ ledgerWritten.set(l.getId());
+ l.addEntry(ENTRY);
+ l.close();
+ }
}
/**
@@ -94,18 +94,19 @@ public class TestAuth extends BookKeeperClusterTestCase {
restartBookies();
- BookKeeper bkc = new BookKeeper(clientConf, zkc);
- LedgerHandle lh = bkc.openLedger(ledgerId, DigestType.CRC32,
- PASSWD);
- if (lh.getLastAddConfirmed() < 0) {
- return 0;
- }
- Enumeration<LedgerEntry> e = lh.readEntries(0, lh.getLastAddConfirmed());
int count = 0;
- while (e.hasMoreElements()) {
- count++;
- assertTrue("Should match what we wrote",
- Arrays.equals(e.nextElement().getEntry(), ENTRY));
+ try (BookKeeper bkc = new BookKeeper(clientConf, zkc);
+ LedgerHandle lh = bkc.openLedger(ledgerId, DigestType.CRC32,
+ PASSWD)) {
+ if (lh.getLastAddConfirmed() < 0) {
+ return 0;
+ }
+ Enumeration<LedgerEntry> e = lh.readEntries(0, lh.getLastAddConfirmed());
+ while (e.hasMoreElements()) {
+ count++;
+ assertTrue("Should match what we wrote",
+ Arrays.equals(e.nextElement().getEntry(), ENTRY));
+ }
}
return count;
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestBookieWatcher.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestBookieWatcher.java
index 63de59c..69fe345 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestBookieWatcher.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestBookieWatcher.java
@@ -104,38 +104,39 @@ public class TestBookieWatcher extends BookKeeperClusterTestCase {
ClientConfiguration conf = new ClientConfiguration();
conf.setMetadataServiceUri(metadataServiceUri);
- BookKeeper bkc = new BookKeeper(conf, zk);
-
- LedgerHandle lh;
- try {
- lh = bkc.createLedger(3, 2, 2, BookKeeper.DigestType.CRC32, new byte[] {});
- fail("Should fail to create ledger due to not enough bookies.");
- } catch (BKException bke) {
- // expected
- }
+ try (BookKeeper bkc = new BookKeeper(conf, zk)) {
+
+ LedgerHandle lh;
+ try {
+ lh = bkc.createLedger(3, 2, 2, BookKeeper.DigestType.CRC32, new byte[]{});
+ fail("Should fail to create ledger due to not enough bookies.");
+ } catch (BKException bke) {
+ // expected
+ }
- // make zookeeper session expired
- expireZooKeeperSession(bkc.getZkHandle(), timeout);
- TimeUnit.MILLISECONDS.sleep(3 * timeout);
+ // make zookeeper session expired
+ expireZooKeeperSession(bkc.getZkHandle(), timeout);
+ TimeUnit.MILLISECONDS.sleep(3 * timeout);
- // start four new bookies
- for (int i = 0; i < 2; i++) {
- startNewBookie();
- }
+ // start four new bookies
+ for (int i = 0; i < 2; i++) {
+ startNewBookie();
+ }
- // wait for bookie watcher backoff time.
- TimeUnit.SECONDS.sleep(1);
+ // wait for bookie watcher backoff time.
+ TimeUnit.SECONDS.sleep(1);
- // should success to detect newly added bookies
- try {
- lh = bkc.createLedger(3, 2, 2, BookKeeper.DigestType.CRC32, new byte[] {});
- lh.close();
- if (!reconnectable) {
- fail("Should fail to create ledger due to bookie watcher could not survive after session expire.");
- }
- } catch (BKException bke) {
- if (reconnectable) {
- fail("Should not fail to create ledger due to bookie watcher could survive after session expire.");
+ // should success to detect newly added bookies
+ try {
+ lh = bkc.createLedger(3, 2, 2, BookKeeper.DigestType.CRC32, new byte[]{});
+ lh.close();
+ if (!reconnectable) {
+ fail("Should fail to create ledger due to bookie watcher could not survive after session expire.");
+ }
+ } catch (BKException bke) {
+ if (reconnectable) {
+ fail("Should not fail to create ledger due to bookie watcher could survive after session expire.");
+ }
}
}
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/UpdateLedgerOpTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/UpdateLedgerOpTest.java
index 7f67242..91628e7 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/UpdateLedgerOpTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/UpdateLedgerOpTest.java
@@ -87,39 +87,40 @@ public class UpdateLedgerOpTest extends BookKeeperClusterTestCase {
}
public void testManyLedgers(boolean useShortHostName) throws Exception {
- BookKeeper bk = new BookKeeper(baseClientConf, zkc);
- BookKeeperAdmin bkadmin = new BookKeeperAdmin(bk);
-
- LOG.info("Create ledger and add entries to it");
- List<LedgerHandle> ledgers = new ArrayList<LedgerHandle>();
- LedgerHandle lh1 = createLedgerWithEntries(bk, 0);
- ledgers.add(lh1);
- for (int i = 0; i < 99; i++) {
- ledgers.add(createLedgerWithEntries(bk, 0));
- }
-
- List<BookieSocketAddress> ensemble = lh1.getLedgerMetadata().getEnsemble(0);
-
- BookieSocketAddress curBookieAddr = ensemble.get(0);
- baseConf.setUseHostNameAsBookieID(true);
- baseConf.setUseShortHostName(useShortHostName);
- BookieSocketAddress curBookieId = Bookie.getBookieAddress(baseConf);
- BookieSocketAddress toBookieAddr = new BookieSocketAddress(curBookieId.getHostName() + ":"
- + curBookieAddr.getPort());
- UpdateLedgerOp updateLedgerOp = new UpdateLedgerOp(bk, bkadmin);
- updateLedgerOp.updateBookieIdInLedgers(curBookieAddr, toBookieAddr, 5, Integer.MIN_VALUE, progressable);
+ try (BookKeeper bk = new BookKeeper(baseClientConf, zkc);
+ BookKeeperAdmin bkadmin = new BookKeeperAdmin(bk)) {
+
+ LOG.info("Create ledger and add entries to it");
+ List<LedgerHandle> ledgers = new ArrayList<LedgerHandle>();
+ LedgerHandle lh1 = createLedgerWithEntries(bk, 0);
+ ledgers.add(lh1);
+ for (int i = 0; i < 99; i++) {
+ ledgers.add(createLedgerWithEntries(bk, 0));
+ }
- for (LedgerHandle lh : ledgers) {
- // ledger#close() would hit BadVersion exception as rename
- // increments cversion. But LedgerMetadata#isConflictWith()
- // gracefully handles this conflicts.
- lh.close();
- LedgerHandle openLedger = bk.openLedger(lh.getId(), digestType, PASSWORD.getBytes());
- ensemble = openLedger.getLedgerMetadata().getEnsemble(0);
- assertTrue("Failed to update the ledger metadata to use bookie host name",
- ensemble.contains(toBookieAddr));
- assertFalse("Failed to update the ledger metadata to use bookie host name",
- ensemble.contains(curBookieAddr));
+ List<BookieSocketAddress> ensemble = lh1.getLedgerMetadata().getEnsemble(0);
+
+ BookieSocketAddress curBookieAddr = ensemble.get(0);
+ baseConf.setUseHostNameAsBookieID(true);
+ baseConf.setUseShortHostName(useShortHostName);
+ BookieSocketAddress curBookieId = Bookie.getBookieAddress(baseConf);
+ BookieSocketAddress toBookieAddr = new BookieSocketAddress(curBookieId.getHostName() + ":"
+ + curBookieAddr.getPort());
+ UpdateLedgerOp updateLedgerOp = new UpdateLedgerOp(bk, bkadmin);
+ updateLedgerOp.updateBookieIdInLedgers(curBookieAddr, toBookieAddr, 5, Integer.MIN_VALUE, progressable);
+
+ for (LedgerHandle lh : ledgers) {
+ // ledger#close() would hit BadVersion exception as rename
+ // increments cversion. But LedgerMetadata#isConflictWith()
+ // gracefully handles this conflicts.
+ lh.close();
+ LedgerHandle openLedger = bk.openLedger(lh.getId(), digestType, PASSWORD.getBytes());
+ ensemble = openLedger.getLedgerMetadata().getEnsemble(0);
+ assertTrue("Failed to update the ledger metadata to use bookie host name",
+ ensemble.contains(toBookieAddr));
+ assertFalse("Failed to update the ledger metadata to use bookie host name",
+ ensemble.contains(curBookieAddr));
+ }
}
}
@@ -128,43 +129,44 @@ public class UpdateLedgerOpTest extends BookKeeperClusterTestCase {
*/
@Test
public void testLimitLessThanTotalLedgers() throws Exception {
- BookKeeper bk = new BookKeeper(baseClientConf, zkc);
- BookKeeperAdmin bkadmin = new BookKeeperAdmin(bk);
-
- LOG.info("Create ledger and add entries to it");
- List<LedgerHandle> ledgers = new ArrayList<LedgerHandle>();
- LedgerHandle lh1 = createLedgerWithEntries(bk, 0);
- ledgers.add(lh1);
- for (int i = 1; i < 10; i++) {
- ledgers.add(createLedgerWithEntries(bk, 0));
- }
+ try (BookKeeper bk = new BookKeeper(baseClientConf, zkc);
+ BookKeeperAdmin bkadmin = new BookKeeperAdmin(bk)) {
+
+ LOG.info("Create ledger and add entries to it");
+ List<LedgerHandle> ledgers = new ArrayList<LedgerHandle>();
+ LedgerHandle lh1 = createLedgerWithEntries(bk, 0);
+ ledgers.add(lh1);
+ for (int i = 1; i < 10; i++) {
+ ledgers.add(createLedgerWithEntries(bk, 0));
+ }
- List<BookieSocketAddress> ensemble = lh1.getLedgerMetadata().getEnsemble(0);
-
- BookieSocketAddress curBookieAddr = ensemble.get(0);
- baseConf.setUseHostNameAsBookieID(true);
- BookieSocketAddress toBookieId = Bookie.getBookieAddress(baseConf);
- BookieSocketAddress toBookieAddr = new BookieSocketAddress(toBookieId.getHostName() + ":"
- + curBookieAddr.getPort());
- UpdateLedgerOp updateLedgerOp = new UpdateLedgerOp(bk, bkadmin);
- updateLedgerOp.updateBookieIdInLedgers(curBookieAddr, toBookieAddr, 7, 4, progressable);
- int updatedLedgersCount = getUpdatedLedgersCount(bk, ledgers, toBookieAddr);
- assertEquals("Failed to update the ledger metadata to use bookie host name", 4, updatedLedgersCount);
-
- // next execution
- updateLedgerOp.updateBookieIdInLedgers(curBookieAddr, toBookieAddr, 2, 10, progressable);
- updatedLedgersCount = getUpdatedLedgersCount(bk, ledgers, toBookieAddr);
- assertEquals("Failed to update the ledger metadata to use bookie host name", 10, updatedLedgersCount);
-
- // no ledgers
- updateLedgerOp.updateBookieIdInLedgers(curBookieAddr, toBookieAddr, 3, 20, progressable);
- updatedLedgersCount = getUpdatedLedgersCount(bk, ledgers, toBookieAddr);
- assertEquals("Failed to update the ledger metadata to use bookie host name", 10, updatedLedgersCount);
-
- // no ledgers
- updateLedgerOp.updateBookieIdInLedgers(curBookieAddr, toBookieAddr, 3, Integer.MIN_VALUE, progressable);
- updatedLedgersCount = getUpdatedLedgersCount(bk, ledgers, toBookieAddr);
- assertEquals("Failed to update the ledger metadata to use bookie host name", 10, updatedLedgersCount);
+ List<BookieSocketAddress> ensemble = lh1.getLedgerMetadata().getEnsemble(0);
+
+ BookieSocketAddress curBookieAddr = ensemble.get(0);
+ baseConf.setUseHostNameAsBookieID(true);
+ BookieSocketAddress toBookieId = Bookie.getBookieAddress(baseConf);
+ BookieSocketAddress toBookieAddr = new BookieSocketAddress(toBookieId.getHostName() + ":"
+ + curBookieAddr.getPort());
+ UpdateLedgerOp updateLedgerOp = new UpdateLedgerOp(bk, bkadmin);
+ updateLedgerOp.updateBookieIdInLedgers(curBookieAddr, toBookieAddr, 7, 4, progressable);
+ int updatedLedgersCount = getUpdatedLedgersCount(bk, ledgers, toBookieAddr);
+ assertEquals("Failed to update the ledger metadata to use bookie host name", 4, updatedLedgersCount);
+
+ // next execution
+ updateLedgerOp.updateBookieIdInLedgers(curBookieAddr, toBookieAddr, 2, 10, progressable);
+ updatedLedgersCount = getUpdatedLedgersCount(bk, ledgers, toBookieAddr);
+ assertEquals("Failed to update the ledger metadata to use bookie host name", 10, updatedLedgersCount);
+
+ // no ledgers
+ updateLedgerOp.updateBookieIdInLedgers(curBookieAddr, toBookieAddr, 3, 20, progressable);
+ updatedLedgersCount = getUpdatedLedgersCount(bk, ledgers, toBookieAddr);
+ assertEquals("Failed to update the ledger metadata to use bookie host name", 10, updatedLedgersCount);
+
+ // no ledgers
+ updateLedgerOp.updateBookieIdInLedgers(curBookieAddr, toBookieAddr, 3, Integer.MIN_VALUE, progressable);
+ updatedLedgersCount = getUpdatedLedgersCount(bk, ledgers, toBookieAddr);
+ assertEquals("Failed to update the ledger metadata to use bookie host name", 10, updatedLedgersCount);
+ }
}
/**
@@ -187,60 +189,61 @@ public class UpdateLedgerOpTest extends BookKeeperClusterTestCase {
public void testChangeEnsembleAfterRenaming(boolean useShortHostName) throws Exception {
- BookKeeper bk = new BookKeeper(baseClientConf, zkc);
- BookKeeperAdmin bkadmin = new BookKeeperAdmin(bk);
+ try (BookKeeper bk = new BookKeeper(baseClientConf, zkc);
+ BookKeeperAdmin bkadmin = new BookKeeperAdmin(bk)) {
- LOG.info("Create ledger and add entries to it");
- LedgerHandle lh = createLedgerWithEntries(bk, 100);
+ LOG.info("Create ledger and add entries to it");
+ LedgerHandle lh = createLedgerWithEntries(bk, 100);
- BookieServer bookieServer = bs.get(0);
- List<BookieSocketAddress> ensemble = lh.getLedgerMetadata().getEnsemble(0);
- BookieSocketAddress curBookieAddr = null;
- for (BookieSocketAddress bookieSocketAddress : ensemble) {
- if (bookieServer.getLocalAddress().equals(bookieSocketAddress)) {
- curBookieAddr = bookieSocketAddress;
+ BookieServer bookieServer = bs.get(0);
+ List<BookieSocketAddress> ensemble = lh.getLedgerMetadata().getEnsemble(0);
+ BookieSocketAddress curBookieAddr = null;
+ for (BookieSocketAddress bookieSocketAddress : ensemble) {
+ if (bookieServer.getLocalAddress().equals(bookieSocketAddress)) {
+ curBookieAddr = bookieSocketAddress;
+ }
}
- }
- assertNotNull("Couldn't find the bookie in ledger metadata!", curBookieAddr);
- baseConf.setUseHostNameAsBookieID(true);
- baseConf.setUseShortHostName(useShortHostName);
- BookieSocketAddress toBookieId = Bookie.getBookieAddress(baseConf);
- BookieSocketAddress toBookieAddr = new BookieSocketAddress(toBookieId.getHostName() + ":"
- + curBookieAddr.getPort());
- UpdateLedgerOp updateLedgerOp = new UpdateLedgerOp(bk, bkadmin);
- updateLedgerOp.updateBookieIdInLedgers(curBookieAddr, toBookieAddr, 5, 100, progressable);
-
- bookieServer.shutdown();
-
- ServerConfiguration serverConf1 = newServerConfiguration();
- bsConfs.add(serverConf1);
- bs.add(startBookie(serverConf1));
-
- // ledger#asyncAddEntry() would hit BadVersion exception as rename incr
- // cversion. But LedgerMetadata#isConflictWith() gracefully handles
- // this conflicts.
- final CountDownLatch latch = new CountDownLatch(1);
- final AtomicInteger rc = new AtomicInteger(BKException.Code.OK);
- lh.asyncAddEntry("foobar".getBytes(), new AddCallback() {
- @Override
- public void addComplete(int rccb, LedgerHandle lh, long entryId, Object ctx) {
- rc.compareAndSet(BKException.Code.OK, rccb);
- latch.countDown();
+ assertNotNull("Couldn't find the bookie in ledger metadata!", curBookieAddr);
+ baseConf.setUseHostNameAsBookieID(true);
+ baseConf.setUseShortHostName(useShortHostName);
+ BookieSocketAddress toBookieId = Bookie.getBookieAddress(baseConf);
+ BookieSocketAddress toBookieAddr = new BookieSocketAddress(toBookieId.getHostName() + ":"
+ + curBookieAddr.getPort());
+ UpdateLedgerOp updateLedgerOp = new UpdateLedgerOp(bk, bkadmin);
+ updateLedgerOp.updateBookieIdInLedgers(curBookieAddr, toBookieAddr, 5, 100, progressable);
+
+ bookieServer.shutdown();
+
+ ServerConfiguration serverConf1 = newServerConfiguration();
+ bsConfs.add(serverConf1);
+ bs.add(startBookie(serverConf1));
+
+ // ledger#asyncAddEntry() would hit BadVersion exception as rename incr
+ // cversion. But LedgerMetadata#isConflictWith() gracefully handles
+ // this conflicts.
+ final CountDownLatch latch = new CountDownLatch(1);
+ final AtomicInteger rc = new AtomicInteger(BKException.Code.OK);
+ lh.asyncAddEntry("foobar".getBytes(), new AddCallback() {
+ @Override
+ public void addComplete(int rccb, LedgerHandle lh, long entryId, Object ctx) {
+ rc.compareAndSet(BKException.Code.OK, rccb);
+ latch.countDown();
+ }
+ }, null);
+ if (!latch.await(30, TimeUnit.SECONDS)) {
+ throw new Exception("Entries took too long to add");
}
- }, null);
- if (!latch.await(30, TimeUnit.SECONDS)) {
- throw new Exception("Entries took too long to add");
- }
- if (rc.get() != BKException.Code.OK) {
- throw BKException.create(rc.get());
+ if (rc.get() != BKException.Code.OK) {
+ throw BKException.create(rc.get());
+ }
+ lh.close();
+ LedgerHandle openLedger = bk.openLedger(lh.getId(), digestType, PASSWORD.getBytes());
+ final LedgerMetadata ledgerMetadata = openLedger.getLedgerMetadata();
+ assertEquals("Failed to reform ensemble!", 2, ledgerMetadata.getEnsembles().size());
+ ensemble = ledgerMetadata.getEnsemble(0);
+ assertTrue("Failed to update the ledger metadata to use bookie host name",
+ ensemble.contains(toBookieAddr));
}
- lh.close();
- LedgerHandle openLedger = bk.openLedger(lh.getId(), digestType, PASSWORD.getBytes());
- final LedgerMetadata ledgerMetadata = openLedger.getLedgerMetadata();
- assertEquals("Failed to reform ensemble!", 2, ledgerMetadata.getEnsembles().size());
- ensemble = ledgerMetadata.getEnsemble(0);
- assertTrue("Failed to update the ledger metadata to use bookie host name",
- ensemble.contains(toBookieAddr));
}
/**
@@ -249,50 +252,51 @@ public class UpdateLedgerOpTest extends BookKeeperClusterTestCase {
*/
@Test
public void testRenameWhenAddEntryInProgress() throws Exception {
- final BookKeeper bk = new BookKeeper(baseClientConf, zkc);
- BookKeeperAdmin bkadmin = new BookKeeperAdmin(bk);
-
- LOG.info("Create ledger and add entries to it");
- final int numOfEntries = 5000;
- final CountDownLatch latch = new CountDownLatch(numOfEntries);
- final AtomicInteger rc = new AtomicInteger(BKException.Code.OK);
- final LedgerHandle lh = createLedgerWithEntries(bk, 1);
- latch.countDown();
- Thread th = new Thread() {
- public void run() {
- final AddCallback cb = new AddCallback() {
- public void addComplete(int rccb, LedgerHandle lh, long entryId, Object ctx) {
- rc.compareAndSet(BKException.Code.OK, rccb);
- if (entryId % 100 == 0) {
- LOG.info("Added entries till entryId:{}", entryId);
+ try (final BookKeeper bk = new BookKeeper(baseClientConf, zkc);
+ BookKeeperAdmin bkadmin = new BookKeeperAdmin(bk)) {
+
+ LOG.info("Create ledger and add entries to it");
+ final int numOfEntries = 5000;
+ final CountDownLatch latch = new CountDownLatch(numOfEntries);
+ final AtomicInteger rc = new AtomicInteger(BKException.Code.OK);
+ final LedgerHandle lh = createLedgerWithEntries(bk, 1);
+ latch.countDown();
+ Thread th = new Thread() {
+ public void run() {
+ final AddCallback cb = new AddCallback() {
+ public void addComplete(int rccb, LedgerHandle lh, long entryId, Object ctx) {
+ rc.compareAndSet(BKException.Code.OK, rccb);
+ if (entryId % 100 == 0) {
+ LOG.info("Added entries till entryId:{}", entryId);
+ }
+ latch.countDown();
}
- latch.countDown();
+ };
+ for (int i = 1; i < numOfEntries; i++) {
+ lh.asyncAddEntry(("foobar" + i).getBytes(), cb, null);
}
- };
- for (int i = 1; i < numOfEntries; i++) {
- lh.asyncAddEntry(("foobar" + i).getBytes(), cb, null);
- }
+ }
+ };
+ th.start();
+ List<BookieSocketAddress> ensemble = lh.getLedgerMetadata().getEnsemble(0);
+ BookieSocketAddress curBookieAddr = ensemble.get(0);
+ BookieSocketAddress toBookieAddr = new BookieSocketAddress("localhost:" + curBookieAddr.getPort());
+ UpdateLedgerOp updateLedgerOp = new UpdateLedgerOp(bk, bkadmin);
+ updateLedgerOp.updateBookieIdInLedgers(curBookieAddr, toBookieAddr, 5, 100, progressable);
+
+ if (!latch.await(120, TimeUnit.SECONDS)) {
+ throw new Exception("Entries took too long to add");
}
- };
- th.start();
- List<BookieSocketAddress> ensemble = lh.getLedgerMetadata().getEnsemble(0);
- BookieSocketAddress curBookieAddr = ensemble.get(0);
- BookieSocketAddress toBookieAddr = new BookieSocketAddress("localhost:" + curBookieAddr.getPort());
- UpdateLedgerOp updateLedgerOp = new UpdateLedgerOp(bk, bkadmin);
- updateLedgerOp.updateBookieIdInLedgers(curBookieAddr, toBookieAddr, 5, 100, progressable);
-
- if (!latch.await(120, TimeUnit.SECONDS)) {
- throw new Exception("Entries took too long to add");
- }
- if (rc.get() != BKException.Code.OK) {
- throw BKException.create(rc.get());
+ if (rc.get() != BKException.Code.OK) {
+ throw BKException.create(rc.get());
+ }
+ lh.close();
+ LedgerHandle openLedger = bk.openLedger(lh.getId(), digestType, PASSWORD.getBytes());
+ ensemble = openLedger.getLedgerMetadata().getEnsemble(0);
+ assertTrue("Failed to update the ledger metadata to use bookie host name",
+ ensemble.contains(toBookieAddr));
}
- lh.close();
- LedgerHandle openLedger = bk.openLedger(lh.getId(), digestType, PASSWORD.getBytes());
- ensemble = openLedger.getLedgerMetadata().getEnsemble(0);
- assertTrue("Failed to update the ledger metadata to use bookie host name",
- ensemble.contains(toBookieAddr));
}
private int getUpdatedLedgersCount(BookKeeper bk, List<LedgerHandle> ledgers, BookieSocketAddress toBookieAddr)
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPeriodicCheckTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPeriodicCheckTest.java
index cc4287d..4e3bda8 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPeriodicCheckTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPeriodicCheckTest.java
@@ -336,12 +336,13 @@ public class AuditorPeriodicCheckTest extends BookKeeperClusterTestCase {
}
lh.close();
}
- final Auditor auditor = new Auditor(
+
+ try (final Auditor auditor = new Auditor(
Bookie.getBookieAddress(bsConfs.get(0)).toString(),
- bsConfs.get(0), zkc, NullStatsLogger.INSTANCE);
- final AtomicBoolean exceptionCaught = new AtomicBoolean(false);
- final CountDownLatch latch = new CountDownLatch(1);
- Thread t = new Thread() {
+ bsConfs.get(0), zkc, NullStatsLogger.INSTANCE)) {
+ final AtomicBoolean exceptionCaught = new AtomicBoolean(false);
+ final CountDownLatch latch = new CountDownLatch(1);
+ Thread t = new Thread() {
public void run() {
try {
latch.countDown();
@@ -354,13 +355,14 @@ public class AuditorPeriodicCheckTest extends BookKeeperClusterTestCase {
}
}
};
- t.start();
- latch.await();
- for (Long id : ids) {
- bkc.deleteLedger(id);
+ t.start();
+ latch.await();
+ for (Long id : ids) {
+ bkc.deleteLedger(id);
+ }
+ t.join();
+ assertFalse("Shouldn't have thrown exception", exceptionCaught.get());
}
- t.join();
- assertFalse("Shouldn't have thrown exception", exceptionCaught.get());
}
private BookieSocketAddress replaceBookieWithWriteFailingBookie(LedgerHandle lh) throws Exception {
diff --git a/pom.xml b/pom.xml
index ee2b745..b29f423 100644
--- a/pom.xml
+++ b/pom.xml
@@ -750,7 +750,7 @@
<artifactId>maven-surefire-plugin</artifactId>
<version>${maven-surefire-plugin.version}</version>
<configuration>
- <argLine>-Xmx2G -Djava.net.preferIPv4Stack=true</argLine>
+ <argLine>-Xmx2G -Djava.net.preferIPv4Stack=true -Dio.netty.leakDetection.level=paranoid</argLine>
<redirectTestOutputToFile>${redirectTestOutputToFile}</redirectTestOutputToFile>
<reuseForks>false</reuseForks>
<forkedProcessTimeoutInSeconds>1800</forkedProcessTimeoutInSeconds>