HDFS-11716. Ozone: SCM: CLI: Revisit delete container API. Contributed by Weiwei Yang.

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/client/ContainerOperationClient.java

@@ -141,6 +141,8 @@ public class ContainerOperationClient implements ScmClient {
       client = xceiverClientManager.acquireClient(pipeline);
       String traceID = UUID.randomUUID().toString();
       ContainerProtocolCalls.deleteContainer(client, force, traceID);
+      storageContainerLocationClient
+          .deleteContainer(pipeline.getContainerName());
       LOG.info("Deleted container {}, leader: {}, machines: {} ",
           pipeline.getContainerName(),
           pipeline.getLeader(),

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/protocol/StorageContainerLocationProtocol.java

@@ -72,4 +72,13 @@ public interface StorageContainerLocationProtocol {
    */
   Pipeline getContainer(String containerName) throws IOException;
 
+  /**
+   * Deletes a container in SCM.
+   *
+   * @param containerName
+   * @throws IOException
+   *   if failed to delete the container mapping from db store
+   *   or container doesn't exist.
+   */
+  void deleteContainer(String containerName) throws IOException;
 }

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/protocolPB/StorageContainerLocationProtocolClientSideTranslatorPB.java

@@ -17,6 +17,7 @@
 package org.apache.hadoop.scm.protocolPB;
 
 import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
 import com.google.common.collect.Sets;
 import com.google.protobuf.RpcController;
 import com.google.protobuf.ServiceException;
@@ -37,6 +38,7 @@ import org.apache.hadoop.ozone.protocol.proto.StorageContainerLocationProtocolPr
 import org.apache.hadoop.ozone.protocol.proto.StorageContainerLocationProtocolProtos.LocatedContainerProto;
 import org.apache.hadoop.ozone.protocol.proto.StorageContainerLocationProtocolProtos.GetContainerRequestProto;
 import org.apache.hadoop.ozone.protocol.proto.StorageContainerLocationProtocolProtos.GetContainerResponseProto;
+import org.apache.hadoop.ozone.protocol.proto.StorageContainerLocationProtocolProtos.DeleteContainerRequestProto;
 import org.apache.hadoop.scm.container.common.helpers.Pipeline;
 
 import java.io.Closeable;
@@ -166,6 +168,29 @@ public final class StorageContainerLocationProtocolClientSideTranslatorPB
     }
   }
 
+  /**
+   * Ask SCM to delete a container by name. SCM will remove
+   * the container mapping in its database.
+   *
+   * @param containerName
+   * @throws IOException
+   */
+  @Override
+  public void deleteContainer(String containerName)
+      throws IOException {
+    Preconditions.checkState(!Strings.isNullOrEmpty(containerName),
+        "Container name cannot be null or empty");
+    DeleteContainerRequestProto request = DeleteContainerRequestProto
+        .newBuilder()
+        .setContainerName(containerName)
+        .build();
+    try {
+      rpcProxy.deleteContainer(NULL_RPC_CONTROLLER, request);
+    } catch (ServiceException e) {
+      throw ProtobufHelper.getRemoteException(e);
+    }
+  }
+
   @Override
   public Object getUnderlyingProxyObject() {
     return rpcProxy;

hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/StorageContainerLocationProtocol.proto

@@ -92,6 +92,14 @@ message GetContainerResponseProto {
   required hadoop.hdfs.ozone.Pipeline pipeline = 1;
 }
 
+message DeleteContainerRequestProto {
+  required string containerName = 1;
+}
+
+message DeleteContainerResponseProto {
+  // Empty response
+}
+
 // SCM Block protocol
 /**
  * keys - batch of block keys to find
@@ -163,6 +171,11 @@ service StorageContainerLocationProtocolService {
    */
   rpc getContainer(GetContainerRequestProto) returns (GetContainerResponseProto);
 
+  /**
+   * Deletes a container in SCM.
+   */
+  rpc deleteContainer(DeleteContainerRequestProto) returns (DeleteContainerResponseProto);
+
   /**
    * Find the set of nodes that currently host the block, as
    * identified by the key.  This method supports batch lookup by

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/impl/Dispatcher.java

@@ -51,7 +51,6 @@ import static org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.Result
 import static org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.Result.GET_SMALL_FILE_ERROR;
 import static org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.Result.NO_SUCH_ALGORITHM;
 import static org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.Result.PUT_SMALL_FILE_ERROR;
-import static org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.Result.UNCLOSED_CONTAINER_IO;
 
 /**
  * Ozone Container dispatcher takes a call from the netty server and routes it
@@ -362,22 +361,11 @@ public class Dispatcher implements ContainerDispatcher {
       return ContainerUtils.malformedRequest(msg);
     }
 
-    String name = msg.getDeleteContainer().getName();
-    boolean forceDelete = msg.getDeleteContainer().getForceDelete();
     Pipeline pipeline = Pipeline.getFromProtoBuf(
         msg.getDeleteContainer().getPipeline());
     Preconditions.checkNotNull(pipeline);
-
-    // If this cmd requires force deleting, then we should
-    // make sure the container is in the state of closed,
-    // otherwise, stop deleting container.
-    if (forceDelete) {
-      if (this.containerManager.isOpen(pipeline.getContainerName())) {
-        throw new StorageContainerException("Attempting to force delete "
-            + "an open container.", UNCLOSED_CONTAINER_IO);
-      }
-    }
-
+    String name = msg.getDeleteContainer().getName();
+    boolean forceDelete = msg.getDeleteContainer().getForceDelete();
     this.containerManager.deleteContainer(pipeline, name, forceDelete);
     return ContainerUtils.getContainerResponse(msg);
   }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerLocationProtocolServerSideTranslatorPB.java

@@ -58,6 +58,10 @@ import org.apache.hadoop.ozone.protocol.proto
     .StorageContainerLocationProtocolProtos.GetContainerRequestProto;
 import org.apache.hadoop.ozone.protocol.proto
     .StorageContainerLocationProtocolProtos.GetContainerResponseProto;
+import org.apache.hadoop.ozone.protocol.proto
+    .StorageContainerLocationProtocolProtos.DeleteContainerRequestProto;
+import org.apache.hadoop.ozone.protocol.proto
+    .StorageContainerLocationProtocolProtos.DeleteContainerResponseProto;
 import org.apache.hadoop.scm.container.common.helpers.Pipeline;
 import org.apache.hadoop.scm.protocolPB.StorageContainerLocationProtocolPB;
 
@@ -148,6 +152,18 @@ public final class StorageContainerLocationProtocolServerSideTranslatorPB
     }
   }
 
+  @Override
+  public DeleteContainerResponseProto deleteContainer(
+      RpcController controller, DeleteContainerRequestProto request)
+      throws ServiceException {
+    try {
+      impl.deleteContainer(request.getContainerName());
+      return DeleteContainerResponseProto.newBuilder().build();
+    } catch (IOException e) {
+      throw new ServiceException(e);
+    }
+  }
+
   @Override
   public GetScmBlockLocationsResponseProto getScmBlockLocations(
       RpcController controller, GetScmBlockLocationsRequestProto req)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/StorageContainerManager.java

@@ -373,6 +373,14 @@ public class StorageContainerManager
     return scmContainerManager.getContainer(containerName);
   }
 
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public void deleteContainer(String containerName) throws IOException {
+    scmContainerManager.deleteContainer(containerName);
+  }
+
   /**
    * Asks SCM where a container should be allocated. SCM responds with the set
    * of datanodes that should be used creating this container.

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/container/ContainerMapping.java

@@ -237,6 +237,31 @@ public class ContainerMapping implements Mapping {
     return pipeline;
   }
 
+  /**
+   * Deletes a container from SCM.
+   *
+   * @param containerName - Container name
+   * @throws IOException
+   *   if container doesn't exist
+   *   or container store failed to delete the specified key.
+   */
+  @Override
+  public void deleteContainer(String containerName) throws IOException {
+    lock.lock();
+    try {
+      byte[] dbKey = containerName.getBytes(encoding);
+      byte[] pipelineBytes =
+          containerStore.get(dbKey);
+      if(pipelineBytes == null) {
+        throw new IOException("Failed to delete container "
+            + containerName + ", reason : container doesn't exist.");
+      }
+      containerStore.delete(dbKey);
+    } finally {
+      lock.unlock();
+    }
+  }
+
   /**
    * Closes this stream and releases any system resources associated with it. If
    * the stream is already closed then invoking this method has no effect.

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/container/Mapping.java

@@ -56,4 +56,12 @@ public interface Mapping extends Closeable {
    */
   Pipeline allocateContainer(String containerName,
       ScmClient.ReplicationFactor replicationFactor) throws IOException;
+
+  /**
+   * Deletes a container from SCM.
+   *
+   * @param containerName - Container Name
+   * @throws IOException
+   */
+  void deleteContainer(String containerName) throws IOException;
 }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerPersistence.java

@@ -206,12 +206,14 @@ public class TestContainerPersistence {
     data.addMetadata("owner)", "bilbo");
     containerManager.createContainer(createSingleNodePipeline(containerName1),
         data);
+    containerManager.closeContainer(containerName1);
 
     data = new ContainerData(containerName2);
     data.addMetadata("VOLUME", "shire");
     data.addMetadata("owner)", "bilbo");
     containerManager.createContainer(createSingleNodePipeline(containerName2),
         data);
+    containerManager.closeContainer(containerName2);
 
     Assert.assertTrue(containerManager.getContainerMap()
         .containsKey(containerName1));
@@ -231,6 +233,7 @@ public class TestContainerPersistence {
     data.addMetadata("owner)", "bilbo");
     containerManager.createContainer(createSingleNodePipeline(containerName1),
         data);
+    containerManager.closeContainer(containerName1);
 
     // Assert we still have both containers.
     Assert.assertTrue(containerManager.getContainerMap()

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainer.java

@@ -421,16 +421,6 @@ public class TestOzoneContainer {
       Assert.assertTrue(
           putKeyRequest.getTraceID().equals(response.getTraceID()));
 
-      // Container cannot be deleted because the container is not empty.
-      request = ContainerTestHelper.getDeleteContainer(
-          client.getPipeline(), false);
-      response = client.sendCommand(request);
-
-      Assert.assertNotNull(response);
-      Assert.assertEquals(ContainerProtos.Result.ERROR_CONTAINER_NOT_EMPTY,
-          response.getResult());
-      Assert.assertTrue(request.getTraceID().equals(response.getTraceID()));
-
       // Container cannot be deleted forcibly because
       // the container is not closed.
       request = ContainerTestHelper.getDeleteContainer(
@@ -449,8 +439,18 @@ public class TestOzoneContainer {
       Assert.assertEquals(ContainerProtos.Result.SUCCESS, response.getResult());
       Assert.assertTrue(request.getTraceID().equals(response.getTraceID()));
 
+      // Container cannot be deleted because the container is not empty.
+      request = ContainerTestHelper.getDeleteContainer(
+          client.getPipeline(), false);
+      response = client.sendCommand(request);
+
+      Assert.assertNotNull(response);
+      Assert.assertEquals(ContainerProtos.Result.ERROR_CONTAINER_NOT_EMPTY,
+          response.getResult());
+      Assert.assertTrue(request.getTraceID().equals(response.getTraceID()));
+
       // Container can be deleted forcibly because
-      // it has been closed.
+      // it is closed and non-empty.
       request = ContainerTestHelper.getDeleteContainer(
           client.getPipeline(), true);
       response = client.sendCommand(request);

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/scm/TestSCMCli.java

@@ -31,6 +31,7 @@ import org.apache.hadoop.scm.client.ScmClient;
 import org.apache.hadoop.scm.container.common.helpers.Pipeline;
 import org.apache.hadoop.scm.protocolPB.StorageContainerLocationProtocolClientSideTranslatorPB;
 import org.junit.AfterClass;
+import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
@@ -120,58 +121,97 @@ public class TestSCMCli {
     assertEquals(containerName, container.getContainerName());
   }
 
+  private boolean containerExist(String containerName) {
+    try {
+      Pipeline scmPipeline = scm.getContainer(containerName);
+      return scmPipeline != null
+          && containerName.equals(scmPipeline.getContainerName());
+    } catch (IOException e) {
+      return false;
+    }
+  }
 
   @Test
   public void testDeleteContainer() throws Exception {
-    final String cname1 = "cname1";
-    final String cname2 = "cname2";
+    String containerName;
+    ContainerData containerData;
+    Pipeline pipeline;
+    String[] delCmd;
+    ByteArrayOutputStream testErr;
+    int exitCode;
 
     // ****************************************
     // 1. Test to delete a non-empty container.
     // ****************************************
     // Create an non-empty container
-    Pipeline pipeline1 = scm.allocateContainer(cname1);
-    ContainerData data1 = new ContainerData(cname1);
-    containerManager.createContainer(pipeline1, data1);
-    ContainerData cdata = containerManager.readContainer(cname1);
-    KeyUtils.getDB(cdata, conf).put(cname1.getBytes(),
+    containerName = "non-empty-container";
+    pipeline = scm.allocateContainer(containerName);
+    containerData = new ContainerData(containerName);
+    containerManager.createContainer(pipeline, containerData);
+    ContainerData cdata = containerManager.readContainer(containerName);
+    KeyUtils.getDB(cdata, conf).put(containerName.getBytes(),
         "someKey".getBytes());
+    Assert.assertTrue(containerExist(containerName));
 
-    // Gracefully delete a container should fail because it is not empty.
-    String[] del1 = {"-container", "-del", cname1};
-    ByteArrayOutputStream testErr1 = new ByteArrayOutputStream();
-    int exitCode1 = runCommandAndGetOutput(del1, null, testErr1);
-    assertEquals(ResultCode.EXECUTION_ERROR, exitCode1);
-    assertTrue(testErr1.toString()
-        .contains("Container cannot be deleted because it is not empty."));
+    // Gracefully delete a container should fail because it is open.
+    delCmd = new String[] {"-container", "-del", containerName};
+    testErr = new ByteArrayOutputStream();
+    exitCode = runCommandAndGetOutput(delCmd, null, testErr);
+    assertEquals(ResultCode.EXECUTION_ERROR, exitCode);
+    assertTrue(testErr.toString()
+        .contains("Deleting an open container is not allowed."));
+    Assert.assertTrue(containerExist(containerName));
 
-    // Delete should fail when attempts to delete an open container.
-    // Even with the force tag.
-    String[] del2 = {"-container", "-del", cname1, "-f"};
-    ByteArrayOutputStream testErr2 = new ByteArrayOutputStream();
-    int exitCode2 = runCommandAndGetOutput(del2, null, testErr2);
-    assertEquals(ResultCode.EXECUTION_ERROR, exitCode2);
-    assertTrue(testErr2.toString()
-        .contains("Attempting to force delete an open container."));
+    // Close the container
+    containerManager.closeContainer(containerName);
 
-    // Close the container and try force delete again.
-    containerManager.closeContainer(cname1);
-    int exitCode3 = runCommandAndGetOutput(del2, null, null);
-    assertEquals(ResultCode.SUCCESS, exitCode3);
+    // Gracefully delete a container should fail because it is not empty.
+    testErr = new ByteArrayOutputStream();
+    int exitCode2 = runCommandAndGetOutput(delCmd, null, testErr);
+    assertEquals(ResultCode.EXECUTION_ERROR, exitCode2);
+    assertTrue(testErr.toString()
+        .contains("Container cannot be deleted because it is not empty."));
+    Assert.assertTrue(containerExist(containerName));
 
+    // Try force delete again.
+    delCmd = new String[] {"-container", "-del", containerName, "-f"};
+    exitCode = runCommandAndGetOutput(delCmd, null, null);
+    assertEquals(ResultCode.SUCCESS, exitCode);
+    Assert.assertFalse(containerExist(containerName));
 
     // ****************************************
     // 2. Test to delete an empty container.
     // ****************************************
     // Create an empty container
-    Pipeline pipeline2 = scm.allocateContainer(cname2);
-    ContainerData data2 = new ContainerData(cname2);
-    containerManager.createContainer(pipeline2, data2);
+    containerName = "empty-container";
+    pipeline = scm.allocateContainer(containerName);
+    containerData = new ContainerData(containerName);
+    containerManager.createContainer(pipeline, containerData);
+    containerManager.closeContainer(containerName);
+    Assert.assertTrue(containerExist(containerName));
 
     // Successfully delete an empty container.
-    String[] del3 = {"-container", "-del", cname2};
-    int exitCode4 = runCommandAndGetOutput(del3, null, null);
-    assertEquals(ResultCode.SUCCESS, exitCode4);
+    delCmd = new String[] {"-container", "-del", containerName};
+    exitCode = runCommandAndGetOutput(delCmd, null, null);
+    assertEquals(ResultCode.SUCCESS, exitCode);
+    Assert.assertFalse(containerExist(containerName));
+
+    // After the container is deleted,
+    // a same name container can now be recreated.
+    pipeline = scm.allocateContainer(containerName);
+    containerManager.createContainer(pipeline, containerData);
+    Assert.assertTrue(containerExist(containerName));
+
+    // ****************************************
+    // 3. Test to delete a non-exist container.
+    // ****************************************
+    containerName = "non-exist-container";
+    delCmd = new String[] {"-container", "-del", containerName};
+    testErr = new ByteArrayOutputStream();
+    exitCode = runCommandAndGetOutput(delCmd, null, testErr);
+    assertEquals(ResultCode.EXECUTION_ERROR, exitCode);
+    assertTrue(testErr.toString()
+        .contains("Specified key does not exist."));
   }
 
   @Test