This is an automated email from the ASF dual-hosted git repository.
wenbingshen 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 a41c70061d Issue 4797:Prevent thread leak when BookKeeper client
constructor fails (#4798)
a41c70061d is described below
commit a41c70061df0bb822af321c6d61c8c525ce7bb88
Author: KROOS <[email protected]>
AuthorDate: Thu Aug 13 15:40:07 2026 +0800
Issue 4797:Prevent thread leak when BookKeeper client constructor fails
(#4798)
* [FIX] Prevent thread leak when BookKeeper client constructor fails
* add test
* fix checkstyle
* add close
* use close directly
* use Try-finally to handle the close.
---------
Co-authored-by: shanxu <[email protected]>
---
.../org/apache/bookkeeper/client/BookKeeper.java | 287 ++++++++++++---------
.../client/BookKeeperConstructorFailureTest.java | 176 +++++++++++++
2 files changed, 335 insertions(+), 128 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 db34b482f2..1542cdcaf3 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
@@ -423,127 +423,144 @@ public class BookKeeper implements
org.apache.bookkeeper.client.api.BookKeeper {
this.internalConf =
ClientInternalConf.fromConfigAndFeatureProvider(conf, this.featureProvider);
- // initialize resources
- this.scheduler =
OrderedScheduler.newSchedulerBuilder().numThreads(1).name("BookKeeperClientScheduler").build();
- this.highPriorityTaskExecutor =
-
OrderedScheduler.newSchedulerBuilder().numThreads(1).name("BookKeeperHighPriorityThread").build();
- this.mainWorkerPool = OrderedExecutor.newBuilder()
- .name("BookKeeperClientWorker")
- .numThreads(conf.getNumWorkerThreads())
- .statsLogger(rootStatsLogger)
- .traceTaskExecution(conf.getEnableTaskExecutionStats())
-
.preserveMdcForTaskExecution(conf.getPreserveMdcForTaskExecution())
-
.traceTaskWarnTimeMicroSec(conf.getTaskExecutionWarnTimeMicros())
- .enableBusyWait(conf.isBusyWaitEnabled())
- .build();
-
- // initialize stats logger
- this.statsLogger =
rootStatsLogger.scope(BookKeeperClientStats.CLIENT_SCOPE);
- this.clientStats = BookKeeperClientStats.newInstance(this.statsLogger);
-
- // initialize metadata driver
+ boolean initialized = false;
try {
- String metadataServiceUriStr = conf.getMetadataServiceUri();
- if (null != metadataServiceUriStr) {
- this.metadataDriver =
MetadataDrivers.getClientDriver(URI.create(metadataServiceUriStr));
- } else {
- checkNotNull(zkc, "No external zookeeper provided when no
metadata service uri is found");
- this.metadataDriver = MetadataDrivers.getClientDriver("zk");
+ // initialize resources
+ this.scheduler =
+
OrderedScheduler.newSchedulerBuilder().numThreads(1).name("BookKeeperClientScheduler").build();
+ this.highPriorityTaskExecutor =
+
OrderedScheduler.newSchedulerBuilder().numThreads(1).name("BookKeeperHighPriorityThread").build();
+ this.mainWorkerPool = OrderedExecutor.newBuilder()
+ .name("BookKeeperClientWorker")
+ .numThreads(conf.getNumWorkerThreads())
+ .statsLogger(rootStatsLogger)
+ .traceTaskExecution(conf.getEnableTaskExecutionStats())
+
.preserveMdcForTaskExecution(conf.getPreserveMdcForTaskExecution())
+
.traceTaskWarnTimeMicroSec(conf.getTaskExecutionWarnTimeMicros())
+ .enableBusyWait(conf.isBusyWaitEnabled())
+ .build();
+
+ // initialize stats logger
+ this.statsLogger =
rootStatsLogger.scope(BookKeeperClientStats.CLIENT_SCOPE);
+ this.clientStats =
BookKeeperClientStats.newInstance(this.statsLogger);
+
+ // initialize metadata driver
+ try {
+ String metadataServiceUriStr = conf.getMetadataServiceUri();
+ if (null != metadataServiceUriStr) {
+ this.metadataDriver =
MetadataDrivers.getClientDriver(URI.create(metadataServiceUriStr));
+ } else {
+ checkNotNull(zkc, "No external zookeeper provided when no
metadata service uri is found");
+ this.metadataDriver =
MetadataDrivers.getClientDriver("zk");
+ }
+ this.metadataDriver.initialize(
+ conf,
+ highPriorityTaskExecutor,
+ rootStatsLogger,
+ Optional.ofNullable(zkc));
+ } catch (ConfigurationException ce) {
+ log.error()
+ .exception(ce)
+ .log("Failed to initialize metadata client driver
using invalid metadata service uri");
+ throw new IOException("Failed to initialize metadata client
driver", ce);
+ } catch (MetadataException me) {
+ log.error().exception(me).log("Encountered metadata exceptions
on initializing metadata client driver");
+ throw new IOException("Failed to initialize metadata client
driver", me);
}
- this.metadataDriver.initialize(
- conf,
- highPriorityTaskExecutor,
- rootStatsLogger,
- Optional.ofNullable(zkc));
- } catch (ConfigurationException ce) {
- log.error()
- .exception(ce)
- .log("Failed to initialize metadata client driver using
invalid metadata service uri");
- throw new IOException("Failed to initialize metadata client
driver", ce);
- } catch (MetadataException me) {
- log.error().exception(me).log("Encountered metadata exceptions on
initializing metadata client driver");
- throw new IOException("Failed to initialize metadata client
driver", me);
- }
- // initialize event loop group
- if (null == eventLoopGroup) {
- this.eventLoopGroup = EventLoopUtil.getClientEventLoopGroup(conf,
- new DefaultThreadFactory("bookkeeper-io"));
- this.ownEventLoopGroup = true;
- } else {
- this.eventLoopGroup = eventLoopGroup;
- this.ownEventLoopGroup = false;
- }
+ // initialize event loop group
+ if (null == eventLoopGroup) {
+ this.eventLoopGroup =
EventLoopUtil.getClientEventLoopGroup(conf,
+ new DefaultThreadFactory("bookkeeper-io"));
+ this.ownEventLoopGroup = true;
+ } else {
+ this.eventLoopGroup = eventLoopGroup;
+ this.ownEventLoopGroup = false;
+ }
- if (byteBufAllocator != null) {
- this.allocator = byteBufAllocator;
- } else {
- this.allocator = ByteBufAllocatorBuilder.create()
- .poolingPolicy(conf.getAllocatorPoolingPolicy())
- .poolingConcurrency(conf.getAllocatorPoolingConcurrency())
- .outOfMemoryPolicy(conf.getAllocatorOutOfMemoryPolicy())
-
.leakDetectionPolicy(conf.getAllocatorLeakDetectionPolicy())
- .exitOnOutOfMemory(conf.exitOnOutOfMemory())
- .build();
- }
+ if (byteBufAllocator != null) {
+ this.allocator = byteBufAllocator;
+ } else {
+ this.allocator = ByteBufAllocatorBuilder.create()
+ .poolingPolicy(conf.getAllocatorPoolingPolicy())
+
.poolingConcurrency(conf.getAllocatorPoolingConcurrency())
+
.outOfMemoryPolicy(conf.getAllocatorOutOfMemoryPolicy())
+
.leakDetectionPolicy(conf.getAllocatorLeakDetectionPolicy())
+ .exitOnOutOfMemory(conf.exitOnOutOfMemory())
+ .build();
+ }
+ if (null == requestTimer) {
+ this.requestTimer = new HashedWheelTimer(
+ new
ThreadFactoryBuilder().setNameFormat("BookieClientTimer-%d").build(),
+ conf.getTimeoutTimerTickDurationMs(),
TimeUnit.MILLISECONDS,
+ conf.getTimeoutTimerNumTicks());
+ this.ownTimer = true;
+ } else {
+ this.requestTimer = requestTimer;
+ this.ownTimer = false;
+ }
- if (null == requestTimer) {
- this.requestTimer = new HashedWheelTimer(
- new
ThreadFactoryBuilder().setNameFormat("BookieClientTimer-%d").build(),
- conf.getTimeoutTimerTickDurationMs(),
TimeUnit.MILLISECONDS,
- conf.getTimeoutTimerNumTicks());
- this.ownTimer = true;
- } else {
- this.requestTimer = requestTimer;
- this.ownTimer = false;
- }
+ BookieAddressResolver bookieAddressResolver =
conf.getBookieAddressResolverEnabled()
+ ? new
DefaultBookieAddressResolver(metadataDriver.getRegistrationClient())
+ : new BookieAddressResolverDisabled();
+ if (dnsResolver != null) {
+ dnsResolver.setBookieAddressResolver(bookieAddressResolver);
+ }
+ // initialize the ensemble placement
+ this.placementPolicy = initializeEnsemblePlacementPolicy(conf,
+ dnsResolver, this.requestTimer, this.featureProvider,
this.statsLogger, bookieAddressResolver);
+
+ this.bookieWatcher = new BookieWatcherImpl(
+ conf, this.placementPolicy,
metadataDriver.getRegistrationClient(), bookieAddressResolver,
+ this.statsLogger.scope(WATCHER_SCOPE));
+
+ // initialize bookie client
+ this.bookieClient = new BookieClientImpl(conf,
this.eventLoopGroup, this.allocator, this.mainWorkerPool,
+ scheduler, rootStatsLogger,
this.bookieWatcher.getBookieAddressResolver());
+
+ if (conf.getDiskWeightBasedPlacementEnabled()) {
+ log.info("Weighted ledger placement enabled");
+ ThreadFactoryBuilder tFBuilder = new ThreadFactoryBuilder()
+ .setNameFormat("BKClientMetaDataPollScheduler-%d");
+ this.bookieInfoScheduler =
Executors.newSingleThreadScheduledExecutor(tFBuilder.build());
+ this.bookieInfoReader = new BookieInfoReader(this, conf,
this.bookieInfoScheduler);
+ this.bookieWatcher.initialBlockingBookieRead();
+ this.bookieInfoReader.start();
+ } else {
+ log.info("Weighted ledger placement is not enabled");
+ this.bookieInfoScheduler = null;
+ this.bookieInfoReader = new BookieInfoReader(this, conf, null);
+ this.bookieWatcher.initialBlockingBookieRead();
+ }
- BookieAddressResolver bookieAddressResolver =
conf.getBookieAddressResolverEnabled()
- ? new
DefaultBookieAddressResolver(metadataDriver.getRegistrationClient())
- : new BookieAddressResolverDisabled();
- if (dnsResolver != null) {
- dnsResolver.setBookieAddressResolver(bookieAddressResolver);
- }
- // initialize the ensemble placement
- this.placementPolicy = initializeEnsemblePlacementPolicy(conf,
- dnsResolver, this.requestTimer, this.featureProvider,
this.statsLogger, bookieAddressResolver);
-
- this.bookieWatcher = new BookieWatcherImpl(
- conf, this.placementPolicy,
metadataDriver.getRegistrationClient(), bookieAddressResolver,
- this.statsLogger.scope(WATCHER_SCOPE));
-
- // initialize bookie client
- this.bookieClient = new BookieClientImpl(conf, this.eventLoopGroup,
this.allocator, this.mainWorkerPool,
- scheduler, rootStatsLogger,
this.bookieWatcher.getBookieAddressResolver());
-
- if (conf.getDiskWeightBasedPlacementEnabled()) {
- log.info("Weighted ledger placement enabled");
- ThreadFactoryBuilder tFBuilder = new ThreadFactoryBuilder()
- .setNameFormat("BKClientMetaDataPollScheduler-%d");
- this.bookieInfoScheduler =
Executors.newSingleThreadScheduledExecutor(tFBuilder.build());
- this.bookieInfoReader = new BookieInfoReader(this, conf,
this.bookieInfoScheduler);
- this.bookieWatcher.initialBlockingBookieRead();
- this.bookieInfoReader.start();
- } else {
- log.info("Weighted ledger placement is not enabled");
- this.bookieInfoScheduler = null;
- this.bookieInfoReader = new BookieInfoReader(this, conf, null);
- this.bookieWatcher.initialBlockingBookieRead();
- }
+ // initialize ledger manager
+ try {
+ this.ledgerManagerFactory =
+ this.metadataDriver.getLedgerManagerFactory();
+ } catch (MetadataException e) {
+ throw new IOException("Failed to initialize ledger manager
factory", e);
+ }
+ this.ledgerManager = new
CleanupLedgerManager(ledgerManagerFactory.newLedgerManager());
+ this.ledgerIdGenerator =
ledgerManagerFactory.newLedgerIdGenerator();
- // initialize ledger manager
- try {
- this.ledgerManagerFactory =
- this.metadataDriver.getLedgerManagerFactory();
- } catch (MetadataException e) {
- throw new IOException("Failed to initialize ledger manager
factory", e);
+ this.bookieQuarantineRatio = conf.getBookieQuarantineRatio();
+ scheduleBookieHealthCheckIfEnabled(conf);
+ initialized = true;
+ } finally {
+ if (!initialized) {
+ try {
+ close();
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ log.warn().exception(ie)
+ .log("Interrupted while closing
partially-initialized BookKeeper client");
+ } catch (Throwable t) {
+ log.warn().exception(t)
+ .log("Failed to close partially-initialized
BookKeeper client");
+ }
+ }
}
- this.ledgerManager = new
CleanupLedgerManager(ledgerManagerFactory.newLedgerManager());
- this.ledgerIdGenerator = ledgerManagerFactory.newLedgerIdGenerator();
-
- this.bookieQuarantineRatio = conf.getBookieQuarantineRatio();
- scheduleBookieHealthCheckIfEnabled(conf);
}
/**
@@ -1520,32 +1537,44 @@ public class BookKeeper implements
org.apache.bookkeeper.client.api.BookKeeper {
// Close bookie client so all pending bookie requests would be failed
// which will reject any incoming bookie requests.
- bookieClient.close();
+ if (bookieClient != null) {
+ bookieClient.close();
+ }
try {
// Close ledger manage so all pending metadata requests would be
failed
// which will reject any incoming metadata requests.
- ledgerManager.close();
- ledgerIdGenerator.close();
+ if (ledgerManager != null) {
+ ledgerManager.close();
+ }
+ if (ledgerIdGenerator != null) {
+ ledgerIdGenerator.close();
+ }
} catch (IOException ie) {
log.error().exception(ie).log("Failed to close ledger manager");
}
// Close the scheduler
- scheduler.shutdown();
- if (!scheduler.awaitTermination(10, TimeUnit.SECONDS)) {
- log.warn("The scheduler did not shutdown cleanly");
+ if (scheduler != null) {
+ scheduler.shutdown();
+ if (!scheduler.awaitTermination(10, TimeUnit.SECONDS)) {
+ log.warn("The scheduler did not shutdown cleanly");
+ }
}
// Close the watchTask scheduler
- highPriorityTaskExecutor.shutdown();
- if (!highPriorityTaskExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
- log.warn("The highPriorityTaskExecutor for WatchTask did not
shutdown cleanly, interrupting");
- highPriorityTaskExecutor.shutdownNow();
+ if (highPriorityTaskExecutor != null) {
+ highPriorityTaskExecutor.shutdown();
+ if (!highPriorityTaskExecutor.awaitTermination(10,
TimeUnit.SECONDS)) {
+ log.warn("The highPriorityTaskExecutor for WatchTask did not
shutdown cleanly, interrupting");
+ highPriorityTaskExecutor.shutdownNow();
+ }
}
- mainWorkerPool.shutdown();
- if (!mainWorkerPool.awaitTermination(10, TimeUnit.SECONDS)) {
- log.warn("The mainWorkerPool did not shutdown cleanly");
+ if (mainWorkerPool != null) {
+ mainWorkerPool.shutdown();
+ if (!mainWorkerPool.awaitTermination(10, TimeUnit.SECONDS)) {
+ log.warn("The mainWorkerPool did not shutdown cleanly");
+ }
}
if (this.bookieInfoScheduler != null) {
this.bookieInfoScheduler.shutdown();
@@ -1554,13 +1583,15 @@ public class BookKeeper implements
org.apache.bookkeeper.client.api.BookKeeper {
}
}
- if (ownTimer) {
+ if (ownTimer && requestTimer != null) {
requestTimer.stop();
}
- if (ownEventLoopGroup) {
+ if (ownEventLoopGroup && eventLoopGroup != null) {
eventLoopGroup.shutdownGracefully();
}
- this.metadataDriver.close();
+ if (metadataDriver != null) {
+ this.metadataDriver.close();
+ }
}
@Override
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookKeeperConstructorFailureTest.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookKeeperConstructorFailureTest.java
new file mode 100644
index 0000000000..2d5db8658b
--- /dev/null
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookKeeperConstructorFailureTest.java
@@ -0,0 +1,176 @@
+/*
+ *
+ * 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.client;
+
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.TreeMap;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Verifies that when {@link BookKeeper} construction fails at metadata-driver
+ * initialization, the {@link
org.apache.bookkeeper.common.util.OrderedExecutor}
+ * pools allocated before that point are released and do not accumulate across
+ * repeated failing {@code build()} calls.
+ *
+ * <p>This test targets the <em>early-failure</em> path (an unreachable
+ * ZooKeeper URI), which is the most common production misconfiguration and the
+ * historical source of thread leaks. It does not cover resources allocated
after
+ * the metadata driver (e.g. {@code eventLoopGroup}, {@code requestTimer},
+ * {@code bookieInfoScheduler}); those code paths require a reachable ZK and
are
+ * exercised by {@link BookKeeperCloseTest} together with the construction
+ * cleanup logic in {@link BookKeeper}.
+ *
+ * <p>Every {@code build()} attempt runs inside a dedicated {@link ThreadGroup}
+ * and only threads living in that group are counted, so concurrently executing
+ * tests in the same JVM cannot pollute the result.
+ */
+public class BookKeeperConstructorFailureTest {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(BookKeeperConstructorFailureTest.class);
+
+ private static final String UNREACHABLE_METADATA_URI =
"zk://127.0.0.1:1/ledgers";
+
+ /**
+ * Thread-name prefixes of the three OrderedExecutor pools that BookKeeper
allocates
+ * <em>before</em> {@code metadataDriver.initialize(...)}. With an
unreachable ZK URI,
+ * construction fails inside that call, so only these three pools can
possibly leak
+ * on this code path.
+ */
+ private static final String[] CLIENT_THREAD_PREFIXES = new String[] {
+ "BookKeeperClientScheduler",
+ "BookKeeperClientWorker",
+ "BookKeeperHighPriorityThread",
+ };
+
+ private static final int FAILED_BUILD_ITERATIONS = 5;
+
+ private static final long THREAD_SHUTDOWN_TIMEOUT_MS = 5_000L;
+
+ /** Repeated failing {@code build()} calls must not accumulate worker-pool
threads. */
+ @Test
+ public void testRepeatedFailedBuildsDoNotAccumulateThreads() throws
Exception {
+ ThreadGroup group = new ThreadGroup("bk-ctor-failure-test");
+ runFailingBuildsIn(group);
+ waitForThreadShutdown(group, THREAD_SHUTDOWN_TIMEOUT_MS);
+
+ Map<String, Integer> leak = snapshotClientThreadCounts(group);
+ if (hasLeak(leak)) {
+ LOG.error("Thread leak detected after {} failed build()s in group
{}: {}",
+ FAILED_BUILD_ITERATIONS, group.getName(), leak);
+ fail("BookKeeper constructor leaked threads after " +
FAILED_BUILD_ITERATIONS
+ + " failed build()s in group " + group.getName() + ": " +
leak);
+ }
+ }
+
+ private static void runFailingBuildsIn(ThreadGroup group) throws
InterruptedException {
+ ClientConfiguration conf = newFailingClientConf();
+ Throwable[] driverError = new Throwable[1];
+ Thread driver = new Thread(group, () -> {
+ for (int i = 0; i < FAILED_BUILD_ITERATIONS; i++) {
+ try {
+ BookKeeper.forConfig(conf).build();
+ driverError[0] = new AssertionError(
+ "BookKeeper construction should have failed at
iteration " + i);
+ return;
+ } catch (Exception expected) {
+ LOG.debug("iteration {} failed as expected: {}", i,
expected.toString());
+ }
+ }
+ }, "bk-ctor-failure-driver");
+ driver.start();
+ driver.join();
+ if (driverError[0] instanceof AssertionError) {
+ throw (AssertionError) driverError[0];
+ }
+ }
+
+ // ---------- helpers ----------
+
+ private static ClientConfiguration newFailingClientConf() {
+ ClientConfiguration conf = new ClientConfiguration();
+ conf.setMetadataServiceUri(UNREACHABLE_METADATA_URI);
+ conf.setZkTimeout(1000);
+ conf.setZkRetryBackoffMaxRetries(0);
+ return conf;
+ }
+
+ /** Returns the live-thread counts inside {@code group} grouped by {@link
#CLIENT_THREAD_PREFIXES}. */
+ private static Map<String, Integer> snapshotClientThreadCounts(ThreadGroup
group) {
+ Thread[] threads = new Thread[Math.max(16, group.activeCount() * 2)];
+ int n;
+ // Loop to handle the race where a thread starts between activeCount()
and enumerate().
+ while ((n = group.enumerate(threads, true)) == threads.length) {
+ threads = new Thread[threads.length * 2];
+ }
+ Map<String, Integer> counts = new TreeMap<>();
+ for (String prefix : CLIENT_THREAD_PREFIXES) {
+ counts.put(prefix, 0);
+ }
+ for (int i = 0; i < n; i++) {
+ Thread t = threads[i];
+ if (t == null || !t.isAlive()) {
+ continue;
+ }
+ String prefix = matchPrefix(t.getName());
+ if (prefix != null) {
+ counts.merge(prefix, 1, Integer::sum);
+ }
+ }
+ return counts;
+ }
+
+ private static String matchPrefix(String name) {
+ if (name == null) {
+ return null;
+ }
+ return Arrays.stream(CLIENT_THREAD_PREFIXES)
+ .filter(name::startsWith)
+ .findFirst()
+ .orElse(null);
+ }
+
+ private static boolean hasLeak(Map<String, Integer> counts) {
+ for (Integer v : counts.values()) {
+ if (v != null && v > 0) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static void waitForThreadShutdown(ThreadGroup group, long
timeoutMs)
+ throws InterruptedException {
+ long deadline = System.currentTimeMillis() + timeoutMs;
+ while (System.currentTimeMillis() < deadline) {
+ if (!hasLeak(snapshotClientThreadCounts(group))) {
+ return;
+ }
+ Thread.sleep(50);
+ }
+ }
+}
\ No newline at end of file
