[GitHub] [ozone] szetszwo commented on a diff in pull request #3493: HDDS-6844. Reduce the number of watch requests in XceiverClientRatis.

[GitBox <gi...@apache.org>]

szetszwo commented on code in PR #3493:
URL: https://github.com/apache/ozone/pull/3493#discussion_r892497306


##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java:
##########
@@ -244,51 +260,58 @@ private void addDatanodetoReply(UUID address, XceiverClientReply reply) {
     reply.addDatanode(builder.build());
   }
 
+  private XceiverClientReply newWatchReply(
+      long watchIndex, Object reason, long replyIndex) {
+    LOG.debug("watchForCommit({}) returns {} {}",
+        watchIndex, reason, replyIndex);
+    final XceiverClientReply reply = new XceiverClientReply(null);
+    reply.setLogIndex(replyIndex);
+    return reply;
+  }
+
   @Override
   public XceiverClientReply watchForCommit(long index)
       throws InterruptedException, ExecutionException, TimeoutException,
       IOException {
-    long commitIndex = getReplicatedMinCommitIndex();
-    XceiverClientReply clientReply = new XceiverClientReply(null);
-    if (commitIndex >= index) {
-      // return the min commit index till which the log has been replicated to
-      // all servers
-      clientReply.setLogIndex(commitIndex);
-      return clientReply;
+    final long replicatedMin = getReplicatedMinCommitIndex();
+    if (replicatedMin >= index) {
+      return newWatchReply(index, "replicatedMin", replicatedMin);
     }
-    RaftClientReply reply;
+
     try {
       CompletableFuture<RaftClientReply> replyFuture = getClient().async()
           .watch(index, RaftProtos.ReplicationLevel.ALL_COMMITTED);
-      replyFuture.get();
+      final RaftClientReply reply = replyFuture.get();
+      final long updated = updateCommitInfosMap(reply);
+      Preconditions.checkState(updated >= index);
+      return newWatchReply(index, ReplicationLevel.ALL_COMMITTED, updated);
     } catch (Exception e) {
       Throwable t = HddsClientUtils.checkForException(e);
       LOG.warn("3 way commit failed on pipeline {}", pipeline, e);
       if (t instanceof GroupMismatchException) {
         throw e;
       }
-      reply = getClient().async()
+      final RaftClientReply reply = getClient().async()
           .watch(index, RaftProtos.ReplicationLevel.MAJORITY_COMMITTED)
           .get();
-      List<RaftProtos.CommitInfoProto> commitInfoProtoList =
-          reply.getCommitInfos().stream()
-              .filter(i -> i.getCommitIndex() < index)
-              .collect(Collectors.toList());
-      commitInfoProtoList.parallelStream().forEach(proto -> {
-        UUID address = RatisHelper.toDatanodeId(proto.getServer());
-        addDatanodetoReply(address, clientReply);
-        // since 3 way commit has failed, the updated map from now on  will
-        // only store entries for those datanodes which have had successful
-        // replication.
-        commitInfoMap.remove(address);
-        LOG.info(
-            "Could not commit index {} on pipeline {} to all the nodes. " +
-            "Server {} has failed. Committed by majority.",
-            index, pipeline, address);
-      });
+      final XceiverClientReply clientReply = newWatchReply(
+          index, ReplicationLevel.MAJORITY_COMMITTED, index);

Review Comment:
   @captainzmc , thanks for the comment.  This is the existing semantic.    Since this is an error case, we probably should not change the semantic.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org