HDDS-20. Ozone: Add support for rename key within a bucket for rpc client. Contributed by Lokesh Jain.

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneBucket.java

@@ -305,6 +305,14 @@ public class OzoneBucket {
     proxy.deleteKey(volumeName, name, key);
   }
 
+  public void renameKey(String fromKeyName, String toKeyName)
+      throws IOException {
+    Preconditions.checkNotNull(proxy, "Client proxy is not set.");
+    Preconditions.checkNotNull(fromKeyName);
+    Preconditions.checkNotNull(toKeyName);
+    proxy.renameKey(volumeName, name, fromKeyName, toKeyName);
+  }
+
   /**
    * An Iterator to iterate over {@link OzoneKey} list.
    */

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java

@@ -285,6 +285,16 @@ public interface ClientProtocol {
   void deleteKey(String volumeName, String bucketName, String keyName)
       throws IOException;
 
+  /**
+   * Renames an existing key within a bucket.
+   * @param volumeName Name of the Volume
+   * @param bucketName Name of the Bucket
+   * @param fromKeyName Name of the Key to be renamed
+   * @param toKeyName New name to be used for the Key
+   * @throws IOException
+   */
+  void renameKey(String volumeName, String bucketName, String fromKeyName,
+      String toKeyName) throws IOException;
 
   /**
    * Returns list of Keys in {Volume/Bucket} that matches the keyPrefix,

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rest/RestClient.java

@@ -675,6 +675,12 @@ public class RestClient implements ClientProtocol {
     }
   }
 
+  @Override
+  public void renameKey(String volumeName, String bucketName,
+      String fromKeyName, String toKeyName) throws IOException {
+    throw new UnsupportedOperationException("Not yet implemented.");
+  }
+
   @Override
   public List<OzoneKey> listKeys(String volumeName, String bucketName,
                                  String keyPrefix, String prevKey,

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java

@@ -519,6 +519,21 @@ public class RpcClient implements ClientProtocol {
     keySpaceManagerClient.deleteKey(keyArgs);
   }
 
+  @Override
+  public void renameKey(String volumeName, String bucketName,
+      String fromKeyName, String toKeyName) throws IOException {
+    Preconditions.checkNotNull(volumeName);
+    Preconditions.checkNotNull(bucketName);
+    Preconditions.checkNotNull(fromKeyName);
+    Preconditions.checkNotNull(toKeyName);
+    KsmKeyArgs keyArgs = new KsmKeyArgs.Builder()
+        .setVolumeName(volumeName)
+        .setBucketName(bucketName)
+        .setKeyName(fromKeyName)
+        .build();
+    keySpaceManagerClient.renameKey(keyArgs, toKeyName);
+  }
+
   @Override
   public List<OzoneKey> listKeys(String volumeName, String bucketName,
                                  String keyPrefix, String prevKey,

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/ksm/helpers/KsmKeyInfo.java

@@ -34,7 +34,7 @@ public final class KsmKeyInfo {
   private final String volumeName;
   private final String bucketName;
   // name of key client specified
-  private final String keyName;
+  private String keyName;
   private long dataSize;
   private List<KsmKeyLocationInfoGroup> keyLocationVersions;
   private final long creationTime;
@@ -75,6 +75,10 @@ public final class KsmKeyInfo {
     return keyName;
   }
 
+  public void setKeyName(String keyName) {
+    this.keyName = keyName;
+  }
+
   public long getDataSize() {
     return dataSize;
   }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/ksm/protocol/KeySpaceManagerProtocol.java

@@ -166,11 +166,18 @@ public interface KeySpaceManagerProtocol {
    * Look up for the container of an existing key.
    *
    * @param args the args of the key.
-   * @return KsmKeyInfo isntacne that client uses to talk to container.
+   * @return KsmKeyInfo instance that client uses to talk to container.
    * @throws IOException
    */
   KsmKeyInfo lookupKey(KsmKeyArgs args) throws IOException;
 
+  /**
+   * Rename an existing key within a bucket
+   * @param args the args of the key.
+   * @param toKeyName New name to be used for the Key
+   */
+  void renameKey(KsmKeyArgs args, String toKeyName) throws IOException;
+
   /**
    * Deletes an existing key.
    *

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/ksm/protocolPB/KeySpaceManagerProtocolClientSideTranslatorPB.java

@@ -69,6 +69,10 @@ import org.apache.hadoop.ozone.protocol.proto
     .KeySpaceManagerProtocolProtos.LocateKeyRequest;
 import org.apache.hadoop.ozone.protocol.proto
     .KeySpaceManagerProtocolProtos.LocateKeyResponse;
+import org.apache.hadoop.ozone.protocol.proto
+    .KeySpaceManagerProtocolProtos.RenameKeyRequest;
+import org.apache.hadoop.ozone.protocol.proto
+    .KeySpaceManagerProtocolProtos.RenameKeyResponse;
 import org.apache.hadoop.ozone.protocol.proto
     .KeySpaceManagerProtocolProtos.KeyArgs;
 import org.apache.hadoop.ozone.protocol.proto
@@ -623,6 +627,29 @@ public final class KeySpaceManagerProtocolClientSideTranslatorPB
     return KsmKeyInfo.getFromProtobuf(resp.getKeyInfo());
   }
 
+  @Override
+  public void renameKey(KsmKeyArgs args, String toKeyName) throws IOException {
+    RenameKeyRequest.Builder req = RenameKeyRequest.newBuilder();
+    KeyArgs keyArgs = KeyArgs.newBuilder()
+        .setVolumeName(args.getVolumeName())
+        .setBucketName(args.getBucketName())
+        .setKeyName(args.getKeyName())
+        .setDataSize(args.getDataSize()).build();
+    req.setKeyArgs(keyArgs);
+    req.setToKeyName(toKeyName);
+
+    final RenameKeyResponse resp;
+    try {
+      resp = rpcProxy.renameKey(NULL_RPC_CONTROLLER, req.build());
+    } catch (ServiceException e) {
+      throw ProtobufHelper.getRemoteException(e);
+    }
+    if (resp.getStatus() != Status.OK) {
+      throw new IOException("Rename key failed, error:" +
+          resp.getStatus());
+    }
+  }
+
   /**
    * Deletes an existing key.
    *

hadoop-ozone/common/src/main/proto/KeySpaceManagerProtocol.proto

@@ -50,8 +50,9 @@ enum Status {
     BUCKET_ALREADY_EXISTS = 10;
     KEY_ALREADY_EXISTS = 11;
     KEY_NOT_FOUND = 12;
-    ACCESS_DENIED = 13;
-    INTERNAL_ERROR = 14;
+    INVALID_KEY_NAME = 13;
+    ACCESS_DENIED = 14;
+    INTERNAL_ERROR = 15;
 }
 
 
@@ -276,6 +277,15 @@ message SetBucketPropertyResponse {
     required Status status = 1;
 }
 
+message RenameKeyRequest{
+    required KeyArgs keyArgs = 1;
+    required string toKeyName = 2;
+}
+
+message RenameKeyResponse{
+    required Status status = 1;
+}
+
 message DeleteBucketRequest {
     required string volumeName = 1;
     required string bucketName = 2;
@@ -412,6 +422,12 @@ service KeySpaceManagerService {
     rpc lookupKey(LocateKeyRequest)
         returns(LocateKeyResponse);
 
+    /**
+       Rename an existing key within a bucket.
+    */
+    rpc renameKey(RenameKeyRequest)
+        returns(RenameKeyResponse);
+
     /**
        Delete an existing key.
     */

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClient.java

@@ -527,6 +527,50 @@ public class TestOzoneRpcClient {
     bucket.getKey(keyName);
   }
 
+  @Test
+  public void testRenameKey()
+      throws IOException, OzoneException {
+    String volumeName = UUID.randomUUID().toString();
+    String bucketName = UUID.randomUUID().toString();
+    String fromKeyName = UUID.randomUUID().toString();
+    String value = "sample value";
+    store.createVolume(volumeName);
+    OzoneVolume volume = store.getVolume(volumeName);
+    volume.createBucket(bucketName);
+    OzoneBucket bucket = volume.getBucket(bucketName);
+    OzoneOutputStream out = bucket.createKey(fromKeyName,
+        value.getBytes().length, ReplicationType.STAND_ALONE,
+        ReplicationFactor.ONE);
+    out.write(value.getBytes());
+    out.close();
+    OzoneKey key = bucket.getKey(fromKeyName);
+    Assert.assertEquals(fromKeyName, key.getName());
+
+    // Rename to empty string should fail.
+    IOException ioe = null;
+    String toKeyName = "";
+    try {
+      bucket.renameKey(fromKeyName, toKeyName);
+    } catch (IOException e) {
+      ioe = e;
+    }
+    Assert.assertTrue(ioe.getMessage().contains("Rename key failed, error"));
+
+    toKeyName = UUID.randomUUID().toString();
+    bucket.renameKey(fromKeyName, toKeyName);
+
+    // Lookup for old key should fail.
+    try {
+      bucket.getKey(fromKeyName);
+    } catch (IOException e) {
+      ioe = e;
+    }
+    Assert.assertTrue(ioe.getMessage().contains("Lookup key failed, error"));
+
+    key = bucket.getKey(toKeyName);
+    Assert.assertEquals(toKeyName, key.getName());
+  }
+
   @Test
   public void testListVolume() throws IOException, OzoneException {
     String volBase = "vol-" + RandomStringUtils.randomNumeric(3);

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/ksm/TestKeySpaceManager.java

@@ -652,6 +652,121 @@ public class TestKeySpaceManager {
         ksmMetrics.getNumKeyDeletesFails());
   }
 
+  /**
+   * Test rename key for ksm.
+   *
+   * @throws IOException
+   * @throws OzoneException
+   */
+  @Test
+  public void testRenameKey() throws IOException, OzoneException {
+    String userName = "user" + RandomStringUtils.randomNumeric(5);
+    String adminName = "admin" + RandomStringUtils.randomNumeric(5);
+    String volumeName = "volume" + RandomStringUtils.randomNumeric(5);
+    String bucketName = "bucket" + RandomStringUtils.randomNumeric(5);
+    String keyName = "key" + RandomStringUtils.randomNumeric(5);
+    long numKeyRenames = ksmMetrics.getNumKeyRenames();
+    long numKeyRenameFails = ksmMetrics.getNumKeyRenameFails();
+    int testRenameFails = 0;
+    int testRenames = 0;
+    IOException ioe = null;
+
+    VolumeArgs createVolumeArgs = new VolumeArgs(volumeName, userArgs);
+    createVolumeArgs.setUserName(userName);
+    createVolumeArgs.setAdminName(adminName);
+    storageHandler.createVolume(createVolumeArgs);
+
+    BucketArgs bucketArgs = new BucketArgs(bucketName, createVolumeArgs);
+    storageHandler.createBucket(bucketArgs);
+
+    KeyArgs keyArgs = new KeyArgs(keyName, bucketArgs);
+    keyArgs.setSize(100);
+    String toKeyName = "key" + RandomStringUtils.randomNumeric(5);
+
+    // Rename from non-existent key should fail
+    try {
+      testRenames++;
+      storageHandler.renameKey(keyArgs, toKeyName);
+    } catch (IOException e) {
+      testRenameFails++;
+      ioe = e;
+    }
+    Assert.assertTrue(ioe.getMessage().contains("Rename key failed, error"));
+
+    // Write the contents of the key to be renamed
+    String dataString = RandomStringUtils.randomAscii(100);
+    try (OutputStream stream = storageHandler.newKeyWriter(keyArgs)) {
+      stream.write(dataString.getBytes());
+    }
+
+    // Rename the key
+    toKeyName = "key" + RandomStringUtils.randomNumeric(5);
+    testRenames++;
+    storageHandler.renameKey(keyArgs, toKeyName);
+    Assert.assertEquals(numKeyRenames + testRenames,
+        ksmMetrics.getNumKeyRenames());
+    Assert.assertEquals(numKeyRenameFails + testRenameFails,
+        ksmMetrics.getNumKeyRenameFails());
+
+    // Try to get the key, should fail as it has been renamed
+    try {
+      storageHandler.newKeyReader(keyArgs);
+    } catch (IOException e) {
+      ioe = e;
+    }
+    Assert.assertTrue(ioe.getMessage().contains("KEY_NOT_FOUND"));
+
+    // Verify the contents of the renamed key
+    keyArgs = new KeyArgs(toKeyName, bucketArgs);
+    InputStream in = storageHandler.newKeyReader(keyArgs);
+    byte[] b = new byte[dataString.getBytes().length];
+    in.read(b);
+    Assert.assertEquals(new String(b), dataString);
+
+    // Rewrite the renamed key. Rename to key which already exists should fail.
+    keyArgs = new KeyArgs(keyName, bucketArgs);
+    keyArgs.setSize(100);
+    dataString = RandomStringUtils.randomAscii(100);
+    try (OutputStream stream = storageHandler.newKeyWriter(keyArgs)) {
+      stream.write(dataString.getBytes());
+      stream.close();
+      testRenames++;
+      storageHandler.renameKey(keyArgs, toKeyName);
+    } catch (IOException e) {
+      testRenameFails++;
+      ioe = e;
+    }
+    Assert.assertTrue(ioe.getMessage().contains("Rename key failed, error"));
+
+    // Rename to empty string should fail
+    toKeyName = "";
+    try {
+      testRenames++;
+      storageHandler.renameKey(keyArgs, toKeyName);
+    } catch (IOException e) {
+      testRenameFails++;
+      ioe = e;
+    }
+    Assert.assertTrue(ioe.getMessage().contains("Rename key failed, error"));
+
+    // Rename from empty string should fail
+    keyArgs = new KeyArgs("", bucketArgs);
+    toKeyName = "key" + RandomStringUtils.randomNumeric(5);
+    try {
+      testRenames++;
+      storageHandler.renameKey(keyArgs, toKeyName);
+    } catch (IOException e) {
+      testRenameFails++;
+      ioe = e;
+    }
+    Assert.assertTrue(ioe.getMessage().contains("Rename key failed, error"));
+
+    Assert.assertEquals(numKeyRenames + testRenames,
+        ksmMetrics.getNumKeyRenames());
+    Assert.assertEquals(numKeyRenameFails + testRenameFails,
+        ksmMetrics.getNumKeyRenameFails());
+  }
+
   @Test(timeout = 60000)
   public void testListBuckets() throws IOException, OzoneException {
     ListBuckets result = null;

hadoop-ozone/objectstore-service/src/main/java/org/apache/hadoop/ozone/web/interfaces/StorageHandler.java

@@ -264,6 +264,15 @@ public interface StorageHandler extends Closeable{
    */
   void deleteKey(KeyArgs args) throws IOException, OzoneException;
 
+  /**
+   * Renames an existing key within a bucket.
+   *
+   * @param args KeyArgs
+   * @param toKeyName New name to be used for the key
+   * @throws OzoneException
+   */
+  void renameKey(KeyArgs args, String toKeyName)
+      throws IOException, OzoneException;
 
   /**
    * Returns a list of Key.

hadoop-ozone/objectstore-service/src/main/java/org/apache/hadoop/ozone/web/localstorage/LocalStorageHandler.java

@@ -339,6 +339,12 @@ public class LocalStorageHandler implements StorageHandler {
     oz.deleteKey(args);
   }
 
+  @Override
+  public void renameKey(KeyArgs args, String toKeyName)
+      throws IOException, OzoneException {
+    throw new UnsupportedOperationException("Not yet implemented");
+  }
+
   /**
    * Returns a list of Key.
    *

hadoop-ozone/objectstore-service/src/main/java/org/apache/hadoop/ozone/web/storage/DistributedStorageHandler.java

@@ -456,6 +456,17 @@ public final class DistributedStorageHandler implements StorageHandler {
     keySpaceManagerClient.deleteKey(keyArgs);
   }
 
+  @Override
+  public void renameKey(KeyArgs args, String toKeyName)
+      throws IOException, OzoneException {
+    KsmKeyArgs keyArgs = new KsmKeyArgs.Builder()
+        .setVolumeName(args.getVolumeName())
+        .setBucketName(args.getBucketName())
+        .setKeyName(args.getKeyName())
+        .build();
+    keySpaceManagerClient.renameKey(keyArgs, toKeyName);
+  }
+
   @Override
   public KeyInfo getKeyInfo(KeyArgs args) throws IOException, OzoneException {
     KsmKeyArgs keyArgs = new KsmKeyArgs.Builder()

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/ksm/KSMMetrics.java

@@ -52,6 +52,7 @@ public class KSMMetrics {
   private @Metric MutableCounterLong numBucketDeletes;
   private @Metric MutableCounterLong numKeyAllocate;
   private @Metric MutableCounterLong numKeyLookup;
+  private @Metric MutableCounterLong numKeyRenames;
   private @Metric MutableCounterLong numKeyDeletes;
   private @Metric MutableCounterLong numBucketLists;
   private @Metric MutableCounterLong numKeyLists;
@@ -72,6 +73,7 @@ public class KSMMetrics {
   private @Metric MutableCounterLong numBucketDeleteFails;
   private @Metric MutableCounterLong numKeyAllocateFails;
   private @Metric MutableCounterLong numKeyLookupFails;
+  private @Metric MutableCounterLong numKeyRenameFails;
   private @Metric MutableCounterLong numKeyDeleteFails;
   private @Metric MutableCounterLong numBucketListFails;
   private @Metric MutableCounterLong numKeyListFails;
@@ -208,6 +210,16 @@ public class KSMMetrics {
     numKeyLookupFails.incr();
   }
 
+  public void incNumKeyRenames() {
+    numKeyOps.incr();
+    numKeyRenames.incr();
+  }
+
+  public void incNumKeyRenameFails() {
+    numKeyOps.incr();
+    numKeyRenameFails.incr();
+  }
+
   public void incNumKeyDeleteFails() {
     numKeyDeleteFails.incr();
   }
@@ -380,6 +392,16 @@ public class KSMMetrics {
     return numKeyLookupFails.value();
   }
 
+  @VisibleForTesting
+  public long getNumKeyRenames() {
+    return numKeyRenames.value();
+  }
+
+  @VisibleForTesting
+  public long getNumKeyRenameFails() {
+    return numKeyRenameFails.value();
+  }
+
   @VisibleForTesting
   public long getNumKeyDeletes() {
     return numKeyDeletes.value();

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/ksm/KeyManager.java

@@ -85,6 +85,16 @@ public interface KeyManager {
    */
   KsmKeyInfo lookupKey(KsmKeyArgs args) throws IOException;
 
+  /**
+   * Renames an existing key within a bucket.
+   *
+   * @param args the args of the key provided by client.
+   * @param toKeyName New name to be used for the key
+   * @throws IOException if specified key doesn't exist or
+   * some other I/O errors while renaming the key.
+   */
+  void renameKey(KsmKeyArgs args, String toKeyName) throws IOException;
+
   /**
    * Deletes an object by an object key. The key will be immediately removed
    * from KSM namespace and become invisible to clients. The object data

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/ksm/KeyManagerImpl.java

@@ -395,6 +395,71 @@ public class KeyManagerImpl implements KeyManager {
     }
   }
 
+  @Override
+  public void renameKey(KsmKeyArgs args, String toKeyName) throws IOException {
+    Preconditions.checkNotNull(args);
+    Preconditions.checkNotNull(toKeyName);
+    String volumeName = args.getVolumeName();
+    String bucketName = args.getBucketName();
+    String fromKeyName = args.getKeyName();
+    if (toKeyName.length() == 0 || fromKeyName.length() == 0) {
+      LOG.error("Rename key failed for volume:{} bucket:{} fromKey:{} toKey:{}.",
+          volumeName, bucketName, fromKeyName, toKeyName);
+      throw new KSMException("Key name is empty",
+          ResultCodes.FAILED_INVALID_KEY_NAME);
+    }
+
+    metadataManager.writeLock().lock();
+    try {
+      // fromKeyName should exist
+      byte[] fromKey = metadataManager.getDBKeyBytes(
+          volumeName, bucketName, fromKeyName);
+      byte[] fromKeyValue = metadataManager.get(fromKey);
+      if (fromKeyValue == null) {
+        // TODO: Add support for renaming open key
+        LOG.error(
+            "Rename key failed for volume:{} bucket:{} fromKey:{} toKey:{}. "
+                + "Key: {} not found.", volumeName, bucketName, fromKeyName,
+            toKeyName, fromKeyName);
+        throw new KSMException("Key not found",
+            KSMException.ResultCodes.FAILED_KEY_NOT_FOUND);
+      }
+
+      // toKeyName should not exist
+      byte[] toKey =
+          metadataManager.getDBKeyBytes(volumeName, bucketName, toKeyName);
+      byte[] toKeyValue = metadataManager.get(toKey);
+      if (toKeyValue != null) {
+        LOG.error(
+            "Rename key failed for volume:{} bucket:{} fromKey:{} toKey:{}. "
+                + "Key: {} already exists.", volumeName, bucketName,
+            fromKeyName, toKeyName, toKeyName);
+        throw new KSMException("Key not found",
+            KSMException.ResultCodes.FAILED_KEY_ALREADY_EXISTS);
+      }
+
+      if (fromKeyName.equals(toKeyName)) {
+        return;
+      }
+
+      KsmKeyInfo newKeyInfo =
+          KsmKeyInfo.getFromProtobuf(KeyInfo.parseFrom(fromKeyValue));
+      newKeyInfo.setKeyName(toKeyName);
+      newKeyInfo.updateModifcationTime();
+      BatchOperation batch = new BatchOperation();
+      batch.delete(fromKey);
+      batch.put(toKey, newKeyInfo.getProtobuf().toByteArray());
+      metadataManager.writeBatch(batch);
+    } catch (DBException ex) {
+      LOG.error("Rename key failed for volume:{} bucket:{} fromKey:{} toKey:{}.",
+          volumeName, bucketName, fromKeyName, toKeyName, ex);
+      throw new KSMException(ex.getMessage(),
+          ResultCodes.FAILED_KEY_RENAME);
+    } finally {
+      metadataManager.writeLock().unlock();
+    }
+  }
+
   @Override
   public void deleteKey(KsmKeyArgs args) throws IOException {
     Preconditions.checkNotNull(args);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/ksm/KeySpaceManager.java

@@ -744,6 +744,17 @@ public final class KeySpaceManager extends ServiceRuntimeInfoImpl
     }
   }
 
+  @Override
+  public void renameKey(KsmKeyArgs args, String toKeyName) throws IOException {
+    try {
+      metrics.incNumKeyRenames();
+      keyManager.renameKey(args, toKeyName);
+    } catch (IOException e) {
+      metrics.incNumKeyRenameFails();
+      throw e;
+    }
+  }
+
   /**
    * Deletes an existing key.
    *

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/ksm/exceptions/KSMException.java

@@ -108,6 +108,8 @@ public class KSMException extends IOException {
     FAILED_KEY_NOT_FOUND,
     FAILED_KEY_ALLOCATION,
     FAILED_KEY_DELETION,
+    FAILED_KEY_RENAME,
+    FAILED_INVALID_KEY_NAME,
     FAILED_METADATA_ERROR,
     FAILED_INTERNAL_ERROR,
     KSM_NOT_INITIALIZED,

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

@@ -62,6 +62,10 @@ import org.apache.hadoop.ozone.protocol.proto
     .KeySpaceManagerProtocolProtos.LocateKeyRequest;
 import org.apache.hadoop.ozone.protocol.proto
     .KeySpaceManagerProtocolProtos.LocateKeyResponse;
+import org.apache.hadoop.ozone.protocol.proto
+    .KeySpaceManagerProtocolProtos.RenameKeyRequest;
+import org.apache.hadoop.ozone.protocol.proto
+    .KeySpaceManagerProtocolProtos.RenameKeyResponse;
 import org.apache.hadoop.ozone.protocol.proto
     .KeySpaceManagerProtocolProtos.KeyArgs;
 import org.apache.hadoop.ozone.protocol.proto
@@ -152,6 +156,8 @@ public class KeySpaceManagerProtocolServerSideTranslatorPB implements
         return Status.KEY_ALREADY_EXISTS;
       case FAILED_KEY_NOT_FOUND:
         return Status.KEY_NOT_FOUND;
+      case FAILED_INVALID_KEY_NAME:
+        return Status.INVALID_KEY_NAME;
       default:
         return Status.INTERNAL_ERROR;
       }
@@ -372,6 +378,26 @@ public class KeySpaceManagerProtocolServerSideTranslatorPB implements
     return resp.build();
   }
 
+  @Override
+  public RenameKeyResponse renameKey(
+      RpcController controller, RenameKeyRequest request)
+      throws ServiceException {
+    RenameKeyResponse.Builder resp = RenameKeyResponse.newBuilder();
+    try {
+      KeyArgs keyArgs = request.getKeyArgs();
+      KsmKeyArgs ksmKeyArgs = new KsmKeyArgs.Builder()
+          .setVolumeName(keyArgs.getVolumeName())
+          .setBucketName(keyArgs.getBucketName())
+          .setKeyName(keyArgs.getKeyName())
+          .build();
+      impl.renameKey(ksmKeyArgs, request.getToKeyName());
+      resp.setStatus(Status.OK);
+    } catch (IOException e){
+      resp.setStatus(exceptionToResponseStatus(e));
+    }
+    return resp.build();
+  }
+
   @Override
   public SetBucketPropertyResponse setBucketProperty(
       RpcController controller, SetBucketPropertyRequest request)

hadoop-tools/hadoop-ozone/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java

@@ -257,30 +257,17 @@ public class OzoneFileSystem extends FileSystem {
 
     boolean processKey(String key) throws IOException {
       String newKeyName = dstKey.concat(key.substring(srcKey.length()));
-      rename(key, newKeyName);
+      bucket.renameKey(key, newKeyName);
       return true;
     }
-
-    // TODO: currently rename work by copying the streams, with changes in KSM,
-    // this operation can be improved by renaming the keys in KSM directly.
-    private void rename(String src, String dst) throws IOException {
-      try (OzoneInputStream inputStream = bucket.readKey(src);
-          OzoneOutputStream outputStream = bucket
-              .createKey(dst, 0, replicationType, replicationFactor)) {
-        IOUtils.copyBytes(inputStream, outputStream, getConf());
-      }
-    }
   }
 
   /**
    * Check whether the source and destination path are valid and then perform
-   * rename by copying the data from source path to destination path.
+   * rename from source path to destination path.
    *
-   * The rename operation is performed by copying data from source key
-   * to destination key. This is done by reading the source key data into a
-   * temporary file and then writing this temporary file to destination key.
-   * The temporary file is deleted after the rename operation.
-   * TODO: Optimize the operation by renaming keys in KSM.
+   * The rename operation is performed by renaming the keys with src as prefix.
+   * For such keys the prefix is changed from src to dst.
    *
    * @param src source path for rename
    * @param dst destination path for rename
@@ -290,8 +277,11 @@ public class OzoneFileSystem extends FileSystem {
    */
   @Override
   public boolean rename(Path src, Path dst) throws IOException {
-    LOG.trace("rename() from:{} to:{}", src, dst);
+    if (src.equals(dst)) {
+      return true;
+    }
 
+    LOG.trace("rename() from:{} to:{}", src, dst);
     if (src.isRoot()) {
       // Cannot rename root of file system
       LOG.trace("Cannot rename the root of a filesystem");
@@ -367,8 +357,7 @@ public class OzoneFileSystem extends FileSystem {
       }
     }
     RenameIterator iterator = new RenameIterator(src, dst);
-    iterator.iterate();
-    return src.equals(dst) || delete(src, true);
+    return iterator.iterate();
   }
 
   private class DeleteIterator extends OzoneListingIterator {