bitflicker64 commented on code in PR #3195:
URL: https://github.com/apache/hugegraph/pull/3195#discussion_r3950916807
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -228,19 +247,57 @@ public PeerId getLeader() {
}
/**
- * Send a message to the leader to get the grpc address;
+ * Build static mapping from Raft endpoint to gRPC address.
+ * Assumes all PD nodes use the same grpc.port (standard deployment).
+ */
+ private void buildPeerGrpcAddressMap() {
+ String grpcPort = String.valueOf(config.getGrpcPort());
+ String[] peers = config.getPeersList().split(",");
+ for (String peer : peers) {
+ peer = peer.trim();
+ if (peer.isEmpty()) {
+ continue;
+ }
+ Endpoint ep = JRaftUtils.getEndPoint(peer);
+ String grpcAddr = ep.getIp() + ":" + grpcPort;
+ peerGrpcAddressMap.put(ep.toString(), grpcAddr);
Review Comment:
‼️ `grpcPort` is the local node's `grpc.port`, applied to every peer, so
this is exactly the constraint you described in #3172: "#2961 的降级地址 = leader
raft IP + 本机 grpc.port, 要求所有 PD 节点 grpc 端口一致". The difference is that #2961
pays that cost only after the RPC fails and logs a warning, whereas here the
map is consulted first, so the constraint now applies on every call.
Four co-located configurations in this repo violate it. On
`docker/configs/application-pd0.yml` (`grpc.port: 8686`), with the shared
`peers-list: 127.0.0.1:8610,127.0.0.1:8611,127.0.0.1:8612`, this loop produces:
```
127.0.0.1:8610 -> 127.0.0.1:8686
127.0.0.1:8611 -> 127.0.0.1:8686 (pd1 is on 8687)
127.0.0.1:8612 -> 127.0.0.1:8686 (pd2 is on 8688)
```
`application-server{2,3}.yml` under `hg-pd-service/src/test/resources` are
the same shape, and `hugegraph-cluster-test/.../ct/config/PDConfig.java:41-43`
gives every node three independent `getAvailablePort()` results, so no two PDs
there ever share a gRPC port. A follower then resolves the leader to its own
address and `redirectToLeader()` loops back to itself, which is quieter but
harder to diagnose than the NPE.
Requested change: try the RPC first with `config.getRpcTimeout()` as #2961
does, fall through to this derived address on failure or timeout, and keep your
cache so a failing RPC is paid once per leader change rather than per request.
That keeps the win you measured without making the uniform-port assumption
unconditional.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -285,28 +340,44 @@ public List<Metapb.Member> getMembers() throws
ExecutionException, InterruptedEx
builder.setRole(role);
- try {
- if (future.isCompletedExceptionally()) {
- log.error("failed to getGrpcAddress of {}",
peerId.getEndpoint().toString());
+ String endpointStr = peerId.getEndpoint().toString();
+
+ // Use static config-based mapping first
+ String grpcAddr = peerGrpcAddressMap.get(endpointStr);
+ if (grpcAddr != null) {
+ builder.setState(Metapb.StoreState.Up);
+ builder.setRaftUrl(endpointStr);
+ builder.setGrpcUrl(grpcAddr);
+ String host = peerId.getIp();
+ builder.setRestUrl(host + ":" + config.getPort());
+ builder.setDataPath(config.getDataPath());
+ members.add(builder.build());
Review Comment:
‼️ Presence in `peerGrpcAddressMap` means "listed in my `raft.peers-list`",
not "reachable", so `Up` is now unconditional. The map is built once in
`init()` from config and never refreshed, not even by `changePeerList()` at
`:385`, so in a normal cluster this branch always wins and the RPC fallback
below is unreachable. `MemberAPI.java:86-87` and `:101-103` build
`stateCountMap` and `numOfNormalService` from `member.getState()`, and
`IndexAPI.java:208` reads it too, so `/v1/members` and the index page can no
longer report a PD as down. #3172 asked for `getMembers()` to gain #2961's
fault tolerance; this removes the liveness signal instead.
Two smaller problems in the same branch. `restUrl` and `dataPath` are taken
from the local `config.getPort()` and `config.getDataPath()` and attributed to
remote peers, so with `docker/configs` all three members report `restUrl:
127.0.0.1:8620`. And `grpcUrl` is now built from the raft endpoint IP, while
`getLeaderGrpcAddress()` on the leader still returns `config.getGrpcAddress()`,
which is `${grpc.host}` based. `MemberAPI.java:89` matches those two strings to
pick `pdLeader`, so wherever `grpc.host` is not spelled identically to the raft
address host, `pdLeader` goes back to null.
`hg-pd-dist/.../conf/application.yml:40-41` tells operators to set `grpc.host`
separately from `raft.address`, so this is reachable.
Requested change: keep a real reachability check behind the reported state,
and fill `grpcUrl`, `restUrl` and `dataPath` only from values the peer itself
reported.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -228,19 +247,57 @@ public PeerId getLeader() {
}
/**
- * Send a message to the leader to get the grpc address;
+ * Build static mapping from Raft endpoint to gRPC address.
+ * Assumes all PD nodes use the same grpc.port (standard deployment).
+ */
+ private void buildPeerGrpcAddressMap() {
+ String grpcPort = String.valueOf(config.getGrpcPort());
+ String[] peers = config.getPeersList().split(",");
+ for (String peer : peers) {
+ peer = peer.trim();
+ if (peer.isEmpty()) {
+ continue;
+ }
+ Endpoint ep = JRaftUtils.getEndPoint(peer);
Review Comment:
⚠️ This makes a `peers-list` entry that is tolerated today into a startup
failure. `Configuration.parse(config.getPeersList())` at `:107` is fed the same
string and, for an entry it cannot handle, logs `Fail to parse peer {} in {},
ignore it.` and carries on. `JRaftUtils.getEndPoint` throws instead (jraft
1.3.13):
```java
final String[] tmps = StringUtils.split(s, ':');
if (tmps.length != 2) {
throw new IllegalArgumentException("Invalid endpoint string: " + s);
}
```
`buildPeerGrpcAddressMap()` runs before that parse and lets the exception
escape `init()`, and `PDService.init()` is `@PostConstruct`, so the Spring
context fails to start. The concrete forms that differ are the `/learner`
postfix `Configuration.parse` strips and the `ip:port:idx[:priority]` shape
`PeerId` accepts.
Requested change: wrap the per-entry conversion in a try/catch that logs and
skips, so a best-effort address cache can never block startup. If you want full
parity, `PeerId.parse` is an instance method rather than a static one; the
postfix handling already exists in this package at `PeerUtil.java:44-51`, which
strips `/leader`, `/learner` and `/follower` before calling
`JRaftUtils.getPeerId`.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -71,6 +73,10 @@ public class RaftEngine {
private RpcServer rpcServer;
private Node raftNode;
private RaftRpcClient raftRpcClient;
+ /** Cached leader gRPC address to avoid repeated Bolt RPC calls */
+ private volatile String cachedLeaderGrpcAddress;
+ /** Static mapping from Raft endpoint to gRPC address, built from config
at init */
+ private Map<String, String> peerGrpcAddressMap = new HashMap<>();
Review Comment:
🧹 `peerGrpcAddressMap` is a plain `HashMap` populated in `init()` and then
read without synchronization from gRPC and REST worker threads, in
`getLeaderGrpcAddress()` and `getMembers()`. `init()` is `synchronized` but the
readers are not, so there is no happens-before edge and a reader can observe an
empty or half-built map, which silently falls through to the RPC path this
change sets out to avoid.
Requested change: `private final Map<String, String> peerGrpcAddressMap =
new ConcurrentHashMap<>();`. Note that this also needs `import
java.util.concurrent.ConcurrentHashMap;` and the removal of the now-unused
`import java.util.HashMap;` added at `:25`, since `style/checkstyle.xml:59`
enables `UnusedImports`.
##########
hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/ServiceGrpc.java:
##########
@@ -90,6 +91,10 @@ default <ReqT, RespT> void redirectToLeader(ManagedChannel
channel,
observer);
} catch (Exception e) {
log.warn("redirect to leader with error:", e);
+ // Properly complete the observer to avoid hanging client requests
+ observer.onError(Status.UNAVAILABLE
+ .withDescription("Failed to redirect to leader: " +
e.getMessage())
+ .asRuntimeException());
Review Comment:
🧹 Completing the observer is the right call: before this, a redirect failure
left the client waiting on its own deadline. The `try` is a little wider than
it needs to be, though. In grpc-stub 1.39.0, `asyncUnaryRequestCall` runs
`startCall` outside its own try and only guards `sendMessage`/`halfClose`,
whose failure path is `cancelThrow`, which calls `call.cancel(null, t)` and
rethrows. That cancellation delivers `onClose` to the same observer on the call
executor, so a failure from `asyncUnaryCall` at `:90` can race this handler and
surface as `IllegalStateException: call already closed`, hiding the original
cause.
A failure before the call starts, which is the case you are fixing, is
unaffected either way.
Requested change: track a `boolean started` set immediately before
`asyncUnaryCall` and only call `observer.onError` when it is false. Moving
`asyncUnaryCall` out of the `try` works too, though both touch `:66` and `:90`,
which are outside this diff.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -228,19 +247,57 @@ public PeerId getLeader() {
}
/**
- * Send a message to the leader to get the grpc address;
+ * Build static mapping from Raft endpoint to gRPC address.
+ * Assumes all PD nodes use the same grpc.port (standard deployment).
+ */
+ private void buildPeerGrpcAddressMap() {
+ String grpcPort = String.valueOf(config.getGrpcPort());
+ String[] peers = config.getPeersList().split(",");
+ for (String peer : peers) {
+ peer = peer.trim();
+ if (peer.isEmpty()) {
+ continue;
+ }
+ Endpoint ep = JRaftUtils.getEndPoint(peer);
+ String grpcAddr = ep.getIp() + ":" + grpcPort;
+ peerGrpcAddressMap.put(ep.toString(), grpcAddr);
+ log.info("Mapped Raft endpoint {} -> gRPC {}", ep, grpcAddr);
+ }
+ }
+
+ /**
+ * Get the leader's gRPC address.
+ * First tries the static config-based mapping (no Bolt RPC needed),
+ * then falls back to cached address, then Bolt RPC as last resort.
*/
public String getLeaderGrpcAddress() throws ExecutionException,
InterruptedException {
if (isLeader()) {
return config.getGrpcAddress();
}
+ // Try cached address first
+ String cached = this.cachedLeaderGrpcAddress;
+ if (cached != null) {
+ return cached;
+ }
+
if (raftNode.getLeaderId() == null) {
waitingForLeader(10000);
}
- return
raftRpcClient.getGrpcAddress(raftNode.getLeaderId().getEndpoint().toString()).get()
- .getGrpcAddress();
+ // Use static config-based mapping (avoids broken Bolt RPC)
+ String leaderEndpoint =
raftNode.getLeaderId().getEndpoint().toString();
Review Comment:
⚠️ `waitingForLeader(long)` at `:407-427` returns null when the timeout
expires without an election, and its return value is discarded at `:284-286`,
so `raftNode.getLeaderId()` here can still be null and this line still throws
the NPE that #3172 reports. `:298` likewise still calls `.get()` with no
timeout and no null check on the response, which the PR description names as a
root cause.
#2961 guards both, at `origin/master:.../RaftEngine.java:252-255`:
```java
PeerId leader = raftNode.getLeaderId();
if (leader == null) {
throw new ExecutionException(new IllegalStateException("Leader is not
ready"));
}
```
and calls `.get(config.getRpcTimeout(), TimeUnit.MILLISECONDS)` at `:263`.
Requested change: port both guards here. The static map makes this path
rarer but does not close it, and if the RPC is restored to first position it
becomes the normal path again.
Separately, and outside this diff so treat it as optional: `:100` still does
`raftRpcClient.init(new RpcOptions())`, so the client keeps the default RPC
timeout rather than `config.getRpcTimeout()`. #2961 changed that line too.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -86,6 +92,10 @@ public synchronized boolean init(PDConfig.Raft config) {
}
this.config = config;
+ // Build static mapping: Raft endpoint → gRPC address from config
Review Comment:
🧹 The arrow here is a non-ASCII character, and the only one in the file.
Worth replacing with `->` to keep the source ASCII-clean.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java:
##########
@@ -131,6 +141,15 @@ public synchronized boolean init(PDConfig.Raft config) {
this.raftGroupService =
new RaftGroupService(groupId, serverId, nodeOptions,
rpcServer, true);
this.raftNode = raftGroupService.start(false);
+
+ // Register listener to invalidate cached leader address on leader
change
+ this.stateMachine.addStateListener(new RaftStateListener() {
+ @Override
+ public void onRaftLeaderChanged() {
+ cachedLeaderGrpcAddress = null;
+ log.info("Raft leader changed, invalidated cached leader gRPC
address");
+ }
+ });
Review Comment:
⚠️ This listener only fires on leader gain. In `RaftStateMachine`,
`onLeaderStart` (`:117-127`) and `onStartFollowing` (`:137-144`) fan out to
`onRaftLeaderChanged`, but `onLeaderStop` (`:129-134`) and `onStopFollowing`
(`:146-149`) do not. So between losing a leader and following the next one,
`getLeaderGrpcAddress()` keeps returning the stale `cachedLeaderGrpcAddress` at
`:279-282`, before the `raftNode.getLeaderId()` check that would have caught
it. During an election that window is the whole point at which redirects need
to be correct.
Registration also happens after `raftGroupService.start(false)` at `:143`,
so a transition completing during startup is missed. That one is harmless today
because the cache starts null, but it is fragile.
Requested change: add the fan-out to `onLeaderStop` and `onStopFollowing`
(or clear the cache whenever `getLeaderId()` disagrees with the cached
endpoint), and register the listener before `start(false)`.
--
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]
