HDDS-12646. Improve OM decommission check (apache#8122)

Author: jojochuang

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestAddRemoveOzoneManager.java

@@ -351,11 +351,18 @@ public void testForceBootstrap() throws Exception {
   @Test
   public void testDecommission() throws Exception {
     setupCluster(3);
-    user = UserGroupInformation.getCurrentUser();
 
-    // Stop the 3rd OM and decommission it
+    user = UserGroupInformation.createUserForTesting("user", new String[]{});
+    // Stop the 3rd OM and decommission it using non-privileged user
     String omNodeId3 = cluster.getOzoneManager(2).getOMNodeId();
     cluster.stopOzoneManager(omNodeId3);
+    // decommission should fail
+    assertThrows(IOException.class, () -> decommissionOM(omNodeId3));
+
+    // Switch to admin user
+    user = UserGroupInformation.getCurrentUser();
+    // Stop the 3rd OM and decommission it
+    cluster.stopOzoneManager(omNodeId3);
     decommissionOM(omNodeId3);
 
     // Decommission the non leader OM and then stop it. Stopping OM before will

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

@@ -17,12 +17,16 @@
 
 package org.apache.hadoop.ozone.protocolPB;
 
+import static org.apache.hadoop.hdds.utils.HddsServerUtil.getRemoteUser;
+import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.PERMISSION_DENIED;
+
 import com.google.protobuf.RpcController;
 import com.google.protobuf.ServiceException;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.helpers.OMNodeDetails;
 import org.apache.hadoop.ozone.om.protocolPB.OMAdminProtocolPB;
 import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer;
@@ -91,6 +95,9 @@ public DecommissionOMResponse decommission(RpcController controller,
     }
 
     try {
+      if (!ozoneManager.isAdmin(getRemoteUser())) {
+        throw new OMException("Only administrators are authorized to perform decommission.", PERMISSION_DENIED);
+      }
       omRatisServer.removeOMFromRatisRing(decommNode);
     } catch (IOException ex) {
       return DecommissionOMResponse.newBuilder()