This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch branch-4.8
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
The following commit(s) were added to refs/heads/branch-4.8 by this push:
new 5358467 Issue #1610: netty leak detector logs 'LEAK:
HashedWheelTimer.release() was not called before it's garbage-collected.' in
unit tests
5358467 is described below
commit 5358467746d154118f4e318c7546298531d01b67
Author: Andrey Yegorov <[email protected]>
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.
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.
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 <[email protected]>
Reviewers: Enrico Olivelli <[email protected]>, Sijie Guo
<[email protected]>
This closes #1613 from dlg99/fix/issue_1610_hashedwheeltimer, closes #1610
(cherry picked from commit 5f2d45c8fa0619d50423f1672303fea2d1a59238)
Signed-off-by: Sijie Guo <[email protected]>
---
.../org/apache/bookkeeper/client/BookKeeper.java | 7 +-
.../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, 239 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 44f970f..d187b41 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
@@ -472,6 +472,10 @@ public class BookKeeper implements
org.apache.bookkeeper.client.api.BookKeeper {
this.ownEventLoopGroup = false;
}
+ // initialize bookie client
+ this.bookieClient = new BookieClient(conf, this.eventLoopGroup,
this.mainWorkerPool,
+ scheduler, statsLogger);
+
if (null == requestTimer) {
this.requestTimer = new HashedWheelTimer(
new
ThreadFactoryBuilder().setNameFormat("BookieClientTimer-%d").build(),
@@ -506,9 +510,6 @@ public class BookKeeper implements
org.apache.bookkeeper.client.api.BookKeeper {
} else {
this.readLACSpeculativeRequestPolicy =
Optional.<SpeculativeRequestExecutionPolicy>absent();
}
- // initialize bookie client
- this.bookieClient = new BookieClient(conf, this.eventLoopGroup,
this.mainWorkerPool,
- scheduler, statsLogger);
this.bookieWatcher = new BookieWatcher(
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 72cddbd..4f31219 100644
--- a/pom.xml
+++ b/pom.xml
@@ -755,7 +755,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>
