Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
gaoyuan5251 wants to merge 1 commit into
base: release-1.7.0
Choose a base branch
from
+99 −23
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
117 changes: 94 additions & 23 deletions hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
Expand Down Expand Up @@ -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<>();

@bitflicker64 bitflicker64 Sep 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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


public RaftEngine() {
this.stateMachine = new RaftStateMachine();
Expand All @@ -86,6 +92,10 @@ public synchronized boolean init(PDConfig.Raft config) {
}
this.config = config;

// Build static mapping: Raft endpoint → gRPC address from config

@bitflicker64 bitflicker64 Sep 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

// This avoids using the broken BoltRpcClient for address resolution
buildPeerGrpcAddressMap();

raftRpcClient = new RaftRpcClient();
raftRpcClient.init(new RpcOptions());

Expand Down Expand Up @@ -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");
}
});
Comment on lines +146 to +152

@bitflicker64 bitflicker64 Sep 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

log.info("RaftEngine start successfully: id = {}, peers list = {}", groupId,
nodeOptions.getInitialConf().getPeers());
return this.raftNode != null;
Expand Down Expand Up @@ -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);

@bitflicker64 bitflicker64 Sep 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

String grpcAddr = ep.getIp() + ":" + grpcPort;
peerGrpcAddressMap.put(ep.toString(), grpcAddr);
Comment on lines +254 to +263

@bitflicker64 bitflicker64 Sep 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

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();

@bitflicker64 bitflicker64 Sep 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

String address = peerGrpcAddressMap.get(leaderEndpoint);
if (address != null) {
this.cachedLeaderGrpcAddress = address;
return address;
}

// Fallback to Bolt RPC (may fail due to BoltRpcClient bug)
log.warn("Leader endpoint {} not in static map, falling back to Bolt RPC", leaderEndpoint);
address = raftRpcClient.getGrpcAddress(leaderEndpoint).get().getGrpcAddress();
this.cachedLeaderGrpcAddress = address;
return address;
}

/**
Expand Down Expand Up @@ -269,8 +326,6 @@ public List<Metapb.Member> getMembers() throws ExecutionException, InterruptedEx
for (PeerId peerId : peers) {
Metapb.Member.Builder builder = Metapb.Member.newBuilder();
builder.setClusterId(config.getClusterId());
CompletableFuture<RaftRpcProcessor.GetMemberResponse> future =
raftRpcClient.getGrpcAddress(peerId.getEndpoint().toString());

Metapb.ShardRole role = Metapb.ShardRole.Follower;
if (PeerUtil.isPeerEquals(peerId, raftNode.getLeaderId())) {
Expand All @@ -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());
Comment on lines +346 to +354

@bitflicker64 bitflicker64 Sep 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

} else {
// Fallback to Bolt RPC
try {
CompletableFuture<RaftRpcProcessor.GetMemberResponse> future =
raftRpcClient.getGrpcAddress(endpointStr);
if (future.isCompletedExceptionally()) {
log.error("failed to getGrpcAddress of {}", endpointStr);
builder.setState(Metapb.StoreState.Offline);
builder.setRaftUrl(endpointStr);
members.add(builder.build());
} else {
RaftRpcProcessor.GetMemberResponse response = future.get();
builder.setState(Metapb.StoreState.Up);
builder.setRaftUrl(response.getRaftAddress());
builder.setDataPath(response.getDatePath());
builder.setGrpcUrl(response.getGrpcAddress());
builder.setRestUrl(response.getRestAddress());
members.add(builder.build());
}
} catch (Exception e) {
log.error("failed to getGrpcAddress of {}.", endpointStr, e);
builder.setState(Metapb.StoreState.Offline);
builder.setRaftUrl(peerId.getEndpoint().toString());
members.add(builder.build());
} else {
RaftRpcProcessor.GetMemberResponse response = future.get();
builder.setState(Metapb.StoreState.Up);
builder.setRaftUrl(response.getRaftAddress());
builder.setDataPath(response.getDatePath());
builder.setGrpcUrl(response.getGrpcAddress());
builder.setRestUrl(response.getRestAddress());
builder.setRaftUrl(endpointStr);
members.add(builder.build());
}
} catch (Exception e) {
log.error("failed to getGrpcAddress of {}.", peerId.getEndpoint().toString(), e);
builder.setState(Metapb.StoreState.Offline);
builder.setRaftUrl(peerId.getEndpoint().toString());
members.add(builder.build());
}

}
return members;
}
Expand Down
5 changes: 5 additions & 0 deletions hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/ServiceGrpc.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.MethodDescriptor;
import io.grpc.Status;

public interface ServiceGrpc extends RaftStateListener {

Expand Down Expand Up @@ -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());
Comment on lines +94 to +97

@bitflicker64 bitflicker64 Sep 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

}
}

Expand Down