This is an automated email from the ASF dual-hosted git repository.
adoroszlai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new a218b3bfea3 HDDS-15849. Force channel shutdown in
XceiverClientGrpc.close (#10747)
a218b3bfea3 is described below
commit a218b3bfea3e1d385873fb39418be3f14b7ba499
Author: Andrey Yarovoy <[email protected]>
AuthorDate: Wed Jul 15 09:53:37 2026 -0400
HDDS-15849. Force channel shutdown in XceiverClientGrpc.close (#10747)
---
.../apache/hadoop/hdds/scm/XceiverClientGrpc.java | 40 ++---
.../TestXceiverClientManagerCloseContention.java | 161 +++++++++++++++++++++
2 files changed, 173 insertions(+), 28 deletions(-)
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
index 8c52f339027..6882595928d 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
@@ -39,7 +39,6 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.LockSupport;
-import java.util.stream.Collectors;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.HddsUtils;
import org.apache.hadoop.hdds.client.BlockID;
@@ -92,7 +91,6 @@
*/
public class XceiverClientGrpc extends XceiverClientSpi {
private static final Logger LOG =
LoggerFactory.getLogger(XceiverClientGrpc.class);
- private static final int SHUTDOWN_WAIT_INTERVAL_MILLIS = 100;
private static final int SHUTDOWN_WAIT_MAX_SECONDS = 5;
private final Pipeline pipeline;
private final ConfigurationSource config;
@@ -258,7 +256,7 @@ public boolean isConnected(DatanodeDetails details) {
/**
* Closes all the communication channels of the client one-by-one.
* When a channel is closed, no further requests can be sent via the channel,
- * and the method waits to finish all ongoing communication.
+ * and any in-flight RPCs are cancelled immediately (shutdownNow semantics).
*/
@Override
public void close() {
@@ -267,39 +265,25 @@ public void close() {
return;
}
+ // Use shutdownNow() (not the graceful shutdown()) so in-flight RPCs are
+ // cancelled and the channel terminates immediately. close() is frequently
+ // invoked from cache eviction while the XceiverClientManager clientCache
+ // monitor is held (HDDS-15849); a blocking graceful drain there serializes
+ // every concurrent acquireClient()/releaseClient() call.
for (ChannelInfo channelInfo : dnChannelInfoMap.values()) {
- channelInfo.getChannel().shutdown();
- }
-
- final long maxWaitNanos =
TimeUnit.SECONDS.toNanos(SHUTDOWN_WAIT_MAX_SECONDS);
- long deadline = System.nanoTime() + maxWaitNanos;
- List<ManagedChannel> nonTerminatedChannels = dnChannelInfoMap.values()
- .stream()
- .map(ChannelInfo::getChannel)
- .filter(Objects::nonNull)
- .collect(Collectors.toList());
-
- while (!nonTerminatedChannels.isEmpty() && System.nanoTime() < deadline) {
- nonTerminatedChannels.removeIf(ManagedChannel::isTerminated);
+ ManagedChannel channel = channelInfo.getChannel();
+ channel.shutdownNow();
try {
- Thread.sleep(SHUTDOWN_WAIT_INTERVAL_MILLIS);
+ if (!channel.awaitTermination(SHUTDOWN_WAIT_MAX_SECONDS,
TimeUnit.SECONDS)) {
+ LOG.warn("Channel {} did not terminate within {}s.", channel,
SHUTDOWN_WAIT_MAX_SECONDS);
+ }
} catch (InterruptedException e) {
- LOG.error("Interrupted while waiting for channels to terminate", e);
+ LOG.error("Interrupted while waiting for channel termination", e);
Thread.currentThread().interrupt();
break;
}
}
- List<DatanodeID> failedChannels = dnChannelInfoMap.entrySet()
- .stream()
- .filter(e -> !e.getValue().getChannel().isTerminated())
- .map(Map.Entry::getKey)
- .collect(Collectors.toList());
-
- if (!failedChannels.isEmpty()) {
- LOG.warn("Channels {} did not terminate within timeout.",
failedChannels);
- }
-
dnChannelInfoMap.clear();
}
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerCloseContention.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerCloseContention.java
new file mode 100644
index 00000000000..b817f73cb76
--- /dev/null
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerCloseContention.java
@@ -0,0 +1,161 @@
+/*
+ * 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.hadoop.hdds.scm;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.scm.XceiverClientManager.ScmClientConfig;
+import
org.apache.hadoop.hdds.scm.XceiverClientManager.XceiverClientManagerConfigBuilder;
+import org.apache.hadoop.hdds.scm.client.ClientTrustManager;
+import org.apache.hadoop.hdds.scm.pipeline.MockPipeline;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Regression test for HDDS-15849.
+ *
+ * <p>HDDS-14571 changed {@link XceiverClientGrpc#close()} from a forced
+ * {@code shutdownNow()} (which cancels in-flight RPCs and terminates in ~1ms)
+ * to a graceful {@code shutdown()} followed by a polling loop that always
+ * sleeps at least {@code SHUTDOWN_WAIT_INTERVAL_MILLIS} (100ms) before it can
+ * observe channel termination. Every {@code close()} therefore costs ≥100ms
+ * instead of ~1ms.
+ *
+ * <p>The damage is amplified by <em>where</em> {@code close()} runs. Clients
+ * are torn down through {@link XceiverClientManager}'s cache eviction, and the
+ * whole acquire / evict / close sequence executes under the {@code
clientCache}
+ * monitor:
+ * <pre>
+ * acquireClient() // synchronized (clientCache)
+ * -> getClient()
+ * -> clientCache.get(key, newClient)
+ * -> (Guava evicts an entry on the calling thread)
+ * -> RemovalListener.onRemoval() // synchronized
(clientCache)
+ * -> XceiverClientSpi.setEvicted() -> cleanup() ->
close()
+ * </pre>
+ * So the blocking graceful-shutdown wait runs while the {@code clientCache}
+ * monitor is held, and every other thread in {@code acquireClient()} /
+ * {@code releaseClient()} serializes behind it.
+ *
+ * <p>This test exercises the <b>real</b> {@code XceiverClientManager} and
+ * {@code XceiverClientGrpc.close()} — nothing is emulated. Real (lazily
+ * connected, plaintext) gRPC channels are created against random local
+ * datanode addresses; no live datanode is required because
+ * {@code close()} only shuts the channels down. {@code maxCacheSize == 1} plus
+ * a distinct pipeline per iteration forces the previously cached client to be
+ * evicted and closed on essentially every acquisition — mirroring the per-file
+ * client churn of EC checksum collection.
+ *
+ * <p>Run it against the pre-fix {@code close()} and the wall-clock is
dominated
+ * by serialized 100ms sleeps ({@code ~ evictions * 100ms}); run it against the
+ * fix (immediate {@code shutdownNow()}) and per-close cost collapses. The
+ * assertion below encodes the fixed expectation, so this test FAILS on the
+ * regression and PASSES once HDDS-15849 restores immediate termination.
+ */
+class TestXceiverClientManagerCloseContention {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(TestXceiverClientManagerCloseContention.class);
+
+ private static final int THREADS = 8;
+ private static final int CYCLES_PER_THREAD = 8;
+
+ /**
+ * Upper bound on acceptable average wall-clock cost per eviction-driven
+ * close(). The pre-fix code has a hard ≥100ms floor per close (the
+ * mandatory Thread.sleep in the termination polling loop); immediate
+ * shutdownNow() is well under a millisecond. 40ms sits safely between the
+ * two so the assertion is not timing-flaky.
+ */
+ private static final long MAX_MILLIS_PER_CLOSE = 40;
+
+ @Test
+ void evictionDrivenCloseDoesNotSerializeAcquisition() throws Exception {
+ ConfigurationSource conf = new OzoneConfiguration();
+ // maxCacheSize == 1 => each acquisition of a new pipeline evicts (and thus
+ // closes) the previously cached client.
+ ScmClientConfig clientConf = new XceiverClientManagerConfigBuilder()
+ .setMaxCacheSize(1)
+ .setStaleThresholdMs(TimeUnit.SECONDS.toMillis(30))
+ .build();
+
+ try (XceiverClientManager manager =
+ new XceiverClientManager(conf, clientConf, (ClientTrustManager)
null)) {
+
+ ExecutorService pool = Executors.newFixedThreadPool(THREADS);
+ CountDownLatch start = new CountDownLatch(1);
+ long wallStartNanos;
+ try {
+ List<Future<?>> futures = IntStream.range(0, THREADS)
+ .mapToObj(t -> pool.submit(() -> {
+ start.await();
+ for (int i = 0; i < CYCLES_PER_THREAD; i++) {
+ // Distinct pipeline per cycle => distinct cache key => the
+ // previously cached client is evicted and closed.
+ Pipeline pipeline = MockPipeline.createPipeline(1);
+ XceiverClientSpi client = manager.acquireClient(pipeline);
+ manager.releaseClient(client, false);
+ }
+ return null;
+ }))
+ .collect(Collectors.toList());
+
+ wallStartNanos = System.nanoTime();
+ start.countDown();
+ for (Future<?> f : futures) {
+ f.get(120, TimeUnit.SECONDS);
+ }
+ } finally {
+ pool.shutdownNow();
+ }
+ long wallMillis =
+ TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - wallStartNanos);
+
+ // recordStats() is enabled on the cache; each eviction drives one
close().
+ long evictions = manager.getClientCache().stats().evictionCount();
+ long millisPerClose = evictions == 0 ? 0 : wallMillis / evictions;
+
+ LOG.info("threads={} cyclesPerThread={} acquisitions={}
evictions(closes)={}",
+ THREADS, CYCLES_PER_THREAD, THREADS * CYCLES_PER_THREAD, evictions);
+ LOG.info("wall={}ms ms-per-close={}ms (regression floor is
~100ms/close)",
+ wallMillis, millisPerClose);
+
+ assertThat(evictions)
+ .as("cache eviction should have driven close() calls")
+ .isPositive();
+
+ assertThat(millisPerClose)
+ .as("HDDS-15849: eviction-driven close() must not serialize "
+ + "acquisition under the clientCache monitor; the pre-fix "
+ + "graceful-shutdown wait forces a ~100ms floor per close")
+ .isLessThan(MAX_MILLIS_PER_CLOSE);
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
