diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/shell/TestOzoneShellHA.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/shell/TestOzoneShellHA.java
index fac01fe7d014..0b91a364909c 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/shell/TestOzoneShellHA.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/shell/TestOzoneShellHA.java
@@ -45,6 +45,8 @@
 import org.apache.hadoop.test.GenericTestUtils;
 import org.apache.hadoop.test.LambdaTestUtils;
 import org.apache.hadoop.util.ToolRunner;
+import org.apache.hadoop.fs.TrashPolicy;
+import org.apache.hadoop.ozone.om.TrashPolicyOzone;
 
 import com.google.common.base.Strings;
 
@@ -496,6 +498,23 @@ private OzoneConfiguration getClientConfForOFS(
     return clientConf;
   }
 
+  /**
+   * Helper function to retrieve Ozone client configuration for ozone
+   * trash testing with TrashPolicyOzone.
+   * @param hostPrefix Scheme + Authority. e.g. ofs://om-service-test1
+   * @param configuration Server config to generate client config from.
+   * @return Config ofs configuration added with fs.trash.classname
+   * = TrashPolicyOzone.
+   */
+  private OzoneConfiguration getClientConfForOzoneTrashPolicy(
+          String hostPrefix, OzoneConfiguration configuration) {
+    OzoneConfiguration clientConf =
+            getClientConfForOFS(hostPrefix, configuration);
+    clientConf.setClass("fs.trash.classname", TrashPolicyOzone.class,
+            TrashPolicy.class);
+    return clientConf;
+  }
+
   @Test
   public void testDeleteToTrashOrSkipTrash() throws Exception {
     final String hostPrefix = OZONE_OFS_URI_SCHEME + "://" + omServiceId;
@@ -555,9 +574,89 @@ public void testDeleteToTrashOrSkipTrash() throws Exception {
       }
     } finally {
       shell.close();
+      fs.close();
     }
   }
 
+  @Test
+  public void testDeleteTrashNoSkipTrash() throws Exception {
+
+    // Test delete from Trash directory removes item from filesystem
+
+    // setup configuration to use TrashPolicyOzone
+    // (default is TrashPolicyDefault)
+    final String hostPrefix = OZONE_OFS_URI_SCHEME + "://" + omServiceId;
+    OzoneConfiguration clientConf =
+            getClientConfForOzoneTrashPolicy(hostPrefix, conf);
+    OzoneFsShell shell = new OzoneFsShell(clientConf);
+
+    int res;
+
+    // create volume: vol1 with bucket: bucket1
+    final String testVolBucket = "/vol1/bucket1";
+    final String testKey = testVolBucket+"/key1";
+
+    final String[] volBucketArgs = new String[] {"-mkdir", "-p", testVolBucket};
+    final String[] keyArgs = new String[] {"-touch", testKey};
+    final String[] listArgs = new String[] {"key", "list", testVolBucket};
+
+    LOG.info("Executing testDeleteTrashNoSkipTrash: FsShell with args {}",
+            Arrays.asList(volBucketArgs));
+    res = ToolRunner.run(shell, volBucketArgs);
+    Assert.assertEquals(0, res);
+
+    // create key: key1 belonging to bucket1
+    res = ToolRunner.run(shell, keyArgs);
+    Assert.assertEquals(0, res);
+
+    // check directory listing for bucket1 contains 1 key
+    out.reset();
+    execute(ozoneShell, listArgs);
+    Assert.assertEquals(1, getNumOfKeys());
+
+    // Test deleting items in trash are discarded (removed from filesystem)
+    // 1.) remove key1 from bucket1 with fs shell rm command
+    // 2.) on rm, item is placed in Trash
+    // 3.) remove Trash directory and all contents, 
+    //     check directory listing = 0 items
+
+    final String[] rmKeyArgs = new String[] {"-rm", "-R", testKey};
+    final String[] rmTrashArgs = new String[] {"-rm", "-R",
+                                               testVolBucket+"/.Trash"};
+    final Path trashPathKey1 = Path.mergePaths(new Path(
+            new OFSPath(testKey).getTrashRoot(), new Path("Current")),
+            new Path(testKey));
+    FileSystem fs = FileSystem.get(clientConf);
+
+    try {
+      // on delete key, item is placed in trash
+      LOG.info("Executing testDeleteTrashNoSkipTrash: FsShell with args {}",
+              Arrays.asList(rmKeyArgs));
+      res = ToolRunner.run(shell, rmKeyArgs);
+      Assert.assertEquals(0, res);
+
+      LOG.info("Executing testDeleteTrashNoSkipTrash: key1 deleted moved to"
+              +" Trash: "+trashPathKey1.toString());
+      fs.getFileStatus(trashPathKey1);
+
+      LOG.info("Executing testDeleteTrashNoSkipTrash: deleting trash FsShell "
+              +"with args{}: ", Arrays.asList(rmTrashArgs));
+      res = ToolRunner.run(shell, rmTrashArgs);
+      Assert.assertEquals(0, res);
+
+      out.reset();
+      // once trash is is removed, trash should be deleted from filesystem
+      execute(ozoneShell, listArgs);
+      Assert.assertEquals(0, getNumOfKeys());
+
+    } finally {
+      shell.close();
+      fs.close();
+    }
+
+  }
+
+
   @Test
   @SuppressWarnings("methodlength")
   public void testShQuota() throws Exception {
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/TrashPolicyOzone.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/TrashPolicyOzone.java
index e751a982802a..83678c6afd4f 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/TrashPolicyOzone.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/TrashPolicyOzone.java
@@ -36,8 +36,10 @@
 import org.apache.hadoop.fs.FileAlreadyExistsException;
 import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.fs.InvalidPathException;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.ozone.conf.OMClientConfig;
+import org.apache.hadoop.ozone.om.helpers.OzoneFSUtils;
 import org.apache.hadoop.util.Time;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -118,6 +120,28 @@ public Runnable getEmptier() throws IOException {
         emptierInterval);
   }
 
+  @Override
+  public boolean moveToTrash(Path path) throws IOException {
+    this.fs.getFileStatus(path);
+    Path trashRoot = this.fs.getTrashRoot(path);
+
+    String key = path.toUri().getPath();
+    LOG.debug("Key path to moveToTrash: "+ key);
+    String trashRootKey = trashRoot.toUri().getPath();
+    LOG.debug("TrashrootKey for moveToTrash: "+ trashRootKey);
+
+    if (!OzoneFSUtils.isValidName(key)) {
+      throw new InvalidPathException("Invalid path Name " + key);
+    }
+    // first condition tests when length key is <= length trash
+    // and second when length key > length trash
+    if ((key.contains(this.fs.TRASH_PREFIX)) && (trashRootKey.startsWith(key))
+            || key.startsWith(trashRootKey)) {
+      return false;
+    } else {
+      return super.moveToTrash(path);
+    }
+  }
 
   protected class Emptier implements Runnable {