You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@bookkeeper.apache.org by GitBox <gi...@apache.org> on 2018/08/21 21:08:25 UTC
[GitHub] sijie closed pull request #1613: Issue #1610: netty leak detector
logs 'LEAK: HashedWheelTimer.release() was not called before it's
garbage-collected.' in unit tests
sijie closed pull request #1613: Issue #1610: netty leak detector logs 'LEAK: HashedWheelTimer.release() was not called before it's garbage-collected.' in unit tests
URL: https://github.com/apache/bookkeeper/pull/1613
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
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 4fc7728bdc..6544e44da5 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 BookKeeper(ClientConfiguration conf, ZooKeeper zk, EventLoopGroup eventLo
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 BookKeeper(ClientConfiguration conf, ZooKeeper zk, EventLoopGroup eventLo
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 aa9e0483ba..8ee363fc78 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 BookieRequestProcessor(ServerConfiguration serverCfg, Bookie bookie,
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 void close() {
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 559873e7e7..27e8ab2eef 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 void start() throws IOException, UnavailableException, InterruptedExcepti
// 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 15868d957f..c5982fa403 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
@@ -76,7 +76,7 @@
*
* <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;
@@ -730,6 +730,11 @@ public void shutdown() {
}
}
+ @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 5d8603679a..cca97e8d40 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 TestAuth() {
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 @@ private int entryCount(long ledgerId, ServerConfiguration bookieConf,
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 63de59cb88..69fe345f2d 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 @@ private void runBookieWatcherWhenSessionExpired(ZooKeeper zk, int timeout, boole
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 7f672420fa..91628e7e8d 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 void testManyLedgersWithShortHostname() throws Exception {
}
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 void testManyLedgers(boolean useShortHostName) throws Exception {
*/
@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 void testChangeEnsembleAfterRenamingToShortHostname() throws Exception {
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 void addComplete(int rccb, LedgerHandle lh, long entryId, Object ctx) {
*/
@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 cc4287d691..4e3bda8eba 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 void testPeriodicCheckWhenLedgerDeleted() throws Exception {
}
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 void run() {
}
}
};
- 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 ee2b745602..b29f423731 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>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services