fix(pd): bound member lookups in RaftEngine.getMembers()

hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java

@@ -569,13 +569,15 @@ public List<Metapb.Member> 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());

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

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<PeerId> 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<Metapb.Member> 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());
+    }
 }