From d425686df6b71b13a02d9ab7cf48476ae14a35d1 Mon Sep 17 00:00:00 2001
From: Ivan Andika <ivan_andika3@rocketmail.com>
Date: Wed, 29 May 2024 02:43:23 +0800
Subject: [PATCH] RATIS-2089. Add CommitInfoProto in NotReplicatedException
 (#1105)

---
 .../ratis/client/impl/ClientProtoUtils.java       |  3 ++-
 .../exceptions/NotReplicatedException.java        | 15 +++++++++++++++
 .../java/org/apache/ratis/WatchRequestTests.java  |  1 +
 3 files changed, 18 insertions(+), 1 deletion(-)

diff --git a/ratis-client/src/main/java/org/apache/ratis/client/impl/ClientProtoUtils.java b/ratis-client/src/main/java/org/apache/ratis/client/impl/ClientProtoUtils.java
index cab9606a0e..5e217e7da2 100644
--- a/ratis-client/src/main/java/org/apache/ratis/client/impl/ClientProtoUtils.java
+++ b/ratis-client/src/main/java/org/apache/ratis/client/impl/ClientProtoUtils.java
@@ -398,7 +398,8 @@ static RaftClientReply toRaftClientReply(RaftClientReplyProto replyProto) {
       e = new NotLeaderException(serverMemberId, suggestedLeader, peers);
     } else if (replyProto.getExceptionDetailsCase() == NOTREPLICATEDEXCEPTION) {
       final NotReplicatedExceptionProto nre = replyProto.getNotReplicatedException();
-      e = new NotReplicatedException(nre.getCallId(), nre.getReplication(), nre.getLogIndex());
+      e = new NotReplicatedException(nre.getCallId(), nre.getReplication(), nre.getLogIndex(),
+          replyProto.getCommitInfosList());
     } else if (replyProto.getExceptionDetailsCase().equals(STATEMACHINEEXCEPTION)) {
       e = toStateMachineException(serverMemberId, replyProto.getStateMachineException());
     } else if (replyProto.getExceptionDetailsCase().equals(DATASTREAMEXCEPTION)) {
diff --git a/ratis-common/src/main/java/org/apache/ratis/protocol/exceptions/NotReplicatedException.java b/ratis-common/src/main/java/org/apache/ratis/protocol/exceptions/NotReplicatedException.java
index 5f48654eec..37ff816245 100644
--- a/ratis-common/src/main/java/org/apache/ratis/protocol/exceptions/NotReplicatedException.java
+++ b/ratis-common/src/main/java/org/apache/ratis/protocol/exceptions/NotReplicatedException.java
@@ -17,12 +17,17 @@
  */
 package org.apache.ratis.protocol.exceptions;
 
+import org.apache.ratis.proto.RaftProtos.CommitInfoProto;
 import org.apache.ratis.proto.RaftProtos.ReplicationLevel;
 
+import java.util.Collection;
+
 public class NotReplicatedException extends RaftException {
   private final long callId;
   private final ReplicationLevel requiredReplication;
   private final long logIndex;
+  /** This is only populated on client-side since RaftClientReply already has commitInfos */
+  private Collection<CommitInfoProto> commitInfos;
 
   public NotReplicatedException(long callId, ReplicationLevel requiredReplication, long logIndex) {
     super("Request with call Id " + callId + " and log index " + logIndex
@@ -32,6 +37,12 @@ public NotReplicatedException(long callId, ReplicationLevel requiredReplication,
     this.logIndex = logIndex;
   }
 
+  public NotReplicatedException(long callId, ReplicationLevel requiredReplication, long logIndex,
+                                Collection<CommitInfoProto> commitInfos) {
+    this(callId, requiredReplication, logIndex);
+    this.commitInfos = commitInfos;
+  }
+
   public long getCallId() {
     return callId;
   }
@@ -43,4 +54,8 @@ public ReplicationLevel getRequiredReplication() {
   public long getLogIndex() {
     return logIndex;
   }
+
+  public Collection<CommitInfoProto> getCommitInfos() {
+    return commitInfos;
+  }
 }
diff --git a/ratis-server/src/test/java/org/apache/ratis/WatchRequestTests.java b/ratis-server/src/test/java/org/apache/ratis/WatchRequestTests.java
index 32e4527580..b842ee9db5 100644
--- a/ratis-server/src/test/java/org/apache/ratis/WatchRequestTests.java
+++ b/ratis-server/src/test/java/org/apache/ratis/WatchRequestTests.java
@@ -559,5 +559,6 @@ static void assertNotReplicatedException(long logIndex, ReplicationLevel replica
     Assert.assertNotNull(nre);
     Assert.assertEquals(logIndex, nre.getLogIndex());
     Assert.assertEquals(replication, nre.getRequiredReplication());
+    Assert.assertNotNull(nre.getCommitInfos());
   }
 }