Skip to content

fix(pd): add timeout and null-safety to getLeaderGrpcAddress() #2961

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
bitflicker64 wants to merge 3 commits into
base: master
Choose a base branch
from
Open

fix(pd): add timeout and null-safety to getLeaderGrpcAddress() #2961

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fec2963
fix(pd): add timeout and null-safety to getLeaderGrpcAddress()
bitflicker64 Mar 5, 2026
9a13191
ci: trigger rerun
bitflicker64 Mar 12, 2026
8576da0
fix: split ExecutionException catch and remove duplicate setEnableMet…
bitflicker64 Mar 17, 2026
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
65 changes: 42 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 @@ -28,6 +28,7 @@
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

Expand All @@ -49,7 +50,6 @@
import com.alipay.sofa.jraft.entity.Task;
import com.alipay.sofa.jraft.error.RaftError;
import com.alipay.sofa.jraft.option.NodeOptions;
import com.alipay.sofa.jraft.option.RaftOptions;
import com.alipay.sofa.jraft.option.RpcOptions;
import com.alipay.sofa.jraft.rpc.RaftRpcServerFactory;
import com.alipay.sofa.jraft.rpc.RpcServer;
Expand Down Expand Up @@ -86,8 +86,12 @@ public synchronized boolean init(PDConfig.Raft config) {
}
this.config = config;

// Wire configured rpc timeout into RaftRpcClient so the Bolt transport
// timeout and the future.get() caller timeout in getLeaderGrpcAddress() are consistent.
raftRpcClient = new RaftRpcClient();
raftRpcClient.init(new RpcOptions());
RpcOptions rpcOptions = new RpcOptions();
rpcOptions.setRpcDefaultTimeout(config.getRpcTimeout());
raftRpcClient.init(rpcOptions);

String raftPath = config.getDataPath() + "/" + groupId;
new File(raftPath).mkdirs();
Expand Down Expand Up @@ -119,10 +123,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();

nodeOptions.setEnableMetrics(true);
// TODO: tune RaftOptions for PD (see hugegraph-store PartitionEngine for reference)

final PeerId serverId = JRaftUtils.getPeerId(config.getAddress());

Expand Down Expand Up @@ -228,19 +229,49 @@ 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 {
Copy link
Member

@imbajin imbajin Mar 18, 2026 •
edited
Loading

Choose a reason for hiding this comment

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

⚠️ There is still no focused test covering the new getLeaderGrpcAddress() failure paths here. The current PD test additions exercise changePeerList() / IpAuthHandler, but not the timeout, RPC-exception, or null-response branches introduced in this method.

Could add a small UIT with a mocked RaftRpcClient so regressions in the leader-resolution path are caught by CI.

RaftRpcProcessor.GetMemberResponse response = raftRpcClient
.getGrpcAddress(leader.getEndpoint().toString())
.get(config.getRpcTimeout(), TimeUnit.MILLISECONDS);
if (response != null && response.getGrpcAddress() != null) {
return response.getGrpcAddress();
}
} catch (TimeoutException 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("Timed out while resolving leader gRPC address for %s", leader),
e);
} catch (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.
Throwable cause = e.getCause() != null ? e.getCause() : e;
throw new ExecutionException(
String.format("Failed to resolve leader gRPC address for %s", leader), cause);
}
Comment on lines +257 to +269
Copy link
Member

@imbajin imbajin Mar 18, 2026 •
edited
Loading

Choose a reason for hiding this comment

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

Yep, the PR said this method should fall back to deriving the leader gRPC address from the Raft leader endpoint plus the local gRPC port when the Bolt RPC hangs / fails / returns null.

However, in the current implementation, all of those paths still end by throwing ExecutionException:

  • TimeoutException -> rethrown as ExecutionException
  • RPC failure (ExecutionException) -> rethrown
  • response == null or response.getGrpcAddress() == null -> throws again after the warning

That means the follower still does not get a usable leader gRPC address in the exact scenario this PR is trying to fix.

Contextually, the call chain still looks like this:

Follower request
    |
    v
getLeaderGrpcAddress()
    |
    +-- Bolt RPC timeout / failure / null response
            |
            v
      throws ExecutionException
            |
            v
redirectToLeader()
    |
    +-- logs / catches exception
    |
    v
request is not forwarded to leader

Current:
  RPC fails/null
      -> throw
      -> no redirect

Expected:
  RPC fails/null
      -> derive leaderIp:grpcPort
      -> redirect continues


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)));
Comment on lines +271 to +274
}

/**
Expand Down Expand Up @@ -322,14 +353,7 @@ public Status changePeerList(String peerList) {
newPeers.parse(peerList);
CountDownLatch latch = new CountDownLatch(1);
this.raftNode.changePeers(newPeers, status -> {
// Use compareAndSet so a late callback does not overwrite a timeout status
result.compareAndSet(null, status);
// Refresh inside callback so it fires even if caller already timed out
// Note: changePeerList() uses Configuration.parse() which only supports
// plain comma-separated peer addresses with no learner syntax.
// getLearners() will always be empty here. Learner support is handled
// in PDService.updatePdRaft() which uses PeerUtil.parseConfig()
// and supports the /learner suffix.
if (status != null && status.isOk()) {
IpAuthHandler handler = IpAuthHandler.getInstance();
if (handler != null) {
Expand All @@ -347,16 +371,12 @@ public Status changePeerList(String peerList) {
}
latch.countDown();
});
// Use 3x configured RPC timeout — bare await() would block forever if
// the callback is never invoked (e.g. node not started / RPC failure)
boolean completed = latch.await(3L * config.getRpcTimeout(),
TimeUnit.MILLISECONDS);
boolean completed = latch.await(3L * config.getRpcTimeout(), TimeUnit.MILLISECONDS);
if (!completed && result.get() == null) {
Status timeoutStatus = new Status(RaftError.EINTERNAL,
"changePeerList timed out after %d ms",
3L * config.getRpcTimeout());
if (!result.compareAndSet(null, timeoutStatus)) {
// Callback arrived just before us — keep its result
timeoutStatus = null;
}
if (timeoutStatus != null) {
Expand Down Expand Up @@ -395,7 +415,6 @@ public PeerId waitingForLeader(long timeOut) {
}
return leader;
}

}

public Node getRaftNode() {
Expand Down
Loading