You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hugegraph.apache.org by "bitflicker64 (via GitHub)" <gi...@apache.org> on 2026/03/05 08:15:11 UTC
[PR] fix(pd): add timeout and null-safety to getLeaderGrpcAddress() [hugegraph]
bitflicker64 opened a new pull request, #2961:
URL: https://github.com/apache/hugegraph/pull/2961
Here's the filled-out PR template:
---
## Purpose of the PR
- close #2959
In a 3-node PD cluster running in Docker bridge network mode, `getLeaderGrpcAddress()` makes a bolt RPC call to discover the leader's gRPC address when the current node is a follower. This call fails in bridge mode — the TCP connection establishes but the bolt RPC response never returns properly, causing `CompletableFuture.get()` to return null and throw NPE.
This causes:
1. `redirectToLeader()` fails with NPE
2. Store registration requests landing on follower PDs are never forwarded
3. Stores register but partitions are never distributed (`partitionCount:0`)
4. HugeGraph servers stuck in `DEADLINE_EXCEEDED` loop indefinitely
The cluster only works when pd0 wins raft leader election (since `isLeader()` returns true and the broken code path is skipped). If pd1 or pd2 wins, the NPE fires on every redirect attempt.
Related PR: #2952
## Main Changes
- Add a bounded timeout to the bolt RPC call using `config.getRpcTimeout()` instead of unbounded `.get()`
- Add null-check on the RPC response before accessing `.getGrpcAddress()`
- Fall back to deriving the leader address from the raft endpoint IP + local gRPC port when the RPC fails or times out
- Add `TimeUnit` and `TimeoutException` imports
## Verifying these changes
- [ ] Trivial rework / code cleanup without any test coverage. (No Need)
- [ ] Already covered by existing tests, such as *(please modify tests here)*.
- [x] Need tests and can be verified as follows:
- Deploy 3-node PD cluster in Docker bridge network mode
- Verify cluster works regardless of which PD node wins raft leader election
- Confirm stores show `partitionCount:12` on all 3 nodes when pd1 or pd2 is leader
- Confirm no NPE in pd logs at `getLeaderGrpcAddress`
## Does this PR potentially affect the following parts?
- [ ] Dependencies ([add/update license](https://hugegraph.apache.org/docs/contribution-guidelines/contribute/#321-check-licenses) info & [regenerate_known_dependencies.sh](../install-dist/scripts/dependency/regenerate_known_dependencies.sh))
- [ ] Modify configurations
- [ ] The public API
- [ ] Other affects (typed here)
- [x] Nope
## Documentation Status
- [ ] `Doc - TODO`
- [ ] `Doc - Done`
- [x] `Doc - No Need`
--
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: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(pd): add timeout and null-safety to getLeaderGrpcAddress() [hugegraph]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2961:
URL: https://github.com/apache/hugegraph/pull/2961#discussion_r2960380363
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -228,19 +229,57 @@ public PeerId getLeader() {
}
/**
- * Send a message to the leader to get the grpc address;
+ * Send a message to the leader to get the grpc address.
*/
public String getLeaderGrpcAddress() throws ExecutionException, InterruptedException {
if (isLeader()) {
return config.getGrpcAddress();
}
if (raftNode.getLeaderId() == null) {
- waitingForLeader(10000);
+ waitingForLeader(config.getRpcTimeout());
Review Comment:
`waitingForLeader()` waits in fixed 1000ms intervals (see `waitingForLeader(long)`), so passing `config.getRpcTimeout()` here can violate the configured timeout when `rpcTimeout < 1000` (it may still block ~1s). Consider either clamping the value passed in (e.g., `Math.max(config.getRpcTimeout(), 1000)`) or updating `waitingForLeader()` to wait for `min(remaining, 1000)` so the timeout is respected.
##########
hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/raft/RaftEngineLeaderAddressTest.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * 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.pd.raft;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.hugegraph.pd.config.PDConfig;
+import org.apache.hugegraph.testutil.Whitebox;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.alipay.sofa.jraft.Node;
+import com.alipay.sofa.jraft.entity.PeerId;
+import com.alipay.sofa.jraft.util.Endpoint;
+
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class RaftEngineLeaderAddressTest {
+
+ private static final String LEADER_IP = "10.0.0.1";
+ private static final int GRPC_PORT = 8686;
+ private static final String LEADER_GRPC_ADDRESS = "10.0.0.1:8686";
+
+ private Node originalRaftNode;
+ private RaftRpcClient originalRaftRpcClient;
+ private PDConfig.Raft originalConfig;
+
+ private Node mockNode;
+ private RaftRpcClient mockRpcClient;
+ private PDConfig.Raft mockConfig;
+ private PeerId mockLeader;
+
+ @Before
+ public void setUp() {
+ RaftEngine engine = RaftEngine.getInstance();
+
+ // Save originals
+ originalRaftNode = engine.getRaftNode();
+ originalRaftRpcClient = Whitebox.getInternalState(engine, "raftRpcClient");
+ originalConfig = Whitebox.getInternalState(engine, "config");
+
+ // Build mock leader PeerId with real Endpoint
+ mockLeader = mock(PeerId.class);
+ Endpoint endpoint = new Endpoint(LEADER_IP, 8610);
+ when(mockLeader.getEndpoint()).thenReturn(endpoint);
+
+ // Build mock Node that reports itself as follower with a known leader
+ mockNode = mock(Node.class);
+ when(mockNode.isLeader(true)).thenReturn(false);
+ when(mockNode.getLeaderId()).thenReturn(mockLeader);
+
+ // Build mock config
+ // Use a short timeout (100ms) so the null-leader test doesn't block for seconds
+ mockConfig = mock(PDConfig.Raft.class);
+ when(mockConfig.getGrpcAddress()).thenReturn("127.0.0.1:" + GRPC_PORT);
+ when(mockConfig.getGrpcPort()).thenReturn(GRPC_PORT);
+ when(mockConfig.getRpcTimeout()).thenReturn(100);
+
Review Comment:
This comment is now misleading: the tests generally don't hit `waitingForLeader()` with the 100ms timeout (the only null-leader test explicitly overrides `getRpcTimeout()` to 0ms). Consider updating/removing the comment to match the actual behavior being tested.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -228,19 +229,57 @@ public PeerId getLeader() {
}
/**
- * Send a message to the leader to get the grpc address;
+ * Send a message to the leader to get the grpc address.
*/
public String getLeaderGrpcAddress() throws ExecutionException, InterruptedException {
if (isLeader()) {
return config.getGrpcAddress();
}
if (raftNode.getLeaderId() == null) {
- waitingForLeader(10000);
+ waitingForLeader(config.getRpcTimeout());
+ }
+
+ // Cache leader to avoid repeated getLeaderId() calls and guard against
+ // waitingForLeader() returning without a leader being elected.
+ PeerId leader = raftNode.getLeaderId();
+ if (leader == null) {
+ throw new ExecutionException(new IllegalStateException("Leader is not ready"));
+ }
+
+ RaftRpcProcessor.GetMemberResponse response = null;
+ try {
+ // TODO: a more complete fix would need a source of truth for the leader's
+ // actual grpcAddress rather than deriving it from the local node's port config.
+ response = raftRpcClient
+ .getGrpcAddress(leader.getEndpoint().toString())
+ .get(config.getRpcTimeout(), TimeUnit.MILLISECONDS);
+ if (response != null && response.getGrpcAddress() != null) {
+ return response.getGrpcAddress();
+ }
+ if (response == null) {
+ log.warn("Leader RPC response is null for {}, falling back to derived address",
+ leader);
+ } else {
+ log.warn("Leader gRPC address field is null in RPC response for {}, "
+ + "falling back to derived address", leader);
+ }
+ } catch (TimeoutException e) {
+ log.warn("Timed out resolving leader gRPC address for {}, falling back to derived "
+ + "address", leader);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause() != null ? e.getCause() : e;
+ log.warn("Failed to resolve leader gRPC address for {}, falling back to derived "
+ + "address", leader, cause);
}
- return raftRpcClient.getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString()).get()
- .getGrpcAddress();
+ // Best-effort fallback: derive from leader raft endpoint IP + local gRPC port.
+ // WARNING: this may be incorrect in clusters where PD nodes use different grpc.port
+ // values, a proper fix requires a cluster-wide source of truth for gRPC addresses.
+ String derived = leader.getEndpoint().getIp() + ":" + config.getGrpcPort();
+ log.warn("Using derived leader gRPC address {} — may be incorrect if nodes use different ports",
Review Comment:
The fallback path logs warnings multiple times per call (one for the failure condition and another for "Using derived leader gRPC address..."). Since `getLeaderGrpcAddress()` is invoked in hot paths like `PDService.redirectToLeader()` (and even called twice there), this can generate excessive WARN logs under failure conditions. Consider consolidating to a single WARN that includes the derived address (or downgrading the final message to INFO/DEBUG) to reduce log volume.
--
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: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] fix(pd): add timeout and null-safety to getLeaderGrpcAddress() [hugegraph]
Posted by "bitflicker64 (via GitHub)" <gi...@apache.org>.
bitflicker64 commented on PR #2961:
URL: https://github.com/apache/hugegraph/pull/2961#issuecomment-4007663614
**How I tested:**
1. Built a local Docker image from source with this fix applied
2. Brought up the 3-node cluster (3 PD + 3 Store + 3 Server) in bridge network mode
3. Confirmed cluster was healthy with pd0 as initial leader
4. Restarted pd0 to force a new leader election — pd1 won
5. Checked partition distribution and cluster health with pd1 as leader
**Results with pd1 as leader:**
```
partitionCount:12 on all 3 stores ✅
leaderCount:12 on all 3 stores ✅
{"graphs":["hugegraph"]} ✅
All 9 containers healthy ✅
```
**Confirmed fallback triggered in pd1 logs:**
```
[WARN] RaftEngine - Failed to get leader gRPC address via RPC, falling back to endpoint derivation
java.util.concurrent.ExecutionException: com.alipay.remoting.exception.RemotingException:
Create connection failed. The address is 172.20.0.10:8610
at RaftEngine.getLeaderGrpcAddress(RaftEngine.java:247)
at PDService.redirectToLeader(PDService.java:1275)
```
**Before this fix:** RPC returns null → NPE → follower PDs can't redirect requests to leader → cluster only worked when pd0 won leader election since it never hit the broken code path.
**After this fix:** RPC failure caught with bounded timeout → fallback to endpoint IP + gRPC port derivation → follower PDs correctly redirect to leader regardless of which PD node wins election.
Related docker bridge networking PR: #2952
--
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: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(pd): add timeout and null-safety to getLeaderGrpcAddress() [hugegraph]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2961:
URL: https://github.com/apache/hugegraph/pull/2961#discussion_r2930111538
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -228,19 +230,42 @@ public PeerId getLeader() {
}
/**
- * Send a message to the leader to get the grpc address;
+ * Send a message to the leader to get the grpc address.
*/
public String getLeaderGrpcAddress() throws ExecutionException, InterruptedException {
if (isLeader()) {
return config.getGrpcAddress();
}
if (raftNode.getLeaderId() == null) {
- waitingForLeader(10000);
+ waitingForLeader(config.getRpcTimeout());
}
- return raftRpcClient.getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString()).get()
- .getGrpcAddress();
+ // Cache leader to avoid repeated getLeaderId() calls and guard against
+ // waitingForLeader() returning without a leader being elected.
+ PeerId leader = raftNode.getLeaderId();
+ if (leader == null) {
+ throw new ExecutionException(new IllegalStateException("Leader is not ready"));
+ }
+
+ try {
+ RaftRpcProcessor.GetMemberResponse response = raftRpcClient
+ .getGrpcAddress(leader.getEndpoint().toString())
+ .get(config.getRpcTimeout(), TimeUnit.MILLISECONDS);
+ if (response != null && response.getGrpcAddress() != null) {
+ return response.getGrpcAddress();
+ }
+ } catch (TimeoutException | ExecutionException e) {
+ // TODO: a more complete fix would need a source of truth for the leader's
+ // actual grpcAddress rather than deriving it from the local node's port config.
+ throw new ExecutionException(
+ String.format("Failed to resolve leader gRPC address for %s", leader), e);
+ }
+
+ log.warn("Leader gRPC address field is null in RPC response for {}", leader);
+ throw new ExecutionException(
+ new IllegalStateException(
+ String.format("Leader gRPC address unavailable for %s", leader)));
Review Comment:
The PR description says we should fall back to deriving the leader gRPC address from the raft endpoint IP + local gRPC port when the bolt RPC times out/fails, but the current implementation still throws ExecutionException on TimeoutException/ExecutionException and also throws when the response/grpcAddress is null. This means callers like PDService.redirectToLeader() can still fail to forward requests when the bolt RPC is broken in bridge networking, so the cluster behavior described in #2959 likely remains broken. Consider returning a derived address (e.g., leader endpoint host + config grpcPort) on timeout/failed response (and only throw if that derivation is impossible).
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -228,19 +230,42 @@ public PeerId getLeader() {
}
/**
- * Send a message to the leader to get the grpc address;
+ * Send a message to the leader to get the grpc address.
*/
public String getLeaderGrpcAddress() throws ExecutionException, InterruptedException {
if (isLeader()) {
return config.getGrpcAddress();
}
if (raftNode.getLeaderId() == null) {
- waitingForLeader(10000);
+ waitingForLeader(config.getRpcTimeout());
}
- return raftRpcClient.getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString()).get()
- .getGrpcAddress();
+ // Cache leader to avoid repeated getLeaderId() calls and guard against
+ // waitingForLeader() returning without a leader being elected.
+ PeerId leader = raftNode.getLeaderId();
+ if (leader == null) {
+ throw new ExecutionException(new IllegalStateException("Leader is not ready"));
+ }
+
+ try {
+ RaftRpcProcessor.GetMemberResponse response = raftRpcClient
+ .getGrpcAddress(leader.getEndpoint().toString())
+ .get(config.getRpcTimeout(), TimeUnit.MILLISECONDS);
+ if (response != null && response.getGrpcAddress() != null) {
+ return response.getGrpcAddress();
+ }
+ } catch (TimeoutException | ExecutionException e) {
+ // TODO: a more complete fix would need a source of truth for the leader's
+ // actual grpcAddress rather than deriving it from the local node's port config.
+ throw new ExecutionException(
+ String.format("Failed to resolve leader gRPC address for %s", leader), e);
Review Comment:
In this catch block, catching ExecutionException and then wrapping it in a new ExecutionException leads to double-wrapping (the real root cause ends up at e.getCause()). It will also make it harder for callers to distinguish timeout vs other failures. Consider handling TimeoutException separately and either rethrow the original ExecutionException or wrap e.getCause() so the root cause is preserved more directly.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -119,8 +122,7 @@ public synchronized boolean init(PDConfig.Raft config) {
nodeOptions.setRpcConnectTimeoutMs(config.getRpcTimeout());
nodeOptions.setRpcDefaultTimeout(config.getRpcTimeout());
nodeOptions.setRpcInstallSnapshotTimeout(config.getRpcTimeout());
- // Set the raft configuration
- RaftOptions raftOptions = nodeOptions.getRaftOptions();
+ // TODO: tune RaftOptions for PD (see hugegraph-store PartitionEngine for reference)
nodeOptions.setEnableMetrics(true);
Review Comment:
nodeOptions.setEnableMetrics(true) is called twice (earlier right after setFsm(), and again here). The second call is redundant and can be removed to reduce noise and avoid implying there are two separate metric toggles.
--
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: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(pd): add timeout and null-safety to getLeaderGrpcAddress() [hugegraph]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #2961:
URL: https://github.com/apache/hugegraph/pull/2961#discussion_r2904690778
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -239,8 +241,20 @@ public String getLeaderGrpcAddress() throws ExecutionException, InterruptedExcep
waitingForLeader(10000);
}
- return raftRpcClient.getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString()).get()
- .getGrpcAddress();
+ try {
+ RaftRpcProcessor.GetMemberResponse response = raftRpcClient
+ .getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString())
+ .get(config.getRpcTimeout(), TimeUnit.MILLISECONDS);
+ if (response != null && response.getGrpcAddress() != null) {
+ return response.getGrpcAddress();
+ }
+ } catch (TimeoutException | ExecutionException e) {
+ log.warn("Failed to get leader gRPC address via RPC, falling back to endpoint derivation", e);
+ }
+
+ // Fallback: derive from raft endpoint IP + local gRPC port (best effort)
+ String leaderIp = raftNode.getLeaderId().getEndpoint().getIp();
+ return leaderIp + ":" + config.getGrpcPort();
Review Comment:
‼️ This fallback is still incorrect for clusters where PD nodes use different `grpc.port` values. In this repo's own multi-node test configs, `application-server1.yml`, `application-server2.yml`, and `application-server3.yml` advertise `8686`, `8687`, and `8688` respectively, so a follower on `8687` will redirect to `leader-ip:8687` even when the elected leader is actually listening on `8686` or `8688`. That turns the original NPE into a silent misroute.
If we can't recover the leader's advertised gRPC endpoint here, I think it's safer to fail fast than to synthesize an address from the local port, for example:
```suggestion
} catch (TimeoutException | ExecutionException e) {
throw new ExecutionException(
String.format("Failed to resolve leader gRPC address for %s", raftNode.getLeaderId()),
e);
}
```
A more complete fix would need a source of truth for the leader's actual `grpcAddress`, not the local node's port.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -239,8 +241,20 @@ public String getLeaderGrpcAddress() throws ExecutionException, InterruptedExcep
waitingForLeader(10000);
}
- return raftRpcClient.getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString()).get()
- .getGrpcAddress();
+ try {
+ RaftRpcProcessor.GetMemberResponse response = raftRpcClient
+ .getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString())
Review Comment:
⚠️ `waitingForLeader(10000)` does not guarantee that `leaderId` is non-null; it just waits up to the timeout and returns whatever it has. That means both the RPC call and this fallback path can still dereference `raftNode.getLeaderId()` and reintroduce an NPE during leader-election gaps.
Could we cache the leader once after waiting and turn the "leader still unknown" case into a controlled exception instead?
```suggestion
PeerId leader = raftNode.getLeaderId();
if (leader == null) {
throw new ExecutionException(new IllegalStateException("Leader is not ready"));
}
```
Then reuse `leader` for both the RPC request and any follow-up handling.
--
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: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(pd): add timeout and null-safety to getLeaderGrpcAddress() [hugegraph]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #2961:
URL: https://github.com/apache/hugegraph/pull/2961#discussion_r2909775766
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -239,8 +241,20 @@ public String getLeaderGrpcAddress() throws ExecutionException, InterruptedExcep
waitingForLeader(10000);
}
- return raftRpcClient.getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString()).get()
- .getGrpcAddress();
+ try {
+ RaftRpcProcessor.GetMemberResponse response = raftRpcClient
+ .getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString())
+ .get(config.getRpcTimeout(), TimeUnit.MILLISECONDS);
Review Comment:
⚠️ `config.getRpcTimeout()` only bounds the callers `CompletableFuture.get(...)` here; it does **not** bound the underlying Bolt RPC. `RaftRpcClient` is still initialized with a fresh `new RpcOptions()` and `invokeAsync(...)` uses `rpcOptions.getRpcDefaultTimeout()`, so in the bridge-network failure case well return from this method quickly but still leave the actual RPC running until the client default timeout fires. On a follower that keeps redirecting traffic, those in-flight requests can accumulate.
Could we wire the configured timeout into `RaftRpcClient` itself (or cancel the future on timeout) so the transport timeout and the caller timeout stay consistent?
--
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: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(pd): add timeout and null-safety to getLeaderGrpcAddress() [hugegraph]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin merged PR #2961:
URL: https://github.com/apache/hugegraph/pull/2961
--
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: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] fix(pd): add timeout and null-safety to getLeaderGrpcAddress() [hugegraph]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2961:
URL: https://github.com/apache/hugegraph/pull/2961#discussion_r2904689605
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -239,8 +241,20 @@ public String getLeaderGrpcAddress() throws ExecutionException, InterruptedExcep
waitingForLeader(10000);
}
- return raftRpcClient.getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString()).get()
- .getGrpcAddress();
+ try {
+ RaftRpcProcessor.GetMemberResponse response = raftRpcClient
+ .getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString())
+ .get(config.getRpcTimeout(), TimeUnit.MILLISECONDS);
Review Comment:
`waitingForLeader(10000)` is called when `raftNode.getLeaderId()` is null, but the result isn’t used/rechecked. If the leader still isn’t elected after the wait, the subsequent `raftNode.getLeaderId().getEndpoint()` (and the fallback path) will still throw an NPE. Capture the leader into a local variable after waiting and handle the `null` case explicitly (e.g., throw a clear exception or return a sentinel that callers can handle).
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -239,8 +241,20 @@ public String getLeaderGrpcAddress() throws ExecutionException, InterruptedExcep
waitingForLeader(10000);
}
- return raftRpcClient.getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString()).get()
- .getGrpcAddress();
+ try {
+ RaftRpcProcessor.GetMemberResponse response = raftRpcClient
+ .getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString())
+ .get(config.getRpcTimeout(), TimeUnit.MILLISECONDS);
+ if (response != null && response.getGrpcAddress() != null) {
+ return response.getGrpcAddress();
+ }
+ } catch (TimeoutException | ExecutionException e) {
+ log.warn("Failed to get leader gRPC address via RPC, falling back to endpoint derivation", e);
Review Comment:
This `log.warn(..., e)` will likely be hit on every follower redirect in environments where the RPC path is consistently broken/blocked (the scenario described in #2959), which can flood logs and add overhead due to repeated stack traces. Consider reducing verbosity (e.g., warn without stack trace, debug with stack trace, or rate-limit) and include key context like leader endpoint/timeout in the message.
```suggestion
String leaderEndpoint = raftNode.getLeaderId() != null
? String.valueOf(raftNode.getLeaderId().getEndpoint())
: "unknown";
long timeoutMs = config.getRpcTimeout();
log.warn(
"Failed to get leader gRPC address via RPC, falling back to endpoint derivation: " +
"leaderEndpoint={}, timeoutMs={}, errorType={}, errorMessage={}",
leaderEndpoint, timeoutMs, e.getClass().getSimpleName(), e.getMessage());
log.debug("Stack trace for failed getLeaderGrpcAddress RPC", e);
```
--
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: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(pd): add timeout and null-safety to getLeaderGrpcAddress() [hugegraph]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2961:
URL: https://github.com/apache/hugegraph/pull/2961#discussion_r2958539228
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -228,19 +229,57 @@ public PeerId getLeader() {
}
/**
- * Send a message to the leader to get the grpc address;
+ * Send a message to the leader to get the grpc address.
*/
public String getLeaderGrpcAddress() throws ExecutionException, InterruptedException {
if (isLeader()) {
return config.getGrpcAddress();
}
if (raftNode.getLeaderId() == null) {
- waitingForLeader(10000);
+ waitingForLeader(config.getRpcTimeout());
Review Comment:
`getLeaderGrpcAddress()` now uses `waitingForLeader(config.getRpcTimeout())`, but `waitingForLeader()` sleeps in fixed 1000ms intervals (`wait(1000)`), so if `raft.rpc-timeout` is configured < 1000ms this call can still block ~1s (and generally exceed the intended timeout). Consider updating `waitingForLeader()` to wait for `min(1000, remainingTime)` or otherwise ensure the method respects the caller-provided timeout more closely.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -228,19 +229,57 @@ public PeerId getLeader() {
}
/**
- * Send a message to the leader to get the grpc address;
+ * Send a message to the leader to get the grpc address.
*/
public String getLeaderGrpcAddress() throws ExecutionException, InterruptedException {
if (isLeader()) {
return config.getGrpcAddress();
}
if (raftNode.getLeaderId() == null) {
- waitingForLeader(10000);
+ waitingForLeader(config.getRpcTimeout());
+ }
+
+ // Cache leader to avoid repeated getLeaderId() calls and guard against
+ // waitingForLeader() returning without a leader being elected.
+ PeerId leader = raftNode.getLeaderId();
+ if (leader == null) {
+ throw new ExecutionException(new IllegalStateException("Leader is not ready"));
+ }
+
+ RaftRpcProcessor.GetMemberResponse response = null;
+ try {
+ // TODO: a more complete fix would need a source of truth for the leader's
+ // actual grpcAddress rather than deriving it from the local node's port config.
+ response = raftRpcClient
+ .getGrpcAddress(leader.getEndpoint().toString())
+ .get(config.getRpcTimeout(), TimeUnit.MILLISECONDS);
+ if (response != null && response.getGrpcAddress() != null) {
+ return response.getGrpcAddress();
+ }
+ if (response == null) {
+ log.warn("Leader RPC response is null for {}, falling back to derived address",
+ leader);
+ } else {
+ log.warn("Leader gRPC address field is null in RPC response for {}, "
+ + "falling back to derived address", leader);
+ }
+ } catch (TimeoutException e) {
+ log.warn("Timed out resolving leader gRPC address for {}, falling back to derived "
+ + "address", leader);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause() != null ? e.getCause() : e;
+ log.warn("Failed to resolve leader gRPC address for {}, falling back to derived "
+ + "address", leader, cause);
}
- return raftRpcClient.getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString()).get()
- .getGrpcAddress();
+ // Best-effort fallback: derive from leader raft endpoint IP + local gRPC port.
+ // WARNING: this may be incorrect in clusters where PD nodes use different grpc.port
+ // values , a proper fix requires a cluster-wide source of truth for gRPC addresses.
Review Comment:
Typo/spacing in the warning comment: `values ,` has an extra space before the comma. Please fix to `values,` (and consider rewrapping the comment so it reads cleanly).
##########
hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/raft/RaftEngineLeaderAddressTest.java:
##########
@@ -0,0 +1,181 @@
+/*
+ * 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.pd.raft;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.hugegraph.pd.config.PDConfig;
+import org.apache.hugegraph.testutil.Whitebox;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.alipay.sofa.jraft.Node;
+import com.alipay.sofa.jraft.entity.PeerId;
+import com.alipay.sofa.jraft.util.Endpoint;
+
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class RaftEngineLeaderAddressTest {
+
+ private static final String LEADER_IP = "10.0.0.1";
+ private static final int GRPC_PORT = 8686;
+ private static final String LEADER_GRPC_ADDRESS = "10.0.0.1:8686";
+
+ private Node originalRaftNode;
+ private RaftRpcClient originalRaftRpcClient;
+ private PDConfig.Raft originalConfig;
+
+ private Node mockNode;
+ private RaftRpcClient mockRpcClient;
+ private PDConfig.Raft mockConfig;
+ private PeerId mockLeader;
+
+ @Before
+ public void setUp() {
+ RaftEngine engine = RaftEngine.getInstance();
+
+ // Save originals
+ originalRaftNode = engine.getRaftNode();
+ originalRaftRpcClient = Whitebox.getInternalState(engine, "raftRpcClient");
+ originalConfig = Whitebox.getInternalState(engine, "config");
+
+ // Build mock leader PeerId with real Endpoint
+ mockLeader = mock(PeerId.class);
+ Endpoint endpoint = new Endpoint(LEADER_IP, 8610);
+ when(mockLeader.getEndpoint()).thenReturn(endpoint);
+
+ // Build mock Node that reports itself as follower with a known leader
+ mockNode = mock(Node.class);
+ when(mockNode.isLeader(true)).thenReturn(false);
+ when(mockNode.getLeaderId()).thenReturn(mockLeader);
+
+ // Build mock config
+ // Use a short timeout (100ms) so the null-leader test doesn't block for seconds
+ mockConfig = mock(PDConfig.Raft.class);
+ when(mockConfig.getGrpcAddress()).thenReturn("127.0.0.1:" + GRPC_PORT);
+ when(mockConfig.getGrpcPort()).thenReturn(GRPC_PORT);
+ when(mockConfig.getRpcTimeout()).thenReturn(100);
+
+ // Build mock RpcClient
+ mockRpcClient = mock(RaftRpcClient.class);
+
+ // Inject mocks
+ Whitebox.setInternalState(engine, "raftNode", mockNode);
+ Whitebox.setInternalState(engine, "raftRpcClient", mockRpcClient);
+ Whitebox.setInternalState(engine, "config", mockConfig);
+ }
+
+ @After
+ public void tearDown() {
+ RaftEngine engine = RaftEngine.getInstance();
+ Whitebox.setInternalState(engine, "raftNode", originalRaftNode);
+ Whitebox.setInternalState(engine, "raftRpcClient", originalRaftRpcClient);
+ Whitebox.setInternalState(engine, "config", originalConfig);
+ }
+
+ @Test
+ public void testSuccessReturnsGrpcAddress() throws Exception {
+ // RPC succeeds and returns a valid gRPC address
+ RaftRpcProcessor.GetMemberResponse response =
+ mock(RaftRpcProcessor.GetMemberResponse.class);
+ when(response.getGrpcAddress()).thenReturn(LEADER_GRPC_ADDRESS);
+
+ CompletableFuture<RaftRpcProcessor.GetMemberResponse> future =
+ CompletableFuture.completedFuture(response);
+ when(mockRpcClient.getGrpcAddress(anyString())).thenReturn(future);
+
+ String result = RaftEngine.getInstance().getLeaderGrpcAddress();
+ Assert.assertEquals(LEADER_GRPC_ADDRESS, result);
+ }
+
+ @Test
+ public void testTimeoutFallsBackToDerivedAddress() throws Exception {
+ // RPC times out — should fall back to leaderIp:grpcPort
+ CompletableFuture<RaftRpcProcessor.GetMemberResponse> future =
+ mock(CompletableFuture.class);
+ when(future.get(anyLong(), eq(TimeUnit.MILLISECONDS)))
+ .thenThrow(new TimeoutException("simulated timeout"));
+ when(mockRpcClient.getGrpcAddress(anyString())).thenReturn(future);
+
+ String result = RaftEngine.getInstance().getLeaderGrpcAddress();
+ Assert.assertEquals(LEADER_IP + ":" + GRPC_PORT, result);
+ }
+
+ @Test
+ public void testRpcExceptionFallsBackToDerivedAddress() throws Exception {
+ // RPC throws ExecutionException — should fall back to leaderIp:grpcPort
+ CompletableFuture<RaftRpcProcessor.GetMemberResponse> future =
+ mock(CompletableFuture.class);
+ when(future.get(anyLong(), eq(TimeUnit.MILLISECONDS)))
+ .thenThrow(new ExecutionException("simulated rpc failure",
+ new RuntimeException("bolt error")));
+ when(mockRpcClient.getGrpcAddress(anyString())).thenReturn(future);
+
+ String result = RaftEngine.getInstance().getLeaderGrpcAddress();
+ Assert.assertEquals(LEADER_IP + ":" + GRPC_PORT, result);
+ }
+
+ @Test
+ public void testNullResponseFallsBackToDerivedAddress() throws Exception {
+ // RPC returns null response — should fall back to leaderIp:grpcPort
+ CompletableFuture<RaftRpcProcessor.GetMemberResponse> future =
+ CompletableFuture.completedFuture(null);
+ when(mockRpcClient.getGrpcAddress(anyString())).thenReturn(future);
+
+ String result = RaftEngine.getInstance().getLeaderGrpcAddress();
+ Assert.assertEquals(LEADER_IP + ":" + GRPC_PORT, result);
+ }
+
+ @Test
+ public void testNullGrpcAddressInResponseFallsBackToDerivedAddress() throws Exception {
+ // RPC returns a response but grpcAddress field is null — should fall back
+ RaftRpcProcessor.GetMemberResponse response =
+ mock(RaftRpcProcessor.GetMemberResponse.class);
+ when(response.getGrpcAddress()).thenReturn(null);
+
+ CompletableFuture<RaftRpcProcessor.GetMemberResponse> future =
+ CompletableFuture.completedFuture(response);
+ when(mockRpcClient.getGrpcAddress(anyString())).thenReturn(future);
+
+ String result = RaftEngine.getInstance().getLeaderGrpcAddress();
+ Assert.assertEquals(LEADER_IP + ":" + GRPC_PORT, result);
+ }
+
+ @Test
+ public void testNullLeaderAfterWaitThrowsExecutionException() throws Exception {
+ // Leader is still null after waitingForLeader() — should throw ExecutionException
+ when(mockNode.getLeaderId()).thenReturn(null);
+
+ try {
+ RaftEngine.getInstance().getLeaderGrpcAddress();
+ Assert.fail("Expected ExecutionException");
Review Comment:
`testNullLeaderAfterWaitThrowsExecutionException()` is still likely to block for ~1s even though `mockConfig.getRpcTimeout()` is set to 100ms, because `RaftEngine.waitingForLeader()` uses `wait(1000)`. To keep the unit test fast, consider overriding `getRpcTimeout()` to return 0 for this test case (so `waitingForLeader(0)` returns immediately) or adjusting `waitingForLeader()` to respect sub-second timeouts.
--
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: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] fix(pd): add timeout and null-safety to getLeaderGrpcAddress() [hugegraph]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #2961:
URL: https://github.com/apache/hugegraph/pull/2961#discussion_r2909775766
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -239,8 +241,20 @@ public String getLeaderGrpcAddress() throws ExecutionException, InterruptedExcep
waitingForLeader(10000);
}
- return raftRpcClient.getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString()).get()
- .getGrpcAddress();
+ try {
+ RaftRpcProcessor.GetMemberResponse response = raftRpcClient
+ .getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString())
+ .get(config.getRpcTimeout(), TimeUnit.MILLISECONDS);
Review Comment:
⚠️ `config.getRpcTimeout()` only bounds the callers `CompletableFuture.get(...)` here; it does **not** bound the underlying Bolt RPC. `RaftRpcClient` is still initialized with a fresh `new RpcOptions()` and `invokeAsync(...)` uses `rpcOptions.getRpcDefaultTimeout()`, so in the bridge-network failure case well return from this method quickly but still leave the actual RPC running until the client default timeout fires. On a follower that keeps redirecting traffic, those in-flight requests can accumulate.
Could we wire the configured timeout into `RaftRpcClient` itself (or cancel the future on timeout) so the transport timeout and the caller timeout stay consistent?
```text
Follower Request
|
v
getLeaderGrpcAddress()
|
+--> future.get(config.rpcTimeout)
| |
| +--> caller returns early on timeout
|
+--> invokeAsync(..., rpcOptions.getRpcDefaultTimeout())
|
+--> underlying Bolt RPC may still be running
|
v
repeated follower redirects
|
v
in-flight RPC requests can accumulate
```
--
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: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org