HDDS-4262. Use ClientID and CallID from Rpc Client to detect retry requests

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAMetadataOnly.java

@@ -30,9 +30,21 @@
 import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.ha.OMFailoverProxyProvider;
 import org.apache.hadoop.ozone.om.ha.OMProxyInfo;
+import org.apache.hadoop.ozone.om.helpers.OMRatisHelper;
 import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer;
+import org.apache.hadoop.ozone.om.request.volume.OMVolumeCreateRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.CreateVolumeRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.VolumeInfo;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.test.GenericTestUtils;
 import org.apache.log4j.Logger;
+import org.apache.ratis.protocol.ClientId;
+import org.apache.ratis.protocol.Message;
+import org.apache.ratis.protocol.RaftClientReply;
+import org.apache.ratis.protocol.RaftClientRequest;
+import org.apache.ratis.server.RaftServer;
 import org.junit.Assert;
 import org.junit.Ignore;
 import org.junit.Test;
@@ -48,6 +60,7 @@
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.Iterator;
+import java.util.UUID;
 
 import static org.apache.hadoop.ozone.MiniOzoneHAClusterImpl.NODE_FAILURE_TIMEOUT;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_WAIT_BETWEEN_RETRIES_MILLIS_DEFAULT;
@@ -348,6 +361,69 @@ public void testJMXMetrics() throws Exception {
     Assert.assertTrue((long) flushCount >= 0);
   }
 
+  @Test
+  public void testOMRetryCache() throws Exception {
+    ObjectStore objectStore = getObjectStore();
+    objectStore.createVolume(UUID.randomUUID().toString());
+
+
+    OMFailoverProxyProvider omFailoverProxyProvider = OmFailoverProxyUtil
+        .getFailoverProxyProvider(objectStore.getClientProxy());
+
+    String currentLeaderNodeId = omFailoverProxyProvider
+        .getCurrentProxyOMNodeId();
+
+    OzoneManagerRatisServer ozoneManagerRatisServer =
+        getCluster().getOzoneManager(currentLeaderNodeId).getOmRatisServer();
+
+    RaftServer raftServer = ozoneManagerRatisServer.getServer();
+
+    ClientId clientId = ClientId.randomId();
+    long callId = 2000L;
+    String userName = UserGroupInformation.getCurrentUser().getUserName();
+    String volumeName = UUID.randomUUID().toString();
+
+
+    GenericTestUtils.LogCapturer logCapturer = GenericTestUtils.LogCapturer
+        .captureLogs(OMVolumeCreateRequest.getLogger());
+    OMRequest omRequest =
+        OMRequest.newBuilder().setCreateVolumeRequest(
+            CreateVolumeRequest.newBuilder().setVolumeInfo(
+                VolumeInfo.newBuilder().setOwnerName(userName)
+                    .setAdminName(userName).setVolume(volumeName).build())
+                .build()).setClientId(UUID.randomUUID().toString())
+            .setCmdType(OzoneManagerProtocolProtos.Type.CreateVolume).build();
+
+    RaftClientReply raftClientReply =
+        raftServer.submitClientRequest(new RaftClientRequest(clientId,
+         raftServer.getId(), ozoneManagerRatisServer.getRaftGroup()
+         .getGroupId(), callId,
+        Message.valueOf(OMRatisHelper.convertRequestToByteString(omRequest)),
+        RaftClientRequest.writeRequestType(), null));
+
+    Assert.assertTrue(raftClientReply.isSuccess());
+
+    Assert.assertTrue(logCapturer.getOutput().contains("created volume:"
+        + volumeName));
+
+    logCapturer.clearOutput();
+
+    raftClientReply =
+        raftServer.submitClientRequest(new RaftClientRequest(clientId,
+            raftServer.getId(), ozoneManagerRatisServer.getRaftGroup()
+            .getGroupId(), callId, Message.valueOf(
+                OMRatisHelper.convertRequestToByteString(omRequest)),
+            RaftClientRequest.writeRequestType(), null));
+
+    Assert.assertTrue(raftClientReply.isSuccess());
+
+    // As second time with same client id and call id, this request should
+    // not be executed ratis server should return from cache.
+    Assert.assertFalse(logCapturer.getOutput().contains("created volume:"
+        + volumeName));
+
+  }
+
   private void validateVolumesList(String userName,
       Set<String> expectedVolumes) throws Exception {
     ObjectStore objectStore = getObjectStore();

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerRatisServer.java

@@ -38,6 +38,7 @@
 import org.apache.hadoop.hdds.conf.StorageUnit;
 import org.apache.hadoop.hdds.server.ServerUtils;
 import org.apache.hadoop.hdds.tracing.TracingUtil;
+import org.apache.hadoop.ipc.ProtobufRpcEngine.Server;
 import org.apache.hadoop.ozone.om.OMConfigKeys;
 import org.apache.hadoop.ozone.om.OzoneManager;
 import org.apache.hadoop.ozone.om.exceptions.OMException;
@@ -143,8 +144,9 @@ public OMResponse submitRequest(OMRequest omRequest) throws ServiceException {
    * ratis server.
    */
   private RaftClientRequest createWriteRaftClientRequest(OMRequest omRequest) {
-    return new RaftClientRequest(clientId, server.getId(), raftGroupId,
-        nextCallId(),
+    return new RaftClientRequest(
+        ClientId.valueOf(UUID.nameUUIDFromBytes(Server.getClientId())),
+        server.getId(), raftGroupId, Server.getCallId(),
         Message.valueOf(OMRatisHelper.convertRequestToByteString(omRequest)),
         RaftClientRequest.writeRequestType(), null);
   }
@@ -339,6 +341,11 @@ public RaftGroup getRaftGroup() {
     return this.raftGroup;
   }
 
+  @VisibleForTesting
+  public RaftServer getServer() {
+    return server;
+  }
+
   /**
    * Initializes and returns OzoneManager StateMachine.
    */

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/volume/OMVolumeCreateRequest.java

@@ -22,6 +22,7 @@
 import java.util.HashMap;
 import java.util.Map;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 
 import org.apache.hadoop.ozone.OmUtils;
@@ -201,6 +202,11 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
     }
     return omClientResponse;
   }
+
+  @VisibleForTesting
+  public static Logger getLogger() {
+    return LOG;
+  }
 }