bitflicker64 commented on code in PR #3130:
URL: https://github.com/apache/hugegraph/pull/3130#discussion_r3769711073
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -58,31 +114,21 @@ public AbstractGrpcClient() {
}
public ManagedChannel[] getChannels(String target) {
- ManagedChannel[] tc;
- if ((tc = channels.get(target)) == null) {
- synchronized (channels) {
- if ((tc = channels.get(target)) == null) {
- try {
- ManagedChannel[] value = new
ManagedChannel[concurrency];
- CountDownLatch latch = new CountDownLatch(concurrency);
- for (int i = 0; i < concurrency; i++) {
- int fi = i;
- executor.execute(() -> {
- try {
- value[fi] = createChannel(target);
- } catch (Exception e) {
- throw new RuntimeException(e);
- } finally {
- latch.countDown();
- }
- });
- }
- latch.await();
- channels.put(target, tc = value);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
+ CompletableFuture<Void> refresh = this.triggerChannelRefresh(target);
+ ManagedChannel[] tc = channels.get(target);
+ if (tc != null) {
+ return tc;
+ }
+
+ /*
+ * Only the very first pool for a target waits, and only for a bounded
time: building it
+ * before its address is known makes the resolution that lands next
rebuild it. Waiting
+ * avoids that in the common case; if the wait expires the rebuild
still happens.
+ */
+ this.awaitInitialResolution(refresh);
+ synchronized (channels) {
+ if ((tc = channels.get(target)) == null) {
+ channels.put(target, tc = this.createChannels(target));
Review Comment:
Fixed in commit 16851352 and hardened in 9bacaad4. First-pool creation is
submitted to the dedicated prestarted channel-initialization executor, so a
cold Gremlin worker no longer creates channel workers on its restricted stack.
testColdTargetCreationStaysOffDeniedCaller starts with no pool, installs the
denying SecurityManager, invokes the public blocking-stub path, and verifies
the caller completes with a live stub while every channel is created
off-thread. The final focused suite is 29/29 with zero failures/errors/skips.
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -169,6 +232,316 @@ protected AbstractStub setStubOption(AbstractStub value) {
config.getGrpcMaxOutboundMessageSize());
}
+ private static boolean usesChannels(HgPair<ManagedChannel, ?>[] pairs,
+ ManagedChannel[] channels) {
+ if (pairs == null || pairs.length != channels.length) {
+ return false;
+ }
+ for (int i = 0; i < pairs.length; i++) {
+ HgPair<ManagedChannel, ?> pair = pairs[i];
+ if (pair == null || pair.getKey() != channels[i]) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Submits a refresh for the target unless one is already in flight or the
refresh interval
+ * has not elapsed. Returns the in-flight refresh, or null when none is
running.
+ */
+ private CompletableFuture<Void> triggerChannelRefresh(String target) {
+ CompletableFuture<Void> inFlight = refreshTasks.get(target);
+ if (inFlight != null) {
+ return inFlight;
+ }
+ if (!this.shouldRefreshChannels(target)) {
+ return null;
+ }
+
+ CompletableFuture<Void> refresh = new CompletableFuture<>();
+ CompletableFuture<Void> running = refreshTasks.putIfAbsent(target,
refresh);
+ if (running != null) {
+ return running;
+ }
+
+ // Throttle before submitting, so that a failing resolver cannot be
retried in a loop.
+ this.postponeNextRefresh(target);
+ try {
+ this.submitChannelRefresh(() -> {
+ try {
+ this.refreshChannelsIfAddressChanged(target);
+ } catch (Throwable e) {
+ // The executor discards what a task throws, so report it
here.
+ log.warn("Failed to refresh channels of target {}",
target, e);
+ } finally {
+ this.completeRefresh(target, refresh);
+ }
+ });
+ } catch (Throwable e) {
+ // Includes a thread creation denied on this thread; never leave
the entry behind.
+ log.warn("Failed to submit a channel refresh for target {}",
target, e);
+ this.completeRefresh(target, refresh);
+ }
+ return refresh;
+ }
+
+ private void completeRefresh(String target, CompletableFuture<Void>
refresh) {
+ /*
+ * Throttle from completion as well as from submission: a resolver
that is slow rather
+ * than failing can outlast its own interval, which would let every
later call queue
+ * another lookup behind it.
+ */
+ this.postponeNextRefresh(target);
+ refreshTasks.remove(target, refresh);
+ refresh.complete(null);
+ }
+
+ void submitChannelRefresh(Runnable task) {
+ CHANNEL_MAINTENANCE_EXECUTOR.execute(task);
+ }
+
+ private void awaitInitialResolution(CompletableFuture<Void> refresh) {
+ if (refresh == null) {
+ return;
+ }
+ try {
+ refresh.get(Math.max(0L, this.initialResolutionTimeoutNanos()),
+ TimeUnit.NANOSECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (Exception ignored) {
+ // A slow or failing resolver must not delay the first pool any
further.
+ }
+ }
+
+ /**
+ * Runs on a maintenance thread, never on a request thread. At most one
runs per target at a
+ * time — that comes from the refreshTasks entry, not from the size of the
executor. Replaces
+ * the target's pool when its resolved address set has changed, publishing
the replacement
+ * before retiring the previous pool.
+ */
+ private void refreshChannelsIfAddressChanged(String target) {
+ String resolvedTarget = this.resolveTarget(target);
+ if (resolvedTarget.isEmpty()) {
+ return;
+ }
+
+ ManagedChannel[] staleChannels = channels.get(target);
+ String previousTarget = resolvedTargets.get(target);
+ if (resolvedTarget.equals(previousTarget)) {
+ return;
+ }
+ if (staleChannels == null) {
+ /*
+ * Nothing to replace yet. Recording the address here is what lets
the common path
+ * build its first pool already knowing the address, instead of
rebuilding it.
+ */
+ resolvedTargets.put(target, resolvedTarget);
+ return;
+ }
+
+ ManagedChannel[] replacementChannels;
+ try {
+ replacementChannels = this.createChannels(target);
+ } catch (RuntimeException e) {
+ // Keep serving from the last healthy pool.
+ log.warn("Failed to create replacement channels of target {}, " +
+ "keeping the current pool", target, e);
+ return;
+ }
+
+ ReentrantReadWriteLock.WriteLock writeLock =
channelLock(target).writeLock();
+ writeLock.lock();
+ try {
+ boolean replaced = false;
+ synchronized (channels) {
+ if (channels.get(target) == staleChannels) {
+ channels.put(target, replacementChannels);
+ resolvedTargets.put(target, resolvedTarget);
+ replaced = true;
+ }
+ }
+ if (replaced) {
+ log.info("Replaced the channel pool of target {}, address
changed from {} to {}",
+ target, previousTarget, resolvedTarget);
+ }
+ this.retireChannels(replaced ? staleChannels :
replacementChannels);
+ } finally {
+ writeLock.unlock();
+ }
+ }
+
+ private boolean shouldRefreshChannels(String target) {
+ AtomicReference<Long> nextResolution =
+ nextResolutions.computeIfAbsent(target, key -> new
AtomicReference<>());
+ Long deadline = nextResolution.get();
+ return deadline == null || this.nanoTime() - deadline >= 0L;
+ }
+
+ private void postponeNextRefresh(String target) {
+ long interval = Math.max(0L, this.channelRefreshIntervalNanos());
+ nextResolutions.computeIfAbsent(target, key -> new AtomicReference<>())
+ .set(this.nanoTime() + interval);
+ }
+
+ protected long nanoTime() {
+ return System.nanoTime();
+ }
+
+ protected long channelRefreshIntervalNanos() {
+ return DEFAULT_CHANNEL_REFRESH_INTERVAL_NANOS;
+ }
+
+ private long initialResolutionTimeoutNanos() {
+ return DEFAULT_INITIAL_RESOLUTION_TIMEOUT_NANOS;
+ }
+
+ protected long channelDrainTimeoutNanos() {
+ return TimeUnit.SECONDS.toNanos(config.getGrpcTimeoutSeconds());
+ }
+
+ private ManagedChannel[] createChannels(String target) {
+ ManagedChannel[] value = new ManagedChannel[concurrency];
+ CountDownLatch latch = new CountDownLatch(concurrency);
+ AtomicReference<RuntimeException> failure = new AtomicReference<>();
+ for (int i = 0; i < concurrency; i++) {
+ int fi = i;
+ try {
+ this.submitChannelCreation(() -> {
+ try {
+ value[fi] = createChannel(target);
+ } catch (Exception e) {
+ failure.compareAndSet(null, new RuntimeException(e));
+ } finally {
+ latch.countDown();
+ }
+ });
+ } catch (RuntimeException e) {
+ failure.compareAndSet(null, e);
+ latch.countDown();
+ }
+ }
+
+ InterruptedException interruption = null;
+ while (latch.getCount() > 0L) {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ interruption = e;
+ }
+ }
+
+ if (failure.get() != null || interruption != null) {
+ forceTerminateChannels(value);
+ }
+ if (interruption != null) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(interruption);
+ }
+ if (failure.get() != null) {
+ throw failure.get();
+ }
+ return value;
+ }
+
+ void submitChannelCreation(Runnable task) {
+ this.executor.execute(task);
+ }
+
+ private void retireChannels(ManagedChannel[] retiredChannels) {
+ Arrays.stream(retiredChannels)
+ .filter(channel -> channel != null && !channel.isShutdown())
+ .forEach(ManagedChannel::shutdown);
+
+ long timeout = Math.max(0L, this.channelDrainTimeoutNanos());
+ try {
+ this.scheduleChannelRetirement(() ->
forceTerminateChannels(retiredChannels), timeout);
+ } catch (RuntimeException e) {
+ log.warn("Failed to schedule forced retirement, forcing channels
immediately", e);
+ forceTerminateChannels(retiredChannels);
+ }
+ }
+
+ void scheduleChannelRetirement(Runnable task, long timeoutNanos) {
+ CHANNEL_RETIREMENT_EXECUTOR.schedule(task, timeoutNanos,
TimeUnit.NANOSECONDS);
+ }
+
+ private void forceTerminateChannels(ManagedChannel[] retiredChannels) {
+ for (ManagedChannel channel : retiredChannels) {
+ if (channel != null && !channel.isTerminated()) {
+ channel.shutdownNow();
+ }
+ }
+ }
+
+ /**
+ * Extracts the host that a gRPC target resolves through, covering the
plain {@code host:port}
+ * form and the {@code dns:} scheme in both its {@code dns:host:port} and
+ * {@code dns://authority/host:port} spellings. Any other resolver scheme
returns an empty
+ * host, leaving that target to gRPC instead of monitoring the wrong
endpoint.
+ */
+ private static String targetHost(String target) {
+ if (target == null || target.isEmpty()) {
+ return "";
+ }
+
+ String endpoint = target;
+ if (target.regionMatches(true, 0, DNS_SCHEME, 0, DNS_SCHEME.length()))
{
Review Comment:
Fixed in 9bacaad4. targetHost now accepts only lowercase dns targets with a
slash-prefixed URI path, matching gRPC 1.39; raw host:port and bracketed IPv6
remain supported, while scheme-only dns:host:port and uppercase DNS forms are
skipped. The target parser test covers the accepted and rejected forms.
##########
hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClientTest.java:
##########
@@ -40,70 +57,1039 @@
import io.grpc.MethodDescriptor;
import io.grpc.stub.AbstractAsyncStub;
import io.grpc.stub.AbstractBlockingStub;
+import io.grpc.stub.AbstractStub;
/**
- * Verifies that the stub pools of {@link AbstractGrpcClient} spread their
entries over every
- * channel created for a target, instead of binding all of them to a single
channel.
+ * Verifies that Store address changes replace channels and cached stubs
safely, and that the
+ * refresh never resolves or creates channels on the thread that asked for a
stub.
*/
public class AbstractGrpcClientTest {
+ private static final String MAINTENANCE_THREAD_PREFIX =
"channel-maintenance";
+ private static final String RETIREMENT_THREAD_PREFIX =
"channel-retirement";
private static final AtomicInteger TARGET_SEQ = new AtomicInteger();
private static String uniqueTarget(String prefix) {
return prefix + "-" + TARGET_SEQ.incrementAndGet() + ":8500";
}
- private static Set<ManagedChannel> identitySet(Collection<ManagedChannel>
channels) {
- Set<ManagedChannel> set = Collections.newSetFromMap(new
IdentityHashMap<>());
- set.addAll(channels);
- return set;
+ private static boolean belongsToPool(Channel channel, ManagedChannel[]
channels) {
+ return Arrays.stream(channels).anyMatch(current -> current == channel);
+ }
+
+ private static boolean allChannelsAreShutdown(ManagedChannel[] channels) {
+ return Arrays.stream(channels).allMatch(ManagedChannel::isShutdown);
+ }
+
+ private static boolean allChannelsAreLive(ManagedChannel[] channels) {
+ return Arrays.stream(channels).noneMatch(ManagedChannel::isShutdown);
+ }
+
+ private static List<FakeManagedChannel> fakeChannels(ManagedChannel[]
channels) {
+ return Arrays.stream(channels)
+ .map(channel -> (FakeManagedChannel) channel)
+ .collect(Collectors.toList());
+ }
+
+ private static void assertUsesEveryChannel(String message,
+ List<ManagedChannel>
stubChannels,
+ ManagedChannel[] channels) {
+ assertEquals(message, channels.length, new
HashSet<>(stubChannels).size());
+ }
+
+ private static void assertCachedChannelsCurrentAndLive(String message,
+
List<ManagedChannel> cached,
+ ManagedChannel[]
current) {
+ assertEquals(message, current.length, cached.size());
+ assertTrue(message, cached.stream().allMatch(channel ->
+ belongsToPool(channel, current) && !channel.isShutdown()));
+ }
+
+ private static void awaitCondition(String message, Condition condition)
throws Exception {
+ awaitCondition(message, 5L, TimeUnit.SECONDS, condition);
+ }
+
+ private static void awaitCondition(String message, long timeout, TimeUnit
unit,
+ Condition condition) throws Exception {
+ long start = System.nanoTime();
+ long timeoutNanos = unit.toNanos(timeout);
+ while (System.nanoTime() - start < timeoutNanos) {
+ if (condition.isTrue()) {
+ return;
+ }
+ Thread.sleep(10L);
+ }
+ assertTrue(message, condition.isTrue());
+ }
+
+ /**
+ * Refresh runs on the maintenance thread, so a replacement pool becomes
visible some time
+ * after the address changes rather than on the call that observes the
change. Retirement
+ * deliberately trails publication, so waiting for both is what marks a
refresh complete.
+ */
+ private static ManagedChannel[] awaitPoolReplacement(RecordingGrpcClient
client,
+ String target,
+ ManagedChannel[]
staleChannels)
+ throws Exception {
+ awaitCondition("refresh must publish a replacement pool",
+ () -> client.getChannels(target) != staleChannels);
+ awaitCondition("refresh must retire the previous pool after
publishing",
+ () -> allChannelsAreShutdown(staleChannels));
+ return client.getChannels(target);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static boolean refreshIsIdle(AbstractGrpcClient client, String
target)
+ throws Exception {
+ Field field =
AbstractGrpcClient.class.getDeclaredField("refreshTasks");
+ field.setAccessible(true);
+ return !((Map<String, ?>) field.get(client)).containsKey(target);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static List<ManagedChannel>
cachedAsyncStubChannels(AbstractGrpcClient client,
+ String target)
+ throws Exception {
+ Field field = AbstractGrpcClient.class.getDeclaredField("asyncStubs");
+ field.setAccessible(true);
+ Map<String, HgPair<ManagedChannel, AbstractAsyncStub>[]> stubs =
+ (Map<String, HgPair<ManagedChannel, AbstractAsyncStub>[]>)
field.get(client);
+ HgPair<ManagedChannel, AbstractAsyncStub>[] pairs = stubs.get(target);
+ assertNotNull("the async stub cache must exist", pairs);
+ return
Arrays.stream(pairs).map(HgPair::getKey).collect(Collectors.toList());
+ }
+
+ private static ThreadPoolExecutor
channelCreationExecutor(AbstractGrpcClient client)
+ throws Exception {
+ Field field = AbstractGrpcClient.class.getDeclaredField("executor");
+ field.setAccessible(true);
+ return (ThreadPoolExecutor) field.get(client);
}
@Test
- public void testBlockingStubPoolCoversEveryChannel() {
- String target = uniqueTarget("blocking");
+ public void testAddressChangeReplacesChannelAndStubPools() throws
Exception {
+ String target = uniqueTarget("address-change");
RecordingGrpcClient client = new RecordingGrpcClient();
- ManagedChannel[] channels = client.getChannels(target);
- assertTrue("pool must hold more than one channel", channels.length >
1);
+ ManagedChannel[] oldChannels = client.getChannels(target);
+ assertNotNull(client.getBlockingStub(target));
+ assertNotNull(client.getAsyncStub(target));
+
+ client.resolvedTarget = "10.0.0.2";
+ ManagedChannel[] newChannels = awaitPoolReplacement(client, target,
oldChannels);
+ assertTrue("every stale channel must be gracefully shut down",
+ allChannelsAreShutdown(oldChannels));
+ assertFalse("refresh must not force close stale channels immediately",
+ fakeChannels(oldChannels).stream()
+
.anyMatch(FakeManagedChannel::isForceShutdown));
- // Pool initialisation: one stub per channel, each bound to a
different channel.
+ client.blockingStubChannels.clear();
+ client.asyncStubChannels.clear();
assertNotNull(client.getBlockingStub(target));
- assertEquals("one stub per channel", channels.length,
client.blockingStubChannels.size());
- Set<ManagedChannel> bound = identitySet(client.blockingStubChannels);
- assertEquals("stubs must not share a channel", channels.length,
bound.size());
- assertTrue("stubs must cover the channels of the target",
- bound.containsAll(Arrays.asList(channels)));
+ assertNotNull(client.getAsyncStub(target));
+ assertCachedChannelsCurrentAndLive("the blocking stub pool must be
rebuilt on the new pool",
+ client.blockingStubChannels,
newChannels);
+ assertUsesEveryChannel("blocking stubs must be spread across the pool",
+ client.blockingStubChannels, newChannels);
+ assertCachedChannelsCurrentAndLive("the async stub pool must be
rebuilt on the new pool",
+ client.asyncStubChannels,
newChannels);
+ assertUsesEveryChannel("async stubs must be spread across the pool",
+ client.asyncStubChannels, newChannels);
}
@Test
- public void testAsyncStubPoolCoversEveryChannel() {
- String target = uniqueTarget("async");
+ public void testStubPoolsCoverEveryChannelOnFirstBuild() {
+ String target = uniqueTarget("initial-stub-spread");
RecordingGrpcClient client = new RecordingGrpcClient();
ManagedChannel[] channels = client.getChannels(target);
- assertTrue("pool must hold more than one channel", channels.length >
1);
+ assertNotNull(client.getBlockingStub(target));
+ assertNotNull(client.getAsyncStub(target));
+ assertUsesEveryChannel("the first blocking stub pool must cover every
channel",
+ client.blockingStubChannels, channels);
+ assertUsesEveryChannel("the first async stub pool must cover every
channel",
+ client.asyncStubChannels, channels);
+ }
+
+ @Test
+ public void testFirstPoolIsBuiltAfterItsAddressIsKnown() throws Exception {
+ String target = uniqueTarget("initial-resolution");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ ManagedChannel[] initialChannels = client.getChannels(target);
+
+ assertEquals("the first pool must be built once its address is known",
+ 1, client.resolutionCount.get());
+ // Let every refresh settle first, otherwise the assertions below race
the swap.
+ awaitCondition("the refresh must settle", () -> refreshIsIdle(client,
target));
+ assertSame("a pool built with a known address must not be replaced",
+ initialChannels, client.getChannels(target));
+ awaitCondition("the settled refresh must leave no further work",
+ () -> refreshIsIdle(client, target));
+ assertTrue("the first pool must not be retired by its own resolution",
+ allChannelsAreLive(initialChannels));
+ }
+
+ @Test
+ public void testInitialRefreshRunsWhenNanoTimeIsNegative() throws
Exception {
+ String target = uniqueTarget("negative-nano-time");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ client.currentNanoTime = -1L;
+ client.refreshIntervalNanos = 5L;
+
+ ManagedChannel[] initialChannels = client.getChannels(target);
+ awaitCondition("the initial refresh must settle", () ->
refreshIsIdle(client, target));
+
+ assertEquals("a negative clock value must not suppress the initial
refresh",
+ 1, client.resolutionCount.get());
+ assertTrue("the initial pool must remain healthy",
allChannelsAreLive(initialChannels));
+ }
+
+ @Test
+ public void testRefreshDeadlineSurvivesNanoTimeWraparound() throws
Exception {
+ String target = uniqueTarget("wrapped-nano-time");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ client.currentNanoTime = Long.MAX_VALUE - 2L;
+ client.refreshIntervalNanos = 5L;
+
+ client.getChannels(target);
+ awaitCondition("the initial refresh must settle", () ->
refreshIsIdle(client, target));
+ assertEquals(1, client.resolutionCount.get());
+
+ client.currentNanoTime = Long.MAX_VALUE - 1L;
+ client.getChannels(target);
+ assertEquals("the wrapped deadline must not fire early", 1,
+ client.resolutionCount.get());
+
+ client.currentNanoTime = Long.MIN_VALUE + 2L;
+ client.getChannels(target);
+ awaitCondition("the wrapped deadline must eventually fire",
+ () -> client.resolutionCount.get() == 2 &&
+ refreshIsIdle(client, target));
+ }
+
+ @Test
+ public void testUnknownAddressPoolIsReplacedOnFirstSuccessfulResolution()
throws Exception {
+ String target = uniqueTarget("first-successful-resolution");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ client.resolvedTarget = "";
+ ManagedChannel[] unknownChannels = client.getChannels(target);
+
+ client.resolvedTarget = "10.0.0.1";
+ ManagedChannel[] resolvedChannels = awaitPoolReplacement(client,
target, unknownChannels);
+ assertTrue("every channel from the unknown pool must be gracefully
shut down",
+ allChannelsAreShutdown(unknownChannels));
+ assertFalse("unknown channels must not be force closed immediately",
+ fakeChannels(unknownChannels).stream()
+
.anyMatch(FakeManagedChannel::isForceShutdown));
+ assertTrue("the resolved channel pool must remain live",
+ allChannelsAreLive(resolvedChannels));
+ }
+
+ /**
+ * HugeSecurityManager denies socket connection and thread creation on
Gremlin worker stacks,
+ * and InetAddress.getAllByName() performs exactly the checkConnect(host,
-1) simulated here.
+ * A refresh triggered by such a caller must therefore resolve somewhere
else entirely.
+ */
+ @Test
+ public void testRefreshSucceedsWhenCallerIsDeniedSocketAndThreadAccess()
throws Exception {
+ String target = uniqueTarget("denied-caller");
+ HostCapturingGrpcClient client = new HostCapturingGrpcClient();
+ client.checkSocketPermission = true;
+ client.activeCallsFinished = new CountDownLatch(1);
+ client.drainTimeoutNanos = 0L;
+ ManagedChannel[] oldChannels = client.getChannels(target);
+ assertNotNull(client.getBlockingStub(target));
+ awaitCondition("the initial refresh must settle before installing the
security manager",
+ () -> refreshIsIdle(client, target));
+ client.resolutionThreads.clear();
+ client.creationThreads.clear();
+ client.retirementThreads.clear();
+
+ SecurityManager previous = System.getSecurityManager();
+ System.setSecurityManager(new DenyingWorkerSecurityManager());
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ AtomicReference<AbstractBlockingStub> stub = new AtomicReference<>();
+ try {
+ client.resolvedAddress = "10.0.0.2";
+ // The name is what HugeSecurityManager keys its Gremlin worker
check on.
+ Thread worker = new Thread(() -> {
+ try {
+ stub.set(client.getBlockingStub(target));
+ } catch (Throwable e) {
+ failure.set(e);
+ }
+ }, "gremlin-server-exec-1");
+ worker.start();
+ worker.join(TimeUnit.SECONDS.toMillis(10L));
+
+ assertFalse("the denied caller must finish", worker.isAlive());
+ assertNotNull("a denied caller must still receive a stub",
stub.get());
+ assertTrue("a denied caller must not observe a security failure: "
+ failure.get(),
+ failure.get() == null);
+ awaitCondition("the refresh must still publish a replacement pool",
+ () -> AbstractGrpcClient.channels.get(target) !=
oldChannels);
+ awaitCondition("forced retirement must finish on the maintenance
executor",
+ () -> fakeChannels(oldChannels).stream()
+
.allMatch(FakeManagedChannel::isTerminated));
+ assertFalse("the assertion below is vacuous unless something
resolved",
+ client.resolutionThreads.isEmpty());
+ assertTrue("every resolution must run on the channel maintenance
thread",
+ client.resolutionThreads.stream()
+ .allMatch(name ->
name.startsWith(
+
MAINTENANCE_THREAD_PREFIX)));
+ assertEquals("the replacement must include every channel",
+ AbstractGrpcClient.concurrency,
client.creationThreads.size());
+ assertTrue("replacement construction must stay off the denied
caller",
+ client.creationThreads.stream().noneMatch(
+ name ->
name.startsWith("gremlin-server-exec")));
+ assertFalse("the retirement assertion must observe lifecycle work",
+ client.retirementThreads.isEmpty());
+ assertTrue("graceful retirement must run on the maintenance
thread",
+ client.retirementThreads.stream().anyMatch(
+ name ->
name.startsWith(MAINTENANCE_THREAD_PREFIX)));
+ assertTrue("forced retirement must run on the retirement thread",
+ client.retirementThreads.stream().anyMatch(
+ name ->
name.startsWith(RETIREMENT_THREAD_PREFIX)));
+ assertTrue("retirement must stay off the denied caller",
+ client.retirementThreads.stream().noneMatch(
+ name ->
name.startsWith("gremlin-server-exec")));
+ } finally {
+ client.activeCallsFinished.countDown();
+ System.setSecurityManager(previous);
+ }
+ }
+
+ @Test
+ public void testResolutionFailureKeepsHealthyPoolAndAllowsRetry() throws
Exception {
+ String target = uniqueTarget("resolution-failure");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ ManagedChannel[] healthyChannels = client.getChannels(target);
+ awaitCondition("the initial refresh must settle", () ->
refreshIsIdle(client, target));
+
+ client.resolutionFailure = new IllegalStateException("injected
resolution failure");
+ client.getChannels(target);
+ awaitCondition("the failed refresh must clear its single-flight entry",
+ () -> refreshIsIdle(client, target));
+ assertSame("a resolution failure must preserve the published pool",
healthyChannels,
+ AbstractGrpcClient.channels.get(target));
+ assertTrue("a resolution failure must leave the published pool live",
+ allChannelsAreLive(healthyChannels));
+
+ client.resolutionFailure = null;
+ client.resolvedTarget = "10.0.0.2";
+ ManagedChannel[] replacement = awaitPoolReplacement(client, target,
healthyChannels);
+ assertTrue("a later successful refresh must replace the pool",
+ allChannelsAreLive(replacement));
+ }
+
+ @Test
+ public void testRejectedRefreshSubmissionKeepsHealthyPoolAndAllowsRetry()
throws Exception {
+ String target = uniqueTarget("refresh-submission-failure");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ ManagedChannel[] healthyChannels = client.getChannels(target);
+ awaitCondition("the initial refresh must settle", () ->
refreshIsIdle(client, target));
+
+ client.rejectNextRefreshSubmission.set(true);
+ client.resolvedTarget = "10.0.0.2";
+ client.getChannels(target);
+ assertTrue("a rejected refresh must clear its single-flight entry",
+ refreshIsIdle(client, target));
+ assertSame("a rejected refresh must preserve the published pool",
healthyChannels,
+ AbstractGrpcClient.channels.get(target));
+ assertTrue("a rejected refresh must leave the published pool live",
+ allChannelsAreLive(healthyChannels));
+
+ ManagedChannel[] replacement = awaitPoolReplacement(client, target,
healthyChannels);
+ assertTrue("a later refresh submission must still succeed",
+ allChannelsAreLive(replacement));
+ }
+
+ @Test
+ public void testReplacementCreationFailureKeepsHealthyPoolAndAllowsRetry()
throws Exception {
+ String target = uniqueTarget("replacement-creation-failure");
+ CreationControlGrpcClient client = new CreationControlGrpcClient();
+ ManagedChannel[] healthyChannels = client.getChannels(target);
+ awaitCondition("the initial refresh must settle", () ->
refreshIsIdle(client, target));
+ int createdBeforeFailure = client.createdChannels.size();
+
+ client.failedAttempt = client.attempt.get() + 5;
+ client.resolvedTarget = "10.0.0.2";
+ client.getChannels(target);
+ awaitCondition("the failed replacement must clear its single-flight
entry",
+ () -> refreshIsIdle(client, target));
+
+ assertSame("a replacement failure must preserve the published pool",
healthyChannels,
+ AbstractGrpcClient.channels.get(target));
+ assertTrue("a replacement failure must leave the published pool live",
+ allChannelsAreLive(healthyChannels));
+ List<ManagedChannel> partialChannels =
+ client.createdChannels.subList(createdBeforeFailure,
+ client.createdChannels.size());
+ assertEquals("all successful replacement tasks must converge before
failure returns",
+ AbstractGrpcClient.concurrency - 1,
partialChannels.size());
+ assertTrue("every partial replacement channel must be force
terminated",
+ partialChannels.stream().allMatch(channel ->
+ channel.isTerminated() &&
+ ((FakeManagedChannel) channel).isForceShutdown()));
+
+ client.failedAttempt = -1;
+ ManagedChannel[] replacement = awaitPoolReplacement(client, target,
healthyChannels);
+ assertTrue("a later replacement attempt must still succeed",
+ allChannelsAreLive(replacement));
+ }
+
+ @Test
+ public void testRejectedChannelCreationSubmissionRetiresPartialPool()
throws Exception {
+ String target = uniqueTarget("channel-submission-failure");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ ManagedChannel[] healthyChannels = client.getChannels(target);
+ awaitCondition("the initial refresh must settle", () ->
refreshIsIdle(client, target));
+ int createdBeforeFailure = client.createdChannels.size();
+
+ client.rejectNextChannelSubmission.set(true);
+ client.resolvedTarget = "10.0.0.2";
+ client.getChannels(target);
+ awaitCondition("the failed replacement must clear its single-flight
entry",
+ () -> refreshIsIdle(client, target));
+
+ assertSame("a task-submission failure must preserve the published
pool", healthyChannels,
+ AbstractGrpcClient.channels.get(target));
+ List<ManagedChannel> partialChannels =
+ client.createdChannels.subList(createdBeforeFailure,
+ client.createdChannels.size());
+ assertEquals("one rejected task must not prevent the others from
converging",
+ AbstractGrpcClient.concurrency - 1,
partialChannels.size());
+ assertTrue("partial channels must be retired after a task-submission
failure",
+ partialChannels.stream().allMatch(channel ->
+ channel.isTerminated() &&
+ ((FakeManagedChannel) channel).isForceShutdown()));
+
+ ManagedChannel[] replacement = awaitPoolReplacement(client, target,
healthyChannels);
+ assertTrue("a later replacement attempt must still succeed",
+ allChannelsAreLive(replacement));
+ }
+
+ @Test
+ public void
testRejectedRetirementSchedulingKeepsReplacementAndAllowsRetry()
+ throws Exception {
+ String target = uniqueTarget("retirement-submission-failure");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ client.activeCallsFinished = new CountDownLatch(1);
+ ManagedChannel[] firstChannels = client.getChannels(target);
+ awaitCondition("the initial refresh must settle", () ->
refreshIsIdle(client, target));
+
+ try {
+ client.rejectNextRetirementSubmission.set(true);
+ client.resolvedTarget = "10.0.0.2";
+ client.getChannels(target);
+ awaitCondition("the replacement must be published despite cleanup
rejection",
+ () -> AbstractGrpcClient.channels.get(target) !=
firstChannels);
+ awaitCondition("the cleanup rejection must clear its single-flight
entry",
+ () -> refreshIsIdle(client, target));
+ ManagedChannel[] secondChannels =
AbstractGrpcClient.channels.get(target);
+ assertTrue("the replacement must remain live after cleanup
rejection",
+ allChannelsAreLive(secondChannels));
+ assertTrue("the unscheduled pool must be force terminated
immediately",
+ fakeChannels(firstChannels).stream().allMatch(channel ->
+ channel.isTerminated() &&
channel.isForceShutdown()));
+
+ client.resolvedTarget = "10.0.0.3";
+ ManagedChannel[] thirdChannels = awaitPoolReplacement(client,
target, secondChannels);
+ assertTrue("a later refresh must still replace the pool",
+ allChannelsAreLive(thirdChannels));
+ } finally {
+ client.activeCallsFinished.countDown();
+ }
+ }
+
+ @Test
+ public void testRetirementDoesNotBlockWhenCreationExecutorIsSaturated()
throws Exception {
+ String target = uniqueTarget("saturated-retirement");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ client.activeCallsFinished = new CountDownLatch(1);
+ client.drainTimeoutNanos = TimeUnit.SECONDS.toNanos(5L);
+ ManagedChannel[] oldChannels = client.getChannels(target);
+ ThreadPoolExecutor channelExecutor = channelCreationExecutor(client);
+ CountDownLatch workersStarted = new
CountDownLatch(AbstractGrpcClient.concurrency);
+ CountDownLatch releaseWorkers = new CountDownLatch(1);
+
+ try {
+ for (int i = 0; i < AbstractGrpcClient.concurrency; i++) {
+ channelExecutor.execute(() -> {
+ workersStarted.countDown();
+ try {
+ releaseWorkers.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ }
+ assertTrue("the shared creation executor must be fully occupied",
+ workersStarted.await(5, TimeUnit.SECONDS));
+
+ client.resolvedTarget = "10.0.0.2";
+ ManagedChannel[] newChannels = awaitPoolReplacement(client,
target, oldChannels);
+
+ assertNotSame("refresh must publish the replacement pool",
oldChannels, newChannels);
+ assertTrue("the retired pool must be shut down once refresh
completes",
+ allChannelsAreShutdown(oldChannels));
+ assertFalse("the scheduled cleanup must preserve the drain window",
+ fakeChannels(oldChannels).stream()
+
.anyMatch(FakeManagedChannel::isForceShutdown));
+ } finally {
+ releaseWorkers.countDown();
+ client.activeCallsFinished.countDown();
+ }
+ }
+
+ @Test
+ public void testRetirementDeadlineSurvivesBlockedRefreshWorkers() throws
Exception {
+ String retiringTarget = uniqueTarget("isolated-retirement");
+ String firstBlockedTarget = uniqueTarget("blocked-refresh");
+ String secondBlockedTarget = uniqueTarget("blocked-refresh");
+ RecordingGrpcClient retiringClient = new RecordingGrpcClient();
+ RecordingGrpcClient firstBlockedClient = new RecordingGrpcClient();
+ RecordingGrpcClient secondBlockedClient = new RecordingGrpcClient();
+ ManagedChannel[] firstBlockedChannels =
firstBlockedClient.getChannels(firstBlockedTarget);
+ ManagedChannel[] secondBlockedChannels =
+ secondBlockedClient.getChannels(secondBlockedTarget);
+ awaitCondition("the first blocking client must initialize",
+ () -> refreshIsIdle(firstBlockedClient,
firstBlockedTarget));
+ awaitCondition("the second blocking client must initialize",
+ () -> refreshIsIdle(secondBlockedClient,
secondBlockedTarget));
+
+ retiringClient.activeCallsFinished = new CountDownLatch(1);
+ retiringClient.drainTimeoutNanos = TimeUnit.SECONDS.toNanos(2L);
+ ManagedChannel[] retiredChannels =
retiringClient.getChannels(retiringTarget);
+ awaitCondition("the retiring client must initialize",
+ () -> refreshIsIdle(retiringClient, retiringTarget));
+ retiringClient.resolvedTarget = "10.0.0.2";
+ ManagedChannel[] replacement =
+ awaitPoolReplacement(retiringClient, retiringTarget,
retiredChannels);
+ awaitCondition("the replacement refresh must settle",
+ () -> refreshIsIdle(retiringClient, retiringTarget));
+
+ firstBlockedClient.delayResolution = true;
+ firstBlockedClient.delayedResolutionTimeoutSeconds = 30L;
+ firstBlockedClient.resolvedTarget = "10.0.0.2";
+ secondBlockedClient.delayResolution = true;
+ secondBlockedClient.delayedResolutionTimeoutSeconds = 30L;
+ secondBlockedClient.resolvedTarget = "10.0.0.2";
+ try {
+ assertSame(firstBlockedChannels,
firstBlockedClient.getChannels(firstBlockedTarget));
+ assertTrue("the first maintenance worker must block in resolution",
+ firstBlockedClient.delayedResolutionStarted.await(5,
TimeUnit.SECONDS));
+ assertSame(secondBlockedChannels,
secondBlockedClient.getChannels(secondBlockedTarget));
+ assertTrue("the second maintenance worker must block in
resolution",
+ secondBlockedClient.delayedResolutionStarted.await(5,
TimeUnit.SECONDS));
+ assertFalse("the drain deadline must still be pending after
workers are blocked",
+ fakeChannels(retiredChannels).stream()
+
.anyMatch(FakeManagedChannel::isForceShutdown));
+
+ awaitCondition("blocked refresh workers must not delay forced
retirement",
+ 4L, TimeUnit.SECONDS,
+ () ->
fakeChannels(retiredChannels).stream().allMatch(channel ->
+ channel.isTerminated() &&
channel.isForceShutdown()));
+ assertTrue("the replacement pool must remain live",
allChannelsAreLive(replacement));
+ } finally {
+ firstBlockedClient.releaseDelayedResolution.countDown();
+ secondBlockedClient.releaseDelayedResolution.countDown();
+ retiringClient.activeCallsFinished.countDown();
+ awaitCondition("the first blocked refresh must settle",
+ () -> refreshIsIdle(firstBlockedClient,
firstBlockedTarget));
+ awaitCondition("the second blocked refresh must settle",
+ () -> refreshIsIdle(secondBlockedClient,
secondBlockedTarget));
+ }
+ }
+
+ @Test
+ public void testBlockedResolutionsDoNotStarveAnotherTarget() throws
Exception {
+ String firstTarget = uniqueTarget("isolated-refresh");
+ String secondTarget = uniqueTarget("isolated-refresh");
+ String thirdTarget = uniqueTarget("isolated-refresh");
+ RecordingGrpcClient first = new RecordingGrpcClient();
+ RecordingGrpcClient second = new RecordingGrpcClient();
+ RecordingGrpcClient third = new RecordingGrpcClient();
+ ManagedChannel[] firstChannels = first.getChannels(firstTarget);
+ ManagedChannel[] secondChannels = second.getChannels(secondTarget);
+ ManagedChannel[] thirdChannels = third.getChannels(thirdTarget);
+ awaitCondition("initial refreshes must settle", () ->
refreshIsIdle(first, firstTarget) &&
+
refreshIsIdle(second, secondTarget) &&
+
refreshIsIdle(third, thirdTarget));
+
+ first.delayResolution = true;
+ second.delayResolution = true;
+ first.delayedResolutionTimeoutSeconds = 30L;
+ second.delayedResolutionTimeoutSeconds = 30L;
+ first.resolvedTarget = "10.0.0.2";
+ second.resolvedTarget = "10.0.0.2";
+ third.resolvedTarget = "10.0.0.2";
+ try {
+ assertSame(firstChannels, first.getChannels(firstTarget));
+ assertTrue(first.delayedResolutionStarted.await(5,
TimeUnit.SECONDS));
+ assertSame(secondChannels, second.getChannels(secondTarget));
+ assertTrue(second.delayedResolutionStarted.await(5,
TimeUnit.SECONDS));
+
+ ManagedChannel[] replacement = awaitPoolReplacement(third,
thirdTarget, thirdChannels);
+ assertNotSame("the third target must refresh while two lookups are
blocked",
+ thirdChannels, replacement);
+ assertTrue("the third target replacement must remain live",
+ allChannelsAreLive(replacement));
+ } finally {
+ first.releaseDelayedResolution.countDown();
+ second.releaseDelayedResolution.countDown();
+ awaitCondition("blocked refreshes must settle",
+ () -> refreshIsIdle(first, firstTarget) &&
+ refreshIsIdle(second, secondTarget));
+ }
+ }
+
+ @Test
+ public void testPartialChannelsAreRetiredAfterMixedCreationFailure() {
+ String target = uniqueTarget("partial-creation-failure");
+ CreationControlGrpcClient client = new CreationControlGrpcClient();
+ client.failedAttempt = 5;
+
+ try {
+ client.getChannels(target);
+ fail("channel creation must propagate the injected failure");
+ } catch (RuntimeException ignored) {
+ // Expected.
+ }
+
+ assertEquals("all creation tasks must converge before failure is
returned",
+ AbstractGrpcClient.concurrency - 1,
client.createdChannels.size());
+ assertTrue("every partial channel must be force terminated before
failure returns",
+ client.createdChannels.stream().allMatch(channel ->
+ channel.isTerminated() &&
+ ((FakeManagedChannel) channel).isForceShutdown()));
+ }
+
+ @Test
+ public void testInterruptedCreationWaitsAndRetiresPartialChannels() throws
Exception {
+ String target = uniqueTarget("interrupted-creation");
+ CreationControlGrpcClient client = new CreationControlGrpcClient();
+ client.releaseCreation = new CountDownLatch(1);
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ AtomicBoolean interrupted = new AtomicBoolean();
+ Thread caller = new Thread(() -> {
+ try {
+ client.getChannels(target);
+ } catch (Throwable e) {
+ failure.set(e);
+ interrupted.set(Thread.currentThread().isInterrupted());
+ }
+ });
+
+ caller.start();
+ assertTrue("all channel creation tasks must start",
+ client.creationStarted.await(5, TimeUnit.SECONDS));
+ caller.interrupt();
+ try {
+ Thread.sleep(100L);
+ assertTrue("an interrupted caller must wait for creation tasks to
converge",
+ caller.isAlive());
+ } finally {
+ client.releaseCreation.countDown();
+ }
+ caller.join(TimeUnit.SECONDS.toMillis(5L));
+
+ assertFalse("the interrupted creation call must finish",
caller.isAlive());
+ assertTrue("interruption must be reported as a runtime failure",
+ failure.get() instanceof RuntimeException);
+ assertTrue("the caller interrupt status must be restored",
interrupted.get());
+ assertEquals("every creation task must finish before interruption is
reported",
+ AbstractGrpcClient.concurrency,
client.createdChannels.size());
+ assertTrue("all partial channels must be force terminated before
interruption returns",
+ client.createdChannels.stream().allMatch(channel ->
+ channel.isTerminated() &&
+ ((FakeManagedChannel) channel).isForceShutdown()));
+ }
+
+ @Test
+ public void testDrainDeadlineForceTerminatesRetiredChannels() throws
Exception {
+ String target = uniqueTarget("drain-deadline");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ client.activeCallsFinished = new CountDownLatch(1);
+ client.drainTimeoutNanos = 0L;
+ ManagedChannel[] oldChannels = client.getChannels(target);
+
+ client.resolvedTarget = "10.0.0.2";
+ ManagedChannel[] newChannels = awaitPoolReplacement(client, target,
oldChannels);
+
+ awaitCondition("expired drain deadline must force terminate the
retired pool",
+ () ->
fakeChannels(oldChannels).stream().allMatch(channel ->
+ channel.isTerminated() &&
channel.isForceShutdown()));
+ assertTrue("the replacement pool must remain live",
allChannelsAreLive(newChannels));
+ }
+
+ @Test
+ public void testStubAcquisitionReusesResolutionWithinRefreshInterval()
throws Exception {
+ String target = uniqueTarget("throttled-refresh");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ client.refreshIntervalNanos = TimeUnit.HOURS.toNanos(1L);
+
+ assertNotNull(client.getBlockingStub(target));
assertNotNull(client.getAsyncStub(target));
- assertEquals("one stub per channel", channels.length,
client.asyncStubChannels.size());
- Set<ManagedChannel> bound = identitySet(client.asyncStubChannels);
- assertEquals("stubs must not share a channel", channels.length,
bound.size());
- assertTrue("stubs must cover the channels of the target",
- bound.containsAll(Arrays.asList(channels)));
+ for (int i = 0; i < 10; i++) {
+ assertNotNull(client.getBlockingStub(target));
+ assertNotNull(client.getAsyncStub(target));
+ }
+
+ Thread.sleep(100L);
+ assertEquals("stub acquisition must not resolve again inside the
refresh interval",
+ 1, client.resolutionCount.get());
+ }
+
+ @Test
+ public void
testConcurrentStubAcquisitionRetainsHealthyPoolDuringDelayedRefresh()
+ throws Exception {
+ String target = uniqueTarget("delayed-refresh");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ ManagedChannel[] oldChannels = client.getChannels(target);
+ assertNotNull(client.getBlockingStub(target));
+ // Let any refresh already in flight settle, so the resolution count
below is stable.
+ awaitCondition("the initial refresh must settle before the count is
captured",
+ () -> refreshIsIdle(client, target));
+ int resolutionsBeforeConcurrentCalls = client.resolutionCount.get();
+
+ client.resolvedTarget = "10.0.0.2";
+ client.delayResolution = true;
+ client.refreshIntervalNanos = TimeUnit.HOURS.toNanos(1L);
+
+ ExecutorService executor = Executors.newFixedThreadPool(6);
+ List<Future<AbstractBlockingStub>> futures = new ArrayList<>();
+ try {
+ for (int i = 0; i < 6; i++) {
+ futures.add(executor.submit(() ->
client.getBlockingStub(target)));
+ }
+ assertTrue("one refresh should be waiting in the delayed resolver",
+ client.delayedResolutionStarted.await(5,
TimeUnit.SECONDS));
+ for (Future<AbstractBlockingStub> future : futures) {
+ assertNotNull("no caller may block on the delayed refresh",
+ future.get(5, TimeUnit.SECONDS));
+ }
+ assertTrue("the existing healthy pool must stay live during
refresh",
+ allChannelsAreLive(oldChannels));
+
+ client.releaseDelayedResolution.countDown();
+ ManagedChannel[] currentChannels =
+ awaitPoolReplacement(client, target, oldChannels);
+ assertTrue("the previous pool must be retired after replacement is
published",
+ allChannelsAreShutdown(oldChannels));
+ assertTrue("the replacement pool must be live",
allChannelsAreLive(currentChannels));
+ assertEquals("concurrent callers must share a single refresh
resolution",
+ resolutionsBeforeConcurrentCalls + 1,
client.resolutionCount.get());
+ } finally {
+ client.releaseDelayedResolution.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testRefreshGracefullyRetiresActiveStreamChannels() throws
Exception {
+ String target = uniqueTarget("active-stream-refresh");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ client.activeCallsFinished = new CountDownLatch(1);
+ client.drainTimeoutNanos = TimeUnit.SECONDS.toNanos(5L);
+ ManagedChannel[] oldChannels = client.getChannels(target);
+ AbstractAsyncStub activeStreamStub = client.getAsyncStub(target);
+ assertTrue("the simulated active stream must be on the old pool",
+ belongsToPool(activeStreamStub.getChannel(), oldChannels));
+
+ client.resolvedTarget = "10.0.0.2";
+ ManagedChannel[] newChannels = awaitPoolReplacement(client, target,
oldChannels);
+ List<FakeManagedChannel> retiredChannels = fakeChannels(oldChannels);
+ assertTrue("the retired pool must receive graceful shutdown",
+
retiredChannels.stream().allMatch(FakeManagedChannel::isShutdown));
+ assertFalse("active streams must not be force closed immediately",
+
retiredChannels.stream().anyMatch(FakeManagedChannel::isForceShutdown));
+ client.activeCallsFinished.countDown();
+ awaitCondition("retired channels should terminate after active calls
finish",
+ () ->
retiredChannels.stream().allMatch(FakeManagedChannel::isTerminated));
+ assertFalse("drained channels must not need forced shutdown",
+
retiredChannels.stream().anyMatch(FakeManagedChannel::isForceShutdown));
+ assertTrue("the replacement pool must remain live",
allChannelsAreLive(newChannels));
}
/**
- * A client whose channels and stubs are local fakes, so the test needs no
PD or store node.
+ * Holds a stub pool build open, refreshes the pool underneath it, and
asserts that both the
+ * interleaved build and a concurrent one return stubs bound to the
published pool. Blocking
+ * and asynchronous acquisition share one implementation, so the
asynchronous path stands in
+ * for both; it is the one that also publishes a stub cache worth
asserting on.
*/
+ @Test
+ public void testStubBuildRetriesAfterChannelRefresh() throws Exception {
+ String target = uniqueTarget("concurrent-stub-refresh");
+ StubInterleavingGrpcClient client = new StubInterleavingGrpcClient();
+ ManagedChannel[] oldChannels = client.getChannels(target);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+
+ try {
+ Future<AbstractAsyncStub> staleStub =
+ executor.submit(() -> client.getAsyncStub(target));
+ assertTrue("the old stub pool build must be in flight",
+ client.staleStubBuildStarted.await(5,
TimeUnit.SECONDS));
+ client.resolvedTarget = "10.0.0.2";
+ Future<AbstractAsyncStub> freshStub =
+ executor.submit(() -> client.getAsyncStub(target));
+ awaitCondition("refresh must retire the old channel pool",
+ () -> allChannelsAreShutdown(oldChannels));
+ client.releaseStaleStubBuild.countDown();
+
+ ManagedChannel[] currentChannels = client.getChannels(target);
+ assertTrue("the stale build must retry against the current pool",
+ belongsToPool(staleStub.get(5,
TimeUnit.SECONDS).getChannel(),
+ currentChannels));
+ assertTrue("the concurrent build must use the current pool",
+ belongsToPool(freshStub.get(5,
TimeUnit.SECONDS).getChannel(),
+ currentChannels));
+ assertTrue("the current channel pool must remain live",
+ allChannelsAreLive(currentChannels));
+ assertCachedChannelsCurrentAndLive(
+ "the final stub cache must only reference current live
channels",
+ cachedAsyncStubChannels(client, target), currentChannels);
+ } finally {
+ client.releaseStaleStubBuild.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ /**
+ * QueryV2 used to take a channel straight from the pool and build its
stub afterwards, which
+ * let a refresh retire that channel in between. It must now go through
the guarded path.
+ */
+ @Test
+ public void testQueryV2StubFollowsPublishedPoolAcrossRefresh() throws
Exception {
+ String target = uniqueTarget("query-v2-refresh");
+ QueryV2TestClient client = new QueryV2TestClient();
+ ManagedChannel[] oldChannels = client.getChannels(target);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+
+ try {
+ /*
+ * Interleave a refresh with the stub build. Taking a channel from
the pool and
+ * building the stub afterwards would bind it to a channel retired
in between.
+ */
+ Future<QueryServiceGrpc.QueryServiceStub> stub =
+ executor.submit(() -> client.getQueryServiceStub(target));
+ assertTrue("the QueryV2 stub build must be in flight",
+ client.stubBuildStarted.await(5, TimeUnit.SECONDS));
+ client.resolvedTarget = "10.0.0.2";
+ // getChannels is what triggers a refresh, and the blocked build
cannot call it.
+ awaitCondition("refresh must retire the old channel pool",
+ () -> client.getChannels(target) != oldChannels &&
+ allChannelsAreShutdown(oldChannels));
+ client.releaseStubBuild.countDown();
+
+ ManagedChannel[] newChannels = client.getChannels(target);
+ Channel channel = stub.get(5, TimeUnit.SECONDS).getChannel();
+ assertTrue("QueryV2 must never return a stub bound to a retired
channel",
+ belongsToPool(channel, newChannels));
+ assertFalse("QueryV2 must never return a stub on a shut down
channel",
+ ((ManagedChannel) channel).isShutdown());
+
+ List<ManagedChannel> stubChannels = new ArrayList<>();
+ for (int i = 0; i < AbstractGrpcClient.concurrency; i++) {
+ stubChannels.add((ManagedChannel)
client.getQueryServiceStub(target).getChannel());
+ }
+ assertCachedChannelsCurrentAndLive("QueryV2 stubs must stay on the
published pool",
+ stubChannels, newChannels);
+ assertUsesEveryChannel("QueryV2 stubs must still spread across the
pool",
+ stubChannels, newChannels);
+ } finally {
+ client.releaseStubBuild.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testInjectedQueryV2ChannelBypassesRefreshRetirement() throws
Exception {
+ String target = uniqueTarget("query-v2-injected-channel");
+ FakeManagedChannel injected = new FakeManagedChannel(target);
+ InjectedChannelQueryV2Client client = new
InjectedChannelQueryV2Client();
+ QueryV2Client.setTestChannel(injected);
+
+ try {
+ QueryServiceGrpc.QueryServiceStub stub =
client.getQueryServiceStub(target);
+ assertSame("the QueryV2 stub must use the injected channel",
injected,
+ stub.getChannel());
+
+ /*
+ * Without the injected-channel resolution guard, the delayed
resolution lands after
+ * the first pool is published, rebuilds that pool with the same
injected channel,
+ * and retires the channel that the replacement still references.
+ */
+ client.releaseResolution.countDown();
+ awaitCondition("the injected-channel refresh must settle",
+ () -> refreshIsIdle(client, target));
+ assertFalse("refresh must not retire an injected channel",
injected.isShutdown());
+ assertSame("the cached stub must retain the live injected
channel", injected,
+ client.getQueryServiceStub(target).getChannel());
+ } finally {
+ client.releaseResolution.countDown();
+ QueryV2Client.setTestChannel(null);
+ }
+ }
+
+ @Test
+ public void testQueryV2WithoutInjectedChannelUsesDnsResolution() {
Review Comment:
Fixed in fad20a3c. The regression now invokes the public
ResolvingQueryV2Client getQueryServiceStub(String) path with deterministic fake
channel construction, then asserts the returned stub belongs to the published
live pool. The final focused suite is 29/29 with zero failures/errors/skips.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
