Skip to content

RATIS-2805. GrpcUtil: shutdownManagedChannels vs shutdownManagedChannel subtle differences. #1686

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
amaliujia merged 1 commit into from
Nov 7, 2026
+52 −115
Merged

RATIS-2805. GrpcUtil: shutdownManagedChannels vs shutdownManagedChannel subtle differences. #1686

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions ratis-common/src/main/java/org/apache/ratis/util/JavaUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -327,4 +327,15 @@ static <OUTPUT, THROWABLE extends Throwable> OUTPUT supplyAndWrapAsCompletionExc
throw new CompletionException(t);
}
}

/** Apply the given method to each value, continuing with the others even if one fails. */
static <V> void acceptEach(Iterable<V> values, Consumer<V> method, String methodName, Logger log) {
for (V v : values) {
try {
method.accept(v);
} catch (Exception e) {
log.error("Failed to {}: {}", methodName, v, e);
}
}
}
}
156 changes: 41 additions & 115 deletions ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,14 +17,9 @@
*/
package org.apache.ratis.grpc;

import java.util.function.Consumer;
import org.apache.ratis.protocol.exceptions.ServerNotReadyException;
import org.apache.ratis.protocol.exceptions.TimeoutIOException;
import org.apache.ratis.security.TlsConf;
import org.apache.ratis.security.TlsConf.TrustManagerConf;
import org.apache.ratis.security.TlsConf.CertificatesConf;
import org.apache.ratis.security.TlsConf.PrivateKeyConf;
import org.apache.ratis.security.TlsConf.KeyManagerConf;
import org.apache.ratis.thirdparty.io.grpc.ManagedChannel;
import org.apache.ratis.thirdparty.io.grpc.Metadata;
import org.apache.ratis.thirdparty.io.grpc.Status;
Expand All @@ -45,21 +40,24 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.net.ssl.KeyManager;
import javax.net.ssl.SSLException;
import javax.net.ssl.TrustManager;
import java.io.IOException;
import java.security.Provider;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import static org.apache.ratis.thirdparty.io.netty.handler.ssl.SslProvider.OPENSSL;
import static org.apache.ratis.util.JavaUtils.acceptEach;
import static org.apache.ratis.util.NettyUtils.getJsseProvider;
import static org.apache.ratis.util.NettyUtils.initSslContextBuilderForServer;
import static org.apache.ratis.util.NettyUtils.setKeyManager;
import static org.apache.ratis.util.NettyUtils.setTrustManager;

public interface GrpcUtil {
Logger LOG = LoggerFactory.getLogger(GrpcUtil.class);
Expand Down Expand Up @@ -236,138 +234,66 @@ Metadata build() {
}
}

/**
* Tries to gracefully shut down the managed channel. Falls back to forceful shutdown if
* graceful shutdown times out.
*/
static void shutdownManagedChannel(ManagedChannel managedChannel) {
shutdownManagedChannel(managedChannel, GrpcConfigKeys.Client.CHANNEL_SHUTDOWN_GRACEFUL_TIMEOUT_DEFAULT,
/** Shut down the given channel using the default timeouts. */
static void shutdownManagedChannel(ManagedChannel channel) {
shutdownManagedChannel(channel,
GrpcConfigKeys.Client.CHANNEL_SHUTDOWN_GRACEFUL_TIMEOUT_DEFAULT,
GrpcConfigKeys.Client.CHANNEL_SHUTDOWN_FORCEFUL_TIMEOUT_DEFAULT);
}

/**
* Shut down the given channel gracefully, waiting up to the graceful timeout for its calls to finish,
* and then forcefully, cancelling the remaining calls, if it has not terminated.
*
* @param gracefulTimeout zero skips the graceful shutdown.
*/
static void shutdownManagedChannel(ManagedChannel managedChannel,
/** The same as shutdownManagedChannels(Collections.singleton(channel), gracefulTimeout, forcefulTimeout). */
static void shutdownManagedChannel(ManagedChannel channel,
TimeDuration gracefulTimeout, TimeDuration forcefulTimeout) {
// Close the gRPC managed-channel if not shut down already.
if (!managedChannel.isShutdown() && gracefulTimeout.isPositive()) {
try {
managedChannel.shutdown();
if (!managedChannel.awaitTermination(gracefulTimeout.getDuration(), gracefulTimeout.getUnit())) {
LOG.warn("Timed out gracefully shutting down connection: {}. ", managedChannel);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} catch (Exception e) {
LOG.error("Unexpected exception while waiting for channel termination", e);
}
}
shutdownManagedChannels(Collections.singleton(channel), gracefulTimeout, forcefulTimeout);
}

// Forceful shut down if still not terminated.
if (!managedChannel.isTerminated()) {
try {
managedChannel.shutdownNow();
if (!managedChannel.awaitTermination(forcefulTimeout.getDuration(), forcefulTimeout.getUnit())) {
LOG.warn("Timed out forcefully shutting down connection: {}. ", managedChannel);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} catch (Exception e) {
LOG.error("Unexpected exception while waiting for channel termination", e);
}
}
/** Shut down the given channels using the default timeouts. */
static void shutdownManagedChannels(Collection<ManagedChannel> channels) {
shutdownManagedChannels(channels,
GrpcConfigKeys.Client.CHANNEL_SHUTDOWN_GRACEFUL_TIMEOUT_DEFAULT,
GrpcConfigKeys.Client.CHANNEL_SHUTDOWN_FORCEFUL_TIMEOUT_DEFAULT);
}

/**
* Shut down the given channels as {@link #shutdownManagedChannel(ManagedChannel)} does, but in parallel:
* all of them are shut down before any is awaited, so the total wait is bounded by one graceful and one forceful
* timeout instead of one of each per channel.
* Shut down the given channels using the given timeouts.
* It first tries {@link ManagedChannel#shutdown()} and then {@link ManagedChannel#shutdownNow()}.
*
* @param gracefulTimeout the wait time after {@link ManagedChannel#shutdown()}; non-positive skips this step.
* @param forcefulTimeout the wait time after {@link ManagedChannel#shutdownNow()}.
*/
static void shutdownManagedChannels(Collection<ManagedChannel> channels) {
// Close the gRPC managed-channels if not shut down already.
final List<ManagedChannel> open = channels.stream().filter(c -> !c.isShutdown()).collect(Collectors.toList());
open.forEach(ManagedChannel::shutdown);
awaitTermination(open, GrpcConfigKeys.Client.CHANNEL_SHUTDOWN_GRACEFUL_TIMEOUT_DEFAULT, "gracefully");

// Forceful shut down if still not terminated.
final List<ManagedChannel> remaining = channels.stream().filter(c -> !c.isTerminated())
.collect(Collectors.toList());
remaining.forEach(ManagedChannel::shutdownNow);
awaitTermination(remaining, GrpcConfigKeys.Client.CHANNEL_SHUTDOWN_FORCEFUL_TIMEOUT_DEFAULT, "forcefully");
static void shutdownManagedChannels(Collection<ManagedChannel> channels,
TimeDuration gracefulTimeout, TimeDuration forcefulTimeout) {
if (gracefulTimeout.isPositive()) {
// Graceful shutdown
acceptEach(channels, ManagedChannel::shutdown, "shutdown", LOG);
awaitTermination(channels, gracefulTimeout, "shutdown");
}

// Forceful shutdown
acceptEach(channels, ManagedChannel::shutdownNow, "shutdownNow", LOG);
awaitTermination(channels, forcefulTimeout, "shutdownNow");
}

/** Wait for the given channels to terminate, up to the given timeout for all of them together. */
static void awaitTermination(List<ManagedChannel> channels, TimeDuration timeout, String mode) {
static void awaitTermination(Collection<ManagedChannel> channels, TimeDuration timeout, String op) {
if (Thread.currentThread().isInterrupted()) {
return;
}
final long deadline = System.nanoTime() + timeout.toLong(TimeUnit.NANOSECONDS);
for (ManagedChannel channel : channels) {
try {
if (!channel.awaitTermination(Math.max(0, deadline - System.nanoTime()), TimeUnit.NANOSECONDS)) {
LOG.warn("Timed out {} shutting down connection: {}. ", mode, channel);
LOG.warn("Timed out ({}) waiting for {} of channel: {}", timeout, op, channel);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
} catch (Exception e) {
LOG.error("Unexpected exception while waiting for channel termination", e);
LOG.error("Failed to awaitTermination for channel {}", channel, e);
}
}
}

static SslContextBuilder initSslContextBuilderForServer(KeyManagerConf keyManagerConfig) {

@amaliujia amaliujia Nov 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We drop these code because they are not used or we have better replacement?

@szetszwo szetszwo Nov 7, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These code are duplicated in NettyUtils.

final KeyManager keyManager = keyManagerConfig.getKeyManager();
if (keyManager != null) {
return SslContextBuilder.forServer(keyManager);
}
final PrivateKeyConf privateKey = keyManagerConfig.getPrivateKey();
final CertificatesConf certificates = keyManagerConfig.getKeyCertificates();

if (keyManagerConfig.isFileBased()) {
return SslContextBuilder.forServer(certificates.getFile(), privateKey.getFile());
} else {
return SslContextBuilder.forServer(privateKey.get(), certificates.get());
}
}

static void setTrustManager(SslContextBuilder b, TrustManagerConf trustManagerConfig) {
if (trustManagerConfig == null) {
return;
}
final TrustManager trustManager = trustManagerConfig.getTrustManager();
if (trustManager != null) {
b.trustManager(trustManager);
return;
}
final CertificatesConf certificates = trustManagerConfig.getTrustCertificates();
if (certificates.isFileBased()) {
b.trustManager(certificates.getFile());
} else {
b.trustManager(certificates.get());
}
}

static void setKeyManager(SslContextBuilder b, KeyManagerConf keyManagerConfig) {
if (keyManagerConfig == null) {
return;
}
final KeyManager keyManager = keyManagerConfig.getKeyManager();
if (keyManager != null) {
b.keyManager(keyManager);
return;
}
final PrivateKeyConf privateKey = keyManagerConfig.getPrivateKey();
final CertificatesConf certificates = keyManagerConfig.getKeyCertificates();

if (keyManagerConfig.isFileBased()) {
b.keyManager(certificates.getFile(), privateKey.getFile());
} else {
b.keyManager(privateKey.get(), certificates.get());
}
}

static SslContextBuilder configureSslContextBuilder(
SslContextBuilder b, TlsConf tlsConf, SslProvider defaultSslProvider) {
final Provider jsseProvider = getJsseProvider(tlsConf);
Expand Down
Loading