RATIS-2333. Fix TestInstallSnapshotNotificationWithGrpc failure. (#1289)

Author: szetszwo

ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcUtil.java

@@ -138,25 +138,27 @@ static Throwable tryUnwrapThrowable(StatusRuntimeException se) {
   static long getCallId(Throwable t) {
     if (t instanceof StatusRuntimeException) {
       final Metadata trailers = ((StatusRuntimeException)t).getTrailers();
-      String callId = trailers.get(CALL_ID);
-      return callId != null ? Integer.parseInt(callId) : -1;
+      if (trailers != null) {
+        final String callId = trailers.get(CALL_ID);
+        return callId != null ? Integer.parseInt(callId) : -1;
+      }
     }
     return -1;
   }
 
   static boolean isHeartbeat(Throwable t) {
     if (t instanceof StatusRuntimeException) {
       final Metadata trailers = ((StatusRuntimeException)t).getTrailers();
-      String isHeartbeat = trailers != null ? trailers.get(HEARTBEAT) : null;
-      return isHeartbeat != null && Boolean.valueOf(isHeartbeat);
+      final String isHeartbeat = trailers != null ? trailers.get(HEARTBEAT) : null;
+      return Boolean.parseBoolean(isHeartbeat);
     }
     return false;
   }
 
   static IOException unwrapIOException(Throwable t) {
     final IOException e;
     if (t instanceof StatusRuntimeException) {
-      e = GrpcUtil.unwrapException((StatusRuntimeException) t);
+      e = unwrapException((StatusRuntimeException) t);
     } else {
       e = IOUtils.asIOException(t);
     }
@@ -172,15 +174,15 @@ static <REPLY, REPLY_PROTO> void asyncCall(
       supplier.get().whenComplete((reply, exception) -> {
         if (exception != null) {
           warning.accept(exception);
-          responseObserver.onError(GrpcUtil.wrapException(exception));
+          responseObserver.onError(wrapException(exception));
         } else {
           responseObserver.onNext(toProto.apply(reply));
           responseObserver.onCompleted();
         }
       });
     } catch (Exception e) {
       warning.accept(e);
-      responseObserver.onError(GrpcUtil.wrapException(e));
+      responseObserver.onError(wrapException(e));
     }
   }
 
@@ -189,7 +191,7 @@ static void warn(Logger log, Supplier<String> message, Throwable t) {
   }
 
   class StatusRuntimeExceptionMetadataBuilder {
-    private Metadata trailers = new Metadata();
+    private final Metadata trailers = new Metadata();
 
     StatusRuntimeExceptionMetadataBuilder(Throwable t) {
       trailers.put(EXCEPTION_TYPE_KEY, t.getClass().getCanonicalName());

ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolService.java

@@ -24,6 +24,8 @@
 import org.apache.ratis.server.RaftServer;
 import org.apache.ratis.server.protocol.RaftServerProtocol;
 import org.apache.ratis.server.util.ServerStringUtils;
+import org.apache.ratis.thirdparty.com.google.protobuf.MessageOrBuilder;
+import org.apache.ratis.thirdparty.com.google.protobuf.TextFormat;
 import org.apache.ratis.thirdparty.io.grpc.Status;
 import org.apache.ratis.thirdparty.io.grpc.StatusRuntimeException;
 import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
@@ -67,7 +69,8 @@ CompletableFuture<Void> getFuture() {
     }
   }
 
-  abstract class ServerRequestStreamObserver<REQUEST, REPLY> implements StreamObserver<REQUEST> {
+  abstract class ServerRequestStreamObserver<REQUEST, REPLY extends MessageOrBuilder>
+      implements StreamObserver<REQUEST> {
     private final RaftServer.Op op;
     private final Supplier<String> nameSupplier;
     private final StreamObserver<REPLY> responseObserver;
@@ -172,7 +175,8 @@ public void onCompleted() {
                 getId(), op, getPreviousRequestString(), suffix));
         requestFuture.get().thenAccept(reply -> {
           BatchLogger.print(BatchLogKey.COMPLETED_REPLY, getName(),
-              suffix -> LOG.info("{}: Completed {}, lastReply: {} {}", getId(), op, reply, suffix));
+              suffix -> LOG.info("{}: Completed {}, lastReply: {} {}",
+                  getId(), op, TextFormat.shortDebugString(reply), suffix));
           responseObserver.onCompleted();
         });
       }

ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java

@@ -36,6 +36,7 @@
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.ratis.server.raftlog.LogProtoUtils;
 import org.apache.ratis.server.util.ServerStringUtils;
+import org.apache.ratis.thirdparty.com.google.protobuf.TextFormat;
 import org.apache.ratis.util.BatchLogger;
 import org.apache.ratis.util.CodeInjectionForTesting;
 import org.apache.ratis.util.LifeCycle;
@@ -144,7 +145,7 @@ private InstallSnapshotReplyProto installSnapshotImpl(InstallSnapshotRequestProt
         final LogEntryProto proto = request.getLastRaftConfigurationLogEntryProto();
         state.truncate(proto.getIndex());
         if (!state.getRaftConf().equals(LogProtoUtils.toRaftConfiguration(proto))) {
-          LOG.info("{}: set new configuration {} from snapshot", getMemberId(), proto);
+          LOG.info("{}: set new configuration {} from snapshot", getMemberId(), TextFormat.shortDebugString(proto));
           state.setRaftConf(proto);
           state.writeRaftConfiguration(proto);
           server.getStateMachine().event().notifyConfigurationChanged(

ratis-server/src/test/java/org/apache/ratis/InstallSnapshotNotificationTests.java

@@ -79,6 +79,7 @@ public abstract class InstallSnapshotNotificationTests<CLUSTER extends MiniRaftC
 
     RaftServerConfigKeys.Log.setPurgeGap(prop, PURGE_GAP);
     RaftServerConfigKeys.Log.setSegmentSizeMax(prop, SizeInBytes.valueOf(1024)); // 1k segment
+    RaftServerConfigKeys.LeaderElection.setMemberMajorityAdd(prop, true);
   }
 
   private static final int SNAPSHOT_TRIGGER_THRESHOLD = 64;
@@ -239,9 +240,8 @@ private void testAddNewFollowers(CLUSTER cluster, int numRequests) throws Except
       final boolean set = LEADER_SNAPSHOT_INFO_REF.compareAndSet(null, leaderSnapshotInfo);
       Assertions.assertTrue(set);
 
-      // add two more peers
-      final MiniRaftCluster.PeerChanges change = cluster.addNewPeers(2, true,
-          true);
+      // Add new peer(s)
+      final MiniRaftCluster.PeerChanges change = cluster.addNewPeers(1, true, true);
       // trigger setConfiguration
       RaftServerTestUtil.runWithMinorityPeers(cluster, Arrays.asList(change.allPeersInNewConf),
           peers -> cluster.setConfiguration(peers.toArray(RaftPeer.emptyArray())));
@@ -389,9 +389,9 @@ private void testInstallSnapshotNotificationCount(CLUSTER cluster) throws Except
             follower.getRaftLog().getNextIndex());
       }
 
-      // Add two more peers who will need snapshots from the leader.
-      final MiniRaftCluster.PeerChanges change = cluster.addNewPeers(2, true,
-          true);
+      // Add new peer(s) who will need snapshots from the leader.
+      final int numNewPeers = 1;
+      final MiniRaftCluster.PeerChanges change = cluster.addNewPeers(numNewPeers, true, true);
       // trigger setConfiguration
       RaftServerTestUtil.runWithMinorityPeers(cluster, Arrays.asList(change.allPeersInNewConf),
           peers -> cluster.setConfiguration(peers.toArray(RaftPeer.emptyArray())));
@@ -412,7 +412,7 @@ private void testInstallSnapshotNotificationCount(CLUSTER cluster) throws Except
       }
 
       // Make sure each new peer got one snapshot notification.
-      Assertions.assertEquals(2, numSnapshotRequests.get());
+      Assertions.assertEquals(numNewPeers, numSnapshotRequests.get());
 
     } finally {
       cluster.shutdown();
@@ -556,9 +556,9 @@ private void testInstallSnapshotDuringBootstrap(CLUSTER cluster) throws Exceptio
       final boolean set = LEADER_SNAPSHOT_INFO_REF.compareAndSet(null, leaderSnapshotInfo);
       Assertions.assertTrue(set);
 
-      // add two more peers
-      final MiniRaftCluster.PeerChanges change = cluster.addNewPeers(2, true,
-          true);
+      // Add new peer(s)
+      final int numNewPeers = 1;
+      final MiniRaftCluster.PeerChanges change = cluster.addNewPeers(numNewPeers, true, true);
       // trigger setConfiguration
       RaftServerTestUtil.runWithMinorityPeers(cluster, Arrays.asList(change.allPeersInNewConf),
           peers -> cluster.setConfiguration(peers.toArray(RaftPeer.emptyArray())));
@@ -573,7 +573,7 @@ private void testInstallSnapshotDuringBootstrap(CLUSTER cluster) throws Exceptio
       }
 
       // Make sure each new peer got at least one snapshot notification.
-      Assertions.assertTrue(2 <= numSnapshotRequests.get());
+      Assertions.assertTrue(numNewPeers <= numSnapshotRequests.get());
     } finally {
       cluster.shutdown();
     }

ratis-server/src/test/java/org/apache/ratis/statemachine/RaftSnapshotBaseTest.java

@@ -96,16 +96,16 @@ public static void assertLeaderContent(MiniRaftCluster cluster) throws Exception
     assertLogContent(leader, true);
   }
 
-  public static void assertLogContent(RaftServer.Division server, boolean isLeader) throws Exception {
+  public static void checkMetadataEntry(RaftServer.Division server) throws Exception {
     final RaftLog log = server.getRaftLog();
     final long lastIndex = log.getLastEntryTermIndex().getIndex();
     final LogEntryProto e = log.get(lastIndex);
     Assertions.assertTrue(e.hasMetadataEntry());
+    Assertions.assertEquals(log.getLastCommittedIndex() - 1, e.getMetadataEntry().getCommitIndex());
+  }
 
-    JavaUtils.attemptRepeatedly(() -> {
-      Assertions.assertEquals(log.getLastCommittedIndex() - 1, e.getMetadataEntry().getCommitIndex());
-      return null;
-    }, 50, BaseTest.HUNDRED_MILLIS, "CheckMetadataEntry", LOG);
+  public static void assertLogContent(RaftServer.Division server, boolean isLeader) throws Exception {
+    JavaUtils.attempt(() -> checkMetadataEntry(server), 50, HUNDRED_MILLIS, "checkMetadataEntry", LOG);
 
     SimpleStateMachine4Testing simpleStateMachine = SimpleStateMachine4Testing.get(server);
     if (isLeader) {