Verify cluster name in remove cluster call (#3715)

* Verify cluster name in remove cluster call

Author: yux0

service/frontend/operator_handler.go

@@ -425,6 +425,17 @@ func (h *OperatorHandlerImpl) RemoveRemoteCluster(
 	scope, startTime := h.startRequestProfile(metrics.OperatorRemoveRemoteClusterScope)
 	defer func() { scope.Timer(metrics.ServiceLatency.GetMetricName()).Record(time.Since(startTime)) }()
 
+	var isClusterNameExist bool
+	for clusterName := range h.clusterMetadata.GetAllClusterInfo() {
+		if clusterName == request.GetClusterName() {
+			isClusterNameExist = true
+			break
+		}
+	}
+	if !isClusterNameExist {
+		return nil, serviceerror.NewNotFound("The cluster to be deleted cannot be found in clusters cache.")
+	}
+
 	if err := h.clusterMetadataManager.DeleteClusterMetadata(
 		ctx,
 		&persistence.DeleteClusterMetadataRequest{ClusterName: request.GetClusterName()},

service/frontend/operator_handler_test.go

@@ -469,6 +469,7 @@ func (s *operatorHandlerSuite) Test_DeleteNamespace() {
 
 func (s *operatorHandlerSuite) Test_RemoveRemoteCluster_Success() {
 	var clusterName = "cluster"
+	s.mockResource.ClusterMetadata.EXPECT().GetAllClusterInfo().Return(map[string]cluster.ClusterInformation{clusterName: {}})
 	s.mockResource.ClusterMetadataMgr.EXPECT().DeleteClusterMetadata(
 		gomock.Any(),
 		&persistence.DeleteClusterMetadataRequest{ClusterName: clusterName},
@@ -480,6 +481,7 @@ func (s *operatorHandlerSuite) Test_RemoveRemoteCluster_Success() {
 
 func (s *operatorHandlerSuite) Test_RemoveRemoteCluster_Error() {
 	var clusterName = "cluster"
+	s.mockResource.ClusterMetadata.EXPECT().GetAllClusterInfo().Return(map[string]cluster.ClusterInformation{clusterName: {}})
 	s.mockResource.ClusterMetadataMgr.EXPECT().DeleteClusterMetadata(
 		gomock.Any(),
 		&persistence.DeleteClusterMetadataRequest{ClusterName: clusterName},