fix(pd): add null-safety and static mapping for leader gRPC address discovery

[gaoyuan5251]

Purpose of the PR

• fix [Bug] describe the main problem #3172

In 1.7.0, after a cluster-wide PD restart the raft leader may land on any node. Followers then fail to redirect requests to the new leader: RaftEngine.getLeaderGrpcAddress() NPEs because the bolt GetMemberRequest RPC response is not null-checked and future.get() has no timeout, and follower PDs keep logging redirect to leader with error: null. Additionally, when the member query bolt RPC fails (even for localhost), getMembers() marks all members Offline, so /v1/members reports stateCountMap: {"Offline": 3} and pdLeader: null while the raft cluster itself is healthy.

This PR fixes the two main symptoms described in #3172 on the release-1.7.0 branch. Note that master already contains partial null-safety for getLeaderGrpcAddress() via #2961, but getMembers() is still unprotected there, and none of these fixes exist in 1.7.0.

Main Changes

• RaftEngine: build a static mapping (raft endpoint -> gRPC address) from the local config at init time, so address resolution no longer depends on the fragile bolt RPC (which can fail even for localhost per [Bug] describe the main problem #3172)
• Cache the resolved leader gRPC address and invalidate the cache on raft leader change via listener
• getMembers(): prefer the static mapping; keep the bolt RPC as fallback for peers missing from the mapping
• ServiceGrpc: complete the gRPC observer with onError(Status.UNAVAILABLE) when redirect-to-leader fails, so client requests fail fast instead of hanging forever

Verifying these changes

• Trivial rework / code cleanup without any test coverage. (No Need)
• Already covered by existing tests, such as (please modify tests here).
• Need tests and can be verified as follows:
  • Deployed on a 3-node PD cluster (CentOS 7.8, JDK 11) built from this branch
  • Restart the whole PD cluster so the leader moves to a non-first node: follower redirect works, no more redirect to leader with error NPE
  • GET /v1/members on the leader returns all members Up with grpcUrl/restUrl filled in and the correct pdLeader

Does this PR potentially affect the following parts?

• Dependencies
• Modify configurations
• The public API
• Other affects (PD internal leader address resolution & follower redirect path)
• Nope

Documentation Status

• Doc - No Need

[bitflicker64]

Blocking: yes. Summary: the static mapping derives every peer's gRPC address from the local node's grpc.port, which is the uniform-port requirement you yourself flagged as a weakness of #2961's fallback in #3172, except that here it is the primary path rather than a post-failure fallback; and getMembers() now reports Up for any peer present in the map, so /v1/members can no longer report a PD as down.

Evidence, all at 2521312e159249ff6136dc5e0f1706fbc3b5d1e9:

• Four co-located PD configurations in this repo already use per-node gRPC ports: docker/configs/application-pd{0,1,2}.yml (grpc.port 8686/8687/8688, server.port 8620/8621/8622, one shared peers-list), hugegraph-pd/hg-pd-service/src/test/resources/application-server{2,3}.yml, and hugegraph-cluster-test/.../ct/config/PDConfig.java:41-43, which draws raftPort, grpcPort and restPort from three independent getAvailablePort() calls per node.
• git show origin/master:.../RaftEngine.java: #2961 reaches the same derived address at :286, but only after the RPC fails, and logs that it "may be incorrect". Its leader == null guard at :252-255 and its .get(config.getRpcTimeout(), TimeUnit.MILLISECONDS) at :263 are both absent here, so the NPE and the unbounded .get() named in the PR description are still on this branch.
• jraft 1.3.13 sources: JRaftUtils.getEndPoint throws unless the string splits into exactly two colon-separated parts, while Configuration.parse at RaftEngine.java:107 logs and ignores such an entry.
• MemberAPI.java:86-87 and :101-103, and IndexAPI.java:208, read member.getState() straight through to /v1/members and the index page.

Findings are static: I did not build the module or stand up a PD cluster, so nothing here is confirmed at runtime. There is also no test covering buildPeerGrpcAddressMap; hugegraph-pd/hg-pd-core/src/test does not exist.

The ServiceGrpc observer completion is a genuine fix and the leader-address cache is a good idea; the comments on those two are about scope, not direction.

[bitflicker64]

‼️ 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.

[bitflicker64]

‼️ 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.

[bitflicker64]

⚠️ 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:

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.

[bitflicker64]

⚠️ 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):

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.

[bitflicker64]

⚠️ 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).

[bitflicker64]

🧹 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.

[bitflicker64]

🧹 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.

[bitflicker64]

🧹 The arrow here is a non-ASCII character, and the only one in the file. Worth replacing with -> to keep the source ASCII-clean.