feat(ui): Batch add & remove Owners to assets via the UI (#5552)

2025-12-24 08:28:12 +00:00 · 2022-08-03 17:02:37 -07:00 · 2022-08-03 17:02:37 -07:00 · 76b40b0946
commit 76b40b0946
parent f1abdc91ee
18 changed files with 939 additions and 92 deletions
--- a/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/GmsGraphQLEngine.java
+++ b/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/GmsGraphQLEngine.java
@ -144,8 +144,10 @@ import com.linkedin.datahub.graphql.resolvers.mutate.AddTagResolver;
 import com.linkedin.datahub.graphql.resolvers.mutate.AddTagsResolver;
 import com.linkedin.datahub.graphql.resolvers.mutate.AddTermResolver;
 import com.linkedin.datahub.graphql.resolvers.mutate.AddTermsResolver;
+import com.linkedin.datahub.graphql.resolvers.mutate.BatchAddOwnersResolver;
 import com.linkedin.datahub.graphql.resolvers.mutate.BatchAddTagsResolver;
 import com.linkedin.datahub.graphql.resolvers.mutate.BatchAddTermsResolver;
+import com.linkedin.datahub.graphql.resolvers.mutate.BatchRemoveOwnersResolver;
 import com.linkedin.datahub.graphql.resolvers.mutate.BatchRemoveTagsResolver;
 import com.linkedin.datahub.graphql.resolvers.mutate.BatchRemoveTermsResolver;
 import com.linkedin.datahub.graphql.resolvers.mutate.MutableTypeResolver;
@ -704,7 +706,9 @@ public class GmsGraphQLEngine {
            .dataFetcher("updateDescription", new UpdateDescriptionResolver(entityService))
            .dataFetcher("addOwner", new AddOwnerResolver(entityService))
            .dataFetcher("addOwners", new AddOwnersResolver(entityService))
+            .dataFetcher("batchAddOwners", new BatchAddOwnersResolver(entityService))
            .dataFetcher("removeOwner", new RemoveOwnerResolver(entityService))
+            .dataFetcher("batchRemoveOwners", new BatchRemoveOwnersResolver(entityService))
            .dataFetcher("addLink", new AddLinkResolver(entityService))
            .dataFetcher("removeLink", new RemoveLinkResolver(entityService))
            .dataFetcher("addGroupMembers", new AddGroupMembersResolver(this.groupService))
--- a/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/mutate/AddOwnerResolver.java
+++ b/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/mutate/AddOwnerResolver.java
@ -1,5 +1,6 @@
 package com.linkedin.datahub.graphql.resolvers.mutate;

+import com.google.common.collect.ImmutableList;
 import com.linkedin.common.urn.CorpuserUrn;

 import com.linkedin.common.urn.Urn;
@ -7,7 +8,9 @@ import com.linkedin.datahub.graphql.QueryContext;
 import com.linkedin.datahub.graphql.exception.AuthorizationException;
 import com.linkedin.datahub.graphql.generated.AddOwnerInput;
 import com.linkedin.datahub.graphql.generated.OwnerEntityType;
+import com.linkedin.datahub.graphql.generated.OwnerInput;
 import com.linkedin.datahub.graphql.generated.OwnershipType;
+import com.linkedin.datahub.graphql.generated.ResourceRefInput;
 import com.linkedin.datahub.graphql.resolvers.mutate.util.OwnerUtils;
 import com.linkedin.metadata.entity.EntityService;
 import graphql.schema.DataFetcher;
@ -50,11 +53,9 @@ public class AddOwnerResolver implements DataFetcher<CompletableFuture<Boolean>>
        log.debug("Adding Owner. input: {}", input.toString());

        Urn actor = CorpuserUrn.createFromString(((QueryContext) environment.getContext()).getActorUrn());
-        OwnerUtils.addOwner(
-            ownerUrn,
-            // Assumption Alert: Assumes that GraphQL ownership type === GMS ownership type
-            com.linkedin.common.OwnershipType.valueOf(type.name()),
-            targetUrn,
+        OwnerUtils.addOwnersToResources(
+            ImmutableList.of(new OwnerInput(input.getOwnerUrn(), ownerEntityType, type)),
+            ImmutableList.of(new ResourceRefInput(input.getResourceUrn(), null, null)),
            actor,
            _entityService
        );
--- a/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/mutate/AddOwnersResolver.java
+++ b/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/mutate/AddOwnersResolver.java
@ -1,5 +1,6 @@
 package com.linkedin.datahub.graphql.resolvers.mutate;

+import com.google.common.collect.ImmutableList;
 import com.linkedin.common.urn.CorpuserUrn;

 import com.linkedin.common.urn.Urn;
@ -7,6 +8,7 @@ import com.linkedin.datahub.graphql.QueryContext;
 import com.linkedin.datahub.graphql.exception.AuthorizationException;
 import com.linkedin.datahub.graphql.generated.AddOwnersInput;
 import com.linkedin.datahub.graphql.generated.OwnerInput;
+import com.linkedin.datahub.graphql.generated.ResourceRefInput;
 import com.linkedin.datahub.graphql.resolvers.mutate.util.OwnerUtils;
 import com.linkedin.metadata.entity.EntityService;
 import graphql.schema.DataFetcher;
@ -47,9 +49,9 @@ public class AddOwnersResolver implements DataFetcher<CompletableFuture<Boolean>
        log.debug("Adding Owners. input: {}", input.toString());

        Urn actor = CorpuserUrn.createFromString(((QueryContext) environment.getContext()).getActorUrn());
-        OwnerUtils.addOwners(
+        OwnerUtils.addOwnersToResources(
            owners,
-            targetUrn,
+            ImmutableList.of(new ResourceRefInput(input.getResourceUrn(), null, null)),
            actor,
            _entityService
        );
--- a/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/mutate/BatchAddOwnersResolver.java
+++ b/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/mutate/BatchAddOwnersResolver.java
@ -0,0 +1,90 @@
+package com.linkedin.datahub.graphql.resolvers.mutate;
+
+import com.linkedin.common.urn.Urn;
+import com.linkedin.common.urn.UrnUtils;
+import com.linkedin.datahub.graphql.QueryContext;
+import com.linkedin.datahub.graphql.exception.AuthorizationException;
+import com.linkedin.datahub.graphql.generated.BatchAddOwnersInput;
+import com.linkedin.datahub.graphql.generated.OwnerInput;
+import com.linkedin.datahub.graphql.generated.ResourceRefInput;
+import com.linkedin.datahub.graphql.resolvers.mutate.util.LabelUtils;
+import com.linkedin.datahub.graphql.resolvers.mutate.util.OwnerUtils;
+import com.linkedin.metadata.entity.EntityService;
+import graphql.schema.DataFetcher;
+import graphql.schema.DataFetchingEnvironment;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import static com.linkedin.datahub.graphql.resolvers.ResolverUtils.*;
+
+
+@Slf4j
+@RequiredArgsConstructor
+public class BatchAddOwnersResolver implements DataFetcher<CompletableFuture<Boolean>> {
+
+  private final EntityService _entityService;
+
+  @Override
+  public CompletableFuture<Boolean> get(DataFetchingEnvironment environment) throws Exception {
+    final BatchAddOwnersInput input = bindArgument(environment.getArgument("input"), BatchAddOwnersInput.class);
+    final List<OwnerInput> owners = input.getOwners();
+    final List<ResourceRefInput> resources = input.getResources();
+    final QueryContext context = environment.getContext();
+
+    return CompletableFuture.supplyAsync(() -> {
+
+      // First, validate the batch
+      validateOwners(owners);
+      validateInputResources(resources, context);
+
+      try {
+        // Then execute the bulk add
+        batchAddOwners(owners, resources, context);
+        return true;
+      } catch (Exception e) {
+        log.error("Failed to perform update against input {}, {}", input.toString(), e.getMessage());
+        throw new RuntimeException(String.format("Failed to perform update against input %s", input.toString()), e);
+      }
+    });
+  }
+
+  private void validateOwners(List<OwnerInput> owners) {
+    for (OwnerInput ownerInput : owners) {
+      OwnerUtils.validateOwner(UrnUtils.getUrn(ownerInput.getOwnerUrn()), ownerInput.getOwnerEntityType(), _entityService);
+    }
+  }
+
+  private void validateInputResources(List<ResourceRefInput> resources, QueryContext context) {
+    for (ResourceRefInput resource : resources) {
+      validateInputResource(resource, context);
+    }
+  }
+
+  private void validateInputResource(ResourceRefInput resource, QueryContext context) {
+    final Urn resourceUrn = UrnUtils.getUrn(resource.getResourceUrn());
+
+    if (resource.getSubResource() != null) {
+      throw new IllegalArgumentException("Malformed input provided: owners cannot be applied to subresources.");
+    }
+
+    if (!OwnerUtils.isAuthorizedToUpdateOwners(context, resourceUrn)) {
+      throw new AuthorizationException("Unauthorized to perform this action. Please contact your DataHub administrator.");
+    }
+    LabelUtils.validateResource(resourceUrn, resource.getSubResource(), resource.getSubResourceType(), _entityService);
+  }
+
+  private void batchAddOwners(List<OwnerInput> owners, List<ResourceRefInput> resources, QueryContext context) {
+    log.debug("Batch adding owners. owners: {}, resources: {}", owners, resources);
+    try {
+      OwnerUtils.addOwnersToResources(owners, resources, UrnUtils.getUrn(context.getActorUrn()), _entityService);
+    } catch (Exception e) {
+      throw new RuntimeException(String.format("Failed to batch add Owners %s to resources with urns %s!",
+          owners,
+          resources.stream().map(ResourceRefInput::getResourceUrn).collect(Collectors.toList())),
+          e);
+    }
+  }
+}
--- a/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/mutate/BatchRemoveOwnersResolver.java
+++ b/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/mutate/BatchRemoveOwnersResolver.java
@ -0,0 +1,83 @@
+package com.linkedin.datahub.graphql.resolvers.mutate;
+
+import com.linkedin.common.urn.Urn;
+import com.linkedin.common.urn.UrnUtils;
+import com.linkedin.datahub.graphql.QueryContext;
+import com.linkedin.datahub.graphql.exception.AuthorizationException;
+import com.linkedin.datahub.graphql.generated.BatchRemoveOwnersInput;
+import com.linkedin.datahub.graphql.generated.ResourceRefInput;
+import com.linkedin.datahub.graphql.resolvers.mutate.util.LabelUtils;
+import com.linkedin.datahub.graphql.resolvers.mutate.util.OwnerUtils;
+import com.linkedin.metadata.entity.EntityService;
+import graphql.schema.DataFetcher;
+import graphql.schema.DataFetchingEnvironment;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import static com.linkedin.datahub.graphql.resolvers.ResolverUtils.*;
+
+
+@Slf4j
+@RequiredArgsConstructor
+public class BatchRemoveOwnersResolver implements DataFetcher<CompletableFuture<Boolean>> {
+
+  private final EntityService _entityService;
+
+  @Override
+  public CompletableFuture<Boolean> get(DataFetchingEnvironment environment) throws Exception {
+    final BatchRemoveOwnersInput input = bindArgument(environment.getArgument("input"), BatchRemoveOwnersInput.class);
+    final List<String> owners = input.getOwnerUrns();
+    final List<ResourceRefInput> resources = input.getResources();
+    final QueryContext context = environment.getContext();
+
+    return CompletableFuture.supplyAsync(() -> {
+
+      // First, validate the batch
+      validateInputResources(resources, context);
+
+      try {
+        // Then execute the bulk remove
+        batchRemoveOwners(owners, resources, context);
+        return true;
+      } catch (Exception e) {
+        log.error("Failed to perform update against input {}, {}", input.toString(), e.getMessage());
+        throw new RuntimeException(String.format("Failed to perform update against input %s", input.toString()), e);
+      }
+    });
+  }
+
+  private void validateInputResources(List<ResourceRefInput> resources, QueryContext context) {
+    for (ResourceRefInput resource : resources) {
+      validateInputResource(resource, context);
+    }
+  }
+
+  private void validateInputResource(ResourceRefInput resource, QueryContext context) {
+    final Urn resourceUrn = UrnUtils.getUrn(resource.getResourceUrn());
+
+    if (resource.getSubResource() != null) {
+      throw new IllegalArgumentException("Malformed input provided: owners cannot be removed from subresources.");
+    }
+
+    if (!OwnerUtils.isAuthorizedToUpdateOwners(context, resourceUrn)) {
+      throw new AuthorizationException("Unauthorized to perform this action. Please contact your DataHub administrator.");
+    }
+    LabelUtils.validateResource(resourceUrn, resource.getSubResource(), resource.getSubResourceType(), _entityService);
+  }
+
+  private void batchRemoveOwners(List<String> ownerUrns, List<ResourceRefInput> resources, QueryContext context) {
+    log.debug("Batch removing owners. owners: {}, resources: {}", ownerUrns, resources);
+    try {
+      OwnerUtils.removeOwnersFromResources(ownerUrns.stream().map(UrnUtils::getUrn).collect(
+          Collectors.toList()), resources, UrnUtils.getUrn(context.getActorUrn()), _entityService);
+    } catch (Exception e) {
+      throw new RuntimeException(String.format("Failed to batch remove Owners %s to resources with urns %s!",
+          ownerUrns,
+          resources.stream().map(ResourceRefInput::getResourceUrn).collect(Collectors.toList())),
+          e);
+    }
+  }
+}
--- a/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/mutate/RemoveOwnerResolver.java
+++ b/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/mutate/RemoveOwnerResolver.java
@ -1,10 +1,12 @@
 package com.linkedin.datahub.graphql.resolvers.mutate;

+import com.google.common.collect.ImmutableList;
 import com.linkedin.common.urn.CorpuserUrn;
 import com.linkedin.common.urn.Urn;
 import com.linkedin.datahub.graphql.QueryContext;
 import com.linkedin.datahub.graphql.exception.AuthorizationException;
 import com.linkedin.datahub.graphql.generated.RemoveOwnerInput;
+import com.linkedin.datahub.graphql.generated.ResourceRefInput;
 import com.linkedin.datahub.graphql.resolvers.mutate.util.OwnerUtils;
 import com.linkedin.metadata.entity.EntityService;
 import graphql.schema.DataFetcher;
@ -39,12 +41,10 @@ public class RemoveOwnerResolver implements DataFetcher<CompletableFuture<Boolea
          _entityService
      );
      try {
-        log.debug("Removing Link input: {}", input);
-
        Urn actor = CorpuserUrn.createFromString(((QueryContext) environment.getContext()).getActorUrn());
-        OwnerUtils.removeOwner(
-            ownerUrn,
-            targetUrn,
+        OwnerUtils.removeOwnersFromResources(
+            ImmutableList.of(ownerUrn),
+            ImmutableList.of(new ResourceRefInput(input.getResourceUrn(), null, null)),
            actor,
            _entityService
        );
--- a/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/mutate/util/OwnerUtils.java
+++ b/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/mutate/util/OwnerUtils.java
@ -16,10 +16,13 @@ import com.linkedin.datahub.graphql.authorization.ConjunctivePrivilegeGroup;
 import com.linkedin.datahub.graphql.authorization.DisjunctivePrivilegeGroup;
 import com.linkedin.datahub.graphql.generated.OwnerEntityType;
 import com.linkedin.datahub.graphql.generated.OwnerInput;
+import com.linkedin.datahub.graphql.generated.ResourceRefInput;
 import com.linkedin.datahub.graphql.resolvers.mutate.MutationUtils;
 import com.linkedin.metadata.Constants;
 import com.linkedin.metadata.authorization.PoliciesConfig;
 import com.linkedin.metadata.entity.EntityService;
+import com.linkedin.mxe.MetadataChangeProposal;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.stream.Collectors;
 import javax.annotation.Nonnull;
@ -36,28 +39,34 @@ public class OwnerUtils {

  private OwnerUtils() { }

-  public static void addOwner(
-      Urn ownerUrn,
-      OwnershipType type,
-      Urn resourceUrn,
+  public static void addOwnersToResources(
+      List<OwnerInput> owners,
+      List<ResourceRefInput> resources,
      Urn actor,
      EntityService entityService
  ) {
-    Ownership ownershipAspect = (Ownership) getAspectFromEntity(
-        resourceUrn.toString(),
-        Constants.OWNERSHIP_ASPECT_NAME,
-        entityService,
-        new Ownership());
-    addOwner(ownershipAspect, ownerUrn, type);
-    persistAspect(resourceUrn, Constants.OWNERSHIP_ASPECT_NAME, ownershipAspect, actor, entityService);
+    final List<MetadataChangeProposal> changes = new ArrayList<>();
+    for (ResourceRefInput resource : resources) {
+      changes.add(buildAddOwnersProposal(owners, UrnUtils.getUrn(resource.getResourceUrn()), actor, entityService));
+    }
+    ingestChangeProposals(changes, entityService, actor);
  }

-  public static void addOwners(
-      List<OwnerInput> owners,
-      Urn resourceUrn,
+  public static void removeOwnersFromResources(
+      List<Urn> ownerUrns,
+      List<ResourceRefInput> resources,
      Urn actor,
      EntityService entityService
  ) {
+    final List<MetadataChangeProposal> changes = new ArrayList<>();
+    for (ResourceRefInput resource : resources) {
+      changes.add(buildRemoveOwnersProposal(ownerUrns, UrnUtils.getUrn(resource.getResourceUrn()), actor, entityService));
+    }
+    ingestChangeProposals(changes, entityService, actor);
+  }
+
+
+  private static MetadataChangeProposal buildAddOwnersProposal(List<OwnerInput> owners, Urn resourceUrn, Urn actor, EntityService entityService) {
    Ownership ownershipAspect = (Ownership) getAspectFromEntity(
        resourceUrn.toString(),
        Constants.OWNERSHIP_ASPECT_NAME,
@ -66,11 +75,11 @@ public class OwnerUtils {
    for (OwnerInput input : owners) {
      addOwner(ownershipAspect, UrnUtils.getUrn(input.getOwnerUrn()), OwnershipType.valueOf(input.getType().toString()));
    }
-    persistAspect(resourceUrn, Constants.OWNERSHIP_ASPECT_NAME, ownershipAspect, actor, entityService);
+    return buildMetadataChangeProposal(resourceUrn, Constants.OWNERSHIP_ASPECT_NAME, ownershipAspect, actor, entityService);
  }

-  public static void removeOwner(
-      Urn ownerUrn,
+  public static MetadataChangeProposal buildRemoveOwnersProposal(
+      List<Urn> ownerUrns,
      Urn resourceUrn,
      Urn actor,
      EntityService entityService
@ -81,8 +90,8 @@ public class OwnerUtils {
        entityService,
        new Ownership());
    ownershipAspect.setLastModified(getAuditStamp(actor));
-    removeOwner(ownershipAspect, ownerUrn);
-    persistAspect(resourceUrn, Constants.OWNERSHIP_ASPECT_NAME, ownershipAspect, actor, entityService);
+    removeOwnersIfExists(ownershipAspect, ownerUrns);
+    return buildMetadataChangeProposal(resourceUrn, Constants.OWNERSHIP_ASPECT_NAME, ownershipAspect, actor, entityService);
  }

  private static void addOwner(Ownership ownershipAspect, Urn ownerUrn, OwnershipType type) {
@ -103,13 +112,15 @@ public class OwnerUtils {
    ownershipAspect.setOwners(ownerArray);
  }

-  private static void removeOwner(Ownership ownership, Urn ownerUrn) {
+  private static void removeOwnersIfExists(Ownership ownership, List<Urn> ownerUrns) {
    if (!ownership.hasOwners()) {
      ownership.setOwners(new OwnerArray());
    }

    OwnerArray ownerArray = ownership.getOwners();
-    ownerArray.removeIf(owner -> owner.getOwner().equals(ownerUrn));
+    for (Urn ownerUrn : ownerUrns) {
+      ownerArray.removeIf(owner -> owner.getOwner().equals(ownerUrn));
+    }
  }

  public static boolean isAuthorizedToUpdateOwners(@Nonnull QueryContext context, Urn resourceUrn) {
@ -170,6 +181,26 @@ public class OwnerUtils {
    return true;
  }

+  public static void validateOwner(
+      Urn ownerUrn,
+      OwnerEntityType ownerEntityType,
+      EntityService entityService
+  ) {
+    if (OwnerEntityType.CORP_GROUP.equals(ownerEntityType) && !Constants.CORP_GROUP_ENTITY_NAME.equals(ownerUrn.getEntityType())) {
+      throw new IllegalArgumentException(
+          String.format("Failed to change ownership for resource(s). Expected a corp group urn, found %s", ownerUrn));
+    }
+
+    if (OwnerEntityType.CORP_USER.equals(ownerEntityType) && !Constants.CORP_USER_ENTITY_NAME.equals(ownerUrn.getEntityType())) {
+      throw new IllegalArgumentException(
+          String.format("Failed to change ownership for resource(s). Expected a corp user urn, found %s.", ownerUrn));
+    }
+
+    if (!entityService.exists(ownerUrn)) {
+      throw new IllegalArgumentException(String.format("Failed to change ownership for resource(s). Owner with urn %s does not exist.", ownerUrn));
+    }
+  }
+
  public static Boolean validateRemoveInput(
      Urn resourceUrn,
      EntityService entityService
@ -179,4 +210,11 @@ public class OwnerUtils {
    }
    return true;
  }
+
+  private static void ingestChangeProposals(List<MetadataChangeProposal> changes, EntityService entityService, Urn actor) {
+    // TODO: Replace this with a batch ingest proposals endpoint.
+    for (MetadataChangeProposal change : changes) {
+      entityService.ingestProposal(change, getAuditStamp(actor));
+    }
+  }
 }
--- a/datahub-graphql-core/src/main/resources/entity.graphql
+++ b/datahub-graphql-core/src/main/resources/entity.graphql
@ -305,6 +305,11 @@ type Mutation {
    """
    addOwner(input: AddOwnerInput!): Boolean

+    """
+    Add owners to multiple Entities
+    """
+    batchAddOwners(input: BatchAddOwnersInput!): Boolean
+
    """
    Add multiple owners to a particular Entity
    """
@ -315,6 +320,11 @@ type Mutation {
    """
    removeOwner(input: RemoveOwnerInput!): Boolean

+    """
+    Remove owners from multiple Entities
+    """
+    batchRemoveOwners(input: BatchRemoveOwnersInput!): Boolean
+
    """
    Add a link, or institutional memory, from a particular Entity
    """
@ -6739,6 +6749,36 @@ input AddOwnersInput {
    resourceUrn: String!
 }

+"""
+Input provided when adding owners to a batch of assets
+"""
+input BatchAddOwnersInput {
+    """
+    The primary key of the owners
+    """
+    owners: [OwnerInput!]!
+
+    """
+    The target assets to attach the owners to
+    """
+    resources: [ResourceRefInput]!
+}
+
+"""
+Input provided when removing owners from a batch of assets
+"""
+input BatchRemoveOwnersInput {
+    """
+    The primary key of the owners
+    """
+    ownerUrns: [String!]!
+
+    """
+    The target assets to remove the owners from
+    """
+    resources: [ResourceRefInput]!
+}
+
 """
 Input provided when removing the association between a Metadata Entity and an user or group owner
 """
--- a/datahub-graphql-core/src/test/java/com/linkedin/datahub/graphql/resolvers/owner/BatchAddOwnersResolverTest.java
+++ b/datahub-graphql-core/src/test/java/com/linkedin/datahub/graphql/resolvers/owner/BatchAddOwnersResolverTest.java
@ -0,0 +1,290 @@
+package com.linkedin.datahub.graphql.resolvers.owner;
+
+import com.google.common.collect.ImmutableList;
+import com.linkedin.common.AuditStamp;
+import com.linkedin.common.Owner;
+import com.linkedin.common.OwnerArray;
+import com.linkedin.common.Ownership;
+import com.linkedin.common.OwnershipType;
+import com.linkedin.common.urn.Urn;
+import com.linkedin.common.urn.UrnUtils;
+import com.linkedin.datahub.graphql.QueryContext;
+import com.linkedin.datahub.graphql.generated.BatchAddOwnersInput;
+import com.linkedin.datahub.graphql.generated.OwnerEntityType;
+import com.linkedin.datahub.graphql.generated.OwnerInput;
+import com.linkedin.datahub.graphql.generated.ResourceRefInput;
+import com.linkedin.datahub.graphql.resolvers.mutate.BatchAddOwnersResolver;
+import com.linkedin.metadata.Constants;
+import com.linkedin.metadata.entity.EntityService;
+import graphql.schema.DataFetchingEnvironment;
+import java.util.concurrent.CompletionException;
+import org.mockito.Mockito;
+import org.testng.annotations.Test;
+
+import static com.linkedin.datahub.graphql.TestUtils.*;
+import static org.testng.Assert.*;
+
+
+public class BatchAddOwnersResolverTest {
+
+  private static final String TEST_ENTITY_URN_1 = "urn:li:dataset:(urn:li:dataPlatform:mysql,my-test,PROD)";
+  private static final String TEST_ENTITY_URN_2 = "urn:li:dataset:(urn:li:dataPlatform:mysql,my-test-2,PROD)";
+  private static final String TEST_OWNER_URN_1 = "urn:li:corpuser:test-id-1";
+  private static final String TEST_OWNER_URN_2 = "urn:li:corpuser:test-id-2";
+
+  @Test
+  public void testGetSuccessNoExistingOwners() throws Exception {
+    EntityService mockService = Mockito.mock(EntityService.class);
+
+    Mockito.when(mockService.getAspect(
+        Mockito.eq(UrnUtils.getUrn(TEST_ENTITY_URN_1)),
+        Mockito.eq(Constants.OWNERSHIP_ASPECT_NAME),
+        Mockito.eq(0L)))
+        .thenReturn(null);
+
+    Mockito.when(mockService.getAspect(
+        Mockito.eq(UrnUtils.getUrn(TEST_ENTITY_URN_2)),
+        Mockito.eq(Constants.OWNERSHIP_ASPECT_NAME),
+        Mockito.eq(0L)))
+        .thenReturn(null);
+
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_ENTITY_URN_1))).thenReturn(true);
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_ENTITY_URN_2))).thenReturn(true);
+
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_OWNER_URN_1))).thenReturn(true);
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_OWNER_URN_2))).thenReturn(true);
+
+    BatchAddOwnersResolver resolver = new BatchAddOwnersResolver(mockService);
+
+    // Execute resolver
+    QueryContext mockContext = getMockAllowContext();
+    DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
+    BatchAddOwnersInput input = new BatchAddOwnersInput(ImmutableList.of(new OwnerInput(
+            TEST_OWNER_URN_1,
+            OwnerEntityType.CORP_USER,
+            com.linkedin.datahub.graphql.generated.OwnershipType.BUSINESS_OWNER),
+        new OwnerInput(
+            TEST_OWNER_URN_2,
+            OwnerEntityType.CORP_USER,
+            com.linkedin.datahub.graphql.generated.OwnershipType.BUSINESS_OWNER)),
+        ImmutableList.of(
+            new ResourceRefInput(TEST_ENTITY_URN_1, null, null),
+            new ResourceRefInput(TEST_ENTITY_URN_2, null, null)));
+    Mockito.when(mockEnv.getArgument(Mockito.eq("input"))).thenReturn(input);
+    Mockito.when(mockEnv.getContext()).thenReturn(mockContext);
+    assertTrue(resolver.get(mockEnv).get());
+
+    Mockito.verify(mockService, Mockito.times(2)).ingestProposal(
+        Mockito.any(), // Ownership has a dynamically generated timestamp
+        Mockito.any(AuditStamp.class)
+    );
+
+    Mockito.verify(mockService, Mockito.times(1)).exists(
+        Mockito.eq(Urn.createFromString(TEST_OWNER_URN_1))
+    );
+
+    Mockito.verify(mockService, Mockito.times(1)).exists(
+        Mockito.eq(Urn.createFromString(TEST_OWNER_URN_2))
+    );
+  }
+
+  @Test
+  public void testGetSuccessExistingOwners() throws Exception {
+    final Ownership originalOwnership = new Ownership().setOwners(new OwnerArray(ImmutableList.of(
+        new Owner().setOwner(Urn.createFromString(TEST_OWNER_URN_1)).setType(OwnershipType.TECHNICAL_OWNER)
+    )));
+    EntityService mockService = Mockito.mock(EntityService.class);
+
+    Mockito.when(mockService.getAspect(
+        Mockito.eq(UrnUtils.getUrn(TEST_ENTITY_URN_1)),
+        Mockito.eq(Constants.OWNERSHIP_ASPECT_NAME),
+        Mockito.eq(0L)))
+        .thenReturn(originalOwnership);
+
+    Mockito.when(mockService.getAspect(
+        Mockito.eq(UrnUtils.getUrn(TEST_ENTITY_URN_2)),
+        Mockito.eq(Constants.OWNERSHIP_ASPECT_NAME),
+        Mockito.eq(0L)))
+        .thenReturn(originalOwnership);
+
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_ENTITY_URN_1))).thenReturn(true);
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_ENTITY_URN_2))).thenReturn(true);
+
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_OWNER_URN_1))).thenReturn(true);
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_OWNER_URN_2))).thenReturn(true);
+
+    BatchAddOwnersResolver resolver = new BatchAddOwnersResolver(mockService);
+
+    // Execute resolver
+    QueryContext mockContext = getMockAllowContext();
+    DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
+    BatchAddOwnersInput input = new BatchAddOwnersInput(ImmutableList.of(new OwnerInput(
+        TEST_OWNER_URN_1,
+        OwnerEntityType.CORP_USER,
+        com.linkedin.datahub.graphql.generated.OwnershipType.BUSINESS_OWNER),
+        new OwnerInput(
+            TEST_OWNER_URN_2,
+            OwnerEntityType.CORP_USER,
+            com.linkedin.datahub.graphql.generated.OwnershipType.BUSINESS_OWNER)),
+        ImmutableList.of(
+          new ResourceRefInput(TEST_ENTITY_URN_1, null, null),
+          new ResourceRefInput(TEST_ENTITY_URN_2, null, null)));
+    Mockito.when(mockEnv.getArgument(Mockito.eq("input"))).thenReturn(input);
+    Mockito.when(mockEnv.getContext()).thenReturn(mockContext);
+    assertTrue(resolver.get(mockEnv).get());
+
+    Mockito.verify(mockService, Mockito.times(2)).ingestProposal(
+        Mockito.any(), // Ownership has a dynamically generated timestamp
+        Mockito.any(AuditStamp.class)
+    );
+
+    Mockito.verify(mockService, Mockito.times(1)).exists(
+        Mockito.eq(Urn.createFromString(TEST_OWNER_URN_1))
+    );
+
+    Mockito.verify(mockService, Mockito.times(1)).exists(
+        Mockito.eq(Urn.createFromString(TEST_OWNER_URN_2))
+    );
+  }
+
+  @Test
+  public void testGetFailureOwnerDoesNotExist() throws Exception {
+    EntityService mockService = Mockito.mock(EntityService.class);
+
+    Mockito.when(mockService.getAspect(
+        Mockito.eq(UrnUtils.getUrn(TEST_ENTITY_URN_1)),
+        Mockito.eq(Constants.OWNERSHIP_ASPECT_NAME),
+        Mockito.eq(0L)))
+        .thenReturn(null);
+
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_ENTITY_URN_1))).thenReturn(true);
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_OWNER_URN_1))).thenReturn(false);
+
+    BatchAddOwnersResolver resolver = new BatchAddOwnersResolver(mockService);
+
+    // Execute resolver
+    QueryContext mockContext = getMockAllowContext();
+    DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
+    BatchAddOwnersInput input = new BatchAddOwnersInput(ImmutableList.of(new OwnerInput(
+            TEST_OWNER_URN_1,
+            OwnerEntityType.CORP_USER,
+            com.linkedin.datahub.graphql.generated.OwnershipType.BUSINESS_OWNER),
+        new OwnerInput(
+            TEST_OWNER_URN_2,
+            OwnerEntityType.CORP_USER,
+            com.linkedin.datahub.graphql.generated.OwnershipType.BUSINESS_OWNER)),
+        ImmutableList.of(
+            new ResourceRefInput(TEST_ENTITY_URN_1, null, null),
+            new ResourceRefInput(TEST_ENTITY_URN_2, null, null)));
+    Mockito.when(mockEnv.getArgument(Mockito.eq("input"))).thenReturn(input);
+    Mockito.when(mockEnv.getContext()).thenReturn(mockContext);
+
+    assertThrows(CompletionException.class, () -> resolver.get(mockEnv).join());
+    Mockito.verify(mockService, Mockito.times(0)).ingestProposal(
+        Mockito.any(),
+        Mockito.any(AuditStamp.class));
+  }
+
+  @Test
+  public void testGetFailureResourceDoesNotExist() throws Exception {
+    EntityService mockService = Mockito.mock(EntityService.class);
+
+    Mockito.when(mockService.getAspect(
+        Mockito.eq(UrnUtils.getUrn(TEST_ENTITY_URN_1)),
+        Mockito.eq(Constants.OWNERSHIP_ASPECT_NAME),
+        Mockito.eq(0L)))
+        .thenReturn(null);
+    Mockito.when(mockService.getAspect(
+        Mockito.eq(UrnUtils.getUrn(TEST_ENTITY_URN_2)),
+        Mockito.eq(Constants.OWNERSHIP_ASPECT_NAME),
+        Mockito.eq(0L)))
+        .thenReturn(null);
+
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_ENTITY_URN_1))).thenReturn(false);
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_ENTITY_URN_2))).thenReturn(true);
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_OWNER_URN_1))).thenReturn(true);
+
+    BatchAddOwnersResolver resolver = new BatchAddOwnersResolver(mockService);
+
+    // Execute resolver
+    QueryContext mockContext = getMockAllowContext();
+    DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
+    BatchAddOwnersInput input = new BatchAddOwnersInput(ImmutableList.of(new OwnerInput(
+            TEST_OWNER_URN_1,
+            OwnerEntityType.CORP_USER,
+            com.linkedin.datahub.graphql.generated.OwnershipType.BUSINESS_OWNER),
+        new OwnerInput(
+            TEST_OWNER_URN_2,
+            OwnerEntityType.CORP_USER,
+            com.linkedin.datahub.graphql.generated.OwnershipType.BUSINESS_OWNER)),
+        ImmutableList.of(
+            new ResourceRefInput(TEST_ENTITY_URN_1, null, null),
+            new ResourceRefInput(TEST_ENTITY_URN_2, null, null)));
+    Mockito.when(mockEnv.getArgument(Mockito.eq("input"))).thenReturn(input);
+    Mockito.when(mockEnv.getContext()).thenReturn(mockContext);
+
+    assertThrows(CompletionException.class, () -> resolver.get(mockEnv).join());
+    Mockito.verify(mockService, Mockito.times(0)).ingestProposal(
+        Mockito.any(),
+        Mockito.any(AuditStamp.class));
+  }
+
+  @Test
+  public void testGetUnauthorized() throws Exception {
+    EntityService mockService = Mockito.mock(EntityService.class);
+
+    BatchAddOwnersResolver resolver = new BatchAddOwnersResolver(mockService);
+
+    // Execute resolver
+    DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
+    BatchAddOwnersInput input = new BatchAddOwnersInput(ImmutableList.of(new OwnerInput(
+            TEST_OWNER_URN_1,
+            OwnerEntityType.CORP_USER,
+            com.linkedin.datahub.graphql.generated.OwnershipType.BUSINESS_OWNER),
+        new OwnerInput(
+            TEST_OWNER_URN_2,
+            OwnerEntityType.CORP_USER,
+            com.linkedin.datahub.graphql.generated.OwnershipType.BUSINESS_OWNER)),
+        ImmutableList.of(
+            new ResourceRefInput(TEST_ENTITY_URN_1, null, null),
+            new ResourceRefInput(TEST_ENTITY_URN_2, null, null)));
+    Mockito.when(mockEnv.getArgument(Mockito.eq("input"))).thenReturn(input);
+    QueryContext mockContext = getMockDenyContext();
+    Mockito.when(mockEnv.getContext()).thenReturn(mockContext);
+
+    assertThrows(CompletionException.class, () -> resolver.get(mockEnv).join());
+    Mockito.verify(mockService, Mockito.times(0)).ingestProposal(
+        Mockito.any(),
+        Mockito.any(AuditStamp.class));
+  }
+
+  @Test
+  public void testGetEntityClientException() throws Exception {
+    EntityService mockService = Mockito.mock(EntityService.class);
+
+    Mockito.doThrow(RuntimeException.class).when(mockService).ingestProposal(
+        Mockito.any(),
+        Mockito.any(AuditStamp.class));
+
+    BatchAddOwnersResolver resolver = new BatchAddOwnersResolver(mockService);
+
+    // Execute resolver
+    DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
+    QueryContext mockContext = getMockAllowContext();
+    BatchAddOwnersInput input = new BatchAddOwnersInput(ImmutableList.of(new OwnerInput(
+            TEST_OWNER_URN_1,
+            OwnerEntityType.CORP_USER,
+            com.linkedin.datahub.graphql.generated.OwnershipType.BUSINESS_OWNER),
+        new OwnerInput(
+            TEST_OWNER_URN_2,
+            OwnerEntityType.CORP_USER,
+            com.linkedin.datahub.graphql.generated.OwnershipType.BUSINESS_OWNER)),
+        ImmutableList.of(
+            new ResourceRefInput(TEST_ENTITY_URN_1, null, null),
+            new ResourceRefInput(TEST_ENTITY_URN_2, null, null)));
+    Mockito.when(mockEnv.getArgument(Mockito.eq("input"))).thenReturn(input);
+    Mockito.when(mockEnv.getContext()).thenReturn(mockContext);
+
+    assertThrows(CompletionException.class, () -> resolver.get(mockEnv).join());
+  }
+}
--- a/datahub-graphql-core/src/test/java/com/linkedin/datahub/graphql/resolvers/owner/BatchRemoveOwnersResolverTest.java
+++ b/datahub-graphql-core/src/test/java/com/linkedin/datahub/graphql/resolvers/owner/BatchRemoveOwnersResolverTest.java
@ -0,0 +1,206 @@
+package com.linkedin.datahub.graphql.resolvers.owner;
+
+import com.google.common.collect.ImmutableList;
+import com.linkedin.common.AuditStamp;
+import com.linkedin.common.Owner;
+import com.linkedin.common.OwnerArray;
+import com.linkedin.common.Ownership;
+import com.linkedin.common.OwnershipType;
+import com.linkedin.common.urn.Urn;
+import com.linkedin.common.urn.UrnUtils;
+import com.linkedin.datahub.graphql.QueryContext;
+import com.linkedin.datahub.graphql.generated.BatchRemoveOwnersInput;
+import com.linkedin.datahub.graphql.generated.ResourceRefInput;
+import com.linkedin.datahub.graphql.resolvers.mutate.BatchRemoveOwnersResolver;
+import com.linkedin.metadata.Constants;
+import com.linkedin.metadata.entity.EntityService;
+import com.linkedin.mxe.MetadataChangeProposal;
+import graphql.schema.DataFetchingEnvironment;
+import java.util.concurrent.CompletionException;
+import org.mockito.Mockito;
+import org.testng.annotations.Test;
+
+import static com.linkedin.datahub.graphql.TestUtils.*;
+import static org.testng.Assert.*;
+
+
+public class BatchRemoveOwnersResolverTest {
+
+  private static final String TEST_ENTITY_URN_1 = "urn:li:dataset:(urn:li:dataPlatform:mysql,my-test,PROD)";
+  private static final String TEST_ENTITY_URN_2 = "urn:li:dataset:(urn:li:dataPlatform:mysql,my-test-2,PROD)";
+  private static final String TEST_OWNER_URN_1 = "urn:li:corpuser:test-id-1";
+  private static final String TEST_OWNER_URN_2 = "urn:li:corpuser:test-id-2";
+
+  @Test
+  public void testGetSuccessNoExistingOwners() throws Exception {
+    EntityService mockService = Mockito.mock(EntityService.class);
+
+    Mockito.when(mockService.getAspect(
+        Mockito.eq(UrnUtils.getUrn(TEST_ENTITY_URN_1)),
+        Mockito.eq(Constants.OWNERSHIP_ASPECT_NAME),
+        Mockito.eq(0L)))
+        .thenReturn(null);
+    Mockito.when(mockService.getAspect(
+        Mockito.eq(UrnUtils.getUrn(TEST_ENTITY_URN_2)),
+        Mockito.eq(Constants.OWNERSHIP_ASPECT_NAME),
+        Mockito.eq(0L)))
+        .thenReturn(null);
+
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_ENTITY_URN_1))).thenReturn(true);
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_ENTITY_URN_2))).thenReturn(true);
+
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_OWNER_URN_1))).thenReturn(true);
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_OWNER_URN_2))).thenReturn(true);
+
+    BatchRemoveOwnersResolver resolver = new BatchRemoveOwnersResolver(mockService);
+
+    // Execute resolver
+    QueryContext mockContext = getMockAllowContext();
+    DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
+    BatchRemoveOwnersInput input = new BatchRemoveOwnersInput(ImmutableList.of(
+        TEST_OWNER_URN_1,
+        TEST_OWNER_URN_2
+    ), ImmutableList.of(
+        new ResourceRefInput(TEST_ENTITY_URN_1, null, null),
+        new ResourceRefInput(TEST_ENTITY_URN_2, null, null)));
+    Mockito.when(mockEnv.getArgument(Mockito.eq("input"))).thenReturn(input);
+    Mockito.when(mockEnv.getContext()).thenReturn(mockContext);
+    assertTrue(resolver.get(mockEnv).get());
+
+    Mockito.verify(mockService, Mockito.times(2)).ingestProposal(
+        Mockito.any(), // Ownership has a dynamically generated timestamp
+        Mockito.any(AuditStamp.class)
+    );
+  }
+
+  @Test
+  public void testGetSuccessExistingOwners() throws Exception {
+    EntityService mockService = Mockito.mock(EntityService.class);
+
+    final Ownership oldOwners1 = new Ownership().setOwners(new OwnerArray(ImmutableList.of(
+        new Owner().setOwner(Urn.createFromString(TEST_OWNER_URN_1)).setType(OwnershipType.TECHNICAL_OWNER)
+    )));
+
+    Mockito.when(mockService.getAspect(
+        Mockito.eq(UrnUtils.getUrn(TEST_ENTITY_URN_1)),
+        Mockito.eq(Constants.OWNERSHIP_ASPECT_NAME),
+        Mockito.eq(0L)))
+        .thenReturn(oldOwners1);
+
+    final Ownership oldOwners2 = new Ownership().setOwners(new OwnerArray(ImmutableList.of(
+        new Owner().setOwner(Urn.createFromString(TEST_OWNER_URN_2)).setType(OwnershipType.TECHNICAL_OWNER)
+    )));
+
+    Mockito.when(mockService.getAspect(
+        Mockito.eq(UrnUtils.getUrn(TEST_ENTITY_URN_2)),
+        Mockito.eq(Constants.OWNERSHIP_ASPECT_NAME),
+        Mockito.eq(0L)))
+        .thenReturn(oldOwners2);
+
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_ENTITY_URN_1))).thenReturn(true);
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_ENTITY_URN_2))).thenReturn(true);
+
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_OWNER_URN_1))).thenReturn(true);
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_OWNER_URN_2))).thenReturn(true);
+
+    BatchRemoveOwnersResolver resolver = new BatchRemoveOwnersResolver(mockService);
+
+    // Execute resolver
+    QueryContext mockContext = getMockAllowContext();
+    DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
+    BatchRemoveOwnersInput input = new BatchRemoveOwnersInput(ImmutableList.of(TEST_OWNER_URN_1, TEST_OWNER_URN_2
+    ), ImmutableList.of(
+        new ResourceRefInput(TEST_ENTITY_URN_1, null, null),
+        new ResourceRefInput(TEST_ENTITY_URN_2, null, null)));
+    Mockito.when(mockEnv.getArgument(Mockito.eq("input"))).thenReturn(input);
+    Mockito.when(mockEnv.getContext()).thenReturn(mockContext);
+    assertTrue(resolver.get(mockEnv).get());
+
+    Mockito.verify(mockService, Mockito.times(2)).ingestProposal(
+        Mockito.any(MetadataChangeProposal.class),
+        Mockito.any(AuditStamp.class)
+    );
+  }
+
+  @Test
+  public void testGetFailureResourceDoesNotExist() throws Exception {
+    EntityService mockService = Mockito.mock(EntityService.class);
+
+    Mockito.when(mockService.getAspect(
+        Mockito.eq(UrnUtils.getUrn(TEST_ENTITY_URN_1)),
+        Mockito.eq(Constants.OWNERSHIP_ASPECT_NAME),
+        Mockito.eq(0L)))
+        .thenReturn(null);
+    Mockito.when(mockService.getAspect(
+        Mockito.eq(UrnUtils.getUrn(TEST_ENTITY_URN_2)),
+        Mockito.eq(Constants.OWNERSHIP_ASPECT_NAME),
+        Mockito.eq(0L)))
+        .thenReturn(null);
+
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_ENTITY_URN_1))).thenReturn(false);
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_ENTITY_URN_2))).thenReturn(true);
+    Mockito.when(mockService.exists(Urn.createFromString(TEST_OWNER_URN_1))).thenReturn(true);
+
+    BatchRemoveOwnersResolver resolver = new BatchRemoveOwnersResolver(mockService);
+
+    // Execute resolver
+    QueryContext mockContext = getMockAllowContext();
+    DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
+    BatchRemoveOwnersInput input = new BatchRemoveOwnersInput(ImmutableList.of(TEST_OWNER_URN_1, TEST_OWNER_URN_2
+    ), ImmutableList.of(
+        new ResourceRefInput(TEST_ENTITY_URN_1, null, null),
+        new ResourceRefInput(TEST_ENTITY_URN_2, null, null)));
+    Mockito.when(mockEnv.getArgument(Mockito.eq("input"))).thenReturn(input);
+    Mockito.when(mockEnv.getContext()).thenReturn(mockContext);
+
+    assertThrows(CompletionException.class, () -> resolver.get(mockEnv).join());
+    Mockito.verify(mockService, Mockito.times(0)).ingestProposal(
+        Mockito.any(),
+        Mockito.any(AuditStamp.class));
+  }
+
+  @Test
+  public void testGetUnauthorized() throws Exception {
+    EntityService mockService = Mockito.mock(EntityService.class);
+
+    BatchRemoveOwnersResolver resolver = new BatchRemoveOwnersResolver(mockService);
+
+    // Execute resolver
+    DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
+    BatchRemoveOwnersInput input = new BatchRemoveOwnersInput(ImmutableList.of(TEST_OWNER_URN_1, TEST_OWNER_URN_2
+    ), ImmutableList.of(
+        new ResourceRefInput(TEST_ENTITY_URN_1, null, null),
+        new ResourceRefInput(TEST_ENTITY_URN_2, null, null)));
+    Mockito.when(mockEnv.getArgument(Mockito.eq("input"))).thenReturn(input);
+    QueryContext mockContext = getMockDenyContext();
+    Mockito.when(mockEnv.getContext()).thenReturn(mockContext);
+
+    assertThrows(CompletionException.class, () -> resolver.get(mockEnv).join());
+    Mockito.verify(mockService, Mockito.times(0)).ingestProposal(
+        Mockito.any(),
+        Mockito.any(AuditStamp.class));
+  }
+
+  @Test
+  public void testGetEntityClientException() throws Exception {
+    EntityService mockService = Mockito.mock(EntityService.class);
+
+    Mockito.doThrow(RuntimeException.class).when(mockService).ingestProposal(
+        Mockito.any(),
+        Mockito.any(AuditStamp.class));
+
+    BatchRemoveOwnersResolver resolver = new BatchRemoveOwnersResolver(mockService);
+
+    // Execute resolver
+    DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
+    QueryContext mockContext = getMockAllowContext();
+    BatchRemoveOwnersInput input = new BatchRemoveOwnersInput(ImmutableList.of(TEST_OWNER_URN_1, TEST_OWNER_URN_2
+    ), ImmutableList.of(
+        new ResourceRefInput(TEST_ENTITY_URN_1, null, null),
+        new ResourceRefInput(TEST_ENTITY_URN_2, null, null)));
+    Mockito.when(mockEnv.getArgument(Mockito.eq("input"))).thenReturn(input);
+    Mockito.when(mockEnv.getContext()).thenReturn(mockContext);
+
+    assertThrows(CompletionException.class, () -> resolver.get(mockEnv).join());
+  }
+}
--- a/datahub-web-react/src/app/analytics/event.ts
+++ b/datahub-web-react/src/app/analytics/event.ts
@ -15,6 +15,7 @@ export enum EventType {
    EntityViewEvent,
    EntitySectionViewEvent,
    EntityActionEvent,
+    BatchEntityActionEvent,
    RecommendationImpressionEvent,
    RecommendationClickEvent,
    SearchAcrossLineageEvent,
@ -156,10 +157,16 @@ export const EntityActionType = {
 export interface EntityActionEvent extends BaseEvent {
    type: EventType.EntityActionEvent;
    actionType: string;
-    entityType: EntityType;
+    entityType?: EntityType;
    entityUrn: string;
 }

+export interface BatchEntityActionEvent extends BaseEvent {
+    type: EventType.BatchEntityActionEvent;
+    actionType: string;
+    entityUrns: string[];
+}
+
 export interface RecommendationImpressionEvent extends BaseEvent {
    type: EventType.RecommendationImpressionEvent;
    renderId: string; // TODO : Determine whether we need a render id to join with click event.
@ -221,4 +228,5 @@ export type Event =
    | SearchAcrossLineageEvent
    | SearchAcrossLineageResultsViewEvent
    | DownloadAsCsvEvent
-    | RecommendationClickEvent;
+    | RecommendationClickEvent
+    | BatchEntityActionEvent;
--- a/datahub-web-react/src/app/entity/group/GroupOwnerSideBarSection.tsx
+++ b/datahub-web-react/src/app/entity/group/GroupOwnerSideBarSection.tsx
@ -4,7 +4,7 @@ import React, { useState } from 'react';
 import styled from 'styled-components';
 import { EntityType, Ownership } from '../../../types.generated';
 import { ExpandedOwner } from '../shared/components/styled/ExpandedOwner';
-import { AddOwnersModal } from '../shared/containers/profile/sidebar/Ownership/AddOwnersModal';
+import { EditOwnersModal } from '../shared/containers/profile/sidebar/Ownership/EditOwnersModal';
 import { DisplayCount, GroupSectionTitle, GroupSectionHeader } from '../shared/SidebarStyledComponents';

 const TITLE = 'Owners';
@ -51,10 +51,10 @@ export default function GroupOwnerSideBarSection({ urn, ownership, refetch }: Pr
                )}
            </SectionWrapper>
            {showAddModal && (
-                <AddOwnersModal
-                    urn={urn}
+                <EditOwnersModal
+                    urns={[urn]}
                    hideOwnerType
-                    type={EntityType.CorpGroup}
+                    entityType={EntityType.CorpGroup}
                    refetch={refetch}
                    onCloseModal={() => {
                        setShowAddModal(false);
--- a/datahub-web-react/src/app/entity/shared/components/styled/search/SearchSelectActions.tsx
+++ b/datahub-web-react/src/app/entity/shared/components/styled/search/SearchSelectActions.tsx
@ -16,7 +16,11 @@ import { SelectActionGroups } from './types';
 *
 * Currently, only the change tags action is implemented.
 */
-const DEFAULT_ACTION_GROUPS = [SelectActionGroups.CHANGE_TAGS, SelectActionGroups.CHANGE_GLOSSARY_TERMS];
+const DEFAULT_ACTION_GROUPS = [
+    SelectActionGroups.CHANGE_TAGS,
+    SelectActionGroups.CHANGE_GLOSSARY_TERMS,
+    SelectActionGroups.CHANGE_OWNERS,
+];

 type Props = {
    selectedEntities: EntityAndType[];
@ -60,6 +64,7 @@ export const SearchSelectActions = ({
                        selectedEntityUrns.length === 0 ||
                        !isEntityCapabilitySupported(EntityCapabilityType.OWNERS, selectedEntityTypes)
                    }
+                    refetch={refetch}
                />
            )}
            {visibleActionGroups.has(SelectActionGroups.CHANGE_GLOSSARY_TERMS) && (
--- a/datahub-web-react/src/app/entity/shared/components/styled/search/action/OwnersDropdown.tsx
+++ b/datahub-web-react/src/app/entity/shared/components/styled/search/action/OwnersDropdown.tsx
@ -1,16 +1,18 @@
 import React, { useState } from 'react';
-import { EntityType } from '../../../../../../../types.generated';
-import { AddOwnersModal } from '../../../../containers/profile/sidebar/Ownership/AddOwnersModal';
+import { EditOwnersModal, OperationType } from '../../../../containers/profile/sidebar/Ownership/EditOwnersModal';
 import ActionDropdown from './ActionDropdown';

 type Props = {
    urns: Array<string>;
    disabled: boolean;
+    refetch?: () => void;
 };

 // eslint-disable-next-line
-export default function OwnersDropdown({ urns, disabled = false }: Props) {
-    const [showAddModal, setShowAddModal] = useState(false);
+export default function OwnersDropdown({ urns, disabled = false, refetch }: Props) {
+    const [isEditModalVisible, setIsEditModalVisible] = useState(false);
+    const [operationType, setOperationType] = useState(OperationType.ADD);
+
    return (
        <>
            <ActionDropdown
@ -18,22 +20,30 @@ export default function OwnersDropdown({ urns, disabled = false }: Props) {
                actions={[
                    {
                        title: 'Add owners',
-                        onClick: () => setShowAddModal(true),
+                        onClick: () => {
+                            setOperationType(OperationType.ADD);
+                            setIsEditModalVisible(true);
+                        },
                    },
                    {
                        title: 'Remove owners',
-                        onClick: () => null,
+                        onClick: () => {
+                            setOperationType(OperationType.REMOVE);
+                            setIsEditModalVisible(true);
+                        },
                    },
                ]}
                disabled={disabled}
            />
-            {showAddModal && urns.length > 0 && (
-                <AddOwnersModal
-                    urn={urns[0]}
-                    type={EntityType.CorpUser}
+            {isEditModalVisible && (
+                <EditOwnersModal
+                    urns={urns}
+                    operationType={operationType}
                    onCloseModal={() => {
-                        setShowAddModal(false);
+                        setIsEditModalVisible(false);
+                        refetch?.();
                    }}
+                    hideOwnerType={operationType === OperationType.REMOVE}
                />
            )}
        </>
--- a/datahub-web-react/src/app/entity/shared/containers/profile/sidebar/Ownership/EditOwnersModal.tsx
+++ b/datahub-web-react/src/app/entity/shared/containers/profile/sidebar/Ownership/EditOwnersModal.tsx
@ -6,7 +6,10 @@ import { CorpUser, Entity, EntityType, OwnerEntityType, OwnershipType } from '..
 import { useEntityRegistry } from '../../../../../../useEntityRegistry';
 import analytics, { EventType, EntityActionType } from '../../../../../../analytics';
 import { OWNERSHIP_DISPLAY_TYPES } from './ownershipUtils';
-import { useAddOwnersMutation } from '../../../../../../../graphql/mutations.generated';
+import {
+    useBatchAddOwnersMutation,
+    useBatchRemoveOwnersMutation,
+} from '../../../../../../../graphql/mutations.generated';
 import { useGetSearchResultsLazyQuery } from '../../../../../../../graphql/search.generated';
 import { useGetRecommendations } from '../../../../../../shared/recommendation';
 import { OwnerLabel } from '../../../../../../shared/OwnerLabel';
@ -25,13 +28,19 @@ const StyleTag = styled(Tag)`
    align-items: center;
 `;

+export enum OperationType {
+    ADD,
+    REMOVE,
+}
+
 type Props = {
-    urn: string;
-    type: EntityType;
+    urns: string[];
    defaultOwnerType?: OwnershipType;
    hideOwnerType?: boolean | undefined;
+    operationType?: OperationType;
    onCloseModal: () => void;
    refetch?: () => Promise<any>;
+    entityType?: EntityType; // Only used for tracking events
 };

 // value: {ownerUrn: string, ownerEntityType: EntityType}
@ -40,10 +49,19 @@ type SelectedOwner = {
    value;
 };

-export const AddOwnersModal = ({ urn, type, hideOwnerType, defaultOwnerType, onCloseModal, refetch }: Props) => {
+export const EditOwnersModal = ({
+    urns,
+    hideOwnerType,
+    defaultOwnerType,
+    operationType = OperationType.ADD,
+    onCloseModal,
+    refetch,
+    entityType,
+}: Props) => {
    const entityRegistry = useEntityRegistry();
    const [inputValue, setInputValue] = useState('');
-    const [addOwnersMutation] = useAddOwnersMutation();
+    const [batchAddOwnersMutation] = useBatchAddOwnersMutation();
+    const [batchRemoveOwnersMutation] = useBatchRemoveOwnersMutation();
    const ownershipTypes = OWNERSHIP_DISPLAY_TYPES;
    const [selectedOwners, setSelectedOwners] = useState<SelectedOwner[]>([]);
    const [selectedOwnerType, setSelectedOwnerType] = useState<OwnershipType>(defaultOwnerType || OwnershipType.None);
@ -64,12 +82,12 @@ export const AddOwnersModal = ({ urn, type, hideOwnerType, defaultOwnerType, onC
    }, [ownershipTypes]);

    // Invokes the search API as the owner types
-    const handleSearch = (entityType: EntityType, text: string, searchQuery: any) => {
+    const handleSearch = (type: EntityType, text: string, searchQuery: any) => {
        if (text.length > 2) {
            searchQuery({
                variables: {
                    input: {
-                        type: entityType,
+                        type,
                        query: text,
                        start: 0,
                        count: 5,
@ -167,6 +185,69 @@ export const AddOwnersModal = ({ urn, type, hideOwnerType, defaultOwnerType, onC
        );
    };

+    const emitAnalytics = async () => {
+        if (urns.length > 1) {
+            analytics.event({
+                type: EventType.BatchEntityActionEvent,
+                actionType: EntityActionType.UpdateOwnership,
+                entityUrns: urns,
+            });
+        } else {
+            analytics.event({
+                type: EventType.EntityActionEvent,
+                actionType: EntityActionType.UpdateOwnership,
+                entityType,
+                entityUrn: urns[0],
+            });
+        }
+    };
+
+    const batchAddOwners = async (inputs) => {
+        try {
+            await batchAddOwnersMutation({
+                variables: {
+                    input: {
+                        owners: inputs,
+                        resources: urns.map((urn) => ({ resourceUrn: urn })),
+                    },
+                },
+            });
+            message.success({ content: 'Owners Added', duration: 2 });
+            emitAnalytics();
+        } catch (e: unknown) {
+            message.destroy();
+            if (e instanceof Error) {
+                message.error({ content: `Failed to add owners: \n ${e.message || ''}`, duration: 3 });
+            }
+        } finally {
+            refetch?.();
+            onModalClose();
+        }
+    };
+
+    const batchRemoveOwners = async (inputs) => {
+        try {
+            await batchRemoveOwnersMutation({
+                variables: {
+                    input: {
+                        ownerUrns: inputs.map((input) => input.ownerUrn),
+                        resources: urns.map((urn) => ({ resourceUrn: urn })),
+                    },
+                },
+            });
+            message.success({ content: 'Owners Removed', duration: 2 });
+            emitAnalytics();
+        } catch (e: unknown) {
+            message.destroy();
+            if (e instanceof Error) {
+                message.error({ content: `Failed to remove owners: \n ${e.message || ''}`, duration: 3 });
+            }
+        } finally {
+            refetch?.();
+            onModalClose();
+        }
+    };
+
    // Function to handle the modal action's
    const onOk = async () => {
        if (selectedOwners.length === 0) {
@ -180,30 +261,11 @@ export const AddOwnersModal = ({ urn, type, hideOwnerType, defaultOwnerType, onC
            };
            return input;
        });
-        try {
-            await addOwnersMutation({
-                variables: {
-                    input: {
-                        owners: inputs,
-                        resourceUrn: urn,
-                    },
-                },
-            });
-            message.success({ content: 'Owners Added', duration: 2 });
-            analytics.event({
-                type: EventType.EntityActionEvent,
-                actionType: EntityActionType.UpdateOwnership,
-                entityType: type,
-                entityUrn: urn,
-            });
-        } catch (e: unknown) {
-            message.destroy();
-            if (e instanceof Error) {
-                message.error({ content: `Failed to add owners: \n ${e.message || ''}`, duration: 3 });
-            }
-        } finally {
-            refetch?.();
-            onModalClose();
+
+        if (operationType === OperationType.ADD) {
+            batchAddOwners(inputs);
+        } else {
+            batchRemoveOwners(inputs);
        }
    };

@ -213,7 +275,7 @@ export const AddOwnersModal = ({ urn, type, hideOwnerType, defaultOwnerType, onC

    return (
        <Modal
-            title="Add Owners"
+            title={`${operationType === OperationType.ADD ? 'Add' : 'Remove'} Owners`}
            visible
            onCancel={onModalClose}
            keyboard
@ -223,7 +285,7 @@ export const AddOwnersModal = ({ urn, type, hideOwnerType, defaultOwnerType, onC
                        Cancel
                    </Button>
                    <Button id="addOwnerButton" disabled={selectedOwners.length === 0} onClick={onOk}>
-                        Add
+                        Done
                    </Button>
                </>
            }
--- a/datahub-web-react/src/app/entity/shared/containers/profile/sidebar/Ownership/SidebarOwnerSection.tsx
+++ b/datahub-web-react/src/app/entity/shared/containers/profile/sidebar/Ownership/SidebarOwnerSection.tsx
@ -5,7 +5,7 @@ import { ExpandedOwner } from '../../../../components/styled/ExpandedOwner';
 import { EMPTY_MESSAGES } from '../../../../constants';
 import { useEntityData, useMutationUrn, useRefetch } from '../../../../EntityContext';
 import { SidebarHeader } from '../SidebarHeader';
-import { AddOwnersModal } from './AddOwnersModal';
+import { EditOwnersModal } from './EditOwnersModal';

 export const SidebarOwnerSection = ({ properties }: { properties?: any }) => {
    const { entityType, entityData } = useEntityData();
@ -38,11 +38,11 @@ export const SidebarOwnerSection = ({ properties }: { properties?: any }) => {
                </Button>
            </div>
            {showAddModal && (
-                <AddOwnersModal
-                    urn={mutationUrn}
+                <EditOwnersModal
+                    urns={[mutationUrn]}
                    defaultOwnerType={properties?.defaultOwnerType}
                    hideOwnerType={properties?.hideOwnerType || false}
-                    type={entityType}
+                    entityType={entityType}
                    refetch={refetch}
                    onCloseModal={() => {
                        setShowAddModal(false);
--- a/datahub-web-react/src/app/shared/TagStyleEntity.tsx
+++ b/datahub-web-react/src/app/shared/TagStyleEntity.tsx
@ -17,7 +17,7 @@ import { useUpdateDescriptionMutation, useSetTagColorMutation } from '../../grap
 import { useGetSearchResultsForMultipleQuery } from '../../graphql/search.generated';
 import analytics, { EventType, EntityActionType } from '../analytics';
 import { GetSearchResultsParams, SearchResultInterface } from '../entity/shared/components/styled/search/types';
-import { AddOwnersModal } from '../entity/shared/containers/profile/sidebar/Ownership/AddOwnersModal';
+import { EditOwnersModal } from '../entity/shared/containers/profile/sidebar/Ownership/EditOwnersModal';
 import CopyUrn from './CopyUrn';
 import EntityDropdown from '../entity/shared/EntityDropdown';
 import { EntityMenuItems } from '../entity/shared/EntityDropdown/EntityDropdown';
@ -420,14 +420,14 @@ export default function TagStyleEntity({ urn, useGetSearchResults = useWrappedSe
                    </div>
                    <div>
                        {showAddModal && (
-                            <AddOwnersModal
+                            <EditOwnersModal
                                hideOwnerType
                                refetch={refetch}
                                onCloseModal={() => {
                                    setShowAddModal(false);
                                }}
-                                urn={urn}
-                                type={EntityType.Tag}
+                                urns={[urn]}
+                                entityType={EntityType.Tag}
                            />
                        )}
                    </div>
--- a/datahub-web-react/src/graphql/mutations.graphql
+++ b/datahub-web-react/src/graphql/mutations.graphql
@ -42,10 +42,18 @@ mutation addOwner($input: AddOwnerInput!) {
    addOwner(input: $input)
 }

+mutation batchAddOwners($input: BatchAddOwnersInput!) {
+    batchAddOwners(input: $input)
+}
+
 mutation removeOwner($input: RemoveOwnerInput!) {
    removeOwner(input: $input)
 }

+mutation batchRemoveOwners($input: BatchRemoveOwnersInput!) {
+    batchRemoveOwners(input: $input)
+}
+
 mutation updateDescription($input: DescriptionUpdateInput!) {
    updateDescription(input: $input)
 }