imbajin commented on code in PR #3130:
URL: https://github.com/apache/hugegraph/pull/3130#discussion_r3686348784
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -169,6 +205,70 @@ 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 (HgPair<ManagedChannel, ?> pair : pairs) {
+ if (pair == null || pair.getKey() == null ||
+ !containsChannel(channels, pair.getKey())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean containsChannel(ManagedChannel[] channels,
+ ManagedChannel expected) {
+ return Arrays.stream(channels).anyMatch(channel -> channel ==
expected);
+ }
+
+ private void refreshChannelsIfAddressChanged(String target) {
+ long resolutionRequest = resolutionRequests.computeIfAbsent(target,
+ key -> new
AtomicLong())
+ .incrementAndGet();
+ String resolvedTarget = this.resolveTarget(target);
+ if (resolvedTarget.isEmpty()) {
+ return;
+ }
+ synchronized (channels) {
+ Long appliedResolution = appliedResolutions.get(target);
+ if (appliedResolution != null && appliedResolution >=
resolutionRequest) {
+ return;
+ }
+ appliedResolutions.put(target, resolutionRequest);
+ String previousTarget = resolvedTargets.put(target,
resolvedTarget);
+ if (previousTarget == null && !channels.containsKey(target)) {
+ return;
+ }
+ if (resolvedTarget.equals(previousTarget)) {
+ return;
+ }
+ ManagedChannel[] staleChannels = channels.remove(target);
+ if (staleChannels != null) {
+ Arrays.stream(staleChannels)
+ .filter(channel -> channel != null &&
!channel.isShutdown())
+ .forEach(ManagedChannel::shutdownNow);
+ }
+ }
+ }
+
+ protected String resolveTarget(String target) {
+ try {
+ String host = URI.create("dns://" + target).getHost();
Review Comment:
⚠️ This no longer matches the target contract used by
`ManagedChannelBuilder.forTarget(target)`. For example, a valid
`dns:///store:8500` target becomes `dns://dns:///store:8500`, so the
fingerprint resolves the literal host `dns`; other gRPC resolver schemes can
yield no host at all. The channel may still connect while refresh silently
monitors the wrong endpoint. Please parse supported gRPC target schemes
consistently (or explicitly validate and restrict the accepted format) and add
URI-form and IPv6 target tests.
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -169,6 +205,70 @@ 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 (HgPair<ManagedChannel, ?> pair : pairs) {
+ if (pair == null || pair.getKey() == null ||
+ !containsChannel(channels, pair.getKey())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean containsChannel(ManagedChannel[] channels,
+ ManagedChannel expected) {
+ return Arrays.stream(channels).anyMatch(channel -> channel ==
expected);
+ }
+
+ private void refreshChannelsIfAddressChanged(String target) {
+ long resolutionRequest = resolutionRequests.computeIfAbsent(target,
+ key -> new
AtomicLong())
+ .incrementAndGet();
+ String resolvedTarget = this.resolveTarget(target);
Review Comment:
⚠️ Every `getChannels()` call reaches this synchronous resolution path
before consulting the cached pool, and Store session methods acquire a stub per
request. When the JVM DNS cache expires or the resolver is slow, application
request threads now block on `InetAddress.getAllByName()`. Please
throttle/cache refreshes using a configurable interval or single-flight TTL, or
refresh asynchronously while retaining the last healthy pool; cover repeated
and concurrent stub acquisition with a delayed resolver.
##########
hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClientTest.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * 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.hugegraph.store.client.grpc;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+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.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import io.grpc.CallOptions;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ManagedChannel;
+import io.grpc.MethodDescriptor;
+import io.grpc.stub.AbstractAsyncStub;
+import io.grpc.stub.AbstractBlockingStub;
+
+/**
+ * Verifies that Store address changes replace channels and their cached stubs
safely.
+ */
+public class AbstractGrpcClientTest {
+
+ private static final AtomicInteger TARGET_SEQ = new AtomicInteger();
+
+ private static String uniqueTarget(String prefix) {
+ return prefix + "-" + TARGET_SEQ.incrementAndGet() + ":8500";
+ }
+
+ private static boolean belongsToPool(Channel channel,
+ ManagedChannel[] channels) {
+ return Arrays.stream(channels).anyMatch(current -> current == channel);
+ }
+
+ @Test
+ public void testAddressChangeReplacesChannelAndStubPools() {
+ String target = uniqueTarget("address-change");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ ManagedChannel[] oldChannels = client.getChannels(target);
+ assertNotNull(client.getBlockingStub(target));
+ assertNotNull(client.getAsyncStub(target));
+
+ client.resolvedTarget = "10.0.0.2";
+ ManagedChannel[] newChannels = client.getChannels(target);
+ assertNotSame("an address change must replace the channel pool",
+ oldChannels, newChannels);
+ assertTrue("every stale channel must be shut down",
+
Arrays.stream(oldChannels).allMatch(ManagedChannel::isShutdown));
+
+ client.blockingStubChannels.clear();
+ client.asyncStubChannels.clear();
+ assertNotNull(client.getBlockingStub(target));
+ assertNotNull(client.getAsyncStub(target));
+ assertEquals("the blocking stub pool must be rebuilt",
+ newChannels.length, client.blockingStubChannels.size());
+ assertTrue("replacement blocking stubs must use the new channel pool",
+ client.blockingStubChannels.stream()
+ .allMatch(channel ->
+ belongsToPool(channel,
newChannels)));
+ assertEquals("the async stub pool must be rebuilt",
+ newChannels.length, client.asyncStubChannels.size());
+ assertTrue("replacement async stubs must use the new channel pool",
+ client.asyncStubChannels.stream()
+ .allMatch(channel ->
+ belongsToPool(channel,
newChannels)));
+ }
+
+ @Test
+ public void testFirstSuccessfulResolutionReplacesUnknownChannels() {
+ 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 = client.getChannels(target);
+ assertNotSame("a pool with unknown addresses must be replaced",
+ unknownChannels, resolvedChannels);
+ assertTrue("every channel from the unknown pool must be shut down",
+
Arrays.stream(unknownChannels).allMatch(ManagedChannel::isShutdown));
+ assertTrue("the resolved channel pool must remain live",
+
Arrays.stream(resolvedChannels).noneMatch(ManagedChannel::isShutdown));
+ }
+
+ @Test
+ public void testOlderResolutionCannotReplaceNewerChannels() throws
Exception {
+ String target = uniqueTarget("concurrent-address-change");
+ OutOfOrderResolverGrpcClient client = new
OutOfOrderResolverGrpcClient();
+ ManagedChannel[] oldChannels = client.getChannels(target);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+
+ try {
+ Future<ManagedChannel[]> staleResolution =
+ executor.submit(() -> client.getChannels(target));
+ assertTrue("the stale resolution must be in flight",
+ client.staleResolutionStarted.await(5,
TimeUnit.SECONDS));
+ Future<ManagedChannel[]> freshResolution =
+ executor.submit(() -> client.getChannels(target));
+ ManagedChannel[] freshChannels = freshResolution.get(5,
TimeUnit.SECONDS);
+
+ assertNotSame("the newer address must replace the old channel
pool",
+ oldChannels, freshChannels);
+ client.releaseStaleResolution.countDown();
+ assertSame("the late stale result must retain the newer channel
pool",
+ freshChannels, staleResolution.get(5,
TimeUnit.SECONDS));
+ assertTrue("the replaced channel pool must be shut down",
+
Arrays.stream(oldChannels).allMatch(ManagedChannel::isShutdown));
+ assertTrue("the newer channel pool must remain live",
+
Arrays.stream(freshChannels).noneMatch(ManagedChannel::isShutdown));
+ } finally {
+ client.releaseStaleResolution.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testStubBuildRetriesAfterChannelRefresh() throws Exception {
Review Comment:
⚠️ This interleaving test only exercises `getBlockingStub()`, while the PR
independently rewrites the async refresh/retry/cache-publication path and
builds async stubs in parallel. The sequential async assertion cannot catch a
stale async pool published during refresh. Please add the equivalent
latch-controlled async interleaving test and assert that both returned stubs
and the final cache reference only current live channels.
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -169,6 +205,70 @@ 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 (HgPair<ManagedChannel, ?> pair : pairs) {
+ if (pair == null || pair.getKey() == null ||
+ !containsChannel(channels, pair.getKey())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean containsChannel(ManagedChannel[] channels,
+ ManagedChannel expected) {
+ return Arrays.stream(channels).anyMatch(channel -> channel ==
expected);
+ }
+
+ private void refreshChannelsIfAddressChanged(String target) {
+ long resolutionRequest = resolutionRequests.computeIfAbsent(target,
+ key -> new
AtomicLong())
+ .incrementAndGet();
+ String resolvedTarget = this.resolveTarget(target);
+ if (resolvedTarget.isEmpty()) {
+ return;
+ }
+ synchronized (channels) {
+ Long appliedResolution = appliedResolutions.get(target);
+ if (appliedResolution != null && appliedResolution >=
resolutionRequest) {
+ return;
+ }
+ appliedResolutions.put(target, resolutionRequest);
+ String previousTarget = resolvedTargets.put(target,
resolvedTarget);
+ if (previousTarget == null && !channels.containsKey(target)) {
+ return;
+ }
+ if (resolvedTarget.equals(previousTarget)) {
+ return;
+ }
+ ManagedChannel[] staleChannels = channels.remove(target);
+ if (staleChannels != null) {
+ Arrays.stream(staleChannels)
+ .filter(channel -> channel != null &&
!channel.isShutdown())
+ .forEach(ManagedChannel::shutdownNow);
Review Comment:
‼️ `shutdownNow()` forcibly terminates existing calls on these channels, but
callers may already hold blocking or streaming stubs after the cache lock is
released. A normal DNS rotation can therefore cancel active Store operations
and leave writes with ambiguous outcomes. Please publish the replacement pool
first, gracefully retire old channels with `shutdown()`, and use bounded
delayed forced termination only after draining; add an active-RPC/stream
refresh test.
--
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]
