HDDS-45. Removal of old OzoneRestClient. Contributed by Lokesh Jain.

15 files changed, 548 insertions, 2647 deletions

diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClient.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClient.java
index 6d33cd4e36..42e02f9b13 100644
--- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClient.java
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClient.java
@@ -54,6 +54,7 @@ public class XceiverClient extends XceiverClientSpi {
   private Bootstrap b;
   private EventLoopGroup group;
   private final Semaphore semaphore;
+  private boolean closed = false;
 
   /**
    * Constructs a client that can communicate with the Container framework on
@@ -74,6 +75,10 @@ public class XceiverClient extends XceiverClientSpi {
 
   @Override
   public void connect() throws Exception {
+    if (closed) {
+      throw new IOException("This channel is not connected.");
+    }
+
     if (channel != null && channel.isActive()) {
       throw new IOException("This client is already connected to a host.");
     }
@@ -97,6 +102,18 @@ public class XceiverClient extends XceiverClientSpi {
     channel = b.connect(leader.getHostName(), port).sync().channel();
   }
 
+  public void reconnect() throws IOException {
+    try {
+      connect();
+      if (channel == null || !channel.isActive()) {
+        throw new IOException("This channel is not connected.");
+      }
+    } catch (Exception e) {
+      LOG.error("Error while connecting: ", e);
+      throw new IOException(e);
+    }
+  }
+
   /**
    * Returns if the exceiver client connects to a server.
    *
@@ -109,6 +126,7 @@ public class XceiverClient extends XceiverClientSpi {
 
   @Override
   public void close() {
+    closed = true;
     if (group != null) {
       group.shutdownGracefully().awaitUninterruptibly();
     }
@@ -124,7 +142,7 @@ public class XceiverClient extends XceiverClientSpi {
       ContainerProtos.ContainerCommandRequestProto request) throws IOException {
     try {
       if ((channel == null) || (!channel.isActive())) {
-        throw new IOException("This channel is not connected.");
+        reconnect();
       }
       XceiverClientHandler handler =
           channel.pipeline().get(XceiverClientHandler.class);
@@ -160,7 +178,7 @@ public class XceiverClient extends XceiverClientSpi {
       sendCommandAsync(ContainerProtos.ContainerCommandRequestProto request)
       throws IOException, ExecutionException, InterruptedException {
     if ((channel == null) || (!channel.isActive())) {
-      throw new IOException("This channel is not connected.");
+      reconnect();
     }
     XceiverClientHandler handler =
         channel.pipeline().get(XceiverClientHandler.class);
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/web/client/OzoneBucket.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/web/client/OzoneBucket.java
deleted file mode 100644
index 3183d035ed..0000000000
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/web/client/OzoneBucket.java
+++ /dev/null
@@ -1,646 +0,0 @@
-/*
- * 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
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- *  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.web.client;
-
-import org.apache.commons.codec.digest.DigestUtils;
-import org.apache.hadoop.fs.StorageType;
-import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
-import org.apache.hadoop.io.IOUtils;
-
-import org.apache.hadoop.ozone.OzoneConsts;
-import org.apache.hadoop.ozone.client.OzoneClientException;
-import org.apache.hadoop.ozone.client.rest.OzoneException;
-import org.apache.hadoop.ozone.client.rest.headers.Header;
-import org.apache.hadoop.ozone.OzoneAcl;
-import org.apache.hadoop.ozone.web.response.BucketInfo;
-import org.apache.hadoop.ozone.web.response.KeyInfo;
-import org.apache.hadoop.ozone.web.response.ListKeys;
-
-import static org.apache.hadoop.hdds.server.ServerUtils.releaseConnection;
-import org.apache.http.HttpEntity;
-import org.apache.http.HttpResponse;
-import org.apache.http.client.methods.HttpDelete;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.client.methods.HttpPut;
-import org.apache.http.client.utils.URIBuilder;
-import org.apache.http.entity.ContentType;
-import org.apache.http.entity.FileEntity;
-import org.apache.http.entity.InputStreamEntity;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.impl.client.HttpClientBuilder;
-import org.apache.http.util.EntityUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Strings;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.nio.file.Path;
-import java.util.LinkedList;
-import java.util.List;
-
-import static java.net.HttpURLConnection.HTTP_CREATED;
-import static java.net.HttpURLConnection.HTTP_OK;
-import static org.apache.hadoop.ozone.web.utils.OzoneUtils.ENCODING;
-import static org.apache.hadoop.ozone.web.utils.OzoneUtils.ENCODING_NAME;
-
-/**
- * A Bucket class the represents an Ozone Bucket.
- */
-public class OzoneBucket {
-  static final Logger LOG = LoggerFactory.getLogger(OzoneBucket.class);
-
-  private BucketInfo bucketInfo;
-  private OzoneVolume volume;
-
-  /**
-   * Constructor for bucket.
-   *
-   * @param info   - BucketInfo
-   * @param volume - OzoneVolume Object that contains this bucket
-   */
-  public OzoneBucket(BucketInfo info, OzoneVolume volume) {
-    this.bucketInfo = info;
-    this.volume = volume;
-  }
-
-  /**
-   * Gets bucket Info.
-   *
-   * @return BucketInfo
-   */
-  public BucketInfo getBucketInfo() {
-    return bucketInfo;
-  }
-
-  /**
-   * Sets Bucket Info.
-   *
-   * @param bucketInfo BucketInfo
-   */
-  public void setBucketInfo(BucketInfo bucketInfo) {
-    this.bucketInfo = bucketInfo;
-  }
-
-  /**
-   * Returns the parent volume class.
-   *
-   * @return - OzoneVolume
-   */
-  OzoneVolume getVolume() {
-    return volume;
-  }
-
-  /**
-   * Returns bucket name.
-   *
-   * @return Bucket Name
-   */
-  public String getBucketName() {
-    return bucketInfo.getBucketName();
-  }
-
-  /**
-   * Returns the Acls on the bucket.
-   *
-   * @return - Acls
-   */
-  public List<OzoneAcl> getAcls() {
-    return bucketInfo.getAcls();
-  }
-
-  /**
-   * Return versioning info on the bucket - Enabled or disabled.
-   *
-   * @return - Version Enum
-   */
-  public OzoneConsts.Versioning getVersioning() {
-    return bucketInfo.getVersioning();
-  }
-
-  /**
-   * Gets the Storage class for the bucket.
-   *
-   * @return Storage Class Enum
-   */
-  public StorageType getStorageType() {
-    return bucketInfo.getStorageType();
-  }
-
-  /**
-   * Gets the creation time of the bucket.
-   *
-   * @return String
-   */
-  public String getCreatedOn() {
-    return bucketInfo.getCreatedOn();
-  }
-
-  /**
-   * Puts an Object in Ozone bucket.
-   *
-   * @param keyName - Name of the key
-   * @param data    - Data that you want to put
-   * @throws OzoneException
-   */
-  public void putKey(String keyName, String data) throws OzoneException {
-    if ((keyName == null) || keyName.isEmpty()) {
-      throw new OzoneClientException("Invalid key Name.");
-    }
-
-    if (data == null) {
-      throw new OzoneClientException("Invalid data.");
-    }
-
-    HttpPut putRequest = null;
-    InputStream is = null;
-    try (CloseableHttpClient httpClient = HddsClientUtils.newHttpClient()) {
-      URIBuilder builder = new URIBuilder(volume.getClient().getEndPointURI());
-      builder.setPath("/" + getVolume().getVolumeName() + "/" + getBucketName()
-          + "/" + keyName).build();
-
-      putRequest = getVolume().getClient().getHttpPut(builder.toString());
-
-      is = new ByteArrayInputStream(data.getBytes(ENCODING));
-      putRequest.setEntity(new InputStreamEntity(is, data.length()));
-      is.mark(data.length());
-      try {
-        putRequest.setHeader(Header.CONTENT_MD5, DigestUtils.md5Hex(is));
-      } finally {
-        is.reset();
-      }
-      executePutKey(putRequest, httpClient);
-    } catch (IOException | URISyntaxException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      IOUtils.closeStream(is);
-      releaseConnection(putRequest);
-    }
-  }
-
-  /**
-   * Puts an Object in Ozone Bucket.
-   *
-   * @param dataFile - File from which you want the data to be put. Key Name
-   *                 will same as the file name, devoid of any path.
-   * @throws OzoneException
-   */
-  public void putKey(File dataFile) throws OzoneException {
-    if (dataFile == null) {
-      throw new OzoneClientException("Invalid file object.");
-    }
-    String keyName = dataFile.getName();
-    putKey(keyName, dataFile);
-  }
-
-  /**
-   * Puts a Key in Ozone Bucket.
-   *
-   * @param keyName - Name of the Key
-   * @param file    - Stream that gets read to be put into Ozone.
-   * @throws OzoneException
-   */
-  public void putKey(String keyName, File file)
-      throws OzoneException {
-
-    if ((keyName == null) || keyName.isEmpty()) {
-      throw new OzoneClientException("Invalid key Name");
-    }
-
-    if (file == null) {
-      throw new OzoneClientException("Invalid data stream");
-    }
-
-    HttpPut putRequest = null;
-    FileInputStream fis = null;
-    try (CloseableHttpClient httpClient = HddsClientUtils.newHttpClient()) {
-      URIBuilder builder = new URIBuilder(volume.getClient().getEndPointURI());
-      builder.setPath("/" + getVolume().getVolumeName() + "/" + getBucketName()
-          + "/" + keyName).build();
-
-      putRequest = getVolume().getClient().getHttpPut(builder.toString());
-
-      FileEntity fileEntity = new FileEntity(file, ContentType
-          .APPLICATION_OCTET_STREAM);
-      putRequest.setEntity(fileEntity);
-
-      fis = new FileInputStream(file);
-      putRequest.setHeader(Header.CONTENT_MD5, DigestUtils.md5Hex(fis));
-      executePutKey(putRequest, httpClient);
-
-    } catch (IOException | URISyntaxException ex) {
-      final OzoneClientException orce = new OzoneClientException(
-          "Failed to putKey: keyName=" + keyName + ", file=" + file);
-      orce.initCause(ex);
-      LOG.trace("", orce);
-      throw orce;
-    } finally {
-      IOUtils.closeStream(fis);
-      releaseConnection(putRequest);
-    }
-  }
-
-  /**
-   * executePutKey executes the Put request against the Ozone Server.
-   *
-   * @param putRequest - Http Put Request
-   * @param httpClient - httpClient
-   * @throws OzoneException
-   * @throws IOException
-   */
-  public static void executePutKey(HttpPut putRequest,
-      CloseableHttpClient httpClient) throws OzoneException, IOException {
-    HttpEntity entity = null;
-    try {
-      HttpResponse response = httpClient.execute(putRequest);
-      int errorCode = response.getStatusLine().getStatusCode();
-      entity = response.getEntity();
-
-      if ((errorCode == HTTP_OK) || (errorCode == HTTP_CREATED)) {
-        return;
-      }
-
-      if (entity == null) {
-        throw new OzoneClientException("Unexpected null in http payload");
-      }
-
-      throw OzoneException.parse(EntityUtils.toString(entity));
-    } finally {
-      if (entity != null) {
-        EntityUtils.consumeQuietly(entity);
-      }
-    }
-  }
-
-  /**
-   * Gets a key from the Ozone server and writes to the file pointed by the
-   * downloadTo PAth.
-   *
-   * @param keyName    - Key Name in Ozone.
-   * @param downloadTo File Name to download the Key's Data to
-   */
-  public void getKey(String keyName, Path downloadTo) throws OzoneException {
-
-    if ((keyName == null) || keyName.isEmpty()) {
-      throw new OzoneClientException("Invalid key Name");
-    }
-
-    if (downloadTo == null) {
-      throw new OzoneClientException("Invalid download path");
-    }
-
-    FileOutputStream outPutFile = null;
-    HttpGet getRequest = null;
-    try (CloseableHttpClient httpClient = HddsClientUtils.newHttpClient()) {
-      outPutFile = new FileOutputStream(downloadTo.toFile());
-
-      URIBuilder builder = new URIBuilder(volume.getClient().getEndPointURI());
-      builder.setPath("/" + getVolume().getVolumeName() + "/" + getBucketName()
-          + "/" + keyName).build();
-
-      getRequest = getVolume().getClient().getHttpGet(builder.toString());
-      executeGetKey(getRequest, httpClient, outPutFile);
-      outPutFile.flush();
-    } catch (IOException | URISyntaxException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      IOUtils.closeStream(outPutFile);
-      releaseConnection(getRequest);
-    }
-  }
-
-  /**
-   * Returns the data part of the key as a string.
-   *
-   * @param keyName - KeyName to get
-   * @return String - Data
-   * @throws OzoneException
-   */
-  public String getKey(String keyName) throws OzoneException {
-
-    if ((keyName == null) || keyName.isEmpty()) {
-      throw new OzoneClientException("Invalid key Name");
-    }
-
-    HttpGet getRequest = null;
-    ByteArrayOutputStream outPutStream = null;
-    try (CloseableHttpClient httpClient = HddsClientUtils.newHttpClient()) {
-      outPutStream = new ByteArrayOutputStream();
-
-      URIBuilder builder = new URIBuilder(volume.getClient().getEndPointURI());
-
-      builder.setPath("/" + getVolume().getVolumeName() + "/" + getBucketName()
-          + "/" + keyName).build();
-
-      getRequest = getVolume().getClient().getHttpGet(builder.toString());
-      executeGetKey(getRequest, httpClient, outPutStream);
-      return outPutStream.toString(ENCODING_NAME);
-    } catch (IOException | URISyntaxException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      IOUtils.closeStream(outPutStream);
-      releaseConnection(getRequest);
-    }
-
-  }
-
-  /**
-   * Executes get key and returns the data.
-   *
-   * @param getRequest - http Get Request
-   * @param httpClient - Client
-   * @param stream     - Stream to write data to.
-   * @throws IOException
-   * @throws OzoneException
-   */
-  public static void executeGetKey(HttpGet getRequest,
-      CloseableHttpClient httpClient, OutputStream stream)
-      throws IOException, OzoneException {
-
-    HttpEntity entity = null;
-    try {
-
-      HttpResponse response = httpClient.execute(getRequest);
-      int errorCode = response.getStatusLine().getStatusCode();
-      entity = response.getEntity();
-
-      if (errorCode == HTTP_OK) {
-        entity.writeTo(stream);
-        return;
-      }
-
-      if (entity == null) {
-        throw new OzoneClientException("Unexpected null in http payload");
-      }
-
-      throw OzoneException.parse(EntityUtils.toString(entity));
-    } finally {
-      if (entity != null) {
-        EntityUtils.consumeQuietly(entity);
-      }
-    }
-  }
-
-  /**
-   * Deletes a key in this bucket.
-   *
-   * @param keyName - Name of the Key
-   * @throws OzoneException
-   */
-  public void deleteKey(String keyName) throws OzoneException {
-
-    if ((keyName == null) || keyName.isEmpty()) {
-      throw new OzoneClientException("Invalid key Name");
-    }
-
-    HttpDelete deleteRequest = null;
-    try (CloseableHttpClient httpClient = HddsClientUtils.newHttpClient()) {
-      URIBuilder builder = new URIBuilder(volume.getClient().getEndPointURI());
-      builder.setPath("/" + getVolume().getVolumeName() + "/" + getBucketName()
-          + "/" + keyName).build();
-
-      deleteRequest = getVolume()
-          .getClient().getHttpDelete(builder.toString());
-      executeDeleteKey(deleteRequest, httpClient);
-    } catch (IOException | URISyntaxException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      releaseConnection(deleteRequest);
-    }
-  }
-
-  /**
-   * Executes deleteKey.
-   *
-   * @param deleteRequest - http Delete Request
-   * @param httpClient    - Client
-   * @throws IOException
-   * @throws OzoneException
-   */
-  private void executeDeleteKey(HttpDelete deleteRequest,
-      CloseableHttpClient httpClient)
-      throws IOException, OzoneException {
-
-    HttpEntity entity = null;
-    try {
-
-      HttpResponse response = httpClient.execute(deleteRequest);
-      int errorCode = response.getStatusLine().getStatusCode();
-      entity = response.getEntity();
-
-      if (errorCode == HTTP_OK) {
-        return;
-      }
-
-      if (entity == null) {
-        throw new OzoneClientException("Unexpected null in http payload");
-      }
-
-      throw OzoneException.parse(EntityUtils.toString(entity));
-    } finally {
-      if (entity != null) {
-        EntityUtils.consumeQuietly(entity);
-      }
-    }
-  }
-
-  /**
-   * List all keys in a bucket.
-   *
-   * @param resultLength The max length of listing result.
-   * @param previousKey The key from where listing should start,
-   *                    this key is excluded in the result.
-   * @param prefix The prefix that return list keys start with.
-   * @return List of OzoneKeys
-   * @throws OzoneException
-   */
-  public List<OzoneKey> listKeys(String resultLength, String previousKey,
-      String prefix) throws OzoneException {
-    HttpGet getRequest = null;
-    try (CloseableHttpClient httpClient = HddsClientUtils.newHttpClient()) {
-      OzoneRestClient client = getVolume().getClient();
-      URIBuilder builder = new URIBuilder(volume.getClient().getEndPointURI());
-      builder.setPath("/" + getVolume().getVolumeName() + "/" + getBucketName())
-          .build();
-
-      if (!Strings.isNullOrEmpty(resultLength)) {
-        builder.addParameter(Header.OZONE_LIST_QUERY_MAXKEYS, resultLength);
-      }
-
-      if (!Strings.isNullOrEmpty(previousKey)) {
-        builder.addParameter(Header.OZONE_LIST_QUERY_PREVKEY, previousKey);
-      }
-
-      if (!Strings.isNullOrEmpty(prefix)) {
-        builder.addParameter(Header.OZONE_LIST_QUERY_PREFIX, prefix);
-      }
-
-      final String uri = builder.toString();
-      getRequest = client.getHttpGet(uri);
-      LOG.trace("listKeys URI={}", uri);
-      return executeListKeys(getRequest, httpClient);
-
-    } catch (IOException | URISyntaxException e) {
-      throw new OzoneClientException(e.getMessage(), e);
-    } finally {
-      releaseConnection(getRequest);
-    }
-  }
-
-  /**
-   * List keys in a bucket with the provided prefix, with paging results.
-   *
-   * @param prefix The prefix of the object keys
-   * @param maxResult max size per response
-   * @param prevKey the previous key for paging
-   */
-  public List<OzoneKey> listKeys(String prefix, int maxResult, String prevKey)
-      throws OzoneException {
-    HttpGet getRequest = null;
-    try {
-      final URI uri =  new URIBuilder(volume.getClient().getEndPointURI())
-          .setPath(OzoneConsts.KSM_KEY_PREFIX + getVolume().getVolumeName() +
-              OzoneConsts.KSM_KEY_PREFIX + getBucketName())
-          .setParameter(Header.OZONE_LIST_QUERY_PREFIX, prefix)
-          .setParameter(Header.OZONE_LIST_QUERY_MAXKEYS,
-              String.valueOf(maxResult))
-          .setParameter(Header.OZONE_LIST_QUERY_PREVKEY, prevKey)
-          .build();
-      final OzoneRestClient client = getVolume().getClient();
-      getRequest = client.getHttpGet(uri.toString());
-      return executeListKeys(getRequest, HttpClientBuilder.create().build());
-    } catch (IOException | URISyntaxException e) {
-      throw new OzoneClientException(e.getMessage());
-    } finally {
-      releaseConnection(getRequest);
-    }
-  }
-
-  /**
-   * Execute list Key.
-   *
-   * @param getRequest - HttpGet
-   * @param httpClient - HttpClient
-   * @return List<OzoneKey>
-   * @throws IOException
-   * @throws OzoneException
-   */
-  public static List<OzoneKey> executeListKeys(HttpGet getRequest,
-      CloseableHttpClient httpClient) throws IOException, OzoneException {
-    HttpEntity entity = null;
-    List<OzoneKey> ozoneKeyList = new LinkedList<OzoneKey>();
-    try {
-      HttpResponse response = httpClient.execute(getRequest);
-      int errorCode = response.getStatusLine().getStatusCode();
-
-      entity = response.getEntity();
-
-      if (entity == null) {
-        throw new OzoneClientException("Unexpected null in http payload");
-      }
-      if (errorCode == HTTP_OK) {
-        String temp = EntityUtils.toString(entity);
-        ListKeys keyList = ListKeys.parse(temp);
-
-        for (KeyInfo info : keyList.getKeyList()) {
-          ozoneKeyList.add(new OzoneKey(info));
-        }
-        return ozoneKeyList;
-
-      } else {
-        throw OzoneException.parse(EntityUtils.toString(entity));
-      }
-    } finally {
-      if (entity != null) {
-        EntityUtils.consumeQuietly(entity);
-      }
-    }
-  }
-
-  /**
-   * Get info of the specified key.
-   */
-  public OzoneKey getKeyInfo(String keyName) throws OzoneException {
-    if ((keyName == null) || keyName.isEmpty()) {
-      throw new OzoneClientException(
-          "Unable to get key info, key name is null or empty");
-    }
-
-    HttpGet getRequest = null;
-    try (CloseableHttpClient httpClient = HddsClientUtils.newHttpClient()) {
-      OzoneRestClient client = getVolume().getClient();
-      URIBuilder builder = new URIBuilder(volume.getClient().getEndPointURI());
-      builder
-          .setPath("/" + getVolume().getVolumeName() + "/" + getBucketName()
-              + "/" + keyName)
-          .setParameter(Header.OZONE_INFO_QUERY_TAG,
-              Header.OZONE_INFO_QUERY_KEY)
-          .build();
-
-      getRequest = client.getHttpGet(builder.toString());
-      return executeGetKeyInfo(getRequest, httpClient);
-    } catch (IOException | URISyntaxException e) {
-      throw new OzoneClientException(e.getMessage(), e);
-    } finally {
-      releaseConnection(getRequest);
-    }
-  }
-
-  /**
-   * Execute get Key info.
-   *
-   * @param getRequest - HttpGet
-   * @param httpClient - HttpClient
-   * @return List<OzoneKey>
-   * @throws IOException
-   * @throws OzoneException
-   */
-  private OzoneKey executeGetKeyInfo(HttpGet getRequest,
-      CloseableHttpClient httpClient) throws IOException, OzoneException {
-    HttpEntity entity = null;
-    try {
-      HttpResponse response = httpClient.execute(getRequest);
-      int errorCode = response.getStatusLine().getStatusCode();
-      entity = response.getEntity();
-      if (entity == null) {
-        throw new OzoneClientException("Unexpected null in http payload");
-      }
-
-      if (errorCode == HTTP_OK) {
-        OzoneKey key = new OzoneKey(
-            KeyInfo.parse(EntityUtils.toString(entity)));
-        return key;
-      }
-      throw OzoneException.parse(EntityUtils.toString(entity));
-    } finally {
-      if (entity != null) {
-        EntityUtils.consumeQuietly(entity);
-      }
-    }
-  }
-}
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/web/client/OzoneKey.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/web/client/OzoneKey.java
deleted file mode 100644
index 5a3a0c4e78..0000000000
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/web/client/OzoneKey.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * 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
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- *  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.web.client;
-
-import org.apache.hadoop.ozone.web.response.KeyInfo;
-
-/**
- * Client side representation of an ozone Key.
- */
-public class OzoneKey {
-  private KeyInfo keyInfo;
-
-  /**
-   * Constructor for Ozone Key.
-   * @param keyInfo - Key Info
-   */
-  public OzoneKey(KeyInfo keyInfo) {
-    this.keyInfo = keyInfo;
-  }
-
-  /**
-   * Returns Key Info.
-   * @return Object Info
-   */
-  public KeyInfo getObjectInfo() {
-    return keyInfo;
-  }
-
-}
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/web/client/OzoneRestClient.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/web/client/OzoneRestClient.java
deleted file mode 100644
index 8373f676cb..0000000000
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/web/client/OzoneRestClient.java
+++ /dev/null
@@ -1,804 +0,0 @@
-/*
- * 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
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- *  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.web.client;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Strings;
-
-import org.apache.commons.codec.digest.DigestUtils;
-import org.apache.commons.lang.StringUtils;
-import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
-import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.ozone.client.OzoneClientException;
-import org.apache.hadoop.ozone.client.rest.OzoneException;
-import org.apache.hadoop.ozone.client.rest.headers.Header;
-import org.apache.hadoop.ozone.web.response.ListVolumes;
-import org.apache.hadoop.ozone.web.response.VolumeInfo;
-import org.apache.hadoop.ozone.web.utils.OzoneUtils;
-import org.apache.hadoop.util.Time;
-
-import static org.apache.hadoop.hdds.server.ServerUtils.releaseConnection;
-import org.apache.http.HttpEntity;
-import org.apache.http.HttpResponse;
-import org.apache.http.client.methods.HttpDelete;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.client.methods.HttpPut;
-import org.apache.http.client.methods.HttpRequestBase;
-import org.apache.http.client.utils.URIBuilder;
-import org.apache.http.entity.ContentType;
-import org.apache.http.entity.FileEntity;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.util.EntityUtils;
-
-import javax.ws.rs.core.HttpHeaders;
-import java.io.Closeable;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.nio.file.Path;
-import java.text.SimpleDateFormat;
-import java.util.Date;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Locale;
-
-import static java.net.HttpURLConnection.HTTP_CREATED;
-import static java.net.HttpURLConnection.HTTP_OK;
-
-/**
- * Ozone client that connects to an Ozone server. Please note that this class is
- * not  thread safe.
- */
-public class OzoneRestClient implements Closeable {
-  private URI endPointURI;
-  private String userAuth;
-
-  /**
-   * Constructor for OzoneRestClient.
-   */
-  public OzoneRestClient() {
-  }
-
-  /**
-   * Constructor for OzoneRestClient.
-   */
-  public OzoneRestClient(String ozoneURI)
-      throws OzoneException, URISyntaxException {
-    setEndPoint(ozoneURI);
-  }
-
-  /**
-   * Constructor for OzoneRestClient.
-   */
-  public OzoneRestClient(String ozoneURI, String userAuth)
-      throws OzoneException, URISyntaxException {
-    setEndPoint(ozoneURI);
-    setUserAuth(userAuth);
-  }
-
-  /**
-   * Returns the end Point.
-   *
-   * @return String
-   */
-  public URI getEndPointURI() {
-    return endPointURI;
-  }
-
-  /**
-   * Sets the End Point info using an URI.
-   *
-   * @param endPointURI - URI
-   * @throws OzoneException
-   */
-  public void setEndPointURI(URI endPointURI) throws OzoneException {
-    if ((endPointURI == null) || (endPointURI.toString().isEmpty())) {
-      throw new OzoneClientException("Invalid ozone URI");
-    }
-    this.endPointURI = endPointURI;
-  }
-
-  /**
-   * Set endPoint.
-   *
-   * @param clusterFQDN - cluster FQDN.
-   */
-  public void setEndPoint(String clusterFQDN) throws
-      OzoneException, URISyntaxException {
-    setEndPointURI(new URI(clusterFQDN));
-  }
-
-  /**
-   * Get user Auth String.
-   *
-   * @return - User Auth String
-   */
-  public String getUserAuth() {
-    return this.userAuth;
-  }
-
-  /**
-   * Set User Auth.
-   *
-   * @param userAuth - User Auth String
-   */
-  public void setUserAuth(String userAuth) {
-    this.userAuth = userAuth;
-  }
-
-  /**
-   * create volume.
-   *
-   * @param volumeName - volume name 3 - 63 chars, small letters.
-   * @param onBehalfOf - The user on behalf we are making the call for
-   * @param quota      - Quota's are specified in a specific format. it is
-   *                   integer(MB|GB|TB), for example 100TB.
-   * @throws OzoneClientException
-   */
-  public OzoneVolume createVolume(String volumeName, String onBehalfOf,
-                                  String quota) throws OzoneException {
-    HttpPost httpPost = null;
-    try (CloseableHttpClient httpClient = newHttpClient()) {
-      OzoneUtils.verifyResourceName(volumeName);
-
-      URIBuilder builder = new URIBuilder(endPointURI);
-      builder.setPath("/" + volumeName);
-      if (quota != null) {
-        builder.setParameter(Header.OZONE_QUOTA_QUERY_TAG, quota);
-      }
-
-      httpPost = getHttpPost(onBehalfOf, builder.build().toString());
-      executeCreateVolume(httpPost, httpClient);
-      return getVolume(volumeName);
-    } catch (IOException | URISyntaxException | IllegalArgumentException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      releaseConnection(httpPost);
-    }
-  }
-
-  /**
-   * Returns information about an existing Volume. if the Volume does not exist,
-   * or if the user does not have access rights OzoneException is thrown
-   *
-   * @param volumeName - volume name 3 - 63 chars, small letters.
-   * @return OzoneVolume Ozone Client Volume Class.
-   * @throws OzoneException
-   */
-  public OzoneVolume getVolume(String volumeName) throws OzoneException {
-    HttpGet httpGet = null;
-    try (CloseableHttpClient httpClient = newHttpClient()) {
-      OzoneUtils.verifyResourceName(volumeName);
-      URIBuilder builder = new URIBuilder(endPointURI);
-      builder.setPath("/" + volumeName)
-          .setParameter(Header.OZONE_INFO_QUERY_TAG,
-              Header.OZONE_INFO_QUERY_VOLUME)
-          .build();
-
-      httpGet = getHttpGet(builder.toString());
-      return executeInfoVolume(httpGet, httpClient);
-    } catch (IOException | URISyntaxException | IllegalArgumentException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      releaseConnection(httpGet);
-    }
-  }
-
-  /**
-   * List all the volumes owned by the user or Owned by the user specified in
-   * the behalf of string.
-   *
-   * @param onBehalfOf
-   *  User Name of the user if it is not the caller. for example,
-   *  an admin wants to list some other users volumes.
-   * @param prefix
-   *   Return only volumes that match this prefix.
-   * @param maxKeys
-   *   Maximum number of results to return, if the result set
-   *   is smaller than requested size, it means that list is
-   *   complete.
-   * @param previousVolume
-   *   The previous volume name.
-   * @return List of Volumes
-   * @throws OzoneException
-   */
-  public List<OzoneVolume> listVolumes(String onBehalfOf, String prefix,
-      int maxKeys, String previousVolume) throws OzoneException {
-    HttpGet httpGet = null;
-    try (CloseableHttpClient httpClient = newHttpClient()) {
-      URIBuilder builder = new URIBuilder(endPointURI);
-      if (!Strings.isNullOrEmpty(prefix)) {
-        builder.addParameter(Header.OZONE_LIST_QUERY_PREFIX, prefix);
-      }
-
-      if (maxKeys > 0) {
-        builder.addParameter(Header.OZONE_LIST_QUERY_MAXKEYS, Integer
-            .toString(maxKeys));
-      }
-
-      if (!Strings.isNullOrEmpty(previousVolume)) {
-        builder.addParameter(Header.OZONE_LIST_QUERY_PREVKEY,
-            previousVolume);
-      }
-
-      builder.setPath("/").build();
-
-      httpGet = getHttpGet(builder.toString());
-      if (onBehalfOf != null) {
-        httpGet.addHeader(Header.OZONE_USER, onBehalfOf);
-      }
-      return executeListVolume(httpGet, httpClient);
-    } catch (IOException | URISyntaxException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      releaseConnection(httpGet);
-    }
-  }
-
-  /**
-   * List all the volumes owned by the user or Owned by the user specified in
-   * the behalf of string.
-   *
-   * @param onBehalfOf - User Name of the user if it is not the caller. for
-   *                   example, an admin wants to list some other users
-   *                   volumes.
-   * @param prefix     - Return only volumes that match this prefix.
-   * @param maxKeys    - Maximum number of results to return, if the result set
-   *                   is smaller than requested size, it means that list is
-   *                   complete.
-   * @param prevKey    - The last key that client got, server will continue
-   *                   returning results from that point.
-   * @return List of Volumes
-   * @throws OzoneException
-   */
-  public List<OzoneVolume> listVolumes(String onBehalfOf, String prefix,
-      int maxKeys, OzoneVolume prevKey) throws OzoneException {
-    String volumeName = null;
-
-    if (prevKey != null) {
-      volumeName = prevKey.getVolumeName();
-    }
-
-    return listVolumes(onBehalfOf, prefix, maxKeys, volumeName);
-  }
-
-  /**
-   * List volumes of the current user or if onBehalfof is not null lists volume
-   * owned by that user. You need admin privilege to read other users volume
-   * lists.
-   *
-   * @param onBehalfOf - Name of the user you want to get volume list
-   * @return - Volume list.
-   * @throws OzoneException
-   */
-  public List<OzoneVolume> listVolumes(String onBehalfOf)
-      throws OzoneException {
-    return listVolumes(onBehalfOf, null,
-        Integer.parseInt(Header.OZONE_DEFAULT_LIST_SIZE), StringUtils.EMPTY);
-  }
-
-  /**
-   * List all volumes in a cluster. This can be invoked only by an Admin.
-   *
-   * @param prefix  - Returns only volumes that match this prefix.
-   * @param maxKeys - Maximum niumber of keys to return
-   * @param prevKey - Last Ozone Volume from the last Iteration.
-   * @return List of Volumes
-   * @throws OzoneException
-   */
-  public List<OzoneVolume> listAllVolumes(String prefix, int maxKeys,
-      OzoneVolume prevKey) throws OzoneException {
-    HttpGet httpGet = null;
-    try (CloseableHttpClient httpClient = newHttpClient()) {
-      URIBuilder builder = new URIBuilder(endPointURI);
-      if (prefix != null) {
-        builder.addParameter(Header.OZONE_LIST_QUERY_PREFIX, prefix);
-      }
-
-      if (maxKeys > 0) {
-        builder.addParameter(Header.OZONE_LIST_QUERY_MAXKEYS, Integer
-            .toString(maxKeys));
-      }
-
-      if (prevKey != null) {
-        builder.addParameter(Header.OZONE_LIST_QUERY_PREVKEY,
-            prevKey.getOwnerName()+ "/" + prevKey.getVolumeName());
-      }
-
-      builder.addParameter(Header.OZONE_LIST_QUERY_ROOTSCAN, "true");
-      builder.setPath("/").build();
-      httpGet = getHttpGet(builder.toString());
-      return executeListVolume(httpGet, httpClient);
-
-    } catch (IOException | URISyntaxException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      releaseConnection(httpGet);
-    }
-  }
-
-    /**
-     * delete a given volume.
-     *
-     * @param volumeName - volume to be deleted.
-     * @throws OzoneException - Ozone Exception
-     */
-  public void deleteVolume(String volumeName) throws OzoneException {
-    HttpDelete httpDelete = null;
-    try (CloseableHttpClient httpClient = newHttpClient()) {
-      OzoneUtils.verifyResourceName(volumeName);
-      URIBuilder builder = new URIBuilder(endPointURI);
-      builder.setPath("/" + volumeName).build();
-
-      httpDelete = getHttpDelete(builder.toString());
-      executeDeleteVolume(httpDelete, httpClient);
-    } catch (IOException | URISyntaxException | IllegalArgumentException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      releaseConnection(httpDelete);
-    }
-  }
-
-  /**
-   * Sets the Volume Owner.
-   *
-   * @param volumeName - Volume Name
-   * @param newOwner   - New Owner Name
-   * @throws OzoneException
-   */
-  public void setVolumeOwner(String volumeName, String newOwner)
-      throws OzoneException {
-    HttpPut putRequest = null;
-    if (newOwner == null || newOwner.isEmpty()) {
-      throw new OzoneClientException("Invalid new owner name");
-    }
-    try (CloseableHttpClient httpClient = newHttpClient()) {
-      OzoneUtils.verifyResourceName(volumeName);
-      URIBuilder builder = new URIBuilder(endPointURI);
-      builder.setPath("/" + volumeName).build();
-
-      putRequest = getHttpPut(builder.toString());
-      putRequest.addHeader(Header.OZONE_USER, newOwner);
-      executePutVolume(putRequest, httpClient);
-
-    } catch (URISyntaxException | IllegalArgumentException | IOException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      releaseConnection(putRequest);
-    }
-  }
-
-  /**
-   * Sets the Volume Quota. Quota's are specified in a specific format. it is
-   * <integer>|(MB|GB|TB. for example 100TB.
-   * <p>
-   * To Remove a quota you can specify Header.OZONE_QUOTA_REMOVE
-   *
-   * @param volumeName - volume name
-   * @param quota      - Quota String or  Header.OZONE_QUOTA_REMOVE
-   * @throws OzoneException
-   */
-  public void setVolumeQuota(String volumeName, String quota)
-      throws OzoneException {
-    if (quota == null || quota.isEmpty()) {
-      throw new OzoneClientException("Invalid quota");
-    }
-    HttpPut putRequest = null;
-    try (CloseableHttpClient httpClient = newHttpClient()) {
-      OzoneUtils.verifyResourceName(volumeName);
-      URIBuilder builder = new URIBuilder(endPointURI);
-      builder.setPath("/" + volumeName)
-          .setParameter(Header.OZONE_QUOTA_QUERY_TAG, quota)
-          .build();
-
-      putRequest = getHttpPut(builder.toString());
-      executePutVolume(putRequest, httpClient);
-
-    } catch (URISyntaxException | IllegalArgumentException | IOException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      releaseConnection(putRequest);
-    }
-  }
-
-  /**
-   * Sends the create Volume request to the server.
-   *
-   * @param httppost   - http post class
-   * @param httpClient - httpClient
-   * @throws IOException    -
-   * @throws OzoneException
-   */
-  private void executeCreateVolume(HttpPost httppost,
-      final CloseableHttpClient httpClient)
-      throws IOException, OzoneException {
-    HttpEntity entity = null;
-    try {
-      HttpResponse response = httpClient.execute(httppost);
-      int errorCode = response.getStatusLine().getStatusCode();
-      entity = response.getEntity();
-
-      if ((errorCode == HTTP_OK) || (errorCode == HTTP_CREATED)) {
-        return;
-      }
-
-      if (entity != null) {
-        throw OzoneException.parse(EntityUtils.toString(entity));
-      } else {
-        throw new OzoneClientException("Unexpected null in http payload");
-      }
-    } finally {
-      if (entity != null) {
-        EntityUtils.consume(entity);
-      }
-    }
-  }
-
-  /**
-   * Sends the create Volume request to the server.
-   *
-   * @param httpGet - httpGet
-   * @return OzoneVolume
-   * @throws IOException    -
-   * @throws OzoneException
-   */
-  private OzoneVolume executeInfoVolume(HttpGet httpGet,
-      final CloseableHttpClient httpClient)
-      throws IOException, OzoneException {
-    HttpEntity entity = null;
-    try {
-      HttpResponse response = httpClient.execute(httpGet);
-      int errorCode = response.getStatusLine().getStatusCode();
-
-      entity = response.getEntity();
-      if (entity == null) {
-        throw new OzoneClientException("Unexpected null in http payload");
-      }
-
-      if (errorCode == HTTP_OK) {
-        OzoneVolume volume = new OzoneVolume(this);
-        volume.setVolumeInfo(EntityUtils.toString(entity));
-        return volume;
-      } else {
-        throw OzoneException.parse(EntityUtils.toString(entity));
-      }
-    } finally {
-      if (entity != null) {
-        EntityUtils.consumeQuietly(entity);
-      }
-    }
-  }
-
-  /**
-   * Sends update volume requests to the server.
-   *
-   * @param putRequest http request
-   * @throws IOException
-   * @throws OzoneException
-   */
-  private void executePutVolume(HttpPut putRequest,
-      final CloseableHttpClient httpClient)
-      throws IOException, OzoneException {
-    HttpEntity entity = null;
-    try {
-      HttpResponse response = httpClient.execute(putRequest);
-      int errorCode = response.getStatusLine().getStatusCode();
-      entity = response.getEntity();
-      if (errorCode != HTTP_OK) {
-        throw OzoneException.parse(EntityUtils.toString(entity));
-      }
-    } finally {
-      if (entity != null) {
-        EntityUtils.consume(entity);
-      }
-    }
-  }
-
-  /**
-   * List Volumes.
-   *
-   * @param httpGet - httpGet
-   * @return OzoneVolume
-   * @throws IOException    -
-   * @throws OzoneException
-   */
-  private List<OzoneVolume> executeListVolume(HttpGet httpGet,
-      final CloseableHttpClient httpClient)
-      throws IOException, OzoneException {
-    HttpEntity entity = null;
-    List<OzoneVolume> volList = new LinkedList<>();
-    try {
-      HttpResponse response = httpClient.execute(httpGet);
-      int errorCode = response.getStatusLine().getStatusCode();
-      entity = response.getEntity();
-
-      if (entity == null) {
-        throw new OzoneClientException("Unexpected null in http payload");
-      }
-
-      String temp = EntityUtils.toString(entity);
-      if (errorCode == HTTP_OK) {
-        ListVolumes listVolumes =
-            ListVolumes.parse(temp);
-
-        for (VolumeInfo info : listVolumes.getVolumes()) {
-          volList.add(new OzoneVolume(info, this));
-        }
-        return volList;
-
-      } else {
-        throw OzoneException.parse(EntityUtils.toString(entity));
-      }
-    } finally {
-      if (entity != null) {
-        EntityUtils.consumeQuietly(entity);
-      }
-    }
-  }
-
-  /**
-   * Delete Volume.
-   *
-   * @param httpDelete - Http Delete Request
-   * @throws IOException
-   * @throws OzoneException
-   */
-  private void executeDeleteVolume(HttpDelete httpDelete,
-      final CloseableHttpClient httpClient)
-      throws IOException, OzoneException {
-    HttpEntity entity = null;
-    try {
-      HttpResponse response = httpClient.execute(httpDelete);
-      int errorCode = response.getStatusLine().getStatusCode();
-      entity = response.getEntity();
-
-      if (errorCode != HTTP_OK) {
-        throw OzoneException.parse(EntityUtils.toString(entity));
-      }
-    } finally {
-      if (entity != null) {
-        EntityUtils.consumeQuietly(entity);
-      }
-    }
-  }
-
-  /**
-   * Puts a Key in Ozone Bucket.
-   *
-   * @param volumeName - Name of the Volume
-   * @param bucketName - Name of the Bucket
-   * @param keyName - Name of the Key
-   * @param file    - Stream that gets read to be put into Ozone.
-   * @throws OzoneException
-   */
-  public void putKey(String volumeName, String bucketName, String keyName,
-      File file) throws OzoneException {
-    OzoneUtils.verifyResourceName(volumeName);
-    OzoneUtils.verifyResourceName(bucketName);
-
-    if (StringUtils.isEmpty(keyName)) {
-      throw new OzoneClientException("Invalid key Name");
-    }
-
-    if (file == null) {
-      throw new OzoneClientException("Invalid data stream");
-    }
-
-    HttpPut putRequest = null;
-    FileInputStream fis = null;
-    try (CloseableHttpClient httpClient = HddsClientUtils.newHttpClient()) {
-      URIBuilder builder = new URIBuilder(getEndPointURI());
-      builder.setPath("/" + volumeName + "/" + bucketName + "/" + keyName)
-          .build();
-
-      putRequest = getHttpPut(builder.toString());
-
-      FileEntity fileEntity = new FileEntity(file, ContentType
-          .APPLICATION_OCTET_STREAM);
-      putRequest.setEntity(fileEntity);
-
-      fis = new FileInputStream(file);
-      putRequest.setHeader(Header.CONTENT_MD5, DigestUtils.md5Hex(fis));
-      OzoneBucket.executePutKey(putRequest, httpClient);
-    } catch (IOException | URISyntaxException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      IOUtils.closeStream(fis);
-      releaseConnection(putRequest);
-    }
-  }
-
-  /**
-   * Gets a key from the Ozone server and writes to the file pointed by the
-   * downloadTo Path.
-   *
-   * @param volumeName - Volume Name in Ozone.
-   * @param bucketName - Bucket Name in Ozone.
-   * @param keyName - Key Name in Ozone.
-   * @param downloadTo File Name to download the Key's Data to
-   */
-  public void getKey(String volumeName, String bucketName, String keyName,
-      Path downloadTo) throws OzoneException {
-    OzoneUtils.verifyResourceName(volumeName);
-    OzoneUtils.verifyResourceName(bucketName);
-
-    if (StringUtils.isEmpty(keyName)) {
-      throw new OzoneClientException("Invalid key Name");
-    }
-
-    if (downloadTo == null) {
-      throw new OzoneClientException("Invalid download path");
-    }
-
-    FileOutputStream outPutFile = null;
-    HttpGet getRequest = null;
-    try (CloseableHttpClient httpClient = HddsClientUtils.newHttpClient()) {
-      outPutFile = new FileOutputStream(downloadTo.toFile());
-
-      URIBuilder builder = new URIBuilder(getEndPointURI());
-      builder.setPath("/" + volumeName + "/" + bucketName + "/" + keyName)
-          .build();
-
-      getRequest = getHttpGet(builder.toString());
-      OzoneBucket.executeGetKey(getRequest, httpClient, outPutFile);
-      outPutFile.flush();
-    } catch (IOException | URISyntaxException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      IOUtils.closeStream(outPutFile);
-      releaseConnection(getRequest);
-    }
-  }
-
-  /**
-   * List all keys in the given bucket.
-   *
-   * @param volumeName - Volume name
-   * @param bucketName - Bucket name
-   * @param resultLength The max length of listing result.
-   * @param previousKey The key from where listing should start,
-   *                    this key is excluded in the result.
-   * @param prefix The prefix that return list keys start with.
-   *
-   * @return List of OzoneKeys
-   */
-  public List<OzoneKey> listKeys(String volumeName, String bucketName,
-      String resultLength, String previousKey, String prefix)
-      throws OzoneException {
-    OzoneUtils.verifyResourceName(volumeName);
-    OzoneUtils.verifyResourceName(bucketName);
-
-    HttpGet getRequest = null;
-    try (CloseableHttpClient httpClient = HddsClientUtils.newHttpClient()) {
-      URIBuilder builder = new URIBuilder(getEndPointURI());
-      builder.setPath("/" + volumeName + "/" + bucketName).build();
-
-      if (!Strings.isNullOrEmpty(resultLength)) {
-        builder.addParameter(Header.OZONE_LIST_QUERY_MAXKEYS, resultLength);
-      }
-
-      if (!Strings.isNullOrEmpty(previousKey)) {
-        builder.addParameter(Header.OZONE_LIST_QUERY_PREVKEY, previousKey);
-      }
-
-      if (!Strings.isNullOrEmpty(prefix)) {
-        builder.addParameter(Header.OZONE_LIST_QUERY_PREFIX, prefix);
-      }
-
-      getRequest = getHttpGet(builder.toString());
-      return OzoneBucket.executeListKeys(getRequest, httpClient);
-    } catch (IOException | URISyntaxException e) {
-      throw new OzoneClientException(e.getMessage(), e);
-    } finally {
-      releaseConnection(getRequest);
-    }
-  }
-
-  /**
-   * Returns a standard HttpPost Object to use for ozone post requests.
-   *
-   * @param onBehalfOf - If the use is being made on behalf of user, that user
-   * @param uriString  - UriString
-   * @return HttpPost
-   */
-  public HttpPost getHttpPost(String onBehalfOf, String uriString) {
-    HttpPost httpPost = new HttpPost(uriString);
-    addOzoneHeaders(httpPost);
-    if (onBehalfOf != null) {
-      httpPost.addHeader(Header.OZONE_USER, onBehalfOf);
-    }
-    return httpPost;
-  }
-
-  /**
-   * Returns a standard HttpGet Object to use for ozone Get requests.
-   *
-   * @param uriString - The full Uri String
-   * @return HttpGet
-   */
-  public HttpGet getHttpGet(String uriString) {
-    HttpGet httpGet = new HttpGet(uriString);
-    addOzoneHeaders(httpGet);
-    return httpGet;
-  }
-
-  /**
-   * Returns httpDelete.
-   *
-   * @param uriString - uri
-   * @return HttpDelete
-   */
-  public HttpDelete getHttpDelete(String uriString) {
-    HttpDelete httpDel = new HttpDelete(uriString);
-    addOzoneHeaders(httpDel);
-    return httpDel;
-  }
-
-  /**
-   * returns an HttpPut Object.
-   *
-   * @param uriString - Uri
-   * @return HttpPut
-   */
-  public HttpPut getHttpPut(String uriString) {
-    HttpPut httpPut = new HttpPut(uriString);
-    addOzoneHeaders(httpPut);
-    return httpPut;
-  }
-
-  /**
-   * Add Ozone Headers.
-   *
-   * @param httpRequest - Http Request
-   */
-  private void addOzoneHeaders(HttpRequestBase httpRequest) {
-    SimpleDateFormat format =
-        new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss ZZZ", Locale.US);
-
-    httpRequest.addHeader(Header.OZONE_VERSION_HEADER,
-        Header.OZONE_V1_VERSION_HEADER);
-    httpRequest.addHeader(HttpHeaders.DATE,
-        format.format(new Date(Time.monotonicNow())));
-    if (getUserAuth() != null) {
-      httpRequest.addHeader(HttpHeaders.AUTHORIZATION,
-          Header.OZONE_SIMPLE_AUTHENTICATION_SCHEME + " " +
-              getUserAuth());
-    }
-  }
-
-  /**
-   * Closes this stream and releases any system resources associated with it. If
-   * the stream is already closed then invoking this method has no effect.
-   *
-   * @throws IOException if an I/O error occurs
-   */
-  @Override
-  public void close() throws IOException {
-    // TODO : Currently we create a new HTTP client. We should switch
-    // This to a Pool and cleanup the pool here.
-  }
-
-  @VisibleForTesting
-  public CloseableHttpClient newHttpClient() {
-    return HddsClientUtils.newHttpClient();
-  }
-}
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/web/client/OzoneVolume.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/web/client/OzoneVolume.java
deleted file mode 100644
index 9d3831c5a2..0000000000
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/web/client/OzoneVolume.java
+++ /dev/null
@@ -1,584 +0,0 @@
-/*
- * 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
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- *  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.web.client;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Strings;
-import org.apache.hadoop.fs.StorageType;
-import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
-import org.apache.hadoop.ozone.client.OzoneClientException;
-import org.apache.hadoop.ozone.client.rest.OzoneException;
-import org.apache.hadoop.ozone.client.rest.headers.Header;
-import org.apache.hadoop.ozone.web.request.OzoneQuota;
-import org.apache.hadoop.ozone.web.response.BucketInfo;
-import org.apache.hadoop.ozone.web.response.ListBuckets;
-import org.apache.hadoop.ozone.web.response.VolumeInfo;
-import org.apache.hadoop.ozone.OzoneConsts;
-import org.apache.hadoop.ozone.web.utils.OzoneUtils;
-
-import static org.apache.hadoop.hdds.server.ServerUtils.releaseConnection;
-import org.apache.http.HttpEntity;
-import org.apache.http.HttpResponse;
-import org.apache.http.client.methods.HttpDelete;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.client.methods.HttpPut;
-import org.apache.http.client.utils.URIBuilder;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.util.EntityUtils;
-
-import java.io.IOException;
-import java.net.URISyntaxException;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-
-import static java.net.HttpURLConnection.HTTP_CREATED;
-import static java.net.HttpURLConnection.HTTP_OK;
-
-/**
- * Ozone Volume Class.
- */
-public class OzoneVolume {
-  private VolumeInfo volumeInfo;
-  private Map<String, String> headerMap;
-  private final OzoneRestClient client;
-
-  /**
-   * Constructor for OzoneVolume.
-   */
-  public OzoneVolume(OzoneRestClient client) {
-    this.client = client;
-    this.headerMap = new HashMap<>();
-  }
-
-  /**
-   * Constructor for OzoneVolume.
-   *
-   * @param volInfo - volume Info.
-   * @param client  Client
-   */
-  public OzoneVolume(VolumeInfo volInfo, OzoneRestClient client) {
-    this.volumeInfo = volInfo;
-    this.client = client;
-  }
-
-  /**
-   * Returns a Json String of this class.
-   * @return String
-   * @throws IOException
-   */
-  public String getJsonString() throws IOException {
-    return volumeInfo.toJsonString();
-  }
-
-  /**
-   * sets the Volume Info.
-   *
-   * @param volInfoString - Volume Info String
-   */
-  public void setVolumeInfo(String volInfoString) throws IOException {
-    this.volumeInfo = VolumeInfo.parse(volInfoString);
-  }
-
-  /**
-   * @return the volume info.
-   */
-  public VolumeInfo getVolumeInfo() {
-    return this.volumeInfo;
-  }
-
-  /**
-   * Returns volume Name.
-   *
-   * @return Volume Name.
-   */
-  public String getVolumeName() {
-    return this.volumeInfo.getVolumeName();
-  }
-
-  /**
-   * Get created by.
-   *
-   * @return String
-   */
-  public String getCreatedby() {
-    return this.volumeInfo.getCreatedBy();
-  }
-
-  /**
-   * returns the Owner name.
-   *
-   * @return String
-   */
-  public String getOwnerName() {
-    return this.volumeInfo.getOwner().getName();
-  }
-
-  /**
-   * Returns Quota Info.
-   *
-   * @return Quota
-   */
-  public OzoneQuota getQuota() {
-    return volumeInfo.getQuota();
-  }
-
-  /**
-   * Returns creation time of Volume.
-   *
-   * @return String
-   */
-  public String getCreatedOn() {
-    return volumeInfo.getCreatedOn();
-  }
-
-  /**
-   * Returns a Http header from the Last Volume related call.
-   *
-   * @param headerName - Name of the header
-   * @return - Header Value
-   */
-  public String getHeader(String headerName) {
-    return headerMap.get(headerName);
-  }
-
-  /**
-   * Gets the Client, this is used by Bucket and Key Classes.
-   *
-   * @return - Ozone Client
-   */
-  OzoneRestClient getClient() {
-    return client;
-  }
-
-  /**
-   * Create Bucket - Creates a bucket under a given volume.
-   *
-   * @param bucketName - Bucket Name
-   * @param acls - Acls - User Acls
-   * @param storageType - Storage Class
-   * @param versioning - enable versioning support on a bucket.
-   *
-   *
-   * @return - a Ozone Bucket Object
-   */
-  public OzoneBucket createBucket(String bucketName, String[] acls,
-                                  StorageType storageType,
-                                  OzoneConsts.Versioning versioning)
-      throws OzoneException {
-
-    HttpPost httpPost = null;
-    try (CloseableHttpClient httpClient = newHttpClient()) {
-      OzoneUtils.verifyResourceName(bucketName);
-      URIBuilder builder = new URIBuilder(getClient().getEndPointURI());
-      builder.setPath("/" + getVolumeName() + "/" + bucketName).build();
-
-      httpPost = client.getHttpPost(null, builder.toString());
-      if (acls != null) {
-        for (String acl : acls) {
-          httpPost
-              .addHeader(Header.OZONE_ACLS, Header.OZONE_ACL_ADD + " " + acl);
-        }
-      }
-
-      httpPost.addHeader(Header.OZONE_STORAGE_TYPE, storageType.toString());
-      httpPost.addHeader(Header.OZONE_BUCKET_VERSIONING, versioning.toString());
-      executeCreateBucket(httpPost, httpClient);
-      return getBucket(bucketName);
-    } catch (IOException | URISyntaxException | IllegalArgumentException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      releaseConnection(httpPost);
-    }
-  }
-
-  /**
-   * Create Bucket.
-   *
-   * @param bucketName - bucket name
-   * @param acls - acls
-   * @param storageType - storage class
-   *
-   * @throws OzoneException
-   */
-  public OzoneBucket createBucket(String bucketName, String[] acls,
-                                  StorageType storageType)
-      throws OzoneException {
-    return createBucket(bucketName, acls, storageType,
-        OzoneConsts.Versioning.DISABLED);
-  }
-
-  /**
-   * Create Bucket.
-   *
-   * @param bucketName - bucket name
-   * @param acls - acls
-   *
-   * @throws OzoneException
-   */
-  public OzoneBucket createBucket(String bucketName, String[] acls)
-      throws OzoneException {
-    return createBucket(bucketName, acls, StorageType.DEFAULT,
-        OzoneConsts.Versioning.DISABLED);
-  }
-
-
-  /**
-   * Create Bucket.
-   *
-   * @param bucketName - bucket name
-   *
-   * @throws OzoneException
-   */
-  public OzoneBucket createBucket(String bucketName) throws OzoneException {
-    return createBucket(bucketName, null,  StorageType.DEFAULT,
-        OzoneConsts.Versioning.DISABLED);
-  }
-
-
-  /**
-   * execute a Create Bucket Request against Ozone server.
-   *
-   * @param httppost - httpPost
-   *
-   * @throws IOException
-   * @throws OzoneException
-   */
-  private void executeCreateBucket(HttpPost httppost,
-      CloseableHttpClient httpClient)
-      throws IOException, OzoneException {
-    HttpEntity entity = null;
-    try {
-      HttpResponse response = httpClient.execute(httppost);
-      int errorCode = response.getStatusLine().getStatusCode();
-      entity = response.getEntity();
-      if ((errorCode == HTTP_OK) || (errorCode == HTTP_CREATED)) {
-        return;
-      }
-
-      if (entity != null) {
-        throw OzoneException.parse(EntityUtils.toString(entity));
-      } else {
-        throw new OzoneClientException("Unexpected null in http payload");
-      }
-    } finally {
-      if (entity != null) {
-        EntityUtils.consumeQuietly(entity);
-      }
-    }
-  }
-
-  /**
-   * Adds Acls to an existing bucket.
-   *
-   * @param bucketName - Name of the bucket
-   * @param acls - Acls
-   *
-   * @throws OzoneException
-   */
-  public void addAcls(String bucketName, String[] acls) throws OzoneException {
-    HttpPut putRequest = null;
-    try (CloseableHttpClient httpClient = newHttpClient()) {
-      OzoneUtils.verifyResourceName(bucketName);
-      URIBuilder builder = new URIBuilder(getClient().getEndPointURI());
-      builder.setPath("/" + getVolumeName() + "/" + bucketName).build();
-      putRequest = client.getHttpPut(builder.toString());
-
-      for (String acl : acls) {
-        putRequest
-            .addHeader(Header.OZONE_ACLS, Header.OZONE_ACL_ADD + " " + acl);
-      }
-      executePutBucket(putRequest, httpClient);
-    } catch (URISyntaxException | IOException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      releaseConnection(putRequest);
-    }
-  }
-
-  /**
-   * Removes ACLs from a bucket.
-   *
-   * @param bucketName - Bucket Name
-   * @param acls - Acls to be removed
-   *
-   * @throws OzoneException
-   */
-  public void removeAcls(String bucketName, String[] acls)
-      throws OzoneException {
-    HttpPut putRequest = null;
-    try (CloseableHttpClient httpClient = newHttpClient()) {
-      OzoneUtils.verifyResourceName(bucketName);
-      URIBuilder builder = new URIBuilder(getClient().getEndPointURI());
-      builder.setPath("/" + getVolumeName() + "/" + bucketName).build();
-      putRequest = client.getHttpPut(builder.toString());
-
-      for (String acl : acls) {
-        putRequest
-            .addHeader(Header.OZONE_ACLS, Header.OZONE_ACL_REMOVE + " " + acl);
-      }
-      executePutBucket(putRequest, httpClient);
-    } catch (URISyntaxException | IOException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      releaseConnection(putRequest);
-    }
-  }
-
-  /**
-   * Returns information about an existing bucket.
-   *
-   * @param bucketName - BucketName
-   *
-   * @return OZoneBucket
-   */
-  public OzoneBucket getBucket(String bucketName) throws OzoneException {
-    HttpGet getRequest = null;
-    try (CloseableHttpClient httpClient = newHttpClient()) {
-      OzoneUtils.verifyResourceName(bucketName);
-      URIBuilder builder = new URIBuilder(getClient().getEndPointURI());
-      builder.setPath("/" + getVolumeName() + "/" + bucketName)
-        .setParameter(Header.OZONE_INFO_QUERY_TAG,
-            Header.OZONE_INFO_QUERY_BUCKET).build();
-      getRequest = client.getHttpGet(builder.toString());
-      return executeInfoBucket(getRequest, httpClient);
-
-    } catch (IOException | URISyntaxException | IllegalArgumentException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      releaseConnection(getRequest);
-    }
-  }
-
-
-  /**
-   * Execute the info bucket call.
-   *
-   * @param getRequest - httpGet Request
-   * @param httpClient - Http Client
-   *
-   * @return OzoneBucket
-   *
-   * @throws IOException
-   * @throws OzoneException
-   */
-  private OzoneBucket executeInfoBucket(HttpGet getRequest,
-      CloseableHttpClient httpClient)
-      throws IOException, OzoneException {
-    HttpEntity entity = null;
-    try {
-      HttpResponse response = httpClient.execute(getRequest);
-      int errorCode = response.getStatusLine().getStatusCode();
-      entity = response.getEntity();
-      if (entity == null) {
-        throw new OzoneClientException("Unexpected null in http payload");
-      }
-      if ((errorCode == HTTP_OK) || (errorCode == HTTP_CREATED)) {
-        OzoneBucket bucket =
-            new OzoneBucket(BucketInfo.parse(EntityUtils.toString(entity)),
-                this);
-        return bucket;
-      }
-      throw OzoneException.parse(EntityUtils.toString(entity));
-    } finally {
-      if (entity != null) {
-        EntityUtils.consumeQuietly(entity);
-      }
-    }
-  }
-
-  /**
-   * Execute the put bucket call.
-   *
-   * @param putRequest - http put request
-   * @param httpClient - Http Client
-   *
-   * @return OzoneBucket
-   *
-   * @throws IOException
-   * @throws OzoneException
-   */
-  private void executePutBucket(HttpPut putRequest,
-      CloseableHttpClient httpClient)
-      throws IOException, OzoneException {
-    HttpEntity entity = null;
-    try {
-      HttpResponse response = httpClient.execute(putRequest);
-      int errorCode = response.getStatusLine().getStatusCode();
-      entity = response.getEntity();
-
-      if (errorCode == HTTP_OK) {
-        return;
-      }
-
-      if (entity != null) {
-        throw OzoneException.parse(EntityUtils.toString(entity));
-      }
-
-      throw new OzoneClientException("Unexpected null in http result");
-    } finally {
-      if (entity != null) {
-        EntityUtils.consumeQuietly(entity);
-      }
-    }
-  }
-
-  /**
-   * Gets a list of buckets on this volume.
-   *
-   * @return - List of buckets
-   *
-   * @throws OzoneException
-   */
-  public List<OzoneBucket> listBuckets(String resultLength,
-      String previousBucket, String prefix) throws OzoneException {
-    HttpGet getRequest = null;
-    try (CloseableHttpClient httpClient = newHttpClient()) {
-      URIBuilder builder = new URIBuilder(getClient().getEndPointURI());
-      builder.setPath("/" + getVolumeName()).build();
-      if (!Strings.isNullOrEmpty(resultLength)) {
-        builder.addParameter(Header.OZONE_LIST_QUERY_MAXKEYS, resultLength);
-      }
-      if (!Strings.isNullOrEmpty(previousBucket)) {
-        builder.addParameter(Header.OZONE_LIST_QUERY_PREVKEY, previousBucket);
-      }
-      if (!Strings.isNullOrEmpty(prefix)) {
-        builder.addParameter(Header.OZONE_LIST_QUERY_PREFIX, prefix);
-      }
-
-      getRequest = client.getHttpGet(builder.toString());
-      return executeListBuckets(getRequest, httpClient);
-
-    } catch (IOException | URISyntaxException e) {
-      throw new OzoneClientException(e.getMessage(), e);
-    } finally {
-      releaseConnection(getRequest);
-    }
-  }
-
-  /**
-   * executes the List Bucket Call.
-   *
-   * @param getRequest - http Request
-   * @param httpClient - http Client
-   *
-   * @return List of OzoneBuckets
-   *
-   * @throws IOException
-   * @throws OzoneException
-   */
-  private List<OzoneBucket> executeListBuckets(HttpGet getRequest,
-      CloseableHttpClient httpClient)
-      throws IOException, OzoneException {
-    HttpEntity entity = null;
-    List<OzoneBucket> ozoneBucketList = new LinkedList<OzoneBucket>();
-    try {
-      HttpResponse response = httpClient.execute(getRequest);
-      int errorCode = response.getStatusLine().getStatusCode();
-
-      entity = response.getEntity();
-
-      if (entity == null) {
-        throw new OzoneClientException("Unexpected null in http payload");
-      }
-      if (errorCode == HTTP_OK) {
-        ListBuckets bucketList =
-            ListBuckets.parse(EntityUtils.toString(entity));
-
-        for (BucketInfo info : bucketList.getBuckets()) {
-          ozoneBucketList.add(new OzoneBucket(info, this));
-        }
-        return ozoneBucketList;
-
-      } else {
-        throw OzoneException.parse(EntityUtils.toString(entity));
-      }
-    } finally {
-      if (entity != null) {
-        EntityUtils.consumeQuietly(entity);
-      }
-    }
-  }
-
-  /**
-   * Delete an empty bucket.
-   *
-   * @param bucketName - Name of the bucket to delete
-   *
-   * @throws OzoneException
-   */
-  public void deleteBucket(String bucketName) throws OzoneException {
-    HttpDelete delRequest = null;
-    try (CloseableHttpClient httpClient = newHttpClient()) {
-      OzoneUtils.verifyResourceName(bucketName);
-      URIBuilder builder = new URIBuilder(getClient().getEndPointURI());
-      builder.setPath("/" + getVolumeName() + "/" + bucketName).build();
-
-      delRequest = client.getHttpDelete(builder.toString());
-      executeDeleteBucket(delRequest, httpClient);
-
-    } catch (IOException | URISyntaxException | IllegalArgumentException ex) {
-      throw new OzoneClientException(ex.getMessage(), ex);
-    } finally {
-      releaseConnection(delRequest);
-    }
-  }
-
-  /**
-   * Executes delete bucket call.
-   *
-   * @param delRequest - Delete Request
-   * @param httpClient - Http Client
-7   *
-   * @throws IOException
-   * @throws OzoneException
-   */
-  private void executeDeleteBucket(HttpDelete delRequest,
-      CloseableHttpClient httpClient)
-      throws IOException, OzoneException {
-    HttpEntity entity = null;
-    try {
-      HttpResponse response = httpClient.execute(delRequest);
-      int errorCode = response.getStatusLine().getStatusCode();
-      entity = response.getEntity();
-
-      if (errorCode == HTTP_OK) {
-        return;
-      }
-
-      if (entity == null) {
-        throw new OzoneClientException("Unexpected null in http payload.");
-      }
-
-      throw OzoneException.parse(EntityUtils.toString(entity));
-
-    } finally {
-      if (entity != null) {
-        EntityUtils.consumeQuietly(entity);
-      }
-    }
-  }
-
-  @VisibleForTesting
-  public CloseableHttpClient newHttpClient() {
-    return HddsClientUtils.newHttpClient();
-  }
-}
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/web/client/package-info.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/web/client/package-info.java
deleted file mode 100644
index 046568b7b3..0000000000
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/web/client/package-info.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- *  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.
- */
-
-
-/**
- * Ozone client library is a java client for the Ozone
- * Object Store.
- */
-package org.apache.hadoop.ozone.web.client;
-
-/**
- This library is  a simple Ozone REST Library.
-
- This library is a very *minimal* client written for tests and
- command line utils that work against Ozone. It does not have
- things like thread-pools and support for extended security models yet.
-
- OzoneClients return OzoneVolumes and OzoneVolumes return OzoneBuckets.
- **/
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java
index 993681511d..ad8b0163b4 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java
@@ -37,7 +37,6 @@ import org.apache.hadoop.ozone.ksm.KSMConfigKeys;
 import org.apache.hadoop.ozone.ksm.KeySpaceManager;
 import org.apache.hadoop.hdds.scm.server.SCMStorage;
 import org.apache.hadoop.ozone.ksm.KSMStorage;
-import org.apache.hadoop.ozone.web.client.OzoneRestClient;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.protocolPB
     .StorageContainerLocationProtocolClientSideTranslatorPB;
@@ -167,7 +166,7 @@ public final class MiniOzoneClusterImpl implements MiniOzoneCluster {
   }
 
   /**
-   * Creates an {@link OzoneRestClient} connected to this cluster's REST
+   * Creates an {@link OzoneClient} connected to this cluster's REST
    * service. Callers take ownership of the client and must close it when done.
    *
    * @return OzoneRestClient connected to this cluster's REST service
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/RatisTestHelper.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/RatisTestHelper.java
index 9aefe9a996..1a35c5008c 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/RatisTestHelper.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/RatisTestHelper.java
@@ -20,8 +20,9 @@ package org.apache.hadoop.ozone;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.ozone.client.protocol.ClientProtocol;
+import org.apache.hadoop.ozone.client.rpc.RpcClient;
 import org.apache.hadoop.ozone.container.ContainerTestHelper;
-import org.apache.hadoop.ozone.web.client.OzoneRestClient;
 import org.apache.hadoop.ozone.client.rest.OzoneException;
 import org.apache.ratis.rpc.RpcType;
 import org.apache.ratis.rpc.SupportedRpcType;
@@ -65,9 +66,9 @@ public interface RatisTestHelper {
       return cluster;
     }
 
-    public OzoneRestClient newOzoneRestClient()
-        throws OzoneException, URISyntaxException {
-      return RatisTestHelper.newOzoneRestClient(getDatanodeOzoneRestPort());
+    public ClientProtocol newOzoneClient()
+        throws OzoneException, URISyntaxException, IOException {
+      return new RpcClient(conf);
     }
 
     @Override
@@ -102,9 +103,4 @@ public interface RatisTestHelper {
         .setNumDatanodes(numDatanodes).build();
     return cluster;
   }
-
-  static OzoneRestClient newOzoneRestClient(int port)
-      throws OzoneException, URISyntaxException {
-    return new OzoneRestClient("http://localhost:" + port);
-  }
 }
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/TestOzoneRestWithMiniCluster.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/TestOzoneRestWithMiniCluster.java
index 5b676570c4..a9b8175f95 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/TestOzoneRestWithMiniCluster.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/TestOzoneRestWithMiniCluster.java
@@ -23,23 +23,31 @@ import static org.apache.hadoop.fs.contract.ContractTestUtils.dataset;
 import static org.apache.hadoop.ozone.OzoneConsts.CHUNK_SIZE;
 import static org.junit.Assert.*;
 
+import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang.RandomStringUtils;
-import org.apache.hadoop.ozone.OzoneConsts;
-import org.apache.hadoop.ozone.web.client.OzoneRestClient;
+import org.apache.hadoop.hdds.client.OzoneQuota;
+import org.apache.hadoop.hdds.client.ReplicationFactor;
+import org.apache.hadoop.hdds.client.ReplicationType;
+import org.apache.hadoop.ozone.client.VolumeArgs;
+import org.apache.hadoop.ozone.client.io.OzoneInputStream;
+import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
+import org.apache.hadoop.ozone.client.protocol.ClientProtocol;
+import org.apache.hadoop.ozone.client.OzoneVolume;
+import org.apache.hadoop.ozone.client.OzoneBucket;
+import org.apache.hadoop.ozone.client.rpc.RpcClient;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
 
-import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.ozone.MiniOzoneCluster;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
-import org.apache.hadoop.ozone.web.client.OzoneBucket;
-import org.apache.hadoop.ozone.web.client.OzoneVolume;
-import org.apache.hadoop.ozone.web.request.OzoneQuota;
 import org.junit.rules.Timeout;
 
+import java.io.IOException;
+import java.io.InputStream;
+
 /**
  * End-to-end testing of Ozone REST operations.
  */
@@ -52,7 +60,9 @@ public class TestOzoneRestWithMiniCluster {
 
   private static MiniOzoneCluster cluster;
   private static OzoneConfiguration conf;
-  private static OzoneRestClient ozoneClient;
+  private static ClientProtocol client;
+  private static ReplicationFactor replicationFactor = ReplicationFactor.ONE;
+  private static ReplicationType replicationType = ReplicationType.STAND_ALONE;
 
   @Rule
   public ExpectedException exception = ExpectedException.none();
@@ -62,180 +72,125 @@ public class TestOzoneRestWithMiniCluster {
     conf = new OzoneConfiguration();
     cluster = MiniOzoneCluster.newBuilder(conf).build();
     cluster.waitForClusterToBeReady();
-    int port = cluster.getHddsDatanodes().get(0)
-        .getDatanodeDetails().getOzoneRestPort();
-    ozoneClient = new OzoneRestClient(
-        String.format("http://localhost:%d", port));
-    ozoneClient.setUserAuth(OzoneConsts.OZONE_SIMPLE_HDFS_USER);
+    client = new RpcClient(conf);
   }
 
   @AfterClass
-  public static void shutdown() throws InterruptedException {
+  public static void shutdown() throws InterruptedException, IOException {
     if (cluster != null) {
       cluster.shutdown();
     }
-    IOUtils.cleanupWithLogger(null, ozoneClient);
+    client.close();
   }
 
   @Test
   public void testCreateAndGetVolume() throws Exception {
-    String volumeName = nextId("volume");
-    OzoneVolume volume = ozoneClient.createVolume(volumeName, "bilbo", "100TB");
-    assertNotNull(volume);
-    assertEquals(volumeName, volume.getVolumeName());
-    assertEquals(ozoneClient.getUserAuth(), volume.getCreatedby());
-    assertEquals("bilbo", volume.getOwnerName());
-    assertNotNull(volume.getQuota());
-    assertEquals(OzoneQuota.parseQuota("100TB").sizeInBytes(),
-        volume.getQuota().sizeInBytes());
-    volume = ozoneClient.getVolume(volumeName);
-    assertNotNull(volume);
-    assertEquals(volumeName, volume.getVolumeName());
-    assertEquals(ozoneClient.getUserAuth(), volume.getCreatedby());
-    assertEquals("bilbo", volume.getOwnerName());
-    assertNotNull(volume.getQuota());
-    assertEquals(OzoneQuota.parseQuota("100TB").sizeInBytes(),
-        volume.getQuota().sizeInBytes());
+    createAndGetVolume();
   }
 
   @Test
   public void testCreateAndGetBucket() throws Exception {
-    String volumeName = nextId("volume");
-    String bucketName = nextId("bucket");
-    OzoneVolume volume = ozoneClient.createVolume(volumeName, "bilbo", "100TB");
-    assertNotNull(volume);
-    assertEquals(volumeName, volume.getVolumeName());
-    assertEquals(ozoneClient.getUserAuth(), volume.getCreatedby());
-    assertEquals("bilbo", volume.getOwnerName());
-    assertNotNull(volume.getQuota());
-    assertEquals(OzoneQuota.parseQuota("100TB").sizeInBytes(),
-        volume.getQuota().sizeInBytes());
-    OzoneBucket bucket = volume.createBucket(bucketName);
-    assertNotNull(bucket);
-    assertEquals(bucketName, bucket.getBucketName());
-    bucket = volume.getBucket(bucketName);
-    assertNotNull(bucket);
-    assertEquals(bucketName, bucket.getBucketName());
+    OzoneVolume volume = createAndGetVolume();
+    createAndGetBucket(volume);
   }
 
   @Test
   public void testPutAndGetKey() throws Exception {
-    String volumeName = nextId("volume");
-    String bucketName = nextId("bucket");
     String keyName = nextId("key");
     String keyData = nextId("data");
-    OzoneVolume volume = ozoneClient.createVolume(volumeName, "bilbo", "100TB");
-    assertNotNull(volume);
-    assertEquals(volumeName, volume.getVolumeName());
-    assertEquals(ozoneClient.getUserAuth(), volume.getCreatedby());
-    assertEquals("bilbo", volume.getOwnerName());
-    assertNotNull(volume.getQuota());
-    assertEquals(OzoneQuota.parseQuota("100TB").sizeInBytes(),
-        volume.getQuota().sizeInBytes());
-    OzoneBucket bucket = volume.createBucket(bucketName);
-    assertNotNull(bucket);
-    assertEquals(bucketName, bucket.getBucketName());
-    bucket.putKey(keyName, keyData);
-    assertEquals(keyData, bucket.getKey(keyName));
+    OzoneVolume volume = createAndGetVolume();
+    OzoneBucket bucket = createAndGetBucket(volume);
+    putKey(bucket, keyName, keyData);
+  }
+
+  private void putKey(OzoneBucket bucket, String keyName, String keyData) throws IOException {
+    try (
+        OzoneOutputStream ozoneOutputStream = bucket
+            .createKey(keyName, 0, replicationType, replicationFactor);
+        InputStream inputStream = IOUtils.toInputStream(keyData, UTF_8)) {
+      IOUtils.copy(inputStream, ozoneOutputStream);
+    }
+    try (
+        InputStream inputStream = IOUtils.toInputStream(keyData, UTF_8);
+        OzoneInputStream ozoneInputStream = bucket.readKey(keyName)) {
+      IOUtils.contentEquals(ozoneInputStream, inputStream);
+    }
   }
 
   @Test
   public void testPutAndGetEmptyKey() throws Exception {
-    String volumeName = nextId("volume");
-    String bucketName = nextId("bucket");
     String keyName = nextId("key");
     String keyData = "";
-    OzoneVolume volume = ozoneClient.createVolume(volumeName, "bilbo", "100TB");
-    assertNotNull(volume);
-    assertEquals(volumeName, volume.getVolumeName());
-    assertEquals(ozoneClient.getUserAuth(), volume.getCreatedby());
-    assertEquals("bilbo", volume.getOwnerName());
-    assertNotNull(volume.getQuota());
-    assertEquals(OzoneQuota.parseQuota("100TB").sizeInBytes(),
-        volume.getQuota().sizeInBytes());
-    OzoneBucket bucket = volume.createBucket(bucketName);
-    assertNotNull(bucket);
-    assertEquals(bucketName, bucket.getBucketName());
-    bucket.putKey(keyName, keyData);
-    assertEquals(keyData, bucket.getKey(keyName));
+    OzoneVolume volume = createAndGetVolume();
+    OzoneBucket bucket = createAndGetBucket(volume);
+    putKey(bucket, keyName, keyData);
   }
 
   @Test
   public void testPutAndGetMultiChunkKey() throws Exception {
-    String volumeName = nextId("volume");
-    String bucketName = nextId("bucket");
     String keyName = nextId("key");
     int keyDataLen = 3 * CHUNK_SIZE;
     String keyData = buildKeyData(keyDataLen);
-    OzoneVolume volume = ozoneClient.createVolume(volumeName, "bilbo", "100TB");
-    assertNotNull(volume);
-    assertEquals(volumeName, volume.getVolumeName());
-    assertEquals(ozoneClient.getUserAuth(), volume.getCreatedby());
-    assertEquals("bilbo", volume.getOwnerName());
-    assertNotNull(volume.getQuota());
-    assertEquals(OzoneQuota.parseQuota("100TB").sizeInBytes(),
-        volume.getQuota().sizeInBytes());
-    OzoneBucket bucket = volume.createBucket(bucketName);
-    assertNotNull(bucket);
-    assertEquals(bucketName, bucket.getBucketName());
-    bucket.putKey(keyName, keyData);
-    assertEquals(keyData, bucket.getKey(keyName));
+    OzoneVolume volume = createAndGetVolume();
+    OzoneBucket bucket = createAndGetBucket(volume);
+    putKey(bucket, keyName, keyData);
   }
 
   @Test
   public void testPutAndGetMultiChunkKeyLastChunkPartial() throws Exception {
-    String volumeName = nextId("volume");
-    String bucketName = nextId("bucket");
     String keyName = nextId("key");
     int keyDataLen = (int)(2.5 * CHUNK_SIZE);
     String keyData = buildKeyData(keyDataLen);
-    OzoneVolume volume = ozoneClient.createVolume(volumeName, "bilbo", "100TB");
-    assertNotNull(volume);
-    assertEquals(volumeName, volume.getVolumeName());
-    assertEquals(ozoneClient.getUserAuth(), volume.getCreatedby());
-    assertEquals("bilbo", volume.getOwnerName());
-    assertNotNull(volume.getQuota());
-    assertEquals(OzoneQuota.parseQuota("100TB").sizeInBytes(),
-        volume.getQuota().sizeInBytes());
-    OzoneBucket bucket = volume.createBucket(bucketName);
-    assertNotNull(bucket);
-    assertEquals(bucketName, bucket.getBucketName());
-    bucket.putKey(keyName, keyData);
-    assertEquals(keyData, bucket.getKey(keyName));
+    OzoneVolume volume = createAndGetVolume();
+    OzoneBucket bucket = createAndGetBucket(volume);
+    putKey(bucket, keyName, keyData);
   }
 
   @Test
   public void testReplaceKey() throws Exception {
-    String volumeName = nextId("volume");
-    String bucketName = nextId("bucket");
     String keyName = nextId("key");
     int keyDataLen = (int)(2.5 * CHUNK_SIZE);
     String keyData = buildKeyData(keyDataLen);
-    OzoneVolume volume = ozoneClient.createVolume(volumeName, "bilbo", "100TB");
-    assertNotNull(volume);
-    assertEquals(volumeName, volume.getVolumeName());
-    assertEquals(ozoneClient.getUserAuth(), volume.getCreatedby());
-    assertEquals("bilbo", volume.getOwnerName());
-    assertNotNull(volume.getQuota());
-    assertEquals(OzoneQuota.parseQuota("100TB").sizeInBytes(),
-        volume.getQuota().sizeInBytes());
-    OzoneBucket bucket = volume.createBucket(bucketName);
-    assertNotNull(bucket);
-    assertEquals(bucketName, bucket.getBucketName());
-    bucket.putKey(keyName, keyData);
-    assertEquals(keyData, bucket.getKey(keyName));
+    OzoneVolume volume = createAndGetVolume();
+    OzoneBucket bucket = createAndGetBucket(volume);
+    putKey(bucket, keyName, keyData);
 
     // Replace key with data consisting of fewer chunks.
     keyDataLen = (int)(1.5 * CHUNK_SIZE);
     keyData = buildKeyData(keyDataLen);
-    bucket.putKey(keyName, keyData);
-    assertEquals(keyData, bucket.getKey(keyName));
+    putKey(bucket, keyName, keyData);
 
     // Replace key with data consisting of more chunks.
     keyDataLen = (int)(3.5 * CHUNK_SIZE);
     keyData = buildKeyData(keyDataLen);
-    bucket.putKey(keyName, keyData);
-    assertEquals(keyData, bucket.getKey(keyName));
+    putKey(bucket, keyName, keyData);
+  }
+
+  private OzoneVolume createAndGetVolume() throws IOException {
+    String volumeName = nextId("volume");
+    VolumeArgs volumeArgs = VolumeArgs.newBuilder()
+        .setOwner("bilbo")
+        .setQuota("100TB")
+        .setAdmin("hdfs")
+        .build();
+    client.createVolume(volumeName, volumeArgs);
+    OzoneVolume volume = client.getVolumeDetails(volumeName);
+    assertEquals(volumeName, volume.getName());
+    assertNotNull(volume);
+    assertEquals("bilbo", volume.getOwner());
+    assertNotNull(volume.getQuota());
+    assertEquals(OzoneQuota.parseQuota("100TB").sizeInBytes(),
+        volume.getQuota());
+    return volume;
+  }
+
+  private OzoneBucket createAndGetBucket(OzoneVolume vol) throws IOException {
+    String bucketName = nextId("bucket");
+    vol.createBucket(bucketName);
+    OzoneBucket bucket = vol.getBucket(bucketName);
+    assertNotNull(bucket);
+    assertEquals(bucketName, bucket.getName());
+    return bucket;
   }
 
   /**
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestBuckets.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestBuckets.java
index 64e5f7103d..684f4d35a3 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestBuckets.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestBuckets.java
@@ -17,12 +17,17 @@
  */
 package org.apache.hadoop.ozone.web.client;
 
-import org.apache.hadoop.fs.StorageType;
 import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.OzoneAcl;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
-import org.apache.hadoop.ozone.client.OzoneClientException;
+import org.apache.hadoop.ozone.client.BucketArgs;
+import org.apache.hadoop.ozone.client.VolumeArgs;
+import org.apache.hadoop.ozone.client.protocol.ClientProtocol;
+import org.apache.hadoop.ozone.client.OzoneVolume;
+import org.apache.hadoop.ozone.client.OzoneBucket;
 import org.apache.hadoop.ozone.client.rest.OzoneException;
+import org.apache.hadoop.ozone.client.rpc.RpcClient;
 import org.apache.hadoop.ozone.web.request.OzoneQuota;
 import org.apache.hadoop.ozone.web.utils.OzoneUtils;
 import org.apache.hadoop.test.GenericTestUtils;
@@ -36,7 +41,10 @@ import org.junit.rules.Timeout;
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.text.ParseException;
+import java.util.Arrays;
+import java.util.Iterator;
 import java.util.List;
+import java.util.stream.Collectors;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
@@ -54,7 +62,7 @@ public class TestBuckets {
   public Timeout testTimeout = new Timeout(300000);
 
   private static MiniOzoneCluster cluster = null;
-  private static OzoneRestClient ozoneRestClient = null;
+  private static ClientProtocol client = null;
 
   /**
    * Create a MiniDFSCluster for testing.
@@ -76,11 +84,10 @@ public class TestBuckets {
         OzoneConfigKeys.OZONE_LOCALSTORAGE_ROOT_DEFAULT);
 
     conf.set(OzoneConfigKeys.OZONE_LOCALSTORAGE_ROOT, path);
-    cluster = MiniOzoneCluster.newBuilder(conf).build();
-    final int port = cluster.getHddsDatanodes().get(0).getDatanodeDetails()
-        .getOzoneRestPort();
-    ozoneRestClient = new OzoneRestClient(
-        String.format("http://localhost:%d", port));
+    cluster = MiniOzoneCluster.newBuilder(conf)
+        .setNumDatanodes(3)
+        .build();
+    client = new RpcClient(conf);
   }
 
   /**
@@ -95,110 +102,151 @@ public class TestBuckets {
 
   @Test
   public void testCreateBucket() throws Exception {
-    runTestCreateBucket(ozoneRestClient);
+    runTestCreateBucket(client);
   }
 
-  static void runTestCreateBucket(OzoneRestClient client)
+  static void runTestCreateBucket(ClientProtocol client)
       throws OzoneException, IOException, ParseException {
     String volumeName = OzoneUtils.getRequestID().toLowerCase();
-    client.setUserAuth("hdfs");
-    OzoneVolume vol = client.createVolume(volumeName, "bilbo", "100TB");
+    VolumeArgs volumeArgs = VolumeArgs.newBuilder()
+        .setOwner("bilbo")
+        .setQuota("100TB")
+        .setAdmin("hdfs")
+        .build();
+    client.createVolume(volumeName, volumeArgs);
+    OzoneVolume vol = client.getVolumeDetails(volumeName);
     String[] acls = {"user:frodo:rw", "user:samwise:rw"};
 
     // create 10 buckets under same volume
     for (int x = 0; x < 10; x++) {
       long currentTime = Time.now();
       String bucketName = OzoneUtils.getRequestID().toLowerCase();
-      OzoneBucket bucket =
-          vol.createBucket(bucketName, acls, StorageType.DEFAULT);
-      assertEquals(bucket.getBucketName(), bucketName);
+
+      List<OzoneAcl> aclList =
+          Arrays.stream(acls).map(acl -> OzoneAcl.parseAcl(acl))
+              .collect(Collectors.toList());
+      BucketArgs bucketArgs = BucketArgs.newBuilder()
+          .setAcls(aclList)
+          .build();
+      vol.createBucket(bucketName, bucketArgs);
+      OzoneBucket bucket = vol.getBucket(bucketName);
+      assertEquals(bucket.getName(), bucketName);
 
       // verify the bucket creation time
-      assertTrue((OzoneUtils.formatDate(bucket.getCreatedOn())
-          / 1000) >= (currentTime / 1000));
+      assertTrue((bucket.getCreationTime() / 1000) >= (currentTime / 1000));
     }
     client.close();
 
-    assertEquals(vol.getVolumeName(), volumeName);
-    assertEquals(vol.getCreatedby(), "hdfs");
-    assertEquals(vol.getOwnerName(), "bilbo");
-    assertEquals(vol.getQuota().getUnit(), OzoneQuota.Units.TB);
-    assertEquals(vol.getQuota().getSize(), 100);
+    assertEquals(vol.getName(), volumeName);
+    assertEquals(vol.getAdmin(), "hdfs");
+    assertEquals(vol.getOwner(), "bilbo");
+    assertEquals(vol.getQuota(), OzoneQuota.parseQuota("100TB").sizeInBytes());
 
     // Test create a bucket with invalid bucket name,
     // not use Rule here because the test method is static.
     try {
       String invalidBucketName = "#" + OzoneUtils.getRequestID().toLowerCase();
-      vol.createBucket(invalidBucketName, acls, StorageType.DEFAULT);
+      vol.createBucket(invalidBucketName);
       fail("Except the bucket creation to be failed because the"
           + " bucket name starts with an invalid char #");
     } catch (Exception e) {
-      assertTrue(e instanceof OzoneClientException);
-      assertTrue(e.getMessage().contains("Bucket or Volume name"
-          + " has an unsupported character : #"));
+      assertTrue(e.getMessage()
+          .contains("Bucket or Volume name has an unsupported character : #"));
     }
   }
 
   @Test
   public void testAddBucketAcls() throws Exception {
-    runTestAddBucketAcls(ozoneRestClient);
+    runTestAddBucketAcls(client);
   }
 
-  static void runTestAddBucketAcls(OzoneRestClient client)
+  static void runTestAddBucketAcls(ClientProtocol client)
       throws OzoneException, IOException, ParseException {
     String volumeName = OzoneUtils.getRequestID().toLowerCase();
-    client.setUserAuth("hdfs");
-    OzoneVolume vol = client.createVolume(volumeName, "bilbo", "100TB");
+    VolumeArgs volumeArgs = VolumeArgs.newBuilder()
+        .setOwner("bilbo")
+        .setQuota("100TB")
+        .setAdmin("hdfs")
+        .build();
+    client.createVolume(volumeName, volumeArgs);
+    OzoneVolume vol = client.getVolumeDetails(volumeName);
     String[] acls = {"user:frodo:rw", "user:samwise:rw"};
     String bucketName = OzoneUtils.getRequestID().toLowerCase();
     vol.createBucket(bucketName);
-    vol.addAcls(bucketName, acls);
+    OzoneBucket bucket = vol.getBucket(bucketName);
+    List<OzoneAcl> aclList =
+        Arrays.stream(acls).map(acl -> OzoneAcl.parseAcl(acl))
+            .collect(Collectors.toList());
+    int numAcls = bucket.getAcls().size();
+    bucket.addAcls(aclList);
     OzoneBucket updatedBucket = vol.getBucket(bucketName);
-    assertEquals(updatedBucket.getAcls().size(), 2);
+    assertEquals(updatedBucket.getAcls().size(), 2 + numAcls);
     // verify if the creation time is missing after update operation
     assertTrue(
-        (OzoneUtils.formatDate(updatedBucket.getCreatedOn()) / 1000) >= 0);
+        (updatedBucket.getCreationTime()) / 1000 >= 0);
     client.close();
   }
 
   @Test
   public void testRemoveBucketAcls() throws Exception {
-    runTestRemoveBucketAcls(ozoneRestClient);
+    runTestRemoveBucketAcls(client);
   }
 
-  static void runTestRemoveBucketAcls(OzoneRestClient client)
+  static void runTestRemoveBucketAcls(ClientProtocol client)
       throws OzoneException, IOException, ParseException {
     String volumeName = OzoneUtils.getRequestID().toLowerCase();
-    client.setUserAuth("hdfs");
-    OzoneVolume vol = client.createVolume(volumeName, "bilbo", "100TB");
+    VolumeArgs volumeArgs = VolumeArgs.newBuilder()
+        .setOwner("bilbo")
+        .setQuota("100TB")
+        .setAdmin("hdfs")
+        .build();
+    client.createVolume(volumeName, volumeArgs);
+    OzoneVolume vol = client.getVolumeDetails(volumeName);
     String[] acls = {"user:frodo:rw", "user:samwise:rw"};
     String bucketName = OzoneUtils.getRequestID().toLowerCase();
-    OzoneBucket bucket = vol.createBucket(bucketName, acls);
-    assertEquals(bucket.getAcls().size(), 2);
-    vol.removeAcls(bucketName, acls);
+    List<OzoneAcl> aclList =
+        Arrays.stream(acls).map(acl -> OzoneAcl.parseAcl(acl))
+            .collect(Collectors.toList());
+    vol.createBucket(bucketName);
+    OzoneBucket bucket = vol.getBucket(bucketName);
+    int numAcls = bucket.getAcls().size();
+    bucket.addAcls(aclList);
+    assertEquals(bucket.getAcls().size(), 2 + numAcls);
+    bucket.removeAcls(aclList);
     OzoneBucket updatedBucket = vol.getBucket(bucketName);
 
     // We removed all acls
-    assertEquals(updatedBucket.getAcls().size(), 0);
+    assertEquals(updatedBucket.getAcls().size(), numAcls);
     // verify if the creation time is missing after update operation
     assertTrue(
-        (OzoneUtils.formatDate(updatedBucket.getCreatedOn()) / 1000) >= 0);
+        (updatedBucket.getCreationTime() / 1000) >= 0);
     client.close();
   }
 
   @Test
   public void testDeleteBucket() throws OzoneException, IOException {
-    runTestDeleteBucket(ozoneRestClient);
+    runTestDeleteBucket(client);
   }
 
-  static void runTestDeleteBucket(OzoneRestClient client)
+  static void runTestDeleteBucket(ClientProtocol client)
       throws OzoneException, IOException {
     String volumeName = OzoneUtils.getRequestID().toLowerCase();
-    client.setUserAuth("hdfs");
-    OzoneVolume vol = client.createVolume(volumeName, "bilbo", "100TB");
+    VolumeArgs volumeArgs = VolumeArgs.newBuilder()
+        .setOwner("bilbo")
+        .setQuota("100TB")
+        .setAdmin("hdfs")
+        .build();
+    client.createVolume(volumeName, volumeArgs);
+    OzoneVolume vol = client.getVolumeDetails(volumeName);
     String[] acls = {"user:frodo:rw", "user:samwise:rw"};
     String bucketName = OzoneUtils.getRequestID().toLowerCase();
-    vol.createBucket(bucketName, acls);
+    List<OzoneAcl> aclList =
+        Arrays.stream(acls).map(acl -> OzoneAcl.parseAcl(acl))
+            .collect(Collectors.toList());
+    BucketArgs bucketArgs = BucketArgs.newBuilder()
+        .setAcls(aclList)
+        .build();
+    vol.createBucket(bucketName, bucketArgs);
     vol.deleteBucket(bucketName);
     try {
       OzoneBucket updatedBucket = vol.getBucket(bucketName);
@@ -212,38 +260,57 @@ public class TestBuckets {
 
   @Test
   public void testListBucket() throws Exception {
-    runTestListBucket(ozoneRestClient);
+    runTestListBucket(client);
   }
 
-  static void runTestListBucket(OzoneRestClient client)
+  static void runTestListBucket(ClientProtocol client)
       throws OzoneException, IOException, ParseException {
     String volumeName = OzoneUtils.getRequestID().toLowerCase();
-    client.setUserAuth("hdfs");
-    OzoneVolume vol = client.createVolume(volumeName, "bilbo", "100TB");
+    VolumeArgs volumeArgs = VolumeArgs.newBuilder()
+        .setOwner("bilbo")
+        .setQuota("100TB")
+        .setAdmin("hdfs")
+        .build();
+    client.createVolume(volumeName, volumeArgs);
+    OzoneVolume vol = client.getVolumeDetails(volumeName);
     String[] acls = {"user:frodo:rw", "user:samwise:rw"};
+    List<OzoneAcl> aclList =
+        Arrays.stream(acls).map(acl -> OzoneAcl.parseAcl(acl))
+            .collect(Collectors.toList());
 
     long currentTime = Time.now();
     for (int x = 0; x < 10; x++) {
       String bucketName = "listbucket-test-" + x;
-      vol.createBucket(bucketName, acls);
+      BucketArgs bucketArgs = BucketArgs.newBuilder()
+          .setAcls(aclList)
+          .build();
+      vol.createBucket(bucketName, bucketArgs);
     }
-    List<OzoneBucket> bucketList = vol.listBuckets("100", null, null);
-    assertEquals(bucketList.size(), 10);
+    Iterator<OzoneBucket> bucketIterator = vol.listBuckets(null);
+    int count = 0;
 
-    for (OzoneBucket bucket : bucketList) {
-      assertTrue((OzoneUtils.formatDate(bucket.getCreatedOn())
+    while (bucketIterator.hasNext()) {
+      assertTrue((bucketIterator.next().getCreationTime()
           / 1000) >= (currentTime / 1000));
+      count++;
     }
+    assertEquals(count, 10);
 
-    bucketList = vol.listBuckets("3", null, null);
-    assertEquals(bucketList.size(), 3);
-
-    bucketList = vol.listBuckets("100", "listbucket-test-4", null);
-    assertEquals(bucketList.size(), 5);
+    bucketIterator = vol.listBuckets(null, "listbucket-test-4");
+    assertEquals(getSize(bucketIterator), 5);
 
-    bucketList = vol.listBuckets("100", null, "listbucket-test-3");
-    assertEquals(bucketList.size(), 1);
+    bucketIterator = vol.listBuckets(null, "listbucket-test-3");
+    assertEquals(getSize(bucketIterator), 6);
 
     client.close();
   }
+
+  private static int getSize(Iterator<OzoneBucket> bucketIterator) {
+    int count = 0;
+    while (bucketIterator.hasNext()) {
+      count++;
+      bucketIterator.next();
+    }
+    return count;
+  }
 }
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestBucketsRatis.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestBucketsRatis.java
index b913a86fbc..9f80184226 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestBucketsRatis.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestBucketsRatis.java
@@ -18,6 +18,7 @@
 package org.apache.hadoop.ozone.web.client;
 
 import org.apache.hadoop.ozone.RatisTestHelper;
+import org.apache.hadoop.ozone.client.protocol.ClientProtocol;
 import org.apache.hadoop.ozone.client.rest.OzoneException;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -35,12 +36,12 @@ public class TestBucketsRatis {
   public Timeout testTimeout = new Timeout(300000);
 
   private static RatisTestHelper.RatisTestSuite suite;
-  private static OzoneRestClient ozoneRestClient;
+  private static ClientProtocol client;
 
   @BeforeClass
   public static void init() throws Exception {
     suite = new RatisTestHelper.RatisTestSuite(TestBucketsRatis.class);
-    ozoneRestClient = suite.newOzoneRestClient();
+    client = suite.newOzoneClient();
   }
 
   @AfterClass
@@ -52,25 +53,25 @@ public class TestBucketsRatis {
 
   @Test
   public void testCreateBucket() throws Exception {
-    TestBuckets.runTestCreateBucket(ozoneRestClient);
+    TestBuckets.runTestCreateBucket(client);
   }
 
   @Test
   public void testAddBucketAcls() throws Exception {
-    TestBuckets.runTestAddBucketAcls(ozoneRestClient);
+    TestBuckets.runTestAddBucketAcls(client);
   }
 
   @Test
   public void testRemoveBucketAcls() throws Exception {
-    TestBuckets.runTestRemoveBucketAcls(ozoneRestClient);
+    TestBuckets.runTestRemoveBucketAcls(client);
   }
 
   @Test
   public void testDeleteBucket() throws OzoneException, IOException {
-    TestBuckets.runTestDeleteBucket(ozoneRestClient);
+    TestBuckets.runTestDeleteBucket(client);
   }
   @Test
   public void testListBucket() throws Exception {
-    TestBuckets.runTestListBucket(ozoneRestClient);
+    TestBuckets.runTestListBucket(client);
   }
 }
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestKeys.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestKeys.java
index 57d4287b63..f8df7dceb3 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestKeys.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestKeys.java
@@ -20,16 +20,29 @@ package org.apache.hadoop.ozone.web.client;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import org.apache.commons.codec.digest.DigestUtils;
+import org.apache.commons.collections.IteratorUtils;
 import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang.RandomStringUtils;
 import org.apache.commons.lang.math.RandomUtils;
 import org.apache.commons.lang3.tuple.ImmutablePair;
 import org.apache.commons.lang3.tuple.Pair;
-import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdds.client.ReplicationFactor;
+import org.apache.hadoop.hdds.client.ReplicationType;
 import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.OzoneAcl;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.ozone.client.BucketArgs;
+import org.apache.hadoop.ozone.client.VolumeArgs;
+import org.apache.hadoop.ozone.client.OzoneBucket;
+import org.apache.hadoop.ozone.client.OzoneKey;
+import org.apache.hadoop.ozone.client.OzoneVolume;
+import org.apache.hadoop.ozone.client.io.OzoneInputStream;
+import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
+import org.apache.hadoop.ozone.client.protocol.ClientProtocol;
+import org.apache.hadoop.ozone.client.rpc.RpcClient;
 import org.apache.hadoop.ozone.container.common.helpers.ContainerData;
 import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils;
 import org.apache.hadoop.ozone.container.common.helpers.KeyData;
@@ -59,15 +72,18 @@ import org.junit.rules.Timeout;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileOutputStream;
+import java.io.InputStream;
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.Random;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
@@ -84,10 +100,13 @@ public class TestKeys {
   @Rule
   public Timeout testTimeout = new Timeout(300000);
 
+  private static OzoneConfiguration conf;
   private static MiniOzoneCluster ozoneCluster = null;
   private static String path;
-  private static OzoneRestClient ozoneRestClient = null;
+  private static ClientProtocol client = null;
   private static long currentTime;
+  private static ReplicationFactor replicationFactor = ReplicationFactor.ONE;
+  private static ReplicationType replicationType = ReplicationType.STAND_ALONE;
 
   /**
    * Create a MiniDFSCluster for testing.
@@ -96,7 +115,7 @@ public class TestKeys {
    */
   @BeforeClass
   public static void init() throws Exception {
-    OzoneConfiguration conf = new OzoneConfiguration();
+    conf = new OzoneConfiguration();
 
     // Set short block deleting service interval to speed up deletions.
     conf.setTimeDuration(OzoneConfigKeys.OZONE_BLOCK_DELETING_SERVICE_INTERVAL,
@@ -105,12 +124,9 @@ public class TestKeys {
     path = GenericTestUtils.getTempPath(TestKeys.class.getSimpleName());
     Logger.getLogger("log4j.logger.org.apache.http").setLevel(Level.DEBUG);
 
-    ozoneCluster = MiniOzoneCluster.newBuilder(conf).build();
+    ozoneCluster = MiniOzoneCluster.newBuilder(conf).setNumDatanodes(1).build();
     ozoneCluster.waitForClusterToBeReady();
-    final int port = ozoneCluster.getHddsDatanodes().get(0)
-        .getDatanodeDetails().getOzoneRestPort();
-    ozoneRestClient = new OzoneRestClient(
-        String.format("http://localhost:%d", port));
+    client = new RpcClient(conf);
     currentTime = Time.now();
   }
 
@@ -166,7 +182,7 @@ public class TestKeys {
   }
 
   static class PutHelper {
-    private final OzoneRestClient client;
+    private final ClientProtocol client;
     private final String dir;
     private final String keyName;
 
@@ -174,11 +190,11 @@ public class TestKeys {
     private OzoneBucket bucket;
     private File file;
 
-    PutHelper(OzoneRestClient client, String dir) {
+    PutHelper(ClientProtocol client, String dir) {
       this(client, dir, OzoneUtils.getRequestID().toLowerCase());
     }
 
-    PutHelper(OzoneRestClient client, String dir, String key) {
+    PutHelper(ClientProtocol client, String dir, String key) {
       this.client = client;
       this.dir = dir;
       this.keyName = key;
@@ -202,111 +218,139 @@ public class TestKeys {
      * @return Returns the name of the new key that was created.
      * @throws OzoneException
      */
-    private KsmKeyArgs putKey() throws Exception {
+    private String putKey() throws Exception {
       String volumeName = OzoneUtils.getRequestID().toLowerCase();
-      client.setUserAuth("hdfs");
 
-      vol = client.createVolume(volumeName, "bilbo", "100TB");
+      VolumeArgs volumeArgs = VolumeArgs.newBuilder()
+          .setOwner("bilbo")
+          .setQuota("100TB")
+          .setAdmin("hdfs")
+          .build();
+      client.createVolume(volumeName, volumeArgs);
+      vol = client.getVolumeDetails(volumeName);
       String[] acls = {"user:frodo:rw", "user:samwise:rw"};
 
       String bucketName = OzoneUtils.getRequestID().toLowerCase();
-      bucket = vol.createBucket(bucketName, acls, StorageType.DEFAULT);
+      List<OzoneAcl> aclList =
+          Arrays.stream(acls).map(acl -> OzoneAcl.parseAcl(acl))
+              .collect(Collectors.toList());
+      BucketArgs bucketArgs = BucketArgs.newBuilder()
+          .setAcls(aclList)
+          .build();
+      vol.createBucket(bucketName, bucketArgs);
+      bucket = vol.getBucket(bucketName);
 
       String fileName = OzoneUtils.getRequestID().toLowerCase();
 
       file = createRandomDataFile(dir, fileName, 1024);
 
-      bucket.putKey(keyName, file);
-      return new KsmKeyArgs.Builder()
-          .setKeyName(keyName)
-          .setVolumeName(volumeName)
-          .setBucketName(bucketName)
-          .setDataSize(1024)
-          .build();
+      try (
+          OzoneOutputStream ozoneOutputStream = bucket
+              .createKey(keyName, 0, replicationType, replicationFactor);
+          InputStream fileInputStream = new FileInputStream(file)) {
+        IOUtils.copy(fileInputStream, ozoneOutputStream);
+      }
+      return keyName;
     }
   }
 
   @Test
   public void testPutKey() throws Exception {
     // Test non-delimited keys
-    runTestPutKey(new PutHelper(ozoneRestClient, path));
+    runTestPutKey(new PutHelper(client, path));
     // Test key delimited by a random delimiter
     String delimiter = RandomStringUtils.randomAscii(1);
-    runTestPutKey(new PutHelper(ozoneRestClient, path,
+    runTestPutKey(new PutHelper(client, path,
         getMultiPartKey(delimiter)));
   }
 
   static void runTestPutKey(PutHelper helper) throws Exception {
-    final OzoneRestClient client = helper.client;
+    final ClientProtocol client = helper.client;
     helper.putKey();
     assertNotNull(helper.getBucket());
     assertNotNull(helper.getFile());
-    List<OzoneKey> keyList = helper.getBucket().listKeys("100", null, null);
+    List<OzoneKey> keyList = client
+        .listKeys(helper.getVol().getName(), helper.getBucket().getName(), null,
+            null, 10);
     Assert.assertEquals(1, keyList.size());
 
     // test list key using a more efficient call
     String newkeyName = OzoneUtils.getRequestID().toLowerCase();
-    client.putKey(helper.getVol().getVolumeName(),
-        helper.getBucket().getBucketName(), newkeyName, helper.getFile());
-    keyList = helper.getBucket().listKeys("100", null, null);
+    OzoneOutputStream ozoneOutputStream = client
+        .createKey(helper.getVol().getName(), helper.getBucket().getName(),
+            newkeyName, 0, replicationType, replicationFactor);
+    ozoneOutputStream.close();
+    keyList = client
+        .listKeys(helper.getVol().getName(), helper.getBucket().getName(), null,
+            null, 10);
     Assert.assertEquals(2, keyList.size());
 
     // test new put key with invalid volume/bucket name
     try {
-      client.putKey("invalid-volume",
-          helper.getBucket().getBucketName(), newkeyName, helper.getFile());
+      ozoneOutputStream = client
+          .createKey("invalid-volume", helper.getBucket().getName(), newkeyName,
+              0, replicationType, replicationFactor);
+      ozoneOutputStream.close();
       fail("Put key should have thrown"
           + " when using invalid volume name.");
-    } catch (OzoneException e) {
+    } catch (IOException e) {
       GenericTestUtils.assertExceptionContains(
           Status.VOLUME_NOT_FOUND.toString(), e);
     }
 
     try {
-      client.putKey(helper.getVol().getVolumeName(), "invalid-bucket",
-          newkeyName, helper.getFile());
+      ozoneOutputStream = client
+          .createKey(helper.getVol().getName(), "invalid-bucket", newkeyName, 0,
+              replicationType, replicationFactor);
+      ozoneOutputStream.close();
       fail("Put key should have thrown "
           + "when using invalid bucket name.");
-    } catch (OzoneException e) {
+    } catch (IOException e) {
       GenericTestUtils.assertExceptionContains(
           Status.BUCKET_NOT_FOUND.toString(), e);
     }
   }
 
-  private static void restartDatanode(
-      MiniOzoneCluster cluster, int datanodeIdx, OzoneRestClient client)
+  private static void restartDatanode(MiniOzoneCluster cluster, int datanodeIdx)
       throws OzoneException, URISyntaxException {
     cluster.restartHddsDatanode(datanodeIdx);
-    // refresh the datanode endpoint uri after datanode restart
-    final int port = ozoneCluster.getHddsDatanodes().get(0)
-        .getDatanodeDetails().getOzoneRestPort();
-    client.setEndPoint(String.format("http://localhost:%d", port));
   }
 
   @Test
   public void testPutAndGetKeyWithDnRestart() throws Exception {
     runTestPutAndGetKeyWithDnRestart(
-        new PutHelper(ozoneRestClient, path), ozoneCluster);
+        new PutHelper(client, path), ozoneCluster);
     String delimiter = RandomStringUtils.randomAscii(1);
     runTestPutAndGetKeyWithDnRestart(
-        new PutHelper(ozoneRestClient, path,
+        new PutHelper(client, path,
             getMultiPartKey(delimiter)), ozoneCluster);
   }
 
   static void runTestPutAndGetKeyWithDnRestart(
       PutHelper helper, MiniOzoneCluster cluster) throws Exception {
-    String keyName = helper.putKey().getKeyName();
+    String keyName = helper.putKey();
     assertNotNull(helper.getBucket());
     assertNotNull(helper.getFile());
 
     // restart the datanode
-    restartDatanode(cluster, 0, helper.client);
+    restartDatanode(cluster, 0);
+    // TODO: Try removing sleep and adding a join for the MiniOzoneCluster start
+    // The ozoneContainer is not started and its metrics are not initialized
+    // which leads to NullPointerException in Dispatcher.
+    Thread.sleep(1000);
+    ozoneCluster.waitForClusterToBeReady();
     // verify getKey after the datanode restart
     String newFileName = helper.dir + "/"
         + OzoneUtils.getRequestID().toLowerCase();
     Path newPath = Paths.get(newFileName);
-
-    helper.getBucket().getKey(keyName, newPath);
+    try (
+        FileOutputStream newOutputStream = new FileOutputStream(
+            newPath.toString());
+        OzoneInputStream ozoneInputStream = helper.client
+            .getKey(helper.getVol().getName(), helper.getBucket().getName(),
+                keyName)) {
+      IOUtils.copy(ozoneInputStream, newOutputStream);
+    }
 
     try (
         FileInputStream original = new FileInputStream(helper.getFile());
@@ -321,16 +365,16 @@ public class TestKeys {
 
   @Test
   public void testPutAndGetKey() throws Exception {
-    runTestPutAndGetKey(new PutHelper(ozoneRestClient, path));
+    runTestPutAndGetKey(new PutHelper(client, path));
     String delimiter = RandomStringUtils.randomAscii(1);
-    runTestPutAndGetKey(new PutHelper(ozoneRestClient, path,
+    runTestPutAndGetKey(new PutHelper(client, path,
         getMultiPartKey(delimiter)));
   }
 
   static void runTestPutAndGetKey(PutHelper helper) throws Exception {
-    final OzoneRestClient client = helper.client;
+    final ClientProtocol client = helper.client;
 
-    String keyName = helper.putKey().getKeyName();
+    String keyName = helper.putKey();
     assertNotNull(helper.getBucket());
     assertNotNull(helper.getFile());
 
@@ -342,10 +386,22 @@ public class TestKeys {
     Path newPath1 = Paths.get(newFileName1);
     Path newPath2 = Paths.get(newFileName2);
 
-    helper.getBucket().getKey(keyName, newPath1);
+    try (
+        FileOutputStream newOutputStream = new FileOutputStream(
+            newPath1.toString());
+        OzoneInputStream ozoneInputStream = helper.getBucket()
+            .readKey(keyName)) {
+      IOUtils.copy(ozoneInputStream, newOutputStream);
+    }
+
     // test get key using a more efficient call
-    client.getKey(helper.getVol().getVolumeName(),
-        helper.getBucket().getBucketName(), keyName, newPath2);
+    try (
+        FileOutputStream newOutputStream = new FileOutputStream(
+            newPath2.toString());
+        OzoneInputStream ozoneInputStream = helper.getBucket()
+            .readKey(keyName)) {
+      IOUtils.copy(ozoneInputStream, newOutputStream);
+    }
 
     try (FileInputStream original = new FileInputStream(helper.getFile());
         FileInputStream downloaded1 = new FileInputStream(newPath1.toFile());
@@ -363,19 +419,17 @@ public class TestKeys {
 
       // test new get key with invalid volume/bucket name
       try {
-        client.getKey("invalid-volume", helper.getBucket().getBucketName(),
-            keyName, newPath1);
+        client.getKey("invalid-volume", helper.getBucket().getName(), keyName);
         fail("Get key should have thrown " + "when using invalid volume name.");
-      } catch (OzoneException e) {
+      } catch (IOException e) {
         GenericTestUtils
             .assertExceptionContains(Status.KEY_NOT_FOUND.toString(), e);
       }
 
       try {
-        client.getKey(helper.getVol().getVolumeName(), "invalid-bucket",
-            keyName, newPath1);
+        client.getKey(helper.getVol().getName(), "invalid-bucket", keyName);
         fail("Get key should have thrown " + "when using invalid bucket name.");
-      } catch (OzoneException e) {
+      } catch (IOException e) {
         GenericTestUtils.assertExceptionContains(
             Status.KEY_NOT_FOUND.toString(), e);
       }
@@ -384,14 +438,14 @@ public class TestKeys {
 
   @Test
   public void testPutAndDeleteKey() throws Exception {
-    runTestPutAndDeleteKey(new PutHelper(ozoneRestClient, path));
+    runTestPutAndDeleteKey(new PutHelper(client, path));
     String delimiter = RandomStringUtils.randomAscii(1);
-    runTestPutAndDeleteKey(new PutHelper(ozoneRestClient, path,
+    runTestPutAndDeleteKey(new PutHelper(client, path,
         getMultiPartKey(delimiter)));
   }
 
   static void runTestPutAndDeleteKey(PutHelper helper) throws Exception {
-    String keyName = helper.putKey().getKeyName();
+    String keyName = helper.putKey();
     assertNotNull(helper.getBucket());
     assertNotNull(helper.getFile());
     helper.getBucket().deleteKey(keyName);
@@ -399,7 +453,7 @@ public class TestKeys {
     try {
       helper.getBucket().getKey(keyName);
       fail("Get Key on a deleted key should have thrown");
-    } catch (OzoneException ex) {
+    } catch (IOException ex) {
       GenericTestUtils.assertExceptionContains(
           Status.KEY_NOT_FOUND.toString(), ex);
     }
@@ -407,14 +461,14 @@ public class TestKeys {
 
   @Test
   public void testPutAndListKey() throws Exception {
-    runTestPutAndListKey(new PutHelper(ozoneRestClient, path));
+    runTestPutAndListKey(new PutHelper(client, path));
     String delimiter = RandomStringUtils.randomAscii(1);
-    runTestPutAndListKey(new PutHelper(ozoneRestClient, path,
+    runTestPutAndListKey(new PutHelper(client, path,
         getMultiPartKey(delimiter)));
   }
 
   static void runTestPutAndListKey(PutHelper helper) throws Exception {
-    final OzoneRestClient client = helper.client;
+    ClientProtocol client = helper.client;
     helper.putKey();
     assertNotNull(helper.getBucket());
     assertNotNull(helper.getFile());
@@ -422,13 +476,20 @@ public class TestKeys {
     // add keys [list-key0, list-key1, ..., list-key9]
     for (int x = 0; x < 10; x++) {
       String newkeyName = "list-key" + x;
-      helper.getBucket().putKey(newkeyName, helper.getFile());
+      try (
+          OzoneOutputStream ozoneOutputStream = helper.getBucket()
+              .createKey(newkeyName, 0, replicationType, replicationFactor);
+          InputStream fileInputStream = new FileInputStream(helper.getFile())) {
+        IOUtils.copy(fileInputStream, ozoneOutputStream);
+      }
     }
 
-    List<OzoneKey> keyList1 = helper.getBucket().listKeys("100", null, null);
+    List<OzoneKey> keyList1 =
+        IteratorUtils.toList(helper.getBucket().listKeys(null, null));
     // test list key using a more efficient call
-    List<OzoneKey> keyList2 = client.listKeys(helper.getVol().getVolumeName(),
-        helper.getBucket().getBucketName(), "100", null, null);
+    List<OzoneKey> keyList2 = client
+        .listKeys(helper.getVol().getName(), helper.getBucket().getName(), null,
+            null, 100);
 
     Assert.assertEquals(11, keyList1.size());
     Assert.assertEquals(11, keyList2.size());
@@ -436,57 +497,56 @@ public class TestKeys {
     // second unit since the date string reparsed to millisecond will
     // lose precision.
     for (OzoneKey key : keyList1) {
-      assertTrue((OzoneUtils.formatDate(key.getObjectInfo().getCreatedOn())
-          / 1000) >= (currentTime / 1000));
-      assertTrue((OzoneUtils.formatDate(key.getObjectInfo().getModifiedOn())
-          / 1000) >= (currentTime / 1000));
+      assertTrue((key.getCreationTime() / 1000) >= (currentTime / 1000));
+      assertTrue((key.getModificationTime() / 1000) >= (currentTime / 1000));
     }
 
     for (OzoneKey key : keyList2) {
-      assertTrue((OzoneUtils.formatDate(key.getObjectInfo().getCreatedOn())
-          / 1000) >= (currentTime / 1000));
-      assertTrue((OzoneUtils.formatDate(key.getObjectInfo().getModifiedOn())
-          / 1000) >= (currentTime / 1000));
+      assertTrue((key.getCreationTime() / 1000) >= (currentTime / 1000));
+      assertTrue((key.getModificationTime() / 1000) >= (currentTime / 1000));
     }
 
     // test maxLength parameter of list keys
-    keyList1 = helper.getBucket().listKeys("1", null, null);
-    keyList2 = client.listKeys(helper.getVol().getVolumeName(),
-        helper.getBucket().getBucketName(), "1", null, null);
-    Assert.assertEquals(1, keyList1.size());
+    keyList2 = client
+        .listKeys(helper.getVol().getName(), helper.getBucket().getName(), null,
+            null, 1);
     Assert.assertEquals(1, keyList2.size());
 
     // test startKey parameter of list keys
-    keyList1 = helper.getBucket().listKeys("100", "list-key4", "list-key");
-    keyList2 = client.listKeys(helper.getVol().getVolumeName(),
-        helper.getBucket().getBucketName(), "100", "list-key4", "list-key");
+    keyList1 = IteratorUtils
+        .toList(helper.getBucket().listKeys("list-key", "list-key4"));
+    keyList2 = client
+        .listKeys(helper.getVol().getName(), helper.getBucket().getName(),
+            "list-key", "list-key4", 100);
     Assert.assertEquals(5, keyList1.size());
     Assert.assertEquals(5, keyList2.size());
 
     // test prefix parameter of list keys
-    keyList1 = helper.getBucket().listKeys("100", null, "list-key2");
-    keyList2 = client.listKeys(helper.getVol().getVolumeName(),
-        helper.getBucket().getBucketName(), "100", null, "list-key2");
-    Assert.assertTrue(keyList1.size() == 1
-        && keyList1.get(0).getObjectInfo().getKeyName().equals("list-key2"));
-    Assert.assertTrue(keyList2.size() == 1
-        && keyList2.get(0).getObjectInfo().getKeyName().equals("list-key2"));
+    keyList1 =
+        IteratorUtils.toList(helper.getBucket().listKeys("list-key2", null));
+    keyList2 = client
+        .listKeys(helper.getVol().getName(), helper.getBucket().getName(),
+            "list-key2", null, 100);
+    Assert.assertTrue(
+        keyList1.size() == 1 && keyList1.get(0).getName().equals("list-key2"));
+    Assert.assertTrue(
+        keyList2.size() == 1 && keyList2.get(0).getName().equals("list-key2"));
 
     // test new list keys with invalid volume/bucket name
     try {
-      client.listKeys("invalid-volume", helper.getBucket().getBucketName(),
-          "100", null, null);
+      client.listKeys("invalid-volume", helper.getBucket().getName(),
+          null, null, 100);
       fail("List keys should have thrown when using invalid volume name.");
-    } catch (OzoneException e) {
+    } catch (IOException e) {
       GenericTestUtils.assertExceptionContains(
           Status.BUCKET_NOT_FOUND.toString(), e);
     }
 
     try {
-      client.listKeys(helper.getVol().getVolumeName(), "invalid-bucket", "100",
-          null, null);
+      client.listKeys(helper.getVol().getName(), "invalid-bucket", null,
+          null, 100);
       fail("List keys should have thrown when using invalid bucket name.");
-    } catch (OzoneException e) {
+    } catch (IOException e) {
       GenericTestUtils.assertExceptionContains(
           Status.BUCKET_NOT_FOUND.toString(), e);
     }
@@ -494,29 +554,27 @@ public class TestKeys {
 
   @Test
   public void testGetKeyInfo() throws Exception {
-    runTestGetKeyInfo(new PutHelper(ozoneRestClient, path));
+    runTestGetKeyInfo(new PutHelper(client, path));
     String delimiter = RandomStringUtils.randomAscii(1);
-    runTestGetKeyInfo(new PutHelper(ozoneRestClient, path,
+    runTestGetKeyInfo(new PutHelper(client, path,
         getMultiPartKey(delimiter)));
   }
 
   static void runTestGetKeyInfo(PutHelper helper) throws Exception {
-    String keyName = helper.putKey().getKeyName();
+    String keyName = helper.putKey();
     assertNotNull(helper.getBucket());
     assertNotNull(helper.getFile());
 
-    OzoneKey keyInfo = helper.getBucket().getKeyInfo(keyName);
-    assertNotNull(keyInfo.getObjectInfo());
-    assertEquals(keyName, keyInfo.getObjectInfo().getKeyName());
+    OzoneKey keyInfo = helper.getBucket().getKey(keyName);
+    assertNotNull(keyInfo);
+    assertEquals(keyName, keyInfo.getName());
 
     // Compare the time in second unit since the date string reparsed to
     // millisecond will lose precision.
+    Assert
+        .assertTrue((keyInfo.getCreationTime() / 1000) >= (currentTime / 1000));
     Assert.assertTrue(
-        (OzoneUtils.formatDate(keyInfo.getObjectInfo().getCreatedOn())
-            / 1000) >= (currentTime / 1000));
-    Assert.assertTrue(
-        (OzoneUtils.formatDate(keyInfo.getObjectInfo().getModifiedOn())
-            / 1000) >= (currentTime / 1000));
+        (keyInfo.getModificationTime() / 1000) >= (currentTime / 1000));
   }
 
   // Volume, bucket, keys info that helps for test create/delete keys.
@@ -593,12 +651,12 @@ public class TestKeys {
     int numOfExistedKeys = countKsmKeys(ksm);
 
     // Keep tracking bucket keys info while creating them
-    PutHelper helper = new PutHelper(ozoneRestClient, path);
+    PutHelper helper = new PutHelper(client, path);
     BucketKeys bucketKeys = new BucketKeys();
     for (int i = 0; i < 20; i++) {
-      KsmKeyArgs keyArgs = helper.putKey();
-      bucketKeys.addKey(keyArgs.getVolumeName(), keyArgs.getBucketName(),
-          keyArgs.getKeyName());
+      String keyName = helper.putKey();
+      bucketKeys.addKey(helper.getVol().getName(), helper.getBucket().getName(),
+          keyName);
     }
 
     // There should be 20 keys in the buckets we just created.
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestKeysRatis.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestKeysRatis.java
index 802cc3dd68..645b86610b 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestKeysRatis.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestKeysRatis.java
@@ -21,6 +21,7 @@ import org.apache.commons.lang.RandomStringUtils;
 import org.apache.hadoop.ozone.MiniOzoneCluster;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.RatisTestHelper;
+import org.apache.hadoop.ozone.client.protocol.ClientProtocol;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Rule;
@@ -52,7 +53,7 @@ public class TestKeysRatis {
   private static RatisTestHelper.RatisTestSuite suite;
   private static MiniOzoneCluster ozoneCluster = null;
   static private String path;
-  private static OzoneRestClient ozoneRestClient = null;
+  private static ClientProtocol client = null;
 
   @BeforeClass
   public static void init() throws Exception {
@@ -60,7 +61,7 @@ public class TestKeysRatis {
     path = suite.getConf().get(OzoneConfigKeys.OZONE_LOCALSTORAGE_ROOT);
     ozoneCluster = suite.getCluster();
     ozoneCluster.waitForClusterToBeReady();
-    ozoneRestClient = suite.newOzoneRestClient();
+    client = suite.newOzoneClient();
   }
 
   /**
@@ -76,9 +77,9 @@ public class TestKeysRatis {
 
   @Test
   public void testPutKey() throws Exception {
-    runTestPutKey(new PutHelper(ozoneRestClient, path));
+    runTestPutKey(new PutHelper(client, path));
     String delimiter = RandomStringUtils.randomAlphanumeric(1);
-    runTestPutKey(new PutHelper(ozoneRestClient, path,
+    runTestPutKey(new PutHelper(client, path,
         getMultiPartKey(delimiter)));
   }
 
@@ -86,42 +87,42 @@ public class TestKeysRatis {
   @Test
   public void testPutAndGetKeyWithDnRestart() throws Exception {
     runTestPutAndGetKeyWithDnRestart(
-        new PutHelper(ozoneRestClient, path), ozoneCluster);
+        new PutHelper(client, path), ozoneCluster);
     String delimiter = RandomStringUtils.randomAlphanumeric(1);
     runTestPutAndGetKeyWithDnRestart(
-        new PutHelper(ozoneRestClient, path, getMultiPartKey(delimiter)),
+        new PutHelper(client, path, getMultiPartKey(delimiter)),
         ozoneCluster);
   }
 
   @Test
   public void testPutAndGetKey() throws Exception {
-    runTestPutAndGetKey(new PutHelper(ozoneRestClient, path));
+    runTestPutAndGetKey(new PutHelper(client, path));
     String delimiter = RandomStringUtils.randomAlphanumeric(1);
-    runTestPutAndGetKey(new PutHelper(ozoneRestClient, path,
+    runTestPutAndGetKey(new PutHelper(client, path,
         getMultiPartKey(delimiter)));
   }
 
   @Test
   public void testPutAndDeleteKey() throws Exception  {
-    runTestPutAndDeleteKey(new PutHelper(ozoneRestClient, path));
+    runTestPutAndDeleteKey(new PutHelper(client, path));
     String delimiter = RandomStringUtils.randomAlphanumeric(1);
-    runTestPutAndDeleteKey(new PutHelper(ozoneRestClient, path,
+    runTestPutAndDeleteKey(new PutHelper(client, path,
         getMultiPartKey(delimiter)));
   }
 
   @Test
   public void testPutAndListKey() throws Exception {
-    runTestPutAndListKey(new PutHelper(ozoneRestClient, path));
+    runTestPutAndListKey(new PutHelper(client, path));
     String delimiter = RandomStringUtils.randomAlphanumeric(1);
-    runTestPutAndListKey(new PutHelper(ozoneRestClient, path,
+    runTestPutAndListKey(new PutHelper(client, path,
         getMultiPartKey(delimiter)));
   }
 
   @Test
   public void testGetKeyInfo() throws Exception {
-    runTestGetKeyInfo(new PutHelper(ozoneRestClient, path));
+    runTestGetKeyInfo(new PutHelper(client, path));
     String delimiter = RandomStringUtils.randomAlphanumeric(1);
-    runTestGetKeyInfo(new PutHelper(ozoneRestClient, path,
+    runTestGetKeyInfo(new PutHelper(client, path,
         getMultiPartKey(delimiter)));
   }
 }
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestVolume.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestVolume.java
index 2d3cea9f15..f8c7eecb45 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestVolume.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestVolume.java
@@ -20,50 +20,42 @@ package org.apache.hadoop.ozone.web.client;
 
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.lang.RandomStringUtils;
-import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.hdds.client.OzoneQuota;
 import org.apache.hadoop.ozone.MiniOzoneCluster;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
-import org.apache.hadoop.ozone.OzoneConsts;
-import org.apache.hadoop.ozone.client.OzoneClientException;
-import org.apache.hadoop.ozone.protocol.proto.KeySpaceManagerProtocolProtos.Status;
-import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
+import org.apache.hadoop.ozone.client.VolumeArgs;
+import org.apache.hadoop.ozone.client.protocol.ClientProtocol;
+import org.apache.hadoop.ozone.client.rpc.RpcClient;
 import org.apache.hadoop.ozone.client.rest.OzoneException;
-import org.apache.hadoop.ozone.web.request.OzoneQuota;
+import org.apache.hadoop.ozone.client.OzoneVolume;
 import org.apache.hadoop.ozone.web.utils.OzoneUtils;
 import org.apache.hadoop.test.GenericTestUtils;
 import org.apache.hadoop.util.Time;
-import org.apache.http.client.methods.HttpUriRequest;
-import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
-import org.mockito.Mockito;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.Ignore;
 
 import java.io.File;
 import java.io.IOException;
 import java.text.ParseException;
-import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
 
-import org.junit.BeforeClass;
-import org.junit.AfterClass;
-import org.junit.Test;
-import org.junit.Ignore;
-import static org.junit.Assert.fail;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
 
 /**
  * Test Ozone Volumes Lifecycle.
  */
 public class TestVolume {
   private static MiniOzoneCluster cluster = null;
-  private static OzoneRestClient ozoneRestClient = null;
+  private static ClientProtocol client = null;
 
   /**
    * Create a MiniDFSCluster for testing.
@@ -92,8 +84,7 @@ public class TestVolume {
     final int port = cluster.getHddsDatanodes().get(0)
         .getDatanodeDetails().getOzoneRestPort();
 
-    ozoneRestClient = new OzoneRestClient(
-        String.format("http://localhost:%d", port));
+    client = new RpcClient(conf);
   }
 
   /**
@@ -108,128 +99,122 @@ public class TestVolume {
 
   @Test
   public void testCreateVolume() throws Exception {
-    runTestCreateVolume(ozoneRestClient);
+    runTestCreateVolume(client);
   }
 
-  static void runTestCreateVolume(OzoneRestClient client)
+  static void runTestCreateVolume(ClientProtocol client)
       throws OzoneException, IOException, ParseException {
     String volumeName = OzoneUtils.getRequestID().toLowerCase();
-    client.setUserAuth(OzoneConsts.OZONE_SIMPLE_HDFS_USER);
 
     long currentTime = Time.now();
-    OzoneRestClient mockClient = Mockito.spy(client);
-    List<CloseableHttpClient> mockedClients = mockHttpClients(mockClient);
-    OzoneVolume vol = mockClient.createVolume(volumeName, "bilbo", "100TB");
-    // Verify http clients are properly closed.
-    verifyHttpConnectionClosed(mockedClients);
-
-    assertEquals(vol.getVolumeName(), volumeName);
-    assertEquals(vol.getCreatedby(), "hdfs");
-    assertEquals(vol.getOwnerName(), "bilbo");
-    assertEquals(vol.getQuota().getUnit(), OzoneQuota.Units.TB);
-    assertEquals(vol.getQuota().getSize(), 100);
+
+    VolumeArgs volumeArgs = VolumeArgs.newBuilder()
+        .setOwner("bilbo")
+        .setQuota("100TB")
+        .setAdmin("hdfs")
+        .build();
+    client.createVolume(volumeName, volumeArgs);
+    OzoneVolume vol = client.getVolumeDetails(volumeName);
+
+    assertEquals(vol.getName(), volumeName);
+    assertEquals(vol.getAdmin(), "hdfs");
+    assertEquals(vol.getOwner(), "bilbo");
+    assertEquals(vol.getQuota(), OzoneQuota.parseQuota("100TB").sizeInBytes());
 
     // verify the key creation time
-    assertTrue((OzoneUtils.formatDate(vol.getCreatedOn())
+    assertTrue((vol.getCreationTime()
         / 1000) >= (currentTime / 1000));
 
     // Test create a volume with invalid volume name,
     // not use Rule here because the test method is static.
     try {
       String invalidVolumeName = "#" + OzoneUtils.getRequestID().toLowerCase();
-      client.setUserAuth(OzoneConsts.OZONE_SIMPLE_HDFS_USER);
-      mockClient.createVolume(invalidVolumeName, "bilbo", "100TB");
+      client.createVolume(invalidVolumeName);
+      /*
+      //TODO: RestClient and RpcClient should use HddsClientUtils to verify name
       fail("Except the volume creation be failed because the"
-          + " volume name starts with an invalid char #");
+          + " volume name starts with an invalid char #");*/
     } catch (Exception e) {
-      assertTrue(e instanceof OzoneClientException);
       assertTrue(e.getMessage().contains("Bucket or Volume name"
           + " has an unsupported character : #"));
     }
   }
 
   @Test
-  public void testCreateDuplicateVolume() throws OzoneException {
-    runTestCreateDuplicateVolume(ozoneRestClient);
+  public void testCreateDuplicateVolume() throws OzoneException, IOException {
+    runTestCreateDuplicateVolume(client);
   }
 
-  static void runTestCreateDuplicateVolume(OzoneRestClient client)
-      throws OzoneException {
+  static void runTestCreateDuplicateVolume(ClientProtocol client)
+      throws OzoneException, IOException {
     try {
-      client.setUserAuth(OzoneConsts.OZONE_SIMPLE_HDFS_USER);
-      client.createVolume("testvol", "bilbo", "100TB");
-      client.createVolume("testvol", "bilbo", "100TB");
+      client.createVolume("testvol");
+      client.createVolume("testvol");
       assertFalse(true);
-    } catch (OzoneException ex) {
-      // Ozone will throw saying volume already exists
-      GenericTestUtils.assertExceptionContains(
-          Status.VOLUME_ALREADY_EXISTS.toString(), ex);
+    } catch (IOException ioe) {
+      Assert.assertTrue(ioe.getMessage()
+          .contains("Volume creation failed, error:VOLUME_ALREADY_EXISTS"));
     }
   }
 
   @Test
-  public void testDeleteVolume() throws OzoneException {
-    runTestDeleteVolume(ozoneRestClient);
+  public void testDeleteVolume() throws OzoneException, IOException {
+    runTestDeleteVolume(client);
   }
 
-  static void runTestDeleteVolume(OzoneRestClient client)
-      throws OzoneException {
+  static void runTestDeleteVolume(ClientProtocol client)
+      throws OzoneException, IOException {
     String volumeName = OzoneUtils.getRequestID().toLowerCase();
-    client.setUserAuth(OzoneConsts.OZONE_SIMPLE_HDFS_USER);
-    OzoneVolume vol = client.createVolume(volumeName, "bilbo", "100TB");
-    client.deleteVolume(vol.getVolumeName());
+    client.createVolume(volumeName);
+    client.deleteVolume(volumeName);
   }
 
   @Test
   public void testChangeOwnerOnVolume() throws Exception {
-    runTestChangeOwnerOnVolume(ozoneRestClient);
+    runTestChangeOwnerOnVolume(client);
   }
 
-  static void runTestChangeOwnerOnVolume(OzoneRestClient client)
-      throws OzoneException, ParseException {
+  static void runTestChangeOwnerOnVolume(ClientProtocol client)
+      throws OzoneException, ParseException, IOException {
     String volumeName = OzoneUtils.getRequestID().toLowerCase();
-    client.setUserAuth(OzoneConsts.OZONE_SIMPLE_HDFS_USER);
-    OzoneVolume vol = client.createVolume(volumeName, "bilbo", "100TB");
+    client.createVolume(volumeName);
+    client.getVolumeDetails(volumeName);
     client.setVolumeOwner(volumeName, "frodo");
-    OzoneVolume newVol = client.getVolume(volumeName);
-    assertEquals(newVol.getOwnerName(), "frodo");
+    OzoneVolume newVol = client.getVolumeDetails(volumeName);
+    assertEquals(newVol.getOwner(), "frodo");
     // verify if the creation time is missing after setting owner operation
-    assertTrue(OzoneUtils.formatDate(newVol.getCreatedOn()) > 0);
+    assertTrue(newVol.getCreationTime() > 0);
   }
 
   @Test
   public void testChangeQuotaOnVolume() throws Exception {
-    runTestChangeQuotaOnVolume(ozoneRestClient);
+    runTestChangeQuotaOnVolume(client);
   }
 
-  static void runTestChangeQuotaOnVolume(OzoneRestClient client)
+  static void runTestChangeQuotaOnVolume(ClientProtocol client)
       throws OzoneException, IOException, ParseException {
     String volumeName = OzoneUtils.getRequestID().toLowerCase();
-    client.setUserAuth(OzoneConsts.OZONE_SIMPLE_HDFS_USER);
-    OzoneVolume vol = client.createVolume(volumeName, "bilbo", "100TB");
-    client.setVolumeQuota(volumeName, "1000MB");
-    OzoneVolume newVol = client.getVolume(volumeName);
-    assertEquals(newVol.getQuota().getSize(), 1000);
-    assertEquals(newVol.getQuota().getUnit(), OzoneQuota.Units.MB);
+    client.createVolume(volumeName);
+    client.setVolumeQuota(volumeName, OzoneQuota.parseQuota("1000MB"));
+    OzoneVolume newVol = client.getVolumeDetails(volumeName);
+    assertEquals(newVol.getQuota(), OzoneQuota.parseQuota("1000MB").sizeInBytes());
     // verify if the creation time is missing after setting quota operation
-    assertTrue(OzoneUtils.formatDate(newVol.getCreatedOn()) > 0);
+    assertTrue(newVol.getCreationTime() > 0);
   }
 
   @Test
   public void testListVolume() throws OzoneException, IOException {
-    runTestListVolume(ozoneRestClient);
+    runTestListVolume(client);
   }
 
-  static void runTestListVolume(OzoneRestClient client)
+  static void runTestListVolume(ClientProtocol client)
       throws OzoneException, IOException {
-    client.setUserAuth(OzoneConsts.OZONE_SIMPLE_HDFS_USER);
     for (int x = 0; x < 10; x++) {
       String volumeName = OzoneUtils.getRequestID().toLowerCase();
-      OzoneVolume vol = client.createVolume(volumeName, "frodo", "100TB");
-      assertNotNull(vol);
+      client.createVolume(volumeName);
     }
 
-    List<OzoneVolume> ovols = client.listVolumes("frodo");
+    List<OzoneVolume> ovols = client.listVolumes(null, null, 100);
     assertTrue(ovols.size() >= 10);
   }
 
@@ -237,27 +222,24 @@ public class TestVolume {
   @Ignore("Takes 3m to run, disable for now.")
   @Test
   public void testListVolumePagination() throws OzoneException, IOException {
-    runTestListVolumePagination(ozoneRestClient);
+    runTestListVolumePagination(client);
   }
 
-  static void runTestListVolumePagination(OzoneRestClient client)
+  static void runTestListVolumePagination(ClientProtocol client)
       throws OzoneException, IOException {
     final int volCount = 2000;
     final int step = 100;
-    client.setUserAuth(OzoneConsts.OZONE_SIMPLE_HDFS_USER);
     for (int x = 0; x < volCount; x++) {
       String volumeName = OzoneUtils.getRequestID().toLowerCase();
-      OzoneVolume vol = client.createVolume(volumeName, "frodo", "100TB");
-      assertNotNull(vol);
+      client.createVolume(volumeName);
     }
-    OzoneVolume prevKey = null;
+    String prevKey = null;
     int count = 0;
     int pagecount = 0;
     while (count < volCount) {
-      List<OzoneVolume> ovols = client.listVolumes("frodo", null, step,
-          prevKey);
+      List<OzoneVolume> ovols = client.listVolumes(null, prevKey, step);
       count += ovols.size();
-      prevKey = ovols.get(ovols.size() - 1);
+      prevKey = ovols.get(ovols.size() - 1).getName();
       pagecount++;
     }
     assertEquals(volCount / step, pagecount);
@@ -267,30 +249,35 @@ public class TestVolume {
   @Ignore
   @Test
   public void testListAllVolumes() throws OzoneException, IOException {
-    runTestListAllVolumes(ozoneRestClient);
+    runTestListAllVolumes(client);
   }
 
-  static void runTestListAllVolumes(OzoneRestClient client)
+  static void runTestListAllVolumes(ClientProtocol client)
       throws OzoneException, IOException {
     final int volCount = 200;
     final int step = 10;
-    client.setUserAuth(OzoneConsts.OZONE_SIMPLE_HDFS_USER);
     for (int x = 0; x < volCount; x++) {
       String userName =
           "frodo" + RandomStringUtils.randomAlphabetic(5).toLowerCase();
       String volumeName =
           "vol" + RandomStringUtils.randomAlphabetic(5).toLowerCase();
-      OzoneVolume vol = client.createVolume(volumeName, userName, "100TB");
+      VolumeArgs volumeArgs = VolumeArgs.newBuilder()
+          .setOwner(userName)
+          .setQuota("100TB")
+          .setAdmin("hdfs")
+          .build();
+      client.createVolume(volumeName, volumeArgs);
+      OzoneVolume vol = client.getVolumeDetails(volumeName);
       assertNotNull(vol);
     }
-    OzoneVolume prevKey = null;
+    String prevKey = null;
     int count = 0;
     int pagecount = 0;
     while (count < volCount) {
-      List<OzoneVolume> ovols = client.listAllVolumes(null, step, prevKey);
+      List<OzoneVolume> ovols = client.listVolumes(null, prevKey, step);
       count += ovols.size();
       if (ovols.size() > 0) {
-        prevKey = ovols.get(ovols.size() - 1);
+        prevKey = ovols.get(ovols.size() - 1).getName();
       }
       pagecount++;
     }
@@ -301,17 +288,16 @@ public class TestVolume {
 
   @Test
   public void testListVolumes() throws Exception {
-    runTestListVolumes(ozoneRestClient);
+    runTestListVolumes(client);
   }
 
-  static void runTestListVolumes(OzoneRestClient client)
+  static void runTestListVolumes(ClientProtocol client)
       throws OzoneException, IOException, ParseException {
     final int volCount = 20;
     final String user1 = "test-user-a";
     final String user2 = "test-user-b";
 
     long currentTime = Time.now();
-    client.setUserAuth(OzoneConsts.OZONE_SIMPLE_HDFS_USER);
     // Create 20 volumes, 10 for user1 and another 10 for user2.
     for (int x = 0; x < volCount; x++) {
       String volumeName;
@@ -326,109 +312,40 @@ public class TestVolume {
         userName = user2;
         volumeName = "test-vol" + x;
       }
-      OzoneVolume vol = client.createVolume(volumeName, userName, "100TB");
+      VolumeArgs volumeArgs = VolumeArgs.newBuilder()
+          .setOwner(userName)
+          .setQuota("100TB")
+          .setAdmin("hdfs")
+          .build();
+      client.createVolume(volumeName, volumeArgs);
+      OzoneVolume vol = client.getVolumeDetails(volumeName);
       assertNotNull(vol);
     }
 
     // list all the volumes belong to user1
-    List<OzoneVolume> volumeList = client.listVolumes(user1,
-        null, 100, StringUtils.EMPTY);
+    List<OzoneVolume> volumeList = client.listVolumes(user1, null, null, 100);
     assertEquals(10, volumeList.size());
     // verify the owner name and creation time of volume
     for (OzoneVolume vol : volumeList) {
-      assertTrue(vol.getOwnerName().equals(user1));
-      assertTrue((OzoneUtils.formatDate(vol.getCreatedOn())
+      assertTrue(vol.getOwner().equals(user1));
+      assertTrue((vol.getCreationTime()
           / 1000) >= (currentTime / 1000));
     }
 
     // test max key parameter of listing volumes
-    volumeList = client.listVolumes(user1, null, 2, StringUtils.EMPTY);
+    volumeList = client.listVolumes(user1, null, null, 2);
     assertEquals(2, volumeList.size());
 
     // test prefix parameter of listing volumes
-    volumeList = client.listVolumes(user1, "test-vol10", 100,
-        StringUtils.EMPTY);
+    volumeList = client.listVolumes(user1, "test-vol10", null, 10);
     assertTrue(volumeList.size() == 1
-        && volumeList.get(0).getVolumeName().equals("test-vol10"));
+        && volumeList.get(0).getName().equals("test-vol10"));
 
-    volumeList = client.listVolumes(user1, "test-vol1",
-        100, StringUtils.EMPTY);
+    volumeList = client.listVolumes(user1, "test-vol1", null, 10);
     assertEquals(5, volumeList.size());
 
     // test start key parameter of listing volumes
-    volumeList = client.listVolumes(user2, null, 100, "test-vol15");
+    volumeList = client.listVolumes(user2, null, "test-vol15", 10);
     assertEquals(2, volumeList.size());
   }
-
-  /**
-   * Returns a list of mocked {@link CloseableHttpClient} used for testing.
-   * The mocked client replaces the actual calls in
-   * {@link OzoneRestClient#newHttpClient()}, it is used to verify
-   * if the invocation of this client is expected. <b>Note</b>, the output
-   * of this method is always used as the input of
-   * {@link TestVolume#verifyHttpConnectionClosed(List)}.
-   *
-   * @param mockedClient mocked ozone client.
-   * @return a list of mocked {@link CloseableHttpClient}.
-   * @throws IOException
-   */
-  private static List<CloseableHttpClient> mockHttpClients(
-      OzoneRestClient mockedClient)
-      throws IOException {
-    List<CloseableHttpClient> spyHttpClients = new ArrayList<>();
-    for (int i = 0; i < 5; i++) {
-      CloseableHttpClient spyHttpClient = Mockito
-          .spy(HddsClientUtils.newHttpClient());
-      spyHttpClients.add(spyHttpClient);
-    }
-
-    List<CloseableHttpClient> nextReturns =
-        new ArrayList<>(spyHttpClients.subList(1, spyHttpClients.size()));
-    Mockito.when(mockedClient.newHttpClient()).thenReturn(
-        spyHttpClients.get(0),
-        nextReturns.toArray(new CloseableHttpClient[nextReturns.size()]));
-    return spyHttpClients;
-  }
-
-  /**
-   * This method is used together with
-   * {@link TestVolume#mockHttpClients(OzoneRestClient)} to verify
-   * if the http client is properly closed. It verifies that as long as
-   * a client calls {@link CloseableHttpClient#execute(HttpUriRequest)} to
-   * send request, then it must calls {@link CloseableHttpClient#close()}
-   * close the http connection.
-   *
-   * @param mockedHttpClients
-   */
-  private static void verifyHttpConnectionClosed(
-      List<CloseableHttpClient> mockedHttpClients) {
-    final AtomicInteger totalCalled = new AtomicInteger();
-    assertTrue(mockedHttpClients.stream().allMatch(closeableHttpClient -> {
-      boolean clientUsed = false;
-      try {
-        verify(closeableHttpClient, times(1)).execute(Mockito.any());
-        totalCalled.incrementAndGet();
-        clientUsed = true;
-      } catch (Throwable e) {
-        // There might be some redundant instances in mockedHttpClients,
-        // it is allowed that a client is not used.
-        return true;
-      }
-
-      if (clientUsed) {
-        try {
-          // If a client is used, ensure the close function is called.
-          verify(closeableHttpClient, times(1)).close();
-          return true;
-        } catch (IOException e) {
-          return false;
-        }
-      } else {
-        return true;
-      }
-    }));
-    System.out.println("Successful connections " + totalCalled.get());
-    assertTrue("The mocked http client should be called at least once.",
-        totalCalled.get() > 0);
-  }
 }
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestVolumeRatis.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestVolumeRatis.java
index 83148514a8..bc4ba25aba 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestVolumeRatis.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/web/client/TestVolumeRatis.java
@@ -23,7 +23,9 @@ import org.apache.hadoop.ozone.MiniOzoneCluster;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 
+import org.apache.hadoop.ozone.client.protocol.ClientProtocol;
 import org.apache.hadoop.ozone.client.rest.OzoneException;
+import org.apache.hadoop.ozone.client.rpc.RpcClient;
 import org.apache.hadoop.test.GenericTestUtils;
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
@@ -38,7 +40,7 @@ import java.io.IOException;
 public class TestVolumeRatis {
   @Rule
   public Timeout testTimeout = new Timeout(300000);
-  private static OzoneRestClient ozoneClient;
+  private static ClientProtocol client;
   private static MiniOzoneCluster cluster;
 
   @BeforeClass
@@ -63,8 +65,7 @@ public class TestVolumeRatis {
     final int port = cluster.getHddsDatanodes().get(0)
         .getDatanodeDetails().getOzoneRestPort();
 
-    ozoneClient = new OzoneRestClient(
-        String.format("http://localhost:%d", port));
+    client = new RpcClient(conf);
   }
 
   @AfterClass
@@ -77,53 +78,53 @@ public class TestVolumeRatis {
 
   @Test
   public void testCreateVolume() throws Exception {
-    TestVolume.runTestCreateVolume(ozoneClient);
+    TestVolume.runTestCreateVolume(client);
   }
 
   @Test
-  public void testCreateDuplicateVolume() throws OzoneException {
-    TestVolume.runTestCreateDuplicateVolume(ozoneClient);
+  public void testCreateDuplicateVolume() throws OzoneException, IOException {
+    TestVolume.runTestCreateDuplicateVolume(client);
   }
 
   @Test
-  public void testDeleteVolume() throws OzoneException {
-    TestVolume.runTestDeleteVolume(ozoneClient);
+  public void testDeleteVolume() throws OzoneException, IOException {
+    TestVolume.runTestDeleteVolume(client);
   }
 
   @Test
   public void testChangeOwnerOnVolume() throws Exception {
-    TestVolume.runTestChangeOwnerOnVolume(ozoneClient);
+    TestVolume.runTestChangeOwnerOnVolume(client);
   }
 
   @Test
   public void testChangeQuotaOnVolume() throws Exception {
-    TestVolume.runTestChangeQuotaOnVolume(ozoneClient);
+    TestVolume.runTestChangeQuotaOnVolume(client);
   }
 
   // TODO: remove @Ignore below once the problem has been resolved.
   @Ignore("listVolumes not implemented in DistributedStorageHandler")
   @Test
   public void testListVolume() throws OzoneException, IOException {
-    TestVolume.runTestListVolume(ozoneClient);
+    TestVolume.runTestListVolume(client);
   }
 
   // TODO: remove @Ignore below once the problem has been resolved.
   @Ignore("See TestVolume.testListVolumePagination()")
   @Test
   public void testListVolumePagination() throws OzoneException, IOException {
-    TestVolume.runTestListVolumePagination(ozoneClient);
+    TestVolume.runTestListVolumePagination(client);
   }
 
   // TODO: remove @Ignore below once the problem has been resolved.
   @Ignore("See TestVolume.testListAllVolumes()")
   @Test
   public void testListAllVolumes() throws Exception {
-    TestVolume.runTestListAllVolumes(ozoneClient);
+    TestVolume.runTestListAllVolumes(client);
   }
 
   @Ignore("Disabling Ratis tests for pipeline work.")
   @Test
   public void testListVolumes() throws Exception {
-    TestVolume.runTestListVolumes(ozoneClient);
+    TestVolume.runTestListVolumes(client);
   }
 }