diff --git a/Documentation/access-control.txt b/Documentation/access-control.txt
index 7d531ff..a179905 100644
--- a/Documentation/access-control.txt
+++ b/Documentation/access-control.txt
@@ -918,6 +918,9 @@
 `on_behalf_of` field in link:rest-api-changes.html#submit-input[SubmitInput]
 when link:rest-api-changes.html#submit-change[submitting using the REST API].
 
+The user in the `on_behalf_of` field, does not need to have `Submit` permission
+themselves, however they should be able to `read` the changes being submitted.
+
 Note that this permission is named `submitAs` in the `project.config`
 file.
 
diff --git a/java/com/google/gerrit/server/git/validators/MergeValidationListener.java b/java/com/google/gerrit/server/git/validators/MergeValidationListener.java
index 79d53ac..a51a425 100644
--- a/java/com/google/gerrit/server/git/validators/MergeValidationListener.java
+++ b/java/com/google/gerrit/server/git/validators/MergeValidationListener.java
@@ -39,7 +39,8 @@
    * @param destProject the destination project
    * @param destBranch the destination branch
    * @param patchSetId the patch set ID
-   * @param caller the user who initiated the merge request
+   * @param caller the identity of the user that is recorded as the one performing the merge. In
+   *     case of impersonation {@code caller.getRealUser()} contains the user triggering the merge.
    * @throws MergeValidationException if the commit fails to validate
    */
   void onPreMerge(
diff --git a/java/com/google/gerrit/server/git/validators/MergeValidators.java b/java/com/google/gerrit/server/git/validators/MergeValidators.java
index 710e688..c8a3d1e 100644
--- a/java/com/google/gerrit/server/git/validators/MergeValidators.java
+++ b/java/com/google/gerrit/server/git/validators/MergeValidators.java
@@ -196,7 +196,9 @@
             if (!oldParent.equals(newParent)) {
               if (!allowProjectOwnersToChangeParent) {
                 try {
-                  if (!permissionBackend.user(caller).test(GlobalPermission.ADMINISTRATE_SERVER)) {
+                  if (!permissionBackend
+                      .user(caller.getRealUser())
+                      .test(GlobalPermission.ADMINISTRATE_SERVER)) {
                     throw new MergeValidationException(SET_BY_ADMIN);
                   }
                 } catch (PermissionBackendException e) {
@@ -206,7 +208,7 @@
               } else {
                 try {
                   permissionBackend
-                      .user(caller)
+                      .user(caller.getRealUser())
                       .project(destProject.getNameKey())
                       .check(ProjectPermission.WRITE_CONFIG);
                 } catch (AuthException e) {
diff --git a/java/com/google/gerrit/server/restapi/change/Submit.java b/java/com/google/gerrit/server/restapi/change/Submit.java
index 0a47d62..6b35175 100644
--- a/java/com/google/gerrit/server/restapi/change/Submit.java
+++ b/java/com/google/gerrit/server/restapi/change/Submit.java
@@ -87,6 +87,14 @@
 import org.eclipse.jgit.revwalk.RevObject;
 import org.eclipse.jgit.revwalk.RevWalk;
 
+/**
+ * REST API handler for triggering submit of the specific revision of the change.
+ *
+ * <p>See /Documentation/rest-api-changes.html#submit-revision for more information.
+ *
+ * <p>Even though the endpoint is defined for url including a revision, only revision corresponding
+ * to the latest patch set is allowed.
+ */
 @Singleton
 public class Submit
     implements RestModifyView<RevisionResource, SubmitInput>, UiAction<RevisionResource> {
@@ -179,12 +187,12 @@
       input = new SubmitInput();
     }
     input.onBehalfOf = Strings.emptyToNull(input.onBehalfOf);
-    IdentifiedUser submitter;
+    IdentifiedUser submitter = rsrc.getUser().asIdentifiedUser();
+    // It's possible that the user does not have permission to submit all changes in the superset,
+    // but we check the current change for an early exit.
+    rsrc.permissions().check(ChangePermission.SUBMIT);
     if (input.onBehalfOf != null) {
       submitter = onBehalfOf(rsrc, input);
-    } else {
-      rsrc.permissions().check(ChangePermission.SUBMIT);
-      submitter = rsrc.getUser().asIdentifiedUser();
     }
     projectCache
         .get(rsrc.getProject())
@@ -212,9 +220,7 @@
     }
 
     try (MergeOp op = mergeOpProvider.get()) {
-      Change updatedChange;
-
-      updatedChange = op.merge(change, submitter, true, input, false);
+      Change updatedChange = op.merge(change, submitter, true, input, false);
       if (updatedChange.isMerged()) {
         return updatedChange;
       }
@@ -296,7 +302,7 @@
     ChangeSet cs =
         mergeSuperSet
             .get()
-            .completeChangeSet(cd.change(), resource.getUser(), /*includingTopicClosure= */ false);
+            .completeChangeSet(cd.change(), resource.getUser(), /* includingTopicClosure= */ false);
     // Replace potentially stale ChangeData for the current change with the fresher one.
     cs =
         new ChangeSet(
@@ -437,7 +443,8 @@
       throws AuthException, UnprocessableEntityException, PermissionBackendException, IOException,
           ConfigInvalidException {
     PermissionBackend.ForChange perm = rsrc.permissions();
-    perm.check(ChangePermission.SUBMIT);
+    // It's possible that the current user or on-behalf-of user does not have permission for all
+    // changes in the superset, but we check the current change for an early exit.
     perm.check(ChangePermission.SUBMIT_AS);
 
     CurrentUser caller = rsrc.getUser();
@@ -449,9 +456,17 @@
       throw new UnprocessableEntityException(
           String.format("on_behalf_of account %s cannot see change", submitter.getAccountId()), e);
     }
+    logger.atFine().log(
+        "Change %d is being submitted by %s on behalf of %s",
+        rsrc.getChange().getChangeId(), rsrc.getUser().getUserName(), submitter.getUserName());
     return submitter;
   }
 
+  /**
+   * Change-level REST API endpoint that calls submit for the latest revision on a change.
+   *
+   * <p>See /Documentation/rest-api-changes.html#submit-change for more information.
+   */
   public static class CurrentRevision implements RestModifyView<ChangeResource, SubmitInput> {
     private final Submit submit;
     private final PatchSetUtil psUtil;
diff --git a/java/com/google/gerrit/server/submit/MergeOp.java b/java/com/google/gerrit/server/submit/MergeOp.java
index 233f00e..d49638a 100644
--- a/java/com/google/gerrit/server/submit/MergeOp.java
+++ b/java/com/google/gerrit/server/submit/MergeOp.java
@@ -154,6 +154,10 @@
   private static final SubmitRuleOptions SUBMIT_RULE_OPTIONS_ALLOW_CLOSED =
       SUBMIT_RULE_OPTIONS.toBuilder().recomputeOnClosedChanges(true).build();
 
+  /**
+   * For each individual change in merge set aggregates issues and other details throughout the
+   * merge process.
+   */
   public static class CommitStatus {
     private final ImmutableMap<Change.Id, ChangeData> changes;
     private final ImmutableSetMultimap<BranchNameKey, Change.Id> byBranch;
@@ -438,6 +442,7 @@
     return cd.submitRecords(submitRuleOptions(/* allowClosed= */ false));
   }
 
+  /** A problem preventing merge and change on which it occurred. */
   @AutoValue
   public abstract static class ChangeProblem {
     public abstract Change.Id getChangeId();
@@ -449,17 +454,131 @@
     }
   }
 
+  private static void addProblemForChange(
+      Change.Id triggeringChangeId,
+      ChangeData cd,
+      boolean allowMerged,
+      PermissionBackend permissionBackend,
+      CurrentUser caller,
+      ImmutableList.Builder<ChangeProblem> problems) {
+    try {
+      Set<ChangePermission> can =
+          permissionBackend
+              .user(caller.getRealUser())
+              .change(cd)
+              .test(
+                  EnumSet.of(
+                      ChangePermission.READ, ChangePermission.SUBMIT, ChangePermission.SUBMIT_AS));
+      if (!can.contains(ChangePermission.READ)) {
+        // The READ permission should already be handled during generation of ChangeSet, however
+        // MergeSuperSetComputation is an interface and on API level doesn't guarantee that this
+        // have been verified for all changes. Additionally, this protects against potential
+        // issues due to staleness.
+        logger.atFine().log(
+            "Change %d cannot be submitted by user %s because it depends on change %d which the"
+                + "user cannot read",
+            triggeringChangeId.get(), caller.getRealUser().getLoggableName(), cd.getId().get());
+        problems.add(
+            ChangeProblem.create(
+                cd.getId(),
+                String.format(
+                    "Change %d depends on other hidden changes", triggeringChangeId.get())));
+        return;
+      }
+      if (!can.contains(ChangePermission.SUBMIT)) {
+        logger.atFine().log(
+            "Change %d cannot be submitted by user %s because it depends on change %d which the"
+                + "user cannot submit",
+            triggeringChangeId.get(), caller.getRealUser().getLoggableName(), cd.getId().get());
+        problems.add(
+            ChangeProblem.create(
+                cd.getId(),
+                String.format("Insufficient permission to submit change %d", cd.getId().get())));
+        return;
+      }
+      if (caller.isImpersonating()) {
+        if (!permissionBackend.user(caller).change(cd).test(ChangePermission.READ)) {
+          logger.atFine().log(
+              "Change %d cannot be submitted by user %s on behalf of user %s because it depends on"
+                  + " change %d which the on-behalf-of user does not have READ permission for",
+              triggeringChangeId.get(),
+              caller.getRealUser().getLoggableName(),
+              caller.getLoggableName(),
+              cd.getId().get());
+          problems.add(
+              ChangeProblem.create(
+                  cd.getId(),
+                  String.format(
+                      "On-behalf-of user %s lacks permission to read change %d",
+                      caller.getLoggableName(), cd.getId().get())));
+          return;
+        }
+        if (!can.contains(ChangePermission.SUBMIT_AS)) {
+          logger.atFine().log(
+              "Change %d cannot be submitted by user %s on behalf of user %s because it depends on"
+                  + " change %d which the user does not have SUBMIT_AS permission for",
+              triggeringChangeId.get(),
+              caller.getRealUser().getLoggableName(),
+              caller.getLoggableName(),
+              cd.getId().get());
+          problems.add(
+              ChangeProblem.create(
+                  cd.getId(),
+                  String.format(
+                      "Insufficient permission to submit change %d on behalf of user %s",
+                      cd.getId().get(), caller.getLoggableName())));
+          return;
+        }
+      }
+      if (!cd.change().isNew()) {
+        if (!(cd.change().isMerged() && allowMerged)) {
+          problems.add(
+              ChangeProblem.create(
+                  cd.getId(),
+                  String.format(
+                      "Change %d is %s", cd.getId().get(), ChangeUtil.status(cd.change()))));
+          return;
+        }
+      }
+      if (cd.change().isWorkInProgress()) {
+        problems.add(
+            ChangeProblem.create(
+                cd.getId(),
+                String.format("Change %d is marked work in progress", cd.getId().get())));
+        return;
+      }
+      try {
+        checkSubmitRequirements(cd);
+      } catch (ResourceConflictException e) {
+        // ResourceConflictException is thrown means submit requirement is not fulfilled.
+        problems.add(
+            ChangeProblem.create(
+                cd.getId(),
+                triggeringChangeId.equals(cd.getId())
+                    ? String.format("Change %s is not ready: %s", cd.getId(), e.getMessage())
+                    : String.format(
+                        "Change %s must be submitted with change %s but %s is not ready: %s",
+                        triggeringChangeId, cd.getId(), cd.getId(), e.getMessage())));
+      }
+    } catch (StorageException | PermissionBackendException e) {
+      String msg = "Error checking submit rules for change";
+      logger.atWarning().withCause(e).log("%s %s", msg, triggeringChangeId);
+      problems.add(ChangeProblem.create(cd.getId(), msg));
+    }
+  }
+
   /**
    * Returns a list of messages describing what prevents the current change from being submitted.
    *
    * <p>The method checks all changes in the {@code cs} for their current status, submitability and
-   * permissions.
+   * permissions and returns one change per change in the set that can't be submitted.
    *
    * @param triggeringChange Change for which merge/submit action was initiated
    * @param cs Set of changes that the current change depends on
    * @param allowMerged True if change being already merged is not a problem to be reported
    * @param permissionBackend Interface for checking user ACLs
-   * @param caller The user who is triggering a merge
+   * @param caller the identity of the user that is recorded as the one performing the merge. In
+   *     case of impersonation {@code caller.getRealUser()} contains the user triggering the merge.
    * @return List of problems preventing merge
    */
   public static ImmutableList<ChangeProblem> checkCommonSubmitProblems(
@@ -472,7 +591,9 @@
     if (cs.furtherHiddenChanges()) {
       logger.atFine().log(
           "Change %d cannot be submitted by user %s because it depends on hidden changes: %s",
-          triggeringChange.getId().get(), caller.getLoggableName(), cs.nonVisibleChanges());
+          triggeringChange.getId().get(),
+          caller.getRealUser().getLoggableName(),
+          cs.nonVisibleChanges());
       problems.add(
           ChangeProblem.create(
               triggeringChange.getId(),
@@ -480,67 +601,8 @@
                   "Change %d depends on other hidden changes", triggeringChange.getId().get())));
     }
     for (ChangeData cd : cs.changes()) {
-      try {
-        Set<ChangePermission> can =
-            permissionBackend
-                .user(caller)
-                .change(cd)
-                .test(EnumSet.of(ChangePermission.READ, ChangePermission.SUBMIT));
-        if (!can.contains(ChangePermission.READ)) {
-          // The READ permission should already be handled during generation of ChangeSet, however
-          // MergeSuperSetComputation is an interface and on API level doesn't guarantee that this
-          // have been verified for all changes. Additionally, this protects against potential
-          // issues due to staleness.
-          logger.atFine().log(
-              "Change %d cannot be submitted by user %s because it depends on change %d which the"
-                  + "user cannot read",
-              triggeringChange.getId().get(), caller.getLoggableName(), cd.getId().get());
-          problems.add(
-              ChangeProblem.create(
-                  cd.getId(),
-                  String.format(
-                      "Change %d depends on other hidden changes",
-                      triggeringChange.getId().get())));
-        } else if (!can.contains(ChangePermission.SUBMIT)) {
-          logger.atFine().log(
-              "Change %d cannot be submitted by user %s because it depends on change %d which the"
-                  + "user cannot submit",
-              triggeringChange.getId().get(), caller.getLoggableName(), cd.getId().get());
-          problems.add(
-              ChangeProblem.create(
-                  cd.getId(),
-                  String.format("Insufficient permission to submit change %d", cd.getId().get())));
-        } else if (!cd.change().isNew()) {
-          if (!(cd.change().isMerged() && allowMerged)) {
-            problems.add(
-                ChangeProblem.create(
-                    cd.getId(),
-                    String.format(
-                        "Change %d is %s", cd.getId().get(), ChangeUtil.status(cd.change()))));
-          }
-        } else if (cd.change().isWorkInProgress()) {
-          problems.add(
-              ChangeProblem.create(
-                  cd.getId(),
-                  String.format("Change %d is marked work in progress", cd.getId().get())));
-        } else {
-          checkSubmitRequirements(cd);
-        }
-      } catch (ResourceConflictException e) {
-        // Exception is thrown means submit requirement is not fulfilled.
-        problems.add(
-            ChangeProblem.create(
-                cd.getId(),
-                triggeringChange.getId().equals(cd.getId())
-                    ? String.format("Change %s is not ready: %s", cd.getId(), e.getMessage())
-                    : String.format(
-                        "Change %s must be submitted with change %s but %s is not ready: %s",
-                        triggeringChange.getId(), cd.getId(), cd.getId(), e.getMessage())));
-      } catch (StorageException | PermissionBackendException e) {
-        String msg = "Error checking submit rules for change";
-        logger.atWarning().withCause(e).log("%s %s", msg, triggeringChange.getId());
-        problems.add(ChangeProblem.create(cd.getId(), msg));
-      }
+      addProblemForChange(
+          triggeringChange.getId(), cd, allowMerged, permissionBackend, caller, problems);
     }
     return problems.build();
   }
@@ -590,14 +652,21 @@
    * integration strategy.
    *
    * @param change the change to be merged.
-   * @param caller the identity of the caller
+   * @param caller the identity of the user that is recorded as the one performing the merge. In
+   *     case of impersonation {@code caller.getRealUser()} contains the user triggering the merge.
    * @param checkSubmitRules whether the prolog submit rules should be evaluated
    * @param submitInput parameters regarding the merge
+   * @param dryrun if true, this includes calculating all projects affected by the submission,
+   *     checking for possible submission problems (ACLs, merge conflicts, etc) but not the merge
+   *     itself.
    * @throws RestApiException if an error occurred.
    * @throws PermissionBackendException if permissions can't be checked
    * @throws IOException an error occurred reading from NoteDb.
    * @return the merged change
    */
+  // TODO: dryrun was introduced in https://gerrit-review.git.corp.google.com/c/gerrit/+/82911 and
+  // has never been used. Consider removing it. Since it was never used  and this file has been
+  // through many refactorings since, it's likely that the implementation is broken.
   @CanIgnoreReturnValue
   public Change merge(
       Change change,
@@ -731,7 +800,7 @@
                     Change reloadChange = change;
                     ChangeSet indexBackedMergeChangeSet =
                         mergeSuperSet.completeChangeSet(
-                            reloadChange, caller, /* includingTopicClosure= */ false);
+                            reloadChange, caller.getRealUser(), /* includingTopicClosure= */ false);
                     if (!indexBackedMergeChangeSet.ids().contains(reloadChange.getId())) {
                       // indexBackedChangeSet contains only open changes, if the change is missing
                       // in this set it might be that the change was concurrently submitted in the
@@ -1123,7 +1192,8 @@
                 .map(c -> ObjectId.toString(c))
                 .collect(joining(", "));
         logger.atWarning().log(
-            "Timeout during hasImplicitMerge calculation. Number of iterations: %s, commitsToSubmit: %s",
+            "Timeout during hasImplicitMerge calculation. Number of iterations: %s,"
+                + " commitsToSubmit: %s",
             iterationCount, allCommits);
         return true;
       }
diff --git a/java/com/google/gerrit/server/update/BatchUpdates.java b/java/com/google/gerrit/server/update/BatchUpdates.java
index 2f9ef84..4716246 100644
--- a/java/com/google/gerrit/server/update/BatchUpdates.java
+++ b/java/com/google/gerrit/server/update/BatchUpdates.java
@@ -48,6 +48,13 @@
 import java.util.Objects;
 import java.util.function.Function;
 
+/**
+ * Runs execute methods of a collection of {@link BatchUpdate}s calling listeners when appropriate.
+ *
+ * <p>This class does not maintain any state about the updates it executes. The only reason it is
+ * non-static is to provide convenient access to {@link ChangeData.Factory} without needing to
+ * provide one as an argument.
+ */
 @Singleton
 public class BatchUpdates {
   public class Result {
diff --git a/java/com/google/gerrit/server/update/SubmissionExecutor.java b/java/com/google/gerrit/server/update/SubmissionExecutor.java
index 762de57..ff46181 100644
--- a/java/com/google/gerrit/server/update/SubmissionExecutor.java
+++ b/java/com/google/gerrit/server/update/SubmissionExecutor.java
@@ -21,6 +21,7 @@
 import java.util.Optional;
 import java.util.stream.Collectors;
 
+/** Wrapper class for calling BatchUpdates.execute() that manages calls to submission listeners. */
 public class SubmissionExecutor {
   private final BatchUpdates batchUpdates;
   private final ImmutableList<SubmissionListener> submissionListeners;
diff --git a/javatests/com/google/gerrit/acceptance/rest/account/ImpersonationIT.java b/javatests/com/google/gerrit/acceptance/rest/account/ImpersonationIT.java
index 2e706b8..7de689d 100644
--- a/javatests/com/google/gerrit/acceptance/rest/account/ImpersonationIT.java
+++ b/javatests/com/google/gerrit/acceptance/rest/account/ImpersonationIT.java
@@ -689,6 +689,15 @@
   }
 
   @Test
+  @UseLocalDisk
+  public void submitOnBehalfOf_usersSubmitPermissionIgnored() throws Exception {
+    // user is part of the newGroup, block their submit permission
+    blockSubmit(newGroup);
+
+    testSubmitOnBehalfOf(project, admin2, user);
+  }
+
+  @Test
   public void submitOnBehalfOfFailsWhenUserCannotSeeDestinationRef() throws Exception {
     blockRead(newGroup);
 
@@ -926,6 +935,14 @@
         .update();
   }
 
+  private void blockSubmit(GroupInfo group) throws Exception {
+    projectOperations
+        .project(project)
+        .forUpdate()
+        .add(block(Permission.SUBMIT).ref("refs/heads/*").group(AccountGroup.uuid(group.id)))
+        .update();
+  }
+
   private void allowRunAs() throws Exception {
     projectOperations
         .allProjectsForUpdate()
diff --git a/javatests/com/google/gerrit/acceptance/server/change/SubmittedTogetherIT.java b/javatests/com/google/gerrit/acceptance/server/change/SubmittedTogetherIT.java
index 4ee5967..6fabd1a 100644
--- a/javatests/com/google/gerrit/acceptance/server/change/SubmittedTogetherIT.java
+++ b/javatests/com/google/gerrit/acceptance/server/change/SubmittedTogetherIT.java
@@ -16,26 +16,33 @@
 
 import static com.google.common.truth.Truth.assertThat;
 import static com.google.gerrit.acceptance.GitUtil.pushHead;
+import static com.google.gerrit.acceptance.testsuite.project.TestProjectUpdate.allow;
 import static com.google.gerrit.acceptance.testsuite.project.TestProjectUpdate.block;
 import static com.google.gerrit.extensions.api.changes.SubmittedTogetherOption.NON_VISIBLE_CHANGES;
 import static com.google.gerrit.server.group.SystemGroupBackend.REGISTERED_USERS;
 import static com.google.gerrit.testing.GerritJUnit.assertThrows;
 
+import com.google.common.collect.ImmutableList;
 import com.google.gerrit.acceptance.AbstractDaemonTest;
 import com.google.gerrit.acceptance.GitUtil;
 import com.google.gerrit.acceptance.Sandboxed;
+import com.google.gerrit.acceptance.TestAccount;
 import com.google.gerrit.acceptance.TestProjectInput;
 import com.google.gerrit.acceptance.config.GerritConfig;
 import com.google.gerrit.acceptance.testsuite.project.ProjectOperations;
 import com.google.gerrit.acceptance.testsuite.request.RequestScopeOperations;
+import com.google.gerrit.entities.AccountGroup;
 import com.google.gerrit.entities.BranchNameKey;
 import com.google.gerrit.entities.Permission;
 import com.google.gerrit.entities.Project;
+import com.google.gerrit.extensions.api.changes.SubmitInput;
 import com.google.gerrit.extensions.api.changes.SubmittedTogetherInfo;
+import com.google.gerrit.extensions.api.groups.GroupInput;
 import com.google.gerrit.extensions.client.ChangeStatus;
 import com.google.gerrit.extensions.client.ListChangesOption;
 import com.google.gerrit.extensions.client.SubmitType;
 import com.google.gerrit.extensions.common.FileInfo;
+import com.google.gerrit.extensions.common.GroupInfo;
 import com.google.gerrit.extensions.common.RevisionInfo;
 import com.google.gerrit.extensions.restapi.ResourceConflictException;
 import com.google.gerrit.testing.ConfigSuite;
@@ -377,6 +384,139 @@
     }
   }
 
+  @Test
+  public void permissionToSubmitAsForSomeChangesInTopic() throws Exception {
+    projectOperations
+        .project(project)
+        .forUpdate()
+        .add(allow(Permission.SUBMIT_AS).ref("refs/heads/master").group(REGISTERED_USERS))
+        .add(block(Permission.SUBMIT_AS).ref("refs/heads/testbranch").group(REGISTERED_USERS))
+        .update();
+
+    createBranch(BranchNameKey.create(getProject(), "testbranch"));
+    RevCommit initialHead = projectOperations.project(project).getHead("master");
+    // Create two independent commits and push.
+    RevCommit c1_1 = commitBuilder().add("a.txt", "1").message("subject: 1").create();
+    String id1 = getChangeId(c1_1);
+    pushHead(testRepo, "refs/for/master%topic=" + name("connectingTopic"), false);
+
+    testRepo.reset(initialHead);
+    RevCommit c2_1 = commitBuilder().add("b.txt", "2").message("subject: 2").create();
+    String id2 = getChangeId(c2_1);
+    pushHead(testRepo, "refs/for/testbranch%topic=" + name("connectingTopic"), false);
+
+    approve(id1);
+    approve(id2);
+    SubmitInput in = new SubmitInput();
+    in.onBehalfOf = accountCreator.user2().email();
+    if (isSubmitWholeTopicEnabled()) {
+      ResourceConflictException e =
+          assertThrows(
+              ResourceConflictException.class, () -> gApi.changes().id(id1).current().submit(in));
+      assertThat(e.getMessage())
+          .contains(
+              String.format(
+                  "Insufficient permission to submit change %d on behalf of user %s",
+                  gApi.changes().id(id2).get()._number, accountCreator.user2().username()));
+    } else {
+      gApi.changes().id(id1).current().submit(in);
+      assertMerged(id1);
+      assertNotMerged(id2);
+    }
+  }
+
+  @Test
+  public void submitAs_onBehalfOfUserNoReadPermissionToSomeChanges() throws Exception {
+    GroupInfo newGroup = createGroupForUser(accountCreator.user2());
+    projectOperations
+        .project(project)
+        .forUpdate()
+        .add(
+            block(Permission.READ)
+                .ref("refs/heads/testbranch")
+                .group(AccountGroup.uuid(newGroup.id)))
+        .add(allow(Permission.SUBMIT_AS).ref("refs/heads/*").group(REGISTERED_USERS))
+        .update();
+
+    createBranch(BranchNameKey.create(getProject(), "testbranch"));
+    RevCommit initialHead = projectOperations.project(project).getHead("master");
+    // Create two independent commits and push.
+    RevCommit c1_1 = commitBuilder().add("a.txt", "1").message("subject: 1").create();
+    String id1 = getChangeId(c1_1);
+    pushHead(testRepo, "refs/for/master%topic=" + name("connectingTopic"), false);
+
+    testRepo.reset(initialHead);
+    RevCommit c2_1 = commitBuilder().add("b.txt", "2").message("subject: 2").create();
+    String id2 = getChangeId(c2_1);
+    pushHead(testRepo, "refs/for/testbranch%topic=" + name("connectingTopic"), false);
+
+    approve(id1);
+    approve(id2);
+    SubmitInput in = new SubmitInput();
+    in.onBehalfOf = accountCreator.user2().email();
+    if (isSubmitWholeTopicEnabled()) {
+      ResourceConflictException e =
+          assertThrows(
+              ResourceConflictException.class, () -> gApi.changes().id(id1).current().submit(in));
+      assertThat(e.getMessage())
+          .contains(
+              String.format(
+                  "On-behalf-of user %s lacks permission to read change %d",
+                  accountCreator.user2().username(), gApi.changes().id(id2).get()._number));
+    } else {
+      gApi.changes().id(id1).current().submit(in);
+      assertMerged(id1);
+      assertNotMerged(id2);
+    }
+  }
+
+  @Test
+  public void submitAs_succeeds() throws Exception {
+    GroupInfo newGroup = createGroupForUser(accountCreator.user2());
+    projectOperations
+        .project(project)
+        .forUpdate()
+        .add(
+            block(Permission.SUBMIT)
+                .ref("refs/heads/testbranch")
+                .group(AccountGroup.uuid(newGroup.id)))
+        .add(allow(Permission.SUBMIT_AS).ref("refs/heads/*").group(REGISTERED_USERS))
+        .update();
+
+    createBranch(BranchNameKey.create(getProject(), "testbranch"));
+    RevCommit initialHead = projectOperations.project(project).getHead("master");
+    // Create two independent commits and push.
+    RevCommit c1_1 = commitBuilder().add("a.txt", "1").message("subject: 1").create();
+    String id1 = getChangeId(c1_1);
+    pushHead(testRepo, "refs/for/master%topic=" + name("connectingTopic"), false);
+
+    testRepo.reset(initialHead);
+    RevCommit c2_1 = commitBuilder().add("b.txt", "2").message("subject: 2").create();
+    String id2 = getChangeId(c2_1);
+    pushHead(testRepo, "refs/for/testbranch%topic=" + name("connectingTopic"), false);
+
+    approve(id1);
+    approve(id2);
+    SubmitInput in = new SubmitInput();
+    in.onBehalfOf = accountCreator.user2().email();
+    if (isSubmitWholeTopicEnabled()) {
+      gApi.changes().id(id1).current().submit(in);
+      assertMerged(id1);
+      assertMerged(id2);
+    } else {
+      gApi.changes().id(id1).current().submit(in);
+      assertMerged(id1);
+      assertNotMerged(id2);
+    }
+  }
+
+  private GroupInfo createGroupForUser(TestAccount user) throws Exception {
+    GroupInput gi = new GroupInput();
+    gi.name = name("New-Group");
+    gi.members = ImmutableList.of(user.id().toString());
+    return gApi.groups().create(gi).get();
+  }
+
   private String getChangeId(RevCommit c) throws Exception {
     return GitUtil.getChangeId(testRepo, c).get();
   }
@@ -388,4 +528,8 @@
   private void assertMerged(String changeId) throws Exception {
     assertThat(gApi.changes().id(changeId).get().status).isEqualTo(ChangeStatus.MERGED);
   }
+
+  private void assertNotMerged(String changeId) throws Exception {
+    assertThat(gApi.changes().id(changeId).get().status).isEqualTo(ChangeStatus.NEW);
+  }
 }