HDDS-1619. Support volume acl operations for OM HA. Contributed by… (#1147)

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

@@ -41,9 +41,12 @@
 import org.apache.hadoop.ozone.om.request.volume.OMVolumeDeleteRequest;
 import org.apache.hadoop.ozone.om.request.volume.OMVolumeSetOwnerRequest;
 import org.apache.hadoop.ozone.om.request.volume.OMVolumeSetQuotaRequest;
+import org.apache.hadoop.ozone.om.request.volume.acl.OMVolumeAddAclRequest;
+import org.apache.hadoop.ozone.om.request.volume.acl.OMVolumeRemoveAclRequest;
+import org.apache.hadoop.ozone.om.request.volume.acl.OMVolumeSetAclRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
-import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
-    .OMRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OzoneObj.ObjectType;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
 
@@ -117,12 +120,38 @@ public static OMClientRequest createClientRequest(OMRequest omRequest) {
       return new S3MultipartUploadAbortRequest(omRequest);
     case CompleteMultiPartUpload:
       return new S3MultipartUploadCompleteRequest(omRequest);
+    case AddAcl:
+    case RemoveAcl:
+    case SetAcl:
+      return getOMAclRequest(omRequest);
     default:
       // TODO: will update once all request types are implemented.
       return null;
     }
   }
 
+  private static OMClientRequest getOMAclRequest(OMRequest omRequest) {
+    Type cmdType = omRequest.getCmdType();
+    if (Type.AddAcl == cmdType) {
+      ObjectType type = omRequest.getAddAclRequest().getObj().getResType();
+      if (ObjectType.VOLUME == type) {
+        return new OMVolumeAddAclRequest(omRequest);
+      }
+    } else if (Type.RemoveAcl == cmdType) {
+      ObjectType type = omRequest.getAddAclRequest().getObj().getResType();
+      if (ObjectType.VOLUME == type) {
+        return new OMVolumeRemoveAclRequest(omRequest);
+      }
+    } else if (Type.SetAcl == cmdType) {
+      ObjectType type = omRequest.getAddAclRequest().getObj().getResType();
+      if (ObjectType.VOLUME == type) {
+        return new OMVolumeSetAclRequest(omRequest);
+      }
+    }
+    //TODO: handle bucket, key and prefix AddAcl
+    return null;
+  }
+
   /**
    * Convert exception result to {@link OzoneManagerProtocolProtos.Status}.
    * @param exception

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

@@ -0,0 +1,157 @@
+package org.apache.hadoop.ozone.om.request.volume.acl;
+
+import com.google.common.base.Optional;
+import org.apache.hadoop.hdds.scm.storage.CheckedBiFunction;
+import org.apache.hadoop.ozone.OzoneAcl;
+import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.OMMetrics;
+import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.exceptions.OMException;
+import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
+import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerDoubleBufferHelper;
+import org.apache.hadoop.ozone.om.request.OMClientRequest;
+import org.apache.hadoop.ozone.om.response.OMClientResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
+import org.apache.hadoop.ozone.security.acl.OzoneObj;
+import org.apache.hadoop.utils.db.cache.CacheKey;
+import org.apache.hadoop.utils.db.cache.CacheValue;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.apache.hadoop.ozone.om.lock.OzoneManagerLock.Resource.VOLUME_LOCK;
+
+/**
+ * Base class for OMVolumeAcl Request.
+ */
+public abstract class OMVolumeAclRequest extends OMClientRequest {
+
+  private CheckedBiFunction<List<OzoneAcl>, OmVolumeArgs, IOException>
+      omVolumeAclOp;
+
+  public OMVolumeAclRequest(OzoneManagerProtocolProtos.OMRequest omRequest,
+      CheckedBiFunction<List<OzoneAcl>, OmVolumeArgs, IOException> aclOp) {
+    super(omRequest);
+    omVolumeAclOp = aclOp;
+  }
+
+  @Override
+  public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
+      long transactionLogIndex,
+      OzoneManagerDoubleBufferHelper ozoneManagerDoubleBufferHelper) {
+    // protobuf guarantees volume and acls are non-null.
+    String volume = getVolumeName();
+    List<OzoneAcl> ozoneAcls = getAcls();
+
+    OMMetrics omMetrics = ozoneManager.getMetrics();
+    omMetrics.incNumVolumeUpdates();
+    OmVolumeArgs omVolumeArgs = null;
+
+    OMResponse.Builder omResponse = onInit();
+    OMClientResponse omClientResponse = null;
+    IOException exception = null;
+
+    OMMetadataManager omMetadataManager = ozoneManager.getMetadataManager();
+    boolean lockAcquired = false;
+    try {
+      // check Acl
+      if (ozoneManager.getAclsEnabled()) {
+        checkAcls(ozoneManager, OzoneObj.ResourceType.VOLUME,
+            OzoneObj.StoreType.OZONE, IAccessAuthorizer.ACLType.WRITE_ACL,
+            volume, null, null);
+      }
+      lockAcquired =
+          omMetadataManager.getLock().acquireLock(VOLUME_LOCK, volume);
+      String dbVolumeKey = omMetadataManager.getVolumeKey(volume);
+      omVolumeArgs = omMetadataManager.getVolumeTable().get(dbVolumeKey);
+      if (omVolumeArgs == null) {
+        throw new OMException(OMException.ResultCodes.VOLUME_NOT_FOUND);
+      }
+
+      // result is false upon add existing acl or remove non-existing acl
+      boolean result = true;
+      try {
+        omVolumeAclOp.apply(ozoneAcls, omVolumeArgs);
+      } catch (OMException ex) {
+        result = false;
+      }
+
+      if (result) {
+        // update cache.
+        omMetadataManager.getVolumeTable().addCacheEntry(
+            new CacheKey<>(dbVolumeKey),
+            new CacheValue<>(Optional.of(omVolumeArgs), transactionLogIndex));
+      }
+
+      omClientResponse = onSuccess(omResponse, omVolumeArgs, result);
+    } catch (IOException ex) {
+      exception = ex;
+      omMetrics.incNumVolumeUpdateFails();
+      omClientResponse = onFailure(omResponse, ex);
+    } finally {
+      if (omClientResponse != null) {
+        omClientResponse.setFlushFuture(
+            ozoneManagerDoubleBufferHelper.add(omClientResponse,
+                transactionLogIndex));
+      }
+      if (lockAcquired) {
+        omMetadataManager.getLock().releaseLock(VOLUME_LOCK, volume);
+      }
+    }
+
+    onComplete(exception);
+
+    return omClientResponse;
+  }
+
+  /**
+   * Get the Acls from the request.
+   * @return List of OzoneAcls, for add/remove it is a single element list
+   * for set it can be non-single element list.
+   */
+  abstract List<OzoneAcl> getAcls();
+
+  /**
+   * Get the volume name from the request.
+   * @return volume name
+   * This is needed for case where volume does not exist and the omVolumeArgs is
+   * null.
+   */
+  abstract String getVolumeName();
+
+  // TODO: Finer grain metrics can be moved to these callbacks. They can also
+  // be abstracted into separate interfaces in future.
+  /**
+   * Get the initial om response builder with lock.
+   * @return om response builder.
+   */
+  abstract OMResponse.Builder onInit();
+
+  /**
+   * Get the om client response on success case with lock.
+   * @param omResponse
+   * @param omVolumeArgs
+   * @param result
+   * @return OMClientResponse
+   */
+  abstract OMClientResponse onSuccess(
+      OMResponse.Builder omResponse, OmVolumeArgs omVolumeArgs, boolean result);
+
+  /**
+   * Get the om client response on failure case with lock.
+   * @param omResponse
+   * @param ex
+   * @return OMClientResponse
+   */
+  abstract OMClientResponse onFailure(OMResponse.Builder omResponse,
+      IOException ex);
+
+  /**
+   * Completion hook for final processing before return without lock.
+   * Usually used for logging without lock.
+   * @param ex
+   */
+  abstract void onComplete(IOException ex);
+}

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

@@ -0,0 +1,110 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.ozone.om.request.volume.acl;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import org.apache.hadoop.hdds.scm.storage.CheckedBiFunction;
+import org.apache.hadoop.ozone.OzoneAcl;
+import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
+import org.apache.hadoop.ozone.om.response.OMClientResponse;
+import org.apache.hadoop.ozone.om.response.volume.OMVolumeAclOpResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Handles volume add acl request.
+ */
+public class OMVolumeAddAclRequest extends OMVolumeAclRequest {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(OMVolumeAddAclRequest.class);
+
+  private static CheckedBiFunction<List<OzoneAcl>,
+      OmVolumeArgs, IOException> volumeAddAclOp;
+
+  static {
+    volumeAddAclOp = (acls, volArgs) -> volArgs.addAcl(acls.get(0));
+  }
+
+  private List<OzoneAcl> ozoneAcls;
+  private String volumeName;
+
+  public OMVolumeAddAclRequest(OMRequest omRequest) {
+    super(omRequest, volumeAddAclOp);
+    OzoneManagerProtocolProtos.AddAclRequest addAclRequest =
+        getOmRequest().getAddAclRequest();
+    Preconditions.checkNotNull(addAclRequest);
+    ozoneAcls = Lists.newArrayList(
+        OzoneAcl.fromProtobuf(addAclRequest.getAcl()));
+    volumeName = addAclRequest.getObj().getPath().substring(1);
+  }
+
+  @Override
+  public List<OzoneAcl> getAcls() {
+    return ozoneAcls;
+  }
+
+  @Override
+  public String getVolumeName() {
+    return volumeName;
+  }
+
+  private OzoneAcl getAcl() {
+    return ozoneAcls.get(0);
+  }
+
+
+  @Override
+  OMResponse.Builder onInit() {
+    return OMResponse.newBuilder().setCmdType(
+        OzoneManagerProtocolProtos.Type.AddAcl)
+        .setStatus(OzoneManagerProtocolProtos.Status.OK).setSuccess(true);
+  }
+
+  @Override
+  OMClientResponse onSuccess(OMResponse.Builder omResponse,
+      OmVolumeArgs omVolumeArgs, boolean result){
+    omResponse.setAddAclResponse(OzoneManagerProtocolProtos.AddAclResponse
+        .newBuilder().setResponse(result).build());
+    return new OMVolumeAclOpResponse(omVolumeArgs, omResponse.build());
+  }
+
+  @Override
+  OMClientResponse onFailure(OMResponse.Builder omResponse,
+      IOException ex) {
+    return new OMVolumeAclOpResponse(null,
+        createErrorOMResponse(omResponse, ex));
+  }
+
+  @Override
+  void onComplete(IOException ex) {
+    if (ex == null) {
+      LOG.debug("Add acl: {} to volume: {} success!",
+          getAcl(), getVolumeName());
+    } else {
+      LOG.error("Add acl {} to volume {} failed!",
+          getAcl(), getVolumeName(), ex);
+    }
+  }
+}

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

@@ -0,0 +1,109 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.ozone.om.request.volume.acl;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import org.apache.hadoop.hdds.scm.storage.CheckedBiFunction;
+import org.apache.hadoop.ozone.OzoneAcl;
+import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
+import org.apache.hadoop.ozone.om.response.OMClientResponse;
+import org.apache.hadoop.ozone.om.response.volume.OMVolumeAclOpResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Handles volume remove acl request.
+ */
+public class OMVolumeRemoveAclRequest extends OMVolumeAclRequest {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(OMVolumeRemoveAclRequest.class);
+
+  private static CheckedBiFunction<List<OzoneAcl>,
+      OmVolumeArgs, IOException> volumeRemoveAclOp;
+
+  static {
+    volumeRemoveAclOp = (acls, volArgs) -> volArgs.removeAcl(acls.get(0));
+  }
+
+  private List<OzoneAcl> ozoneAcls;
+  private String volumeName;
+
+  public OMVolumeRemoveAclRequest(OMRequest omRequest) {
+    super(omRequest, volumeRemoveAclOp);
+    OzoneManagerProtocolProtos.RemoveAclRequest removeAclRequest =
+        getOmRequest().getRemoveAclRequest();
+    Preconditions.checkNotNull(removeAclRequest);
+    ozoneAcls = Lists.newArrayList(
+        OzoneAcl.fromProtobuf(removeAclRequest.getAcl()));
+    volumeName = removeAclRequest.getObj().getPath().substring(1);
+  }
+
+  @Override
+  public List<OzoneAcl> getAcls() {
+    return ozoneAcls;
+  }
+
+  @Override
+  public String getVolumeName() {
+    return volumeName;
+  }
+
+  private OzoneAcl getAcl() {
+    return ozoneAcls.get(0);
+  }
+
+  @Override
+  OMResponse.Builder onInit() {
+    return OMResponse.newBuilder().setCmdType(
+        OzoneManagerProtocolProtos.Type.RemoveAcl)
+        .setStatus(OzoneManagerProtocolProtos.Status.OK).setSuccess(true);
+  }
+
+  @Override
+  OMClientResponse onSuccess(OMResponse.Builder omResponse,
+      OmVolumeArgs omVolumeArgs, boolean result){
+    omResponse.setRemoveAclResponse(OzoneManagerProtocolProtos.RemoveAclResponse
+        .newBuilder().setResponse(result).build());
+    return new OMVolumeAclOpResponse(omVolumeArgs, omResponse.build());
+  }
+
+  @Override
+  OMClientResponse onFailure(OMResponse.Builder omResponse,
+      IOException ex) {
+    return new OMVolumeAclOpResponse(null,
+        createErrorOMResponse(omResponse, ex));
+  }
+
+  @Override
+  void onComplete(IOException ex) {
+    if (ex == null) {
+      LOG.debug("Remove acl: {} from volume: {} success!",
+          getAcl(), getVolumeName());
+    } else {
+      LOG.error("Remove acl {} from volume {} failed!",
+          getAcl(), getVolumeName(), ex);
+    }
+  }
+}

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

@@ -0,0 +1,106 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.ozone.om.request.volume.acl;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.hdds.scm.storage.CheckedBiFunction;
+import org.apache.hadoop.ozone.OzoneAcl;
+import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
+import org.apache.hadoop.ozone.om.response.OMClientResponse;
+import org.apache.hadoop.ozone.om.response.volume.OMVolumeAclOpResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Handles volume set acl request.
+ */
+public class OMVolumeSetAclRequest extends OMVolumeAclRequest {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(OMVolumeSetAclRequest.class);
+
+  private static CheckedBiFunction<List<OzoneAcl>,
+      OmVolumeArgs, IOException> volumeSetAclOp;
+
+  static {
+    volumeSetAclOp = (acls, volArgs) -> volArgs.setAcls(acls);
+  }
+
+  private List<OzoneAcl> ozoneAcls;
+  private String volumeName;
+
+  public OMVolumeSetAclRequest(OMRequest omRequest) {
+    super(omRequest, volumeSetAclOp);
+    OzoneManagerProtocolProtos.SetAclRequest setAclRequest =
+        getOmRequest().getSetAclRequest();
+    Preconditions.checkNotNull(setAclRequest);
+    ozoneAcls = new ArrayList<>();
+    setAclRequest.getAclList().forEach(oai ->
+        ozoneAcls.add(OzoneAcl.fromProtobuf(oai)));
+    volumeName = setAclRequest.getObj().getPath().substring(1);
+  }
+
+  @Override
+  public List<OzoneAcl> getAcls() {
+    return ozoneAcls;
+  }
+
+  @Override
+  public String getVolumeName() {
+    return volumeName;
+  }
+
+  @Override
+  OMResponse.Builder onInit() {
+    return OMResponse.newBuilder().setCmdType(
+        OzoneManagerProtocolProtos.Type.RemoveAcl)
+        .setStatus(OzoneManagerProtocolProtos.Status.OK).setSuccess(true);
+  }
+
+  @Override
+  OMClientResponse onSuccess(OMResponse.Builder omResponse,
+      OmVolumeArgs omVolumeArgs, boolean result){
+    omResponse.setSetAclResponse(OzoneManagerProtocolProtos.SetAclResponse
+        .newBuilder().setResponse(result).build());
+    return new OMVolumeAclOpResponse(omVolumeArgs, omResponse.build());
+  }
+
+  @Override
+  OMClientResponse onFailure(OMResponse.Builder omResponse,
+      IOException ex) {
+    return new OMVolumeAclOpResponse(null,
+        createErrorOMResponse(omResponse, ex));
+  }
+
+  @Override
+  void onComplete(IOException ex) {
+    if (ex == null) {
+      LOG.debug("Set acls: {} to volume: {} success!",
+          getAcls(), getVolumeName());
+    } else {
+      LOG.error("Set acls {} to volume {} failed!",
+          getAcls(), getVolumeName(), ex);
+    }
+  }
+}

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/volume/acl/package-info.java

@@ -0,0 +1,22 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Package contains classes related to volume acl requests and responses.
+ */
+package org.apache.hadoop.ozone.om.request.volume.acl;

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/volume/OMVolumeAclOpResponse.java

@@ -0,0 +1,68 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.response.volume;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
+import org.apache.hadoop.ozone.om.response.OMClientResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.apache.hadoop.utils.db.BatchOperation;
+
+import java.io.IOException;
+
+/**
+ * Response for om volume acl operation request.
+ */
+public class OMVolumeAclOpResponse extends OMClientResponse {
+
+  private OmVolumeArgs omVolumeArgs;
+
+  public OMVolumeAclOpResponse(OmVolumeArgs omVolumeArgs,
+      OMResponse omResponse) {
+    super(omResponse);
+    this.omVolumeArgs = omVolumeArgs;
+  }
+
+  @Override
+  public void addToDBBatch(OMMetadataManager omMetadataManager,
+      BatchOperation batchOperation) throws IOException {
+
+    // For OmResponse with failure, this should do nothing. This method is
+    // not called in failure scenario in OM code.
+    if (getOMResponse().getSuccess()) {
+      if ((getOMResponse().hasAddAclResponse() &&
+          getOMResponse().getAddAclResponse().getResponse()) ||
+          (getOMResponse().hasRemoveAclResponse() &&
+              getOMResponse().getRemoveAclResponse().getResponse()) ||
+          (getOMResponse().hasSetAclResponse() &&
+              getOMResponse().getSetAclResponse().getResponse())) {
+        omMetadataManager.getVolumeTable().putWithBatch(batchOperation,
+            omMetadataManager.getVolumeKey(omVolumeArgs.getVolume()),
+            omVolumeArgs);
+      }
+    }
+  }
+
+  @VisibleForTesting
+  public OmVolumeArgs getOmVolumeArgs() {
+    return omVolumeArgs;
+  }
+}
+

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerHARequestHandlerImpl.java

@@ -71,6 +71,9 @@ public OMResponse handleApplyTransaction(OMRequest omRequest,
     case CommitMultiPartUpload:
     case AbortMultiPartUpload:
     case CompleteMultiPartUpload:
+    case AddAcl:
+    case RemoveAcl:
+    case SetAcl:
       //TODO: We don't need to pass transactionID, this will be removed when
       // complete write requests is changed to new model. And also we can
       // return OMClientResponse, then adding to doubleBuffer can be taken
@@ -78,10 +81,16 @@ public OMResponse handleApplyTransaction(OMRequest omRequest,
       // paths.
       OMClientRequest omClientRequest =
           OzoneManagerRatisUtils.createClientRequest(omRequest);
-      OMClientResponse omClientResponse =
+      if (omClientRequest != null) {
-          omClientRequest.validateAndUpdateCache(getOzoneManager(),
+        OMClientResponse omClientResponse =
-              transactionLogIndex, ozoneManagerDoubleBuffer::add);
+            omClientRequest.validateAndUpdateCache(getOzoneManager(),
-      return omClientResponse.getOMResponse();
+                transactionLogIndex, ozoneManagerDoubleBuffer::add);
+        return omClientResponse.getOMResponse();
+      } else {
+        //TODO: remove this once we have all HA support for all write request.
+        return handle(omRequest);
+      }
+
     default:
       // As all request types are not changed so we need to call handle
       // here.

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/TestOMRequestUtils.java

@@ -28,6 +28,7 @@
 
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.ozone.OmUtils;
+import org.apache.hadoop.ozone.OzoneAcl;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
@@ -49,6 +50,17 @@
     .OMRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
     .SetVolumePropertyRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
+    .AddAclRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
+    .RemoveAclRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
+    .SetAclRequest;
+import org.apache.hadoop.ozone.security.acl.OzoneObj;
+import org.apache.hadoop.ozone.security.acl.OzoneObj.ResourceType;
+import org.apache.hadoop.ozone.security.acl.OzoneObj.StoreType;
+
+import org.apache.hadoop.ozone.security.acl.OzoneObjInfo;
 import org.apache.hadoop.util.Time;
 
 /**
@@ -285,6 +297,58 @@ public static OMRequest createSetVolumePropertyRequest(String volumeName,
         .setSetVolumePropertyRequest(setVolumePropertyRequest).build();
   }
 
+  public static OMRequest createVolumeAddAclRequest(String volumeName,
+      OzoneAcl acl) {
+    AddAclRequest.Builder addAclRequestBuilder = AddAclRequest.newBuilder();
+    addAclRequestBuilder.setObj(OzoneObj.toProtobuf(new OzoneObjInfo.Builder()
+        .setVolumeName(volumeName)
+        .setResType(ResourceType.VOLUME)
+        .setStoreType(StoreType.OZONE)
+        .build()));
+    if (acl != null) {
+      addAclRequestBuilder.setAcl(OzoneAcl.toProtobuf(acl));
+    }
+    return OMRequest.newBuilder().setClientId(UUID.randomUUID().toString())
+        .setCmdType(OzoneManagerProtocolProtos.Type.AddAcl)
+        .setAddAclRequest(addAclRequestBuilder.build()).build();
+  }
+
+  public static OMRequest createVolumeRemoveAclRequest(String volumeName,
+      OzoneAcl acl) {
+    RemoveAclRequest.Builder removeAclRequestBuilder =
+        RemoveAclRequest.newBuilder();
+    removeAclRequestBuilder.setObj(OzoneObj.toProtobuf(
+        new OzoneObjInfo.Builder()
+            .setVolumeName(volumeName)
+            .setResType(ResourceType.VOLUME)
+            .setStoreType(StoreType.OZONE)
+            .build()));
+    if (acl != null) {
+      removeAclRequestBuilder.setAcl(OzoneAcl.toProtobuf(acl));
+    }
+    return OMRequest.newBuilder().setClientId(UUID.randomUUID().toString())
+        .setCmdType(OzoneManagerProtocolProtos.Type.RemoveAcl)
+        .setRemoveAclRequest(removeAclRequestBuilder.build()).build();
+  }
+
+  public static OMRequest createVolumeSetAclRequest(String volumeName,
+      List<OzoneAcl> acls) {
+    SetAclRequest.Builder setAclRequestBuilder = SetAclRequest.newBuilder();
+    setAclRequestBuilder.setObj(OzoneObj.toProtobuf(new OzoneObjInfo.Builder()
+        .setVolumeName(volumeName)
+        .setResType(ResourceType.VOLUME)
+        .setStoreType(StoreType.OZONE)
+        .build()));
+    if (acls != null) {
+      acls.forEach(
+          acl -> setAclRequestBuilder.addAcl(OzoneAcl.toProtobuf(acl)));
+    }
+
+    return OMRequest.newBuilder().setClientId(UUID.randomUUID().toString())
+        .setCmdType(OzoneManagerProtocolProtos.Type.SetAcl)
+        .setSetAclRequest(setAclRequestBuilder.build()).build();
+  }
+
   /**
    * Deletes key from Key table and adds it to DeletedKeys table.
    * @return the deletedKey name
@@ -387,5 +451,4 @@ public static OMRequest createCompleteMPURequest(String volumeName,
         .build();
 
   }
-
 }

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/volume/acl/TestOMVolumeAddAclRequest.java

@@ -0,0 +1,123 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.request.volume.acl;
+
+import org.apache.hadoop.ozone.OzoneAcl;
+import org.apache.hadoop.ozone.om.helpers.OmOzoneAclMap;
+import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
+import org.apache.hadoop.ozone.om.request.TestOMRequestUtils;
+import org.apache.hadoop.ozone.om.request.volume.TestOMVolumeRequest;
+import org.apache.hadoop.ozone.om.response.OMClientResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.UUID;
+
+/**
+ * Tests volume addAcl request.
+ */
+public class TestOMVolumeAddAclRequest extends TestOMVolumeRequest {
+
+  @Test
+  public void testPreExecute() throws Exception {
+    String volumeName = UUID.randomUUID().toString();
+    OzoneAcl acl = OzoneAcl.parseAcl("user:bilbo:rw");
+    OMRequest originalRequest =
+        TestOMRequestUtils.createVolumeAddAclRequest(volumeName, acl);
+
+    OMVolumeAddAclRequest omVolumeAddAclRequest =
+        new OMVolumeAddAclRequest(originalRequest);
+
+    OMRequest modifiedRequest = omVolumeAddAclRequest.preExecute(
+        ozoneManager);
+    Assert.assertNotEquals(modifiedRequest, originalRequest);
+  }
+
+  @Test
+  public void testValidateAndUpdateCacheSuccess() throws Exception {
+    String volumeName = UUID.randomUUID().toString();
+    String ownerName = "user1";
+
+    TestOMRequestUtils.addUserToDB(volumeName, ownerName, omMetadataManager);
+    TestOMRequestUtils.addVolumeToDB(volumeName, ownerName, omMetadataManager);
+
+    OzoneAcl acl = OzoneAcl.parseAcl("user:bilbo:rwdlncxy[ACCESS]");
+
+    OMRequest originalRequest =
+        TestOMRequestUtils.createVolumeAddAclRequest(volumeName, acl);
+
+    OMVolumeAddAclRequest omVolumeAddAclRequest =
+        new OMVolumeAddAclRequest(originalRequest);
+
+    omVolumeAddAclRequest.preExecute(ozoneManager);
+
+    String volumeKey = omMetadataManager.getVolumeKey(volumeName);
+
+    // Get Acl before validateAndUpdateCache.
+    OmVolumeArgs omVolumeArgs =
+        omMetadataManager.getVolumeTable().get(volumeKey);
+    // As request is valid volume table should have entry.
+    Assert.assertNotNull(omVolumeArgs);
+    OmOzoneAclMap aclMapBeforeSet = omVolumeArgs.getAclMap();
+
+    OMClientResponse omClientResponse =
+        omVolumeAddAclRequest.validateAndUpdateCache(ozoneManager, 1,
+            ozoneManagerDoubleBufferHelper);
+
+    OMResponse omResponse = omClientResponse.getOMResponse();
+    Assert.assertNotNull(omResponse.getAddAclResponse());
+    Assert.assertEquals(OzoneManagerProtocolProtos.Status.OK,
+        omResponse.getStatus());
+
+    OmOzoneAclMap aclMapAfterSet = omMetadataManager
+        .getVolumeTable().get(volumeKey).getAclMap();
+    Assert.assertNotEquals(aclMapBeforeSet, aclMapAfterSet);
+
+    // acl is added to aclMapAfterSet
+    Assert.assertEquals(1, aclMapAfterSet.getAcl().size());
+    Assert.assertEquals(acl, aclMapAfterSet.getAcl().get(0));
+  }
+
+  @Test
+  public void testValidateAndUpdateCacheWithVolumeNotFound()
+      throws Exception {
+    String volumeName = UUID.randomUUID().toString();
+    OzoneAcl acl = OzoneAcl.parseAcl("user:bilbo:rw");
+    OMRequest originalRequest =
+        TestOMRequestUtils.createVolumeAddAclRequest(volumeName, acl);
+
+    OMVolumeAddAclRequest omVolumeAddAclRequest =
+        new OMVolumeAddAclRequest(originalRequest);
+
+    omVolumeAddAclRequest.preExecute(ozoneManager);
+
+    OMClientResponse omClientResponse =
+        omVolumeAddAclRequest.validateAndUpdateCache(ozoneManager, 1,
+            ozoneManagerDoubleBufferHelper);
+
+    OMResponse omResponse = omClientResponse.getOMResponse();
+    Assert.assertNotNull(omResponse.getAddAclResponse());
+    Assert.assertEquals(OzoneManagerProtocolProtos.Status.VOLUME_NOT_FOUND,
+        omResponse.getStatus());
+  }
+}

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/volume/acl/TestOMVolumeRemoveAclRequest.java

@@ -0,0 +1,133 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.request.volume.acl;
+
+import org.apache.hadoop.ozone.OzoneAcl;
+import org.apache.hadoop.ozone.om.helpers.OmOzoneAclMap;
+import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
+import org.apache.hadoop.ozone.om.request.TestOMRequestUtils;
+import org.apache.hadoop.ozone.om.request.volume.TestOMVolumeRequest;
+import org.apache.hadoop.ozone.om.response.OMClientResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.UUID;
+
+/**
+ * Tests volume removeAcl request.
+ */
+public class TestOMVolumeRemoveAclRequest extends TestOMVolumeRequest {
+
+  @Test
+  public void testPreExecute() throws Exception {
+    String volumeName = UUID.randomUUID().toString();
+    OzoneAcl acl = OzoneAcl.parseAcl("user:bilbo:rw");
+    OMRequest originalRequest =
+        TestOMRequestUtils.createVolumeRemoveAclRequest(volumeName, acl);
+
+    OMVolumeRemoveAclRequest omVolumeRemoveAclRequest =
+        new OMVolumeRemoveAclRequest(originalRequest);
+
+    OMRequest modifiedRequest = omVolumeRemoveAclRequest.preExecute(
+        ozoneManager);
+    Assert.assertNotEquals(modifiedRequest, originalRequest);
+  }
+
+  @Test
+  public void testValidateAndUpdateCacheSuccess() throws Exception {
+    String volumeName = UUID.randomUUID().toString();
+    String ownerName = "user1";
+
+    TestOMRequestUtils.addUserToDB(volumeName, ownerName, omMetadataManager);
+    TestOMRequestUtils.addVolumeToDB(volumeName, ownerName, omMetadataManager);
+
+    OzoneAcl acl = OzoneAcl.parseAcl("user:bilbo:rwdlncxy[ACCESS]");
+    // add acl first
+    OMRequest addAclRequest =
+        TestOMRequestUtils.createVolumeAddAclRequest(volumeName, acl);
+    OMVolumeAddAclRequest omVolumeAddAclRequest =
+        new OMVolumeAddAclRequest(addAclRequest);
+    omVolumeAddAclRequest.preExecute(ozoneManager);
+    OMClientResponse omClientAddResponse =
+        omVolumeAddAclRequest.validateAndUpdateCache(ozoneManager, 1,
+            ozoneManagerDoubleBufferHelper);
+    OMResponse omAddAclResponse = omClientAddResponse.getOMResponse();
+    Assert.assertNotNull(omAddAclResponse.getAddAclResponse());
+    Assert.assertEquals(OzoneManagerProtocolProtos.Status.OK,
+        omAddAclResponse.getStatus());
+
+
+    // remove acl
+    OMRequest removeAclRequest =
+        TestOMRequestUtils.createVolumeRemoveAclRequest(volumeName, acl);
+    OMVolumeRemoveAclRequest omVolumeRemoveAclRequest =
+        new OMVolumeRemoveAclRequest(removeAclRequest);
+    omVolumeRemoveAclRequest.preExecute(ozoneManager);
+
+    String volumeKey = omMetadataManager.getVolumeKey(volumeName);
+
+    // Get Acl before Remove.
+    OmVolumeArgs omVolumeArgs =
+        omMetadataManager.getVolumeTable().get(volumeKey);
+    // As request is valid volume table should have entry.
+    Assert.assertNotNull(omVolumeArgs);
+    OmOzoneAclMap aclMapBeforeRemove = omVolumeArgs.getAclMap();
+    Assert.assertEquals(acl, aclMapBeforeRemove.getAcl().get(0));
+
+    OMClientResponse omClientRemoveResponse =
+        omVolumeRemoveAclRequest.validateAndUpdateCache(ozoneManager, 1,
+            ozoneManagerDoubleBufferHelper);
+
+    OMResponse omRemoveAclResponse = omClientRemoveResponse.getOMResponse();
+    Assert.assertNotNull(omRemoveAclResponse.getRemoveAclResponse());
+    Assert.assertEquals(OzoneManagerProtocolProtos.Status.OK,
+        omRemoveAclResponse.getStatus());
+
+    // acl is removed from aclMapAfterSet
+    OmOzoneAclMap aclMapAfterRemove = omMetadataManager
+        .getVolumeTable().get(volumeKey).getAclMap();
+    Assert.assertEquals(0, aclMapAfterRemove.getAcl().size());
+  }
+
+  @Test
+  public void testValidateAndUpdateCacheWithVolumeNotFound()
+      throws Exception {
+    String volumeName = UUID.randomUUID().toString();
+    OzoneAcl acl = OzoneAcl.parseAcl("user:bilbo:rw");
+    OMRequest originalRequest =
+        TestOMRequestUtils.createVolumeRemoveAclRequest(volumeName, acl);
+
+    OMVolumeRemoveAclRequest omVolumeRemoveAclRequest =
+        new OMVolumeRemoveAclRequest(originalRequest);
+
+    omVolumeRemoveAclRequest.preExecute(ozoneManager);
+
+    OMClientResponse omClientResponse =
+        omVolumeRemoveAclRequest.validateAndUpdateCache(ozoneManager, 1,
+            ozoneManagerDoubleBufferHelper);
+
+    OMResponse omResponse = omClientResponse.getOMResponse();
+    Assert.assertNotNull(omResponse.getRemoveAclResponse());
+    Assert.assertEquals(OzoneManagerProtocolProtos.Status.VOLUME_NOT_FOUND,
+        omResponse.getStatus());
+  }
+}

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/volume/acl/TestOMVolumeSetAclRequest.java

@@ -0,0 +1,136 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.request.volume.acl;
+
+import com.google.common.collect.Lists;
+import org.apache.hadoop.ozone.OzoneAcl;
+import org.apache.hadoop.ozone.om.helpers.OmOzoneAclMap;
+import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
+import org.apache.hadoop.ozone.om.request.TestOMRequestUtils;
+import org.apache.hadoop.ozone.om.request.volume.TestOMVolumeRequest;
+import org.apache.hadoop.ozone.om.response.OMClientResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.List;
+import java.util.UUID;
+
+import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OzoneAclInfo.OzoneAclScope.ACCESS;
+import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OzoneAclInfo.OzoneAclScope.DEFAULT;
+
+/**
+ * Tests volume setAcl request.
+ */
+public class TestOMVolumeSetAclRequest extends TestOMVolumeRequest {
+
+  @Test
+  public void testPreExecute() throws Exception {
+    String volumeName = UUID.randomUUID().toString();
+    OzoneAcl acl = OzoneAcl.parseAcl("user:bilbo:rw");
+    OMRequest originalRequest =
+        TestOMRequestUtils.createVolumeSetAclRequest(volumeName,
+            Lists.newArrayList(acl));
+
+    OMVolumeSetAclRequest omVolumeSetAclRequest =
+        new OMVolumeSetAclRequest(originalRequest);
+
+    OMRequest modifiedRequest = omVolumeSetAclRequest.preExecute(
+        ozoneManager);
+    Assert.assertNotEquals(modifiedRequest, originalRequest);
+  }
+
+  @Test
+  public void testValidateAndUpdateCacheSuccess() throws Exception {
+    String volumeName = UUID.randomUUID().toString();
+    String ownerName = "user1";
+
+    TestOMRequestUtils.addUserToDB(volumeName, ownerName, omMetadataManager);
+    TestOMRequestUtils.addVolumeToDB(volumeName, ownerName, omMetadataManager);
+
+    OzoneAcl userAccessAcl = OzoneAcl.parseAcl("user:bilbo:rw[ACCESS]");
+    OzoneAcl groupDefaultAcl =
+        OzoneAcl.parseAcl("group:admin:rwdlncxy[DEFAULT]");
+
+    List<OzoneAcl> acls = Lists.newArrayList(userAccessAcl, groupDefaultAcl);
+
+    OMRequest originalRequest =
+        TestOMRequestUtils.createVolumeSetAclRequest(volumeName, acls);
+
+    OMVolumeSetAclRequest omVolumeSetAclRequest =
+        new OMVolumeSetAclRequest(originalRequest);
+
+    omVolumeSetAclRequest.preExecute(ozoneManager);
+
+    String volumeKey = omMetadataManager.getVolumeKey(volumeName);
+
+    // Get Acl before validateAndUpdateCache.
+    OmVolumeArgs omVolumeArgs =
+        omMetadataManager.getVolumeTable().get(volumeKey);
+    // As request is valid volume table should have entry.
+    Assert.assertNotNull(omVolumeArgs);
+    OmOzoneAclMap aclMapBeforeSet = omVolumeArgs.getAclMap();
+
+    OMClientResponse omClientResponse =
+        omVolumeSetAclRequest.validateAndUpdateCache(ozoneManager, 1,
+            ozoneManagerDoubleBufferHelper);
+
+    OMResponse omResponse = omClientResponse.getOMResponse();
+    Assert.assertNotNull(omResponse.getSetAclResponse());
+    Assert.assertEquals(OzoneManagerProtocolProtos.Status.OK,
+        omResponse.getStatus());
+
+    OmOzoneAclMap aclMapAfterSet = omMetadataManager
+        .getVolumeTable().get(volumeKey).getAclMap();
+    Assert.assertNotEquals(aclMapBeforeSet, aclMapAfterSet);
+
+    // Acl is added to aclMapAfterSet
+    Assert.assertEquals(2, aclMapAfterSet.getAcl().size());
+    Assert.assertTrue("Default Acl should be set.",
+        aclMapAfterSet.getAclsByScope(ACCESS).contains(userAccessAcl));
+    Assert.assertTrue("Default Acl should be set.",
+        aclMapAfterSet.getAclsByScope(DEFAULT).contains(groupDefaultAcl));
+  }
+
+  @Test
+  public void testValidateAndUpdateCacheWithVolumeNotFound()
+      throws Exception {
+    String volumeName = UUID.randomUUID().toString();
+    OzoneAcl acl = OzoneAcl.parseAcl("user:bilbo:rw");
+    OMRequest originalRequest =
+        TestOMRequestUtils.createVolumeSetAclRequest(volumeName,
+            Lists.newArrayList(acl));
+
+    OMVolumeSetAclRequest omVolumeSetAclRequest =
+        new OMVolumeSetAclRequest(originalRequest);
+
+    omVolumeSetAclRequest.preExecute(ozoneManager);
+
+    OMClientResponse omClientResponse =
+        omVolumeSetAclRequest.validateAndUpdateCache(ozoneManager, 1,
+            ozoneManagerDoubleBufferHelper);
+
+    OMResponse omResponse = omClientResponse.getOMResponse();
+    Assert.assertNotNull(omResponse.getSetAclResponse());
+    Assert.assertEquals(OzoneManagerProtocolProtos.Status.VOLUME_NOT_FOUND,
+        omResponse.getStatus());
+  }
+}

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/volume/acl/package-info.java

@@ -0,0 +1,21 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+/**
+ * Package contains test classes for volume acl requests.
+ */
+package org.apache.hadoop.ozone.om.request.volume.acl;