From d8eedd0f0667e2640a730e59c488ce3c01d6915c Mon Sep 17 00:00:00 2001 From: ShryukGrandhi <85529990+ShryukGrandhi@users.noreply.github.com> Date: Sat, 24 Oct 2026 20:16:40 -0700 Subject: [PATCH] fix(pd): bound member lookups in RaftEngine.getMembers() getMembers() waited on each peer's GetMemberRequest future without a timeout, so an unresponsive peer could stall /v1/members. Use the same rpc timeout as getLeaderGrpcAddress(), treat a null response as offline, and log the full exception when the bolt call fails. Signed-off-by: ShryukGrandhi <85529990+ShryukGrandhi@users.noreply.github.com> --- .../apache/hugegraph/pd/raft/RaftEngine.java | 8 +++-- .../hugegraph/pd/raft/RaftRpcClient.java | 2 +- .../pd/raft/RaftEngineLeaderAddressTest.java | 35 +++++++++++++++++++ 3 files changed, 41 insertions(+), 4 deletions(-) diff --git a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java index 117f631201..4962200aac 100644 --- a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java +++ b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java @@ -569,13 +569,15 @@ public List getMembers() throws ExecutionException, InterruptedEx builder.setRole(role); try { - if (future.isCompletedExceptionally()) { - log.error("failed to getGrpcAddress of {}", peerId.getEndpoint().toString()); + RaftRpcProcessor.GetMemberResponse response = + future.get(config.getRpcTimeout(), TimeUnit.MILLISECONDS); + if (response == null) { + log.error("failed to getGrpcAddress of {}, null response", + peerId.getEndpoint().toString()); 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()); diff --git a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftRpcClient.java b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftRpcClient.java index 18602101f0..14aa74f70c 100644 --- a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftRpcClient.java +++ b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftRpcClient.java @@ -87,7 +87,7 @@ public void complete(final Object result, final Throwable err) { this.rpcClient.invokeAsync(endpoint, request, invokeCtx, invokeCallback, this.rpcOptions.getRpcDefaultTimeout()); } catch (final Throwable t) { - log.error("failed to call rpc to {}. {}", endpoint, t.getMessage()); + log.error("failed to call rpc to {}.", endpoint, t); closure.failure(t); closure.run(new Status(-1, t.getMessage())); } diff --git a/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/raft/RaftEngineLeaderAddressTest.java b/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/raft/RaftEngineLeaderAddressTest.java index 420b106a27..b592730707 100644 --- a/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/raft/RaftEngineLeaderAddressTest.java +++ b/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/raft/RaftEngineLeaderAddressTest.java @@ -17,12 +17,16 @@ package org.apache.hugegraph.pd.raft; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; 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.pd.grpc.Metapb; import org.apache.hugegraph.testutil.Whitebox; import org.junit.After; import org.junit.Assert; @@ -180,4 +184,35 @@ public void testNullLeaderAfterWaitThrowsExecutionException() throws Exception { Assert.assertEquals("Leader is not ready", e.getCause().getMessage()); } } + + @Test(timeout = 10000) + public void testGetMembersMarksUnresponsivePeerOffline() throws Exception { + PeerId hanging = new PeerId("10.0.0.2", 8610); + PeerId alive = new PeerId(LEADER_IP, 8610); + List peers = new ArrayList<>(); + peers.add(hanging); + peers.add(alive); + when(mockNode.listPeers()).thenReturn(peers); + when(mockNode.listLearners()).thenReturn(Collections.emptyList()); + when(mockNode.getLeaderId()).thenReturn(alive); + + RaftRpcProcessor.GetMemberResponse response = new RaftRpcProcessor.GetMemberResponse(); + response.setRaftAddress(LEADER_IP + ":8610"); + response.setGrpcAddress(LEADER_GRPC_ADDRESS); + response.setDatePath("./pd_data"); + response.setRestAddress(LEADER_IP + ":8620"); + // A peer that never answers must not block the whole member listing + when(mockRpcClient.getGrpcAddress(hanging.getEndpoint().toString())) + .thenReturn(new CompletableFuture<>()); + when(mockRpcClient.getGrpcAddress(alive.getEndpoint().toString())) + .thenReturn(CompletableFuture.completedFuture(response)); + + List members = RaftEngine.getInstance().getMembers(); + Assert.assertEquals(2, members.size()); + Assert.assertEquals(Metapb.StoreState.Offline, members.get(0).getState()); + Assert.assertEquals("10.0.0.2:8610", members.get(0).getRaftUrl()); + Assert.assertEquals(Metapb.StoreState.Up, members.get(1).getState()); + Assert.assertEquals(Metapb.ShardRole.Leader, members.get(1).getRole()); + Assert.assertEquals(LEADER_GRPC_ADDRESS, members.get(1).getGrpcUrl()); + } }