Closes #2491: Set Owner of Task when Transferring

history/taskana-simplehistory-provider/src/test/java/acceptance/events/task/CreateHistoryEventOnTaskTransferAccTest.java

@@ -181,6 +181,156 @@ class CreateHistoryEventOnTaskTransferAccTest extends AbstractAccTest {
     return DynamicTest.stream(testCases.iterator(), Triplet::getLeft, test);
   }
 
+  @WithAccessId(user = "admin")
+  @TestFactory
+  Stream<DynamicTest> should_CreateTransferredHistoryEvent_When_TaskIsTransferredWithOwner() {
+    List<Quadruple<String, String, String, Consumer<String>>> testCases =
+        List.of(
+            /*
+            The workbasketId of the source Workbasket is parametrized. Putting the tested Tasks
+            into the same Workbasket would result in changes to the test data. This would require
+            changing tests that already use the tested Tasks. That's why workbasketId is
+            parametrized.
+            */
+            Quadruple.of(
+                "Using WorkbasketId; Task doesn't have an Attachment"
+                    + " or any secondary Object References",
+                "TKI:000000000000000000000000000000000005",
+                "WBI:100000000000000000000000000000000001",
+                wrap(
+                    (String taskId) ->
+                        taskService.transferWithOwner(
+                                taskId, "WBI:100000000000000000000000000000000007", "user-1-2"))),
+            Quadruple.of(
+                "Using WorkbasketId; Task has Attachment and secondary Object Reference",
+                "TKI:000000000000000000000000000000000001",
+                "WBI:100000000000000000000000000000000006",
+                wrap(
+                    (String taskId) ->
+                        taskService.transferWithOwner(
+                                taskId, "WBI:100000000000000000000000000000000007", "user-1-2"))),
+            Quadruple.of(
+                "Using WorkbasketKey and Domain",
+                "TKI:000000000000000000000000000000000006",
+                "WBI:100000000000000000000000000000000001",
+                wrap(
+                    (String taskId) ->
+                        taskService.transferWithOwner(
+                                taskId, "USER-1-2", "DOMAIN_A", "user-1-2"))));
+    ThrowingConsumer<Quadruple<String, String, String, Consumer<String>>> test =
+        q -> {
+          String taskId = q.getSecond();
+          Consumer<String> transferMethod = q.getFourth();
+
+          TaskHistoryQueryMapper taskHistoryQueryMapper = getHistoryQueryMapper();
+
+          List<TaskHistoryEvent> events =
+              taskHistoryQueryMapper.queryHistoryEvents(
+                  (TaskHistoryQueryImpl) historyService.createTaskHistoryQuery().taskIdIn(taskId));
+
+          assertThat(events).isEmpty();
+
+          transferMethod.accept(taskId);
+
+          events =
+              taskHistoryQueryMapper.queryHistoryEvents(
+                  (TaskHistoryQueryImpl) historyService.createTaskHistoryQuery().taskIdIn(taskId));
+
+          assertThat(events).hasSize(1);
+          String sourceWorkbasketId = q.getThird();
+          assertTransferHistoryEvent(
+              events.get(0).getId(),
+              sourceWorkbasketId,
+              "WBI:100000000000000000000000000000000007",
+              "admin");
+        };
+    return DynamicTest.stream(testCases.iterator(), Quadruple::getFirst, test);
+  }
+
+  @WithAccessId(user = "admin")
+  @TestFactory
+  Stream<DynamicTest> should_CreateTransferredHistoryEvents_When_TaskBulkTransferWithOwner() {
+    List<Triplet<String, Map<String, String>, Consumer<List<String>>>> testCases =
+        List.of(
+            /*
+            The workbasketId of the source Workbasket is parametrized. Putting the tested Tasks
+            into the same Workbasket would result in changes to the test data. This would require
+            changing tests that already use the tested Tasks. That's why workbasketId is
+            parametrized.
+            */
+            Triplet.of(
+                "Using WorkbasketId",
+                Map.ofEntries(
+                    Map.entry(
+                        "TKI:000000000000000000000000000000000010",
+                        "WBI:100000000000000000000000000000000001"),
+                    Map.entry(
+                        "TKI:000000000000000000000000000000000011",
+                        "WBI:100000000000000000000000000000000001"),
+                    Map.entry(
+                        "TKI:000000000000000000000000000000000012",
+                        "WBI:100000000000000000000000000000000001")),
+                wrap(
+                    (List<String> taskIds) ->
+                        taskService.transferTasksWithOwner(
+                            "WBI:100000000000000000000000000000000007", taskIds, "user-1-2"))),
+            Triplet.of(
+                "Using WorkbasketKey and Domain",
+                Map.ofEntries(
+                    Map.entry(
+                        "TKI:000000000000000000000000000000000013",
+                        "WBI:100000000000000000000000000000000001"),
+                    Map.entry(
+                        "TKI:000000000000000000000000000000000014",
+                        "WBI:100000000000000000000000000000000001"),
+                    Map.entry(
+                        "TKI:000000000000000000000000000000000015",
+                        "WBI:100000000000000000000000000000000001")),
+                wrap(
+                    (List<String> taskIds) ->
+                        taskService.transferTasksWithOwner(
+                                "USER-1-2", "DOMAIN_A", taskIds, "user-1-2"))));
+    ThrowingConsumer<Triplet<String, Map<String, String>, Consumer<List<String>>>> test =
+        t -> {
+          Map<String, String> taskIds = t.getMiddle();
+          Consumer<List<String>> transferMethod = t.getRight();
+
+          TaskHistoryQueryMapper taskHistoryQueryMapper = getHistoryQueryMapper();
+
+          List<TaskHistoryEvent> events =
+              taskHistoryQueryMapper.queryHistoryEvents(
+                  (TaskHistoryQueryImpl)
+                      historyService
+                          .createTaskHistoryQuery()
+                          .taskIdIn(taskIds.keySet().toArray(new String[0])));
+
+          assertThat(events).isEmpty();
+
+          transferMethod.accept(new ArrayList<>(taskIds.keySet()));
+
+          events =
+              taskHistoryQueryMapper.queryHistoryEvents(
+                  (TaskHistoryQueryImpl)
+                      historyService
+                          .createTaskHistoryQuery()
+                          .taskIdIn(taskIds.keySet().toArray(new String[0])));
+
+          assertThat(events)
+              .extracting(TaskHistoryEvent::getTaskId)
+              .containsExactlyInAnyOrderElementsOf(taskIds.keySet());
+
+          for (TaskHistoryEvent event : events) {
+            assertTransferHistoryEvent(
+                event.getId(),
+                taskIds.get(event.getTaskId()),
+                "WBI:100000000000000000000000000000000007",
+                "admin");
+          }
+        };
+
+    return DynamicTest.stream(testCases.iterator(), Triplet::getLeft, test);
+  }
+
   private void assertTransferHistoryEvent(
       String eventId, String expectedOldValue, String expectedNewValue, String expectedUser)
       throws Exception {

lib/taskana-core/src/main/java/pro/taskana/task/api/TaskService.java

@@ -471,6 +471,11 @@ public interface TaskService {
    * Transfers a {@linkplain Task} to another {@linkplain Workbasket} while always setting
    * {@linkplain Task#isTransferred() isTransferred} to true.
    *
+   * @param taskId the {@linkplain Task#getId() id} of the {@linkplain Task} which should be
+   *     transferred
+   * @param destinationWorkbasketId the {@linkplain Workbasket#getId() id} of the target {@linkplain
+   *     Workbasket}
+   * @return the transferred {@linkplain Task}
    * @see #transfer(String, String, boolean)
    */
   @SuppressWarnings("checkstyle:JavadocMethod")
@@ -513,6 +518,13 @@ public interface TaskService {
    * Transfers a {@linkplain Task} to another {@linkplain Workbasket} while always setting
    * {@linkplain Task#isTransferred isTransferred} .
    *
+   * @param taskId the {@linkplain Task#getId() id} of the {@linkplain Task} which should be
+   *     transferred
+   * @param workbasketKey the {@linkplain Workbasket#getKey() key} of the target {@linkplain
+   *     Workbasket}
+   * @param domain the {@linkplain Workbasket#getDomain() domain} of the target {@linkplain
+   *     Workbasket}
+   * @return the transferred {@linkplain Task}
    * @see #transfer(String, String, String, boolean)
    */
   @SuppressWarnings("checkstyle:JavadocMethod")
@@ -553,10 +565,122 @@ public interface TaskService {
           NotAuthorizedOnWorkbasketException,
           InvalidTaskStateException;
 
+  /**
+   * Transfers a {@linkplain Task} to another {@linkplain Workbasket} and sets the owner of the
+   * {@linkplain Task} in the new {@linkplain Workbasket} to owner while always setting {@linkplain
+   * Task#isTransferred() isTransferred} to true.
+   *
+   * @param taskId the {@linkplain Task#getId() id} of the {@linkplain Task} which should be
+   *     transferred
+   * @param destinationWorkbasketId the {@linkplain Workbasket#getId() id} of the target {@linkplain
+   *     Workbasket}
+   * @param owner the owner of the {@linkplain Task} after it has been transferred
+   * @return the transferred {@linkplain Task}
+   * @see #transferWithOwner(String, String, String, boolean)
+   */
+  @SuppressWarnings("checkstyle:JavadocMethod")
+  default Task transferWithOwner(String taskId, String destinationWorkbasketId, String owner)
+      throws TaskNotFoundException,
+          WorkbasketNotFoundException,
+          NotAuthorizedOnWorkbasketException,
+          InvalidTaskStateException {
+    return transferWithOwner(taskId, destinationWorkbasketId, owner, true);
+  }
+
+  /**
+   * Transfers a {@linkplain Task} to another {@linkplain Workbasket} and sets the owner of the
+   * {@linkplain Task}.
+   *
+   * <p>The transfer resets {@linkplain Task#isRead() isRead} and sets {@linkplain
+   * Task#isTransferred() isTransferred} if setTransferFlag is true.
+   *
+   * @param taskId the {@linkplain Task#getId() id} of the {@linkplain Task} which should be
+   *     transferred
+   * @param destinationWorkbasketId the {@linkplain Workbasket#getId() id} of the target {@linkplain
+   *     Workbasket}
+   * @param owner the owner of the {@linkplain Task} after it has been transferred
+   * @param setTransferFlag controls whether to set {@linkplain Task#isTransferred() isTransferred}
+   *     to true or not
+   * @return the transferred {@linkplain Task}
+   * @throws TaskNotFoundException if the {@linkplain Task} with taskId wasn't found
+   * @throws WorkbasketNotFoundException if the target {@linkplain Workbasket} was not found
+   * @throws NotAuthorizedOnWorkbasketException if the current user has no {@linkplain
+   *     WorkbasketPermission#READ} for the source {@linkplain Workbasket} or no {@linkplain
+   *     WorkbasketPermission#TRANSFER} for the target {@linkplain Workbasket}
+   * @throws InvalidTaskStateException if the {@linkplain Task} is in one of the {@linkplain
+   *     TaskState#END_STATES}
+   */
+  Task transferWithOwner(
+      String taskId, String destinationWorkbasketId, String owner, boolean setTransferFlag)
+      throws TaskNotFoundException,
+          WorkbasketNotFoundException,
+          NotAuthorizedOnWorkbasketException,
+          InvalidTaskStateException;
+
+  /**
+   * Transfers a {@linkplain Task} to another {@linkplain Workbasket} and sets the owner of the
+   * {@linkplain Task} in new {@linkplain Workbasket} to owner while always setting {@linkplain
+   * Task#isTransferred isTransferred} .
+   *
+   * @param taskId the {@linkplain Task#getId() id} of the {@linkplain Task} which should be
+   *     transferred
+   * @param workbasketKey the {@linkplain Workbasket#getKey() key} of the target {@linkplain
+   *     Workbasket}
+   * @param domain the {@linkplain Workbasket#getDomain() domain} of the target {@linkplain
+   *     Workbasket}
+   * @param owner the owner of the {@linkplain Task} after it has been transferred
+   * @return the transferred {@linkplain Task}
+   * @see #transferWithOwner(String, String, String, String, boolean)
+   */
+  @SuppressWarnings("checkstyle:JavadocMethod")
+  default Task transferWithOwner(String taskId, String workbasketKey, String domain, String owner)
+      throws TaskNotFoundException,
+          WorkbasketNotFoundException,
+          NotAuthorizedOnWorkbasketException,
+          InvalidTaskStateException {
+    return transferWithOwner(taskId, workbasketKey, domain, owner, true);
+  }
+
+  /**
+   * Transfers a {@linkplain Task} to another {@linkplain Workbasket} and sets the owner of the
+   * {@linkplain Task}.
+   *
+   * <p>The transfer resets {@linkplain Task#isRead() isRead} and sets {@linkplain
+   * Task#isTransferred() isTransferred} if setTransferFlag is true.
+   *
+   * @param taskId the {@linkplain Task#getId() id} of the {@linkplain Task} which should be
+   *     transferred
+   * @param workbasketKey the {@linkplain Workbasket#getKey() key} of the target {@linkplain
+   *     Workbasket}
+   * @param domain the {@linkplain Workbasket#getDomain() domain} of the target {@linkplain
+   *     Workbasket}
+   * @param owner the owner of the {@linkplain Task} after it has been transferred
+   * @param setTransferFlag controls whether to set {@linkplain Task#isTransferred() isTransferred}
+   *     or not
+   * @return the transferred {@linkplain Task}
+   * @throws TaskNotFoundException if the {@linkplain Task} with taskId was not found
+   * @throws WorkbasketNotFoundException if the target {@linkplain Workbasket} was not found
+   * @throws NotAuthorizedOnWorkbasketException if the current user has no {@linkplain
+   *     WorkbasketPermission#READ} for the source {@linkplain Workbasket} or no {@linkplain
+   *     WorkbasketPermission#TRANSFER} for the target {@linkplain Workbasket}
+   * @throws InvalidTaskStateException if the {@linkplain Task} is in one of the {@linkplain
+   *     TaskState#END_STATES}
+   */
+  Task transferWithOwner(
+      String taskId, String workbasketKey, String domain, String owner, boolean setTransferFlag)
+      throws TaskNotFoundException,
+          WorkbasketNotFoundException,
+          NotAuthorizedOnWorkbasketException,
+          InvalidTaskStateException;
+
   /**
    * Transfers a List of {@linkplain Task Tasks} to another {@linkplain Workbasket} while always
    * setting {@linkplain Task#isTransferred isTransferred} to true.
    *
+   * @param destinationWorkbasketId {@linkplain Workbasket#getId() id} of the target {@linkplain
+   *     Workbasket}
+   * @param taskIds List of source {@linkplain Task Tasks} which will be moved
+   * @return Bulkresult with {@linkplain Task#getId() ids} and Error for each failed transactions
    * @see #transferTasks(String, List, boolean)
    */
   @SuppressWarnings("checkstyle:JavadocMethod")
@@ -598,6 +722,10 @@ public interface TaskService {
    * Transfers a List of {@linkplain Task Tasks} to another {@linkplain Workbasket} while always
    * setting {@linkplain Task#isTransferred() isTransferred} to true.
    *
+   * @param destinationWorkbasketKey target {@linkplain Workbasket#getKey() key}
+   * @param destinationWorkbasketDomain target {@linkplain Workbasket#getDomain() domain}
+   * @param taskIds List of source {@linkplain Task Tasks} which will be moved
+   * @return Bulkresult with {@linkplain Task#getId() ids} and Error for each failed transactions
    * @see #transferTasks(String, String, List, boolean)
    */
   @SuppressWarnings("checkstyle:JavadocMethod")
@@ -639,6 +767,112 @@ public interface TaskService {
           WorkbasketNotFoundException,
           NotAuthorizedOnWorkbasketException;
 
+  /**
+   * Transfers a List of {@linkplain Task Tasks} to another {@linkplain Workbasket} and sets the
+   * owner of the {@linkplain Task Tasks} to owner while always setting {@linkplain
+   * Task#isTransferred isTransferred} to true.
+   *
+   * @param destinationWorkbasketId {@linkplain Workbasket#getId() id} of the target {@linkplain
+   *     Workbasket}
+   * @param taskIds List of source {@linkplain Task Tasks} which will be moved
+   * @param owner the owner of the {@linkplain Task Tasks} after they have been transferred
+   * @see #transferTasksWithOwner(String, List, String, boolean)
+   */
+  @SuppressWarnings("checkstyle:JavadocMethod")
+  default BulkOperationResults<String, TaskanaException> transferTasksWithOwner(
+      String destinationWorkbasketId, List<String> taskIds, String owner)
+      throws InvalidArgumentException,
+          WorkbasketNotFoundException,
+          NotAuthorizedOnWorkbasketException {
+    return transferTasksWithOwner(destinationWorkbasketId, taskIds, owner, true);
+  }
+
+  /**
+   * Transfers a List of {@linkplain Task Tasks} to another {@linkplain Workbasket} and sets the
+   * owner of the {@linkplain Task Tasks}.
+   *
+   * <p>The transfer resets {@linkplain Task#isRead() isRead} and sets {@linkplain
+   * Task#isTransferred() isTransferred} if setTransferFlag is true. Exceptions will be thrown if
+   * the caller got no {@linkplain WorkbasketPermission} on the target or if the target {@linkplain
+   * Workbasket} doesn't exist. Other Exceptions will be stored and returned in the end.
+   *
+   * @param destinationWorkbasketId {@linkplain Workbasket#getId() id} of the target {@linkplain
+   *     Workbasket}
+   * @param taskIds List of source {@linkplain Task Tasks} which will be moved
+   * @param owner the owner of the {@linkplain Task Tasks} after they have been transferred
+   * @param setTransferFlag controls whether to set {@linkplain Task#isTransferred() isTransferred}
+   *     or not
+   * @return Bulkresult with {@linkplain Task#getId() ids} and Error for each failed transactions
+   * @throws NotAuthorizedOnWorkbasketException if the current user has no {@linkplain
+   *     WorkbasketPermission#READ} for the source {@linkplain Workbasket} or no {@linkplain
+   *     WorkbasketPermission#TRANSFER} for the target {@linkplain Workbasket}
+   * @throws InvalidArgumentException if the method parameters are empty or NULL
+   * @throws WorkbasketNotFoundException if the target {@linkplain Workbasket} can't be found
+   */
+  BulkOperationResults<String, TaskanaException> transferTasksWithOwner(
+      String destinationWorkbasketId, List<String> taskIds, String owner, boolean setTransferFlag)
+      throws InvalidArgumentException,
+          WorkbasketNotFoundException,
+          NotAuthorizedOnWorkbasketException;
+
+  /**
+   * Transfers a List of {@linkplain Task Tasks} to another {@linkplain Workbasket} and sets the
+   * owner of the {@linkplain Task Tasks} while always setting {@linkplain Task#isTransferred()
+   * isTransferred} to true.
+   *
+   * @param destinationWorkbasketKey target {@linkplain Workbasket#getKey() key}
+   * @param destinationWorkbasketDomain target {@linkplain Workbasket#getDomain() domain}
+   * @param taskIds List of source {@linkplain Task Tasks} which will be moved
+   * @param owner the new owner of the transferred tasks
+   * @return Bulkresult with {@linkplain Task#getId() ids} and Error for each failed transactions
+   * @see #transferTasksWithOwner(String, String, List, String, boolean)
+   */
+  @SuppressWarnings("checkstyle:JavadocMethod")
+  default BulkOperationResults<String, TaskanaException> transferTasksWithOwner(
+      String destinationWorkbasketKey,
+      String destinationWorkbasketDomain,
+      List<String> taskIds,
+      String owner)
+      throws InvalidArgumentException,
+          WorkbasketNotFoundException,
+          NotAuthorizedOnWorkbasketException {
+    return transferTasksWithOwner(
+        destinationWorkbasketKey, destinationWorkbasketDomain, taskIds, owner, true);
+  }
+
+  /**
+   * Transfers a List of {@linkplain Task Tasks} to another {@linkplain Workbasket} and sets the
+   * owner of the {@linkplain Task Tasks}.
+   *
+   * <p>The transfer resets {@linkplain Task#isRead() isRead} and sets {@linkplain
+   * Task#isTransferred() isTransferred} if setTransferFlag is true. Exceptions will be thrown if
+   * the caller got no {@linkplain WorkbasketPermission} on the target {@linkplain Workbasket} or if
+   * it doesn't exist. Other Exceptions will be stored and returned in the end.
+   *
+   * @param destinationWorkbasketKey target {@linkplain Workbasket#getKey() key}
+   * @param destinationWorkbasketDomain target {@linkplain Workbasket#getDomain() domain}
+   * @param taskIds List of {@linkplain Task#getId() ids} of source {@linkplain Task Tasks} which
+   *     will be moved
+   * @param owner the new owner of the transferred tasks
+   * @param setTransferFlag controls whether to set {@linkplain Task#isTransferred() isTransferred}
+   *     or not
+   * @return BulkResult with {@linkplain Task#getId() ids} and Error for each failed transactions
+   * @throws NotAuthorizedOnWorkbasketException if the current user has no {@linkplain
+   *     WorkbasketPermission#READ} for the source {@linkplain Workbasket} or no {@linkplain
+   *     WorkbasketPermission#TRANSFER} for the target {@linkplain Workbasket}
+   * @throws InvalidArgumentException if the method parameters are empty or NULL
+   * @throws WorkbasketNotFoundException if the target {@linkplain Workbasket} can't be found
+   */
+  BulkOperationResults<String, TaskanaException> transferTasksWithOwner(
+      String destinationWorkbasketKey,
+      String destinationWorkbasketDomain,
+      List<String> taskIds,
+      String owner,
+      boolean setTransferFlag)
+      throws InvalidArgumentException,
+          WorkbasketNotFoundException,
+          NotAuthorizedOnWorkbasketException;
+
   /**
    * Update a {@linkplain Task}.
    *

lib/taskana-core/src/main/java/pro/taskana/task/internal/TaskServiceImpl.java

@@ -474,6 +474,27 @@ public class TaskServiceImpl implements TaskService {
     return taskTransferrer.transfer(taskId, workbasketKey, domain, setTransferFlag);
   }
 
+  @Override
+  public Task transferWithOwner(
+      String taskId, String destinationWorkbasketId, String owner, boolean setTransferFlag)
+      throws TaskNotFoundException,
+          WorkbasketNotFoundException,
+          NotAuthorizedOnWorkbasketException,
+          InvalidTaskStateException {
+    return taskTransferrer.transferWithOwner(
+        taskId, destinationWorkbasketId, owner, setTransferFlag);
+  }
+
+  @Override
+  public Task transferWithOwner(
+      String taskId, String workbasketKey, String domain, String owner, boolean setTransferFlag)
+      throws TaskNotFoundException,
+          WorkbasketNotFoundException,
+          NotAuthorizedOnWorkbasketException,
+          InvalidTaskStateException {
+    return taskTransferrer.transferWithOwner(taskId, workbasketKey, domain, owner, setTransferFlag);
+  }
+
   @Override
   public Task setTaskRead(String taskId, boolean isRead)
       throws TaskNotFoundException, NotAuthorizedOnWorkbasketException {
@@ -631,6 +652,30 @@ public class TaskServiceImpl implements TaskService {
         taskIds, destinationWorkbasketKey, destinationWorkbasketDomain, setTransferFlag);
   }
 
+  @Override
+  public BulkOperationResults<String, TaskanaException> transferTasksWithOwner(
+      String destinationWorkbasketId, List<String> taskIds, String owner, boolean setTransferFlag)
+      throws InvalidArgumentException,
+          WorkbasketNotFoundException,
+          NotAuthorizedOnWorkbasketException {
+    return taskTransferrer.transferWithOwner(
+        taskIds, destinationWorkbasketId, owner, setTransferFlag);
+  }
+
+  @Override
+  public BulkOperationResults<String, TaskanaException> transferTasksWithOwner(
+      String destinationWorkbasketKey,
+      String destinationWorkbasketDomain,
+      List<String> taskIds,
+      String owner,
+      boolean setTransferFlag)
+      throws InvalidArgumentException,
+          WorkbasketNotFoundException,
+          NotAuthorizedOnWorkbasketException {
+    return taskTransferrer.transferWithOwner(
+        taskIds, destinationWorkbasketKey, destinationWorkbasketDomain, owner, setTransferFlag);
+  }
+
   @Override
   public void deleteTask(String taskId)
       throws TaskNotFoundException,

lib/taskana-core/src/main/java/pro/taskana/task/internal/TaskTransferrer.java

@@ -60,7 +60,7 @@ final class TaskTransferrer {
           InvalidTaskStateException {
     WorkbasketSummary destinationWorkbasket =
         workbasketService.getWorkbasket(destinationWorkbasketId).asSummary();
-    return transferSingleTask(taskId, destinationWorkbasket, setTransferFlag);
+    return transferSingleTask(taskId, destinationWorkbasket, null, setTransferFlag);
   }
 
   Task transfer(
@@ -74,7 +74,7 @@ final class TaskTransferrer {
           InvalidTaskStateException {
     WorkbasketSummary destinationWorkbasket =
         workbasketService.getWorkbasket(destinationWorkbasketKey, destinationDomain).asSummary();
-    return transferSingleTask(taskId, destinationWorkbasket, setTransferFlag);
+    return transferSingleTask(taskId, destinationWorkbasket, null, setTransferFlag);
   }
 
   BulkOperationResults<String, TaskanaException> transfer(
@@ -86,7 +86,7 @@ final class TaskTransferrer {
         workbasketService.getWorkbasket(destinationWorkbasketId).asSummary();
     checkDestinationWorkbasket(destinationWorkbasket);
 
-    return transferMultipleTasks(taskIds, destinationWorkbasket, setTransferFlag);
+    return transferMultipleTasks(taskIds, destinationWorkbasket, null, setTransferFlag);
   }
 
   BulkOperationResults<String, TaskanaException> transfer(
@@ -101,11 +101,65 @@ final class TaskTransferrer {
         workbasketService.getWorkbasket(destinationWorkbasketKey, destinationDomain).asSummary();
     checkDestinationWorkbasket(destinationWorkbasket);
 
-    return transferMultipleTasks(taskIds, destinationWorkbasket, setTransferFlag);
+    return transferMultipleTasks(taskIds, destinationWorkbasket, null, setTransferFlag);
+  }
+
+  BulkOperationResults<String, TaskanaException> transferWithOwner(
+      List<String> taskIds, String destinationWorkbasketId, String owner, boolean setTransferFlag)
+      throws WorkbasketNotFoundException,
+          InvalidArgumentException,
+          NotAuthorizedOnWorkbasketException {
+    WorkbasketSummary destinationWorkbasket =
+        workbasketService.getWorkbasket(destinationWorkbasketId).asSummary();
+    checkDestinationWorkbasket(destinationWorkbasket);
+
+    return transferMultipleTasks(taskIds, destinationWorkbasket, owner, setTransferFlag);
+  }
+
+  BulkOperationResults<String, TaskanaException> transferWithOwner(
+      List<String> taskIds,
+      String destinationWorkbasketKey,
+      String destinationDomain,
+      String owner,
+      boolean setTransferFlag)
+      throws WorkbasketNotFoundException,
+          InvalidArgumentException,
+          NotAuthorizedOnWorkbasketException {
+    WorkbasketSummary destinationWorkbasket =
+        workbasketService.getWorkbasket(destinationWorkbasketKey, destinationDomain).asSummary();
+    checkDestinationWorkbasket(destinationWorkbasket);
+
+    return transferMultipleTasks(taskIds, destinationWorkbasket, owner, setTransferFlag);
+  }
+
+  Task transferWithOwner(
+      String taskId, String destinationWorkbasketId, String owner, boolean setTransferFlag)
+      throws TaskNotFoundException,
+          WorkbasketNotFoundException,
+          NotAuthorizedOnWorkbasketException,
+          InvalidTaskStateException {
+    WorkbasketSummary destinationWorkbasket =
+        workbasketService.getWorkbasket(destinationWorkbasketId).asSummary();
+    return transferSingleTask(taskId, destinationWorkbasket, owner, setTransferFlag);
+  }
+
+  Task transferWithOwner(
+      String taskId,
+      String destinationWorkbasketKey,
+      String destinationDomain,
+      String owner,
+      boolean setTransferFlag)
+      throws TaskNotFoundException,
+          WorkbasketNotFoundException,
+          NotAuthorizedOnWorkbasketException,
+          InvalidTaskStateException {
+    WorkbasketSummary destinationWorkbasket =
+        workbasketService.getWorkbasket(destinationWorkbasketKey, destinationDomain).asSummary();
+    return transferSingleTask(taskId, destinationWorkbasket, owner, setTransferFlag);
   }
 
   private Task transferSingleTask(
-      String taskId, WorkbasketSummary destinationWorkbasket, boolean setTransferFlag)
+      String taskId, WorkbasketSummary destinationWorkbasket, String owner, boolean setTransferFlag)
       throws TaskNotFoundException,
           WorkbasketNotFoundException,
           NotAuthorizedOnWorkbasketException,
@@ -120,7 +174,7 @@ final class TaskTransferrer {
       WorkbasketSummary originWorkbasket = task.getWorkbasketSummary();
       checkPreconditionsForTransferTask(task, destinationWorkbasket, originWorkbasket);
 
-      applyTransferValuesForTask(task, destinationWorkbasket, setTransferFlag);
+      applyTransferValuesForTask(task, destinationWorkbasket, owner, setTransferFlag);
       taskMapper.update(task);
       if (historyEventManager.isEnabled()) {
         createTransferredEvent(
@@ -136,6 +190,7 @@ final class TaskTransferrer {
   private BulkOperationResults<String, TaskanaException> transferMultipleTasks(
       List<String> taskToBeTransferred,
       WorkbasketSummary destinationWorkbasket,
+      String owner,
       boolean setTransferFlag)
       throws InvalidArgumentException {
     if (taskToBeTransferred == null || taskToBeTransferred.isEmpty()) {
@@ -154,7 +209,7 @@ final class TaskTransferrer {
                   () -> taskService.createTaskQuery().idIn(taskIds.toArray(new String[0])).list());
       taskSummaries =
           filterOutTasksWhichDoNotMatchTransferCriteria(taskIds, taskSummaries, bulkLog);
-      updateTransferableTasks(taskSummaries, destinationWorkbasket, setTransferFlag);
+      updateTransferableTasks(taskSummaries, destinationWorkbasket, owner, setTransferFlag);
 
       return bulkLog;
     } finally {
@@ -254,6 +309,7 @@ final class TaskTransferrer {
   private void updateTransferableTasks(
       List<TaskSummary> taskSummaries,
       WorkbasketSummary destinationWorkbasket,
+      String owner,
       boolean setTransferFlag) {
     Map<TaskState, List<TaskSummary>> summariesByState = groupTasksByState(taskSummaries);
     for (Map.Entry<TaskState, List<TaskSummary>> entry : summariesByState.entrySet()) {
@@ -262,7 +318,7 @@ final class TaskTransferrer {
       if (!taskSummariesWithSameGoalState.isEmpty()) {
         TaskImpl updateObject = new TaskImpl();
         updateObject.setState(goalState);
-        applyTransferValuesForTask(updateObject, destinationWorkbasket, setTransferFlag);
+        applyTransferValuesForTask(updateObject, destinationWorkbasket, owner, setTransferFlag);
         taskMapper.updateTransfered(
             taskSummariesWithSameGoalState.stream()
                 .map(TaskSummary::getId)
@@ -275,7 +331,8 @@ final class TaskTransferrer {
                 TaskSummaryImpl newSummary = (TaskSummaryImpl) oldSummary.copy();
                 newSummary.setId(oldSummary.getId());
                 newSummary.setExternalId(oldSummary.getExternalId());
-                applyTransferValuesForTask(newSummary, destinationWorkbasket, setTransferFlag);
+                applyTransferValuesForTask(
+                    newSummary, destinationWorkbasket, owner, setTransferFlag);
 
                 createTransferredEvent(
                     oldSummary,
@@ -289,11 +346,11 @@ final class TaskTransferrer {
   }
 
   private void applyTransferValuesForTask(
-      TaskSummaryImpl task, WorkbasketSummary workbasket, boolean setTransferFlag) {
+      TaskSummaryImpl task, WorkbasketSummary workbasket, String owner, boolean setTransferFlag) {
     task.setRead(false);
     task.setTransferred(setTransferFlag);
     task.setState(getStateAfterTransfer(task));
-    task.setOwner(null);
+    task.setOwner(owner);
     task.setWorkbasketSummary(workbasket);
     task.setDomain(workbasket.getDomain());
     task.setModified(Instant.now());

lib/taskana-core/src/test/java/acceptance/task/transfer/TransferTaskAccTest.java

@@ -239,33 +239,21 @@ class TransferTaskAccTest extends AbstractAccTest {
         taskService.transferTasks("WBI:100000000000000000000000000000000006", taskIdList);
     assertThat(results.containsErrors()).isFalse();
 
-    final Workbasket wb =
+    final Workbasket destinationWb =
         taskanaEngine.getWorkbasketService().getWorkbasket("USER-1-1", "DOMAIN_A");
     Task transferredTask = taskService.getTask("TKI:000000000000000000000000000000000004");
-    assertThat(transferredTask).isNotNull();
+    assertTaskIsTransferred(
-    assertThat(transferredTask.isTransferred()).isTrue();
+        transferredTask.asSummary(), destinationWb, true, TaskState.READY, before, null);
-    assertThat(transferredTask.isRead()).isFalse();
-    assertThat(transferredTask.getState()).isEqualTo(TaskState.READY);
-    assertThat(transferredTask.getWorkbasketKey()).isEqualTo(wb.getKey());
-    assertThat(transferredTask.getDomain()).isEqualTo(wb.getDomain());
-    assertThat(transferredTask.getModified().isBefore(before)).isFalse();
-    assertThat(transferredTask.getOwner()).isNull();
     transferredTask = taskService.getTask("TKI:000000000000000000000000000000000005");
-    assertThat(transferredTask).isNotNull();
+    assertTaskIsTransferred(
-    assertThat(transferredTask.isTransferred()).isTrue();
+        transferredTask.asSummary(), destinationWb, true, TaskState.READY, before, null);
-    assertThat(transferredTask.isRead()).isFalse();
-    assertThat(transferredTask.getState()).isEqualTo(TaskState.READY);
-    assertThat(transferredTask.getWorkbasketKey()).isEqualTo(wb.getKey());
-    assertThat(transferredTask.getDomain()).isEqualTo(wb.getDomain());
-    assertThat(transferredTask.getModified().isBefore(before)).isFalse();
-    assertThat(transferredTask.getOwner()).isNull();
   }
 
   @WithAccessId(user = "teamlead-1", groups = GROUP_1_DN)
   @Test
   void should_BulkTransferOnlyValidTasks_When_SomeTasksToTransferCauseExceptions()
       throws Exception {
-    final Workbasket wb =
+    final Workbasket destinationWb =
         taskanaEngine.getWorkbasketService().getWorkbasket("USER-1-1", "DOMAIN_A");
     final Instant before = Instant.now().truncatedTo(ChronoUnit.MILLIS);
     // we can't use List.of because of the null value we insert
@@ -298,14 +286,8 @@ class TransferTaskAccTest extends AbstractAccTest {
 
     // verify valid requests
     Task transferredTask = taskService.getTask("TKI:000000000000000000000000000000000006");
-    assertThat(transferredTask).isNotNull();
+    assertTaskIsTransferred(
-    assertThat(transferredTask.isTransferred()).isTrue();
+        transferredTask.asSummary(), destinationWb, true, TaskState.READY, before, null);
-    assertThat(transferredTask.isRead()).isFalse();
-    assertThat(transferredTask.getState()).isEqualTo(TaskState.READY);
-    assertThat(transferredTask.getWorkbasketKey()).isEqualTo(wb.getKey());
-    assertThat(transferredTask.getDomain()).isEqualTo(wb.getDomain());
-    assertThat(transferredTask.getModified().isBefore(before)).isFalse();
-    assertThat(transferredTask.getOwner()).isNull();
 
     transferredTask = taskService.getTask("TKI:200000000000000000000000000000000008");
     assertThat(transferredTask).isNotNull();
@@ -377,26 +359,14 @@ class TransferTaskAccTest extends AbstractAccTest {
         taskService.transferTasks("GPK_B_KSC_1", "DOMAIN_B", taskIdList);
     assertThat(results.containsErrors()).isFalse();
 
-    final Workbasket wb =
+    final Workbasket destinationWb =
         taskanaEngine.getWorkbasketService().getWorkbasket("GPK_B_KSC_1", "DOMAIN_B");
     Task transferredTask = taskService.getTask("TKI:000000000000000000000000000000000023");
-    assertThat(transferredTask).isNotNull();
+    assertTaskIsTransferred(
-    assertThat(transferredTask.isTransferred()).isTrue();
+        transferredTask.asSummary(), destinationWb, true, TaskState.READY, before, null);
-    assertThat(transferredTask.isRead()).isFalse();
-    assertThat(transferredTask.getState()).isEqualTo(TaskState.READY);
-    assertThat(transferredTask.getWorkbasketKey()).isEqualTo(wb.getKey());
-    assertThat(transferredTask.getDomain()).isEqualTo(wb.getDomain());
-    assertThat(transferredTask.getModified().isBefore(before)).isFalse();
-    assertThat(transferredTask.getOwner()).isNull();
     transferredTask = taskService.getTask("TKI:000000000000000000000000000000000024");
-    assertThat(transferredTask).isNotNull();
+    assertTaskIsTransferred(
-    assertThat(transferredTask.isTransferred()).isTrue();
+        transferredTask.asSummary(), destinationWb, true, TaskState.READY, before, null);
-    assertThat(transferredTask.isRead()).isFalse();
-    assertThat(transferredTask.getState()).isEqualTo(TaskState.READY);
-    assertThat(transferredTask.getWorkbasketKey()).isEqualTo(wb.getKey());
-    assertThat(transferredTask.getDomain()).isEqualTo(wb.getDomain());
-    assertThat(transferredTask.getModified().isBefore(before)).isFalse();
-    assertThat(transferredTask.getOwner()).isNull();
   }
 
   @WithAccessId(user = "admin")
@@ -435,4 +405,237 @@ class TransferTaskAccTest extends AbstractAccTest {
 
     assertThat(transferredTasks).extracting(TaskSummary::isTransferred).containsOnly(false);
   }
+
+  @WithAccessId(user = "teamlead-1")
+  @Test
+  void should_SetOwner_When_TransferringTask() throws Exception {
+    final Workbasket destinationWb =
+        taskanaEngine
+            .getWorkbasketService()
+            .getWorkbasket("WBI:100000000000000000000000000000000005");
+    final Instant before = Instant.now().truncatedTo(ChronoUnit.MILLIS);
+
+    taskService.transferWithOwner(
+        "TKI:000000000000000000000000000000000021",
+        "WBI:100000000000000000000000000000000005",
+        "teamlead-1");
+
+    Task transferredTask = taskService.getTask("TKI:000000000000000000000000000000000021");
+    assertTaskIsTransferred(
+        transferredTask.asSummary(), destinationWb, true, TaskState.READY, before, "teamlead-1");
+  }
+
+  @WithAccessId(user = "teamlead-1", groups = GROUP_1_DN)
+  @Test
+  void should_BulkTransferOnlyValidTasksAndSetOwner_When_SomeTasksToTransferCauseExceptions()
+      throws Exception {
+    final Workbasket destinationWb =
+        taskanaEngine
+            .getWorkbasketService()
+            .getWorkbasket("WBI:100000000000000000000000000000000009");
+    final Instant before = Instant.now().truncatedTo(ChronoUnit.MILLIS);
+    List<String> taskIdList =
+        Arrays.asList(
+            "TKI:000000000000000000000000000000000007", // working
+            "TKI:000000000000000000000000000000000041", // NotAuthorized READ
+            "TKI:000000000000000000000000000000000041", // NotAuthorized READ
+            "TKI:200000000000000000000000000000000008", // NotAuthorized TRANSFER
+            "", // InvalidArgument
+            null, // InvalidArgument
+            "TKI:000000000000000000000000000000000099", // not existing
+            "TKI:100000000000000000000000000000000006"); // already completed
+
+    BulkOperationResults<String, TaskanaException> results =
+        taskService.transferTasksWithOwner(
+            "WBI:100000000000000000000000000000000009", taskIdList, "teamlead-1");
+
+    // check for exceptions in bulk
+    assertThat(results.containsErrors()).isTrue();
+    assertThat(results.getErrorMap().values()).hasSize(6);
+    assertThat(results.getErrorForId("TKI:000000000000000000000000000000000041").getClass())
+        .isEqualTo(NotAuthorizedOnWorkbasketException.class);
+    assertThat(results.getErrorForId("TKI:200000000000000000000000000000000008").getClass())
+        .isEqualTo(NotAuthorizedOnWorkbasketException.class);
+    assertThat(results.getErrorForId("TKI:000000000000000000000000000000000099").getClass())
+        .isEqualTo(TaskNotFoundException.class);
+    assertThat(results.getErrorForId("TKI:100000000000000000000000000000000006").getClass())
+        .isEqualTo(InvalidTaskStateException.class);
+    assertThat(results.getErrorForId("").getClass()).isEqualTo(TaskNotFoundException.class);
+    assertThat(results.getErrorForId(null).getClass()).isEqualTo(TaskNotFoundException.class);
+
+    Task transferredTask = taskService.getTask("TKI:000000000000000000000000000000000007");
+    assertTaskIsTransferred(
+        transferredTask.asSummary(), destinationWb, true, TaskState.READY, before, "teamlead-1");
+
+    Task notTransferredTask = taskService.getTask("TKI:200000000000000000000000000000000008");
+    assertThat(notTransferredTask).isNotNull();
+    assertThat(notTransferredTask.isTransferred()).isFalse();
+    assertThat(notTransferredTask.getWorkbasketKey()).isEqualTo("TPK_VIP");
+
+    notTransferredTask = taskService.getTask("TKI:100000000000000000000000000000000006");
+    assertThat(notTransferredTask).isNotNull();
+    assertThat(notTransferredTask.isTransferred()).isFalse();
+    assertThat(notTransferredTask.getWorkbasketKey()).isEqualTo("TEAMLEAD-1");
+
+    notTransferredTask = taskanaEngine.runAsAdmin(() -> {
+      try {
+        return taskService.getTask("TKI:000000000000000000000000000000000041");
+      } catch (NotAuthorizedOnWorkbasketException e) {
+        throw new RuntimeException(e);
+      } catch (TaskNotFoundException e) {
+        throw new RuntimeException(e);
+      }
+    });
+    assertThat(notTransferredTask).isNotNull();
+    assertThat(notTransferredTask.isTransferred()).isFalse();
+    assertThat(notTransferredTask.getWorkbasketKey()).isEqualTo("USER-B-2");
+  }
+
+  @WithAccessId(user = "teamlead-1", groups = GROUP_1_DN)
+  @Test
+  void should_TransferTaskAndSetOwner_When_WorkbasketKeyAndDomainIsProvided() throws Exception {
+    final Workbasket destinationWb =
+        taskanaEngine.getWorkbasketService().getWorkbasket("USER-1-2", "DOMAIN_A");
+    final Instant before = Instant.now().truncatedTo(ChronoUnit.MILLIS);
+
+    taskService.transferWithOwner(
+        "TKI:200000000000000000000000000000000066", "USER-1-2", "DOMAIN_A", "teamlead-1");
+
+    Task transferredTask = taskService.getTask("TKI:200000000000000000000000000000000066");
+    assertTaskIsTransferred(
+        transferredTask.asSummary(), destinationWb, true, TaskState.READY, before, "teamlead-1");
+  }
+
+  @WithAccessId(user = "teamlead-1", groups = GROUP_1_DN)
+  @Test
+  void should_BulkTransferTasksAndSetOwner_When_WorkbasketKeyAndDomainIsProvided()
+      throws Exception {
+    final Instant before = Instant.now().truncatedTo(ChronoUnit.MILLIS);
+    List<String> taskIdList =
+        List.of(
+            "TKI:000000000000000000000000000000000008", "TKI:000000000000000000000000000000000009");
+
+    BulkOperationResults<String, TaskanaException> results =
+        taskService.transferTasksWithOwner("GPK_B_KSC_1", "DOMAIN_B", taskIdList, "teamlead-1");
+    assertThat(results.containsErrors()).isFalse();
+
+    final Workbasket destinationWb =
+        taskanaEngine.getWorkbasketService().getWorkbasket("GPK_B_KSC_1", "DOMAIN_B");
+    Task transferredTask = taskService.getTask("TKI:000000000000000000000000000000000008");
+    assertTaskIsTransferred(
+        transferredTask.asSummary(), destinationWb, true, TaskState.READY, before, "teamlead-1");
+    transferredTask = taskService.getTask("TKI:000000000000000000000000000000000009");
+    assertTaskIsTransferred(
+        transferredTask.asSummary(), destinationWb, true, TaskState.READY, before, "teamlead-1");
+  }
+
+  @WithAccessId(user = "admin")
+  @Test
+  void should_SetTransferFlagAsSpecified_When_TransferringTaskWithOwnerUsingWorkbasketId()
+      throws Exception {
+    final Workbasket destinationWb =
+        taskanaEngine
+            .getWorkbasketService()
+            .getWorkbasket("WBI:100000000000000000000000000000000006");
+    final Instant before = Instant.now().truncatedTo(ChronoUnit.MILLIS);
+
+    taskService.transferWithOwner(
+        "TKI:000000000000000000000000000000000010",
+        "WBI:100000000000000000000000000000000006",
+        "user-1-1",
+        false);
+
+    Task transferredTask = taskService.getTask("TKI:000000000000000000000000000000000010");
+    assertTaskIsTransferred(
+        transferredTask.asSummary(), destinationWb, false, TaskState.READY, before, "user-1-1");
+  }
+
+  @WithAccessId(user = "admin")
+  @Test
+  void should_SetTransferFlagAsSpecifiedWithinBulkTransferWithOwner_When_WorkbasketIdGiven()
+      throws Exception {
+    taskService.transferTasksWithOwner(
+        "WBI:100000000000000000000000000000000006",
+        List.of(
+            "TKI:000000000000000000000000000000000010",
+            "TKI:000000000000000000000000000000000011",
+            "TKI:000000000000000000000000000000000012"),
+        "user-1-1",
+        false);
+
+    List<TaskSummary> transferredTasks =
+        taskService
+            .createTaskQuery()
+            .idIn(
+                "TKI:000000000000000000000000000000000010",
+                "TKI:000000000000000000000000000000000011",
+                "TKI:000000000000000000000000000000000012")
+            .list();
+
+    assertThat(transferredTasks).extracting(TaskSummary::isTransferred).containsOnly(false);
+  }
+
+  @WithAccessId(user = "admin")
+  @Test
+  void should_SetTransferFlagAsSpecified_When_WithOwnerAndWorkbasketKeyAndDomainIsGiven()
+      throws Exception {
+    final Workbasket destinationWb =
+        taskanaEngine.getWorkbasketService().getWorkbasket("USER-1-1", "DOMAIN_A");
+    final Instant before = Instant.now().truncatedTo(ChronoUnit.MILLIS);
+
+    taskService.transferWithOwner(
+        "TKI:000000000000000000000000000000000011", "USER-1-1", "DOMAIN_A", "user-1-1", false);
+
+    Task transferredTask = taskService.getTask("TKI:000000000000000000000000000000000011");
+    assertTaskIsTransferred(
+        transferredTask.asSummary(), destinationWb, false, TaskState.READY, before, "user-1-1");
+  }
+
+  @WithAccessId(user = "admin")
+  @Test
+  void should_SetTransferFlagAsSpecifiedWithinBulkTransferWithOwner_When_WorkbasketKeyDomainGiven()
+      throws Exception {
+    final Workbasket destinationWb =
+        taskanaEngine.getWorkbasketService().getWorkbasket("USER-1-1", "DOMAIN_A");
+    final Instant before = Instant.now().truncatedTo(ChronoUnit.MILLIS);
+    taskService.transferTasksWithOwner(
+        "USER-1-1",
+        "DOMAIN_A",
+        List.of(
+            "TKI:000000000000000000000000000000000013",
+            "TKI:000000000000000000000000000000000014",
+            "TKI:000000000000000000000000000000000015"),
+        "user-1-1",
+        false);
+
+    List<TaskSummary> transferredTasks =
+        taskService
+            .createTaskQuery()
+            .idIn(
+                "TKI:000000000000000000000000000000000013",
+                "TKI:000000000000000000000000000000000014",
+                "TKI:000000000000000000000000000000000015")
+            .list();
+
+    for (TaskSummary transferredTask : transferredTasks) {
+      assertTaskIsTransferred(
+          transferredTask, destinationWb, false, TaskState.READY, before, "user-1-1");
+    }
+  }
+
+  private void assertTaskIsTransferred(
+      TaskSummary transferredTask,
+      Workbasket wb,
+      boolean setTransferFlag,
+      TaskState taskStateAfterTransfer,
+      Instant before,
+      String owner) {
+    assertThat(transferredTask).isNotNull();
+    assertThat(transferredTask.isTransferred()).isEqualTo(setTransferFlag);
+    assertThat(transferredTask.isRead()).isFalse();
+    assertThat(transferredTask.getState()).isEqualTo(taskStateAfterTransfer);
+    assertThat(transferredTask.getWorkbasketSummary()).isEqualTo(wb.asSummary());
+    assertThat(transferredTask.getModified().isBefore(before)).isFalse();
+    assertThat(transferredTask.getOwner()).isEqualTo(owner);
+  }
 }

rest/taskana-rest-spring/src/docs/asciidoc/rest-api.adoc

@@ -120,6 +120,7 @@ include::{snippets}/TaskControllerRestDocTest/setTaskReadDocTest/auto-section.ad
 include::{snippets}/TaskControllerRestDocTest/deleteTaskDocTest/auto-section.adoc[]
 include::{snippets}/TaskControllerRestDocTest/forceDeleteTaskDocTest/auto-section.adoc[]
 include::{snippets}/TaskControllerRestDocTest/deleteTasksDocTest/auto-section.adoc[]
+include::{snippets}/TaskControllerRestDocTest/transferTasksDocTest/auto-section.adoc[]
 
 == Task Comment Resource
 

rest/taskana-rest-spring/src/main/java/pro/taskana/common/rest/RestEndpoints.java

@@ -60,6 +60,7 @@ public final class RestEndpoints {
   public static final String URL_TASKS_ID_TERMINATE = API_V1 + "tasks/{taskId}/terminate";
   public static final String URL_TASKS_ID_TRANSFER_WORKBASKET_ID =
       API_V1 + "tasks/{taskId}/transfer/{workbasketId}";
+  public static final String URL_TRANSFER_WORKBASKET_ID = API_V1 + "tasks/transfer/{workbasketId}";
   public static final String URL_TASKS_ID_SET_READ = API_V1 + "tasks/{taskId}/set-read";
 
   // task comment endpoints

rest/taskana-rest-spring/src/main/java/pro/taskana/task/rest/TaskController.java

@@ -49,12 +49,15 @@ import pro.taskana.task.api.exceptions.TaskAlreadyExistException;
 import pro.taskana.task.api.exceptions.TaskNotFoundException;
 import pro.taskana.task.api.models.Task;
 import pro.taskana.task.api.models.TaskSummary;
+import pro.taskana.task.rest.assembler.BulkOperationResultsRepresentationModelAssembler;
 import pro.taskana.task.rest.assembler.TaskRepresentationModelAssembler;
 import pro.taskana.task.rest.assembler.TaskSummaryRepresentationModelAssembler;
+import pro.taskana.task.rest.models.BulkOperationResultsRepresentationModel;
 import pro.taskana.task.rest.models.IsReadRepresentationModel;
 import pro.taskana.task.rest.models.TaskRepresentationModel;
 import pro.taskana.task.rest.models.TaskSummaryCollectionRepresentationModel;
 import pro.taskana.task.rest.models.TaskSummaryPagedRepresentationModel;
+import pro.taskana.task.rest.models.TransferTaskRepresentationModel;
 import pro.taskana.workbasket.api.exceptions.NotAuthorizedOnWorkbasketException;
 import pro.taskana.workbasket.api.exceptions.WorkbasketNotFoundException;
 
@@ -66,15 +69,21 @@ public class TaskController {
   private final TaskService taskService;
   private final TaskRepresentationModelAssembler taskRepresentationModelAssembler;
   private final TaskSummaryRepresentationModelAssembler taskSummaryRepresentationModelAssembler;
+  private final BulkOperationResultsRepresentationModelAssembler
+      bulkOperationResultsRepresentationModelAssembler;
 
   @Autowired
   TaskController(
       TaskService taskService,
       TaskRepresentationModelAssembler taskRepresentationModelAssembler,
-      TaskSummaryRepresentationModelAssembler taskSummaryRepresentationModelAssembler) {
+      TaskSummaryRepresentationModelAssembler taskSummaryRepresentationModelAssembler,
+      BulkOperationResultsRepresentationModelAssembler
+          bulkOperationResultsRepresentationModelAssembler) {
     this.taskService = taskService;
     this.taskRepresentationModelAssembler = taskRepresentationModelAssembler;
     this.taskSummaryRepresentationModelAssembler = taskSummaryRepresentationModelAssembler;
+    this.bulkOperationResultsRepresentationModelAssembler =
+        bulkOperationResultsRepresentationModelAssembler;
   }
 
   // region CREATE
@@ -531,7 +540,8 @@ public class TaskController {
    * @title Transfer a Task to another Workbasket
    * @param taskId the Id of the Task which should be transferred
    * @param workbasketId the Id of the destination Workbasket
-   * @param setTransferFlag sets the tansfer flag of the task (default: true)
+   * @param transferTaskRepresentationModel sets the transfer flag of the Task (default: true) and
+   *     owner of the task
    * @return the successfully transferred Task.
    * @throws TaskNotFoundException if the requested Task does not exist
    * @throws WorkbasketNotFoundException if the requested Workbasket does not exist
@@ -544,17 +554,63 @@ public class TaskController {
   public ResponseEntity<TaskRepresentationModel> transferTask(
       @PathVariable String taskId,
       @PathVariable String workbasketId,
-      @RequestBody(required = false) Boolean setTransferFlag)
+      @RequestBody(required = false)
+          TransferTaskRepresentationModel transferTaskRepresentationModel)
       throws TaskNotFoundException,
           WorkbasketNotFoundException,
           NotAuthorizedOnWorkbasketException,
           InvalidTaskStateException {
-    Task updatedTask =
+    Task updatedTask;
-        taskService.transfer(taskId, workbasketId, setTransferFlag == null || setTransferFlag);
+    if (transferTaskRepresentationModel == null) {
+      updatedTask = taskService.transfer(taskId, workbasketId);
+    } else {
+      updatedTask =
+          taskService.transferWithOwner(
+              taskId,
+              workbasketId,
+              transferTaskRepresentationModel.getOwner(),
+              transferTaskRepresentationModel.getSetTransferFlag());
+    }
 
     return ResponseEntity.ok(taskRepresentationModelAssembler.toModel(updatedTask));
   }
 
+  /**
+   * This endpoint transfers a list of Tasks listed in the body to a given Workbasket, if possible.
+   * Tasks that can be transfered without throwing an exception get transferred independent of other
+   * Tasks. If the transfer of a Task throws an exception, then the Task will remain in the old
+   * Workbasket.
+   *
+   * @title Transfer Tasks to another Workbasket
+   * @param workbasketId the Id of the destination Workbasket
+   * @param transferTaskRepresentationModel JSON formatted request body containing the TaskIds,
+   *     owner and setTransferFlag of tasks to be transferred; owner and setTransferFlag are
+   *     optional, while the TaskIds are mandatory
+   * @return the taskIds and corresponding ErrorCode of tasks failed to be transferred
+   * @throws WorkbasketNotFoundException if the requested Workbasket does not exist
+   * @throws NotAuthorizedOnWorkbasketException if the current user has no authorization to transfer
+   *     the Task
+   */
+  @PostMapping(path = RestEndpoints.URL_TRANSFER_WORKBASKET_ID)
+  @Transactional(rollbackFor = Exception.class)
+  public ResponseEntity<BulkOperationResultsRepresentationModel> transferTasks(
+      @PathVariable String workbasketId,
+      @RequestBody TransferTaskRepresentationModel transferTaskRepresentationModel)
+      throws NotAuthorizedOnWorkbasketException, WorkbasketNotFoundException {
+    List<String> taskIds = transferTaskRepresentationModel.getTaskIds();
+    BulkOperationResults<String, TaskanaException> result =
+        taskService.transferTasksWithOwner(
+            workbasketId,
+            taskIds,
+            transferTaskRepresentationModel.getOwner(),
+            transferTaskRepresentationModel.getSetTransferFlag());
+
+    BulkOperationResultsRepresentationModel repModel =
+        bulkOperationResultsRepresentationModelAssembler.toModel(result);
+
+    return ResponseEntity.ok(repModel);
+  }
+
   /**
    * This endpoint updates a requested Task.
    *

rest/taskana-rest-spring/src/main/java/pro/taskana/task/rest/assembler/BulkOperationResultsRepresentationModelAssembler.java

@@ -0,0 +1,32 @@
+package pro.taskana.task.rest.assembler;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.springframework.hateoas.server.RepresentationModelAssembler;
+import org.springframework.lang.NonNull;
+import org.springframework.stereotype.Component;
+import pro.taskana.common.api.BulkOperationResults;
+import pro.taskana.common.api.exceptions.ErrorCode;
+import pro.taskana.common.api.exceptions.TaskanaException;
+import pro.taskana.task.rest.models.BulkOperationResultsRepresentationModel;
+
+@Component
+public class BulkOperationResultsRepresentationModelAssembler
+    implements RepresentationModelAssembler<
+        BulkOperationResults<String, TaskanaException>,
+        BulkOperationResultsRepresentationModel> {
+
+  @NonNull
+  @Override
+  public BulkOperationResultsRepresentationModel toModel(
+      BulkOperationResults<String, TaskanaException> entity) {
+    BulkOperationResultsRepresentationModel repModel =
+        new BulkOperationResultsRepresentationModel();
+    Map<String, ErrorCode> newErrorMap = new HashMap<>();
+    for (Map.Entry<String, TaskanaException> entry : entity.getErrorMap().entrySet()) {
+      newErrorMap.put(entry.getKey(), entry.getValue().getErrorCode());
+    }
+    repModel.setTasksWithErrors(newErrorMap);
+    return repModel;
+  }
+}

rest/taskana-rest-spring/src/main/java/pro/taskana/task/rest/models/BulkOperationResultsRepresentationModel.java

@@ -0,0 +1,22 @@
+package pro.taskana.task.rest.models;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.springframework.hateoas.RepresentationModel;
+import pro.taskana.common.api.exceptions.ErrorCode;
+
+/** EntityModel class for BulkOperationResults. */
+public class BulkOperationResultsRepresentationModel
+    extends RepresentationModel<BulkOperationResultsRepresentationModel> {
+
+  /** Map of keys to the stored information. */
+  protected Map<String, ErrorCode> tasksWithErrors = new HashMap<>();
+
+  public Map<String, ErrorCode> getTasksWithErrors() {
+    return tasksWithErrors;
+  }
+
+  public void setTasksWithErrors(Map<String, ErrorCode> tasksWithErrors) {
+    this.tasksWithErrors = tasksWithErrors;
+  }
+}

rest/taskana-rest-spring/src/main/java/pro/taskana/task/rest/models/TransferTaskRepresentationModel.java

@@ -0,0 +1,40 @@
+package pro.taskana.task.rest.models;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.beans.ConstructorProperties;
+import java.util.List;
+
+public class TransferTaskRepresentationModel {
+
+  /** The value to set the Task property owner. */
+  @JsonProperty("owner")
+  private final String owner;
+
+  /** The value to set the Task property setTransferFlag. */
+  @JsonProperty("setTransferFlag")
+  private final Boolean setTransferFlag;
+
+  /** The value to set the Task property taskIds. */
+  @JsonProperty("taskIds")
+  private final List<String> taskIds;
+
+  @ConstructorProperties({"setTransferFlag", "owner", "taskIds"})
+  public TransferTaskRepresentationModel(
+      Boolean setTransferFlag, String owner, List<String> taskIds) {
+    this.setTransferFlag = setTransferFlag == null || setTransferFlag;
+    this.owner = owner;
+    this.taskIds = taskIds;
+  }
+
+  public Boolean getSetTransferFlag() {
+    return setTransferFlag;
+  }
+
+  public String getOwner() {
+    return owner;
+  }
+
+  public List<String> getTaskIds() {
+    return taskIds;
+  }
+}

rest/taskana-rest-spring/src/test/java/pro/taskana/task/rest/TaskControllerIntTest.java

@@ -15,7 +15,9 @@ import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.Arrays;
 import java.util.Iterator;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Stream;
 import javax.sql.DataSource;
 import org.assertj.core.api.ThrowableAssert.ThrowingCallable;
@@ -53,6 +55,7 @@ import pro.taskana.task.rest.models.TaskRepresentationModel.CustomAttribute;
 import pro.taskana.task.rest.models.TaskSummaryCollectionRepresentationModel;
 import pro.taskana.task.rest.models.TaskSummaryPagedRepresentationModel;
 import pro.taskana.task.rest.models.TaskSummaryRepresentationModel;
+import pro.taskana.task.rest.models.TransferTaskRepresentationModel;
 import pro.taskana.task.rest.routing.IntegrationTestTaskRouter;
 import pro.taskana.workbasket.rest.models.WorkbasketSummaryRepresentationModel;
 
@@ -60,16 +63,18 @@ import pro.taskana.workbasket.rest.models.WorkbasketSummaryRepresentationModel;
 @TaskanaSpringBootTest
 class TaskControllerIntTest {
 
-  @Autowired TaskanaConfiguration taskanaConfiguration;
   private static final ParameterizedTypeReference<TaskSummaryPagedRepresentationModel>
       TASK_SUMMARY_PAGE_MODEL_TYPE = new ParameterizedTypeReference<>() {};
   private static final ParameterizedTypeReference<TaskSummaryCollectionRepresentationModel>
       TASK_SUMMARY_COLLECTION_MODEL_TYPE = new ParameterizedTypeReference<>() {};
+  private static final ParameterizedTypeReference<Map<String, Object>>
+      BULK_RESULT_TASKS_MODEL_TYPE = new ParameterizedTypeReference<>() {};
   private static final ParameterizedTypeReference<TaskRepresentationModel> TASK_MODEL_TYPE =
       ParameterizedTypeReference.forType(TaskRepresentationModel.class);
   private final RestHelper restHelper;
   private final DataSource dataSource;
   private final String schemaName;
+  @Autowired TaskanaConfiguration taskanaConfiguration;
 
   @Autowired
   TaskControllerIntTest(
@@ -1995,33 +2000,36 @@ class TaskControllerIntTest {
   @TestInstance(Lifecycle.PER_CLASS)
   class TransferTasks {
     @TestFactory
-    Stream<DynamicTest> should_SetTransferFlagDependentOnRequestBody_When_TransferringTask() {
+    Stream<DynamicTest> should_SetTransferFlagAndOwnerDependentOnBody_When_TransferringTask() {
-      Iterator<Boolean> iterator = Arrays.asList(true, false).iterator();
+      Iterator<Pair<Boolean, String>> iterator =
+          Arrays.asList(Pair.of(false, "user-1-1"), Pair.of(true, "user-1-1")).iterator();
       String url =
           restHelper.toUrl(
               RestEndpoints.URL_TASKS_ID_TRANSFER_WORKBASKET_ID,
               "TKI:000000000000000000000000000000000003",
               "WBI:100000000000000000000000000000000006");
 
-      ThrowingConsumer<Boolean> test =
+      ThrowingConsumer<Pair<Boolean, String>> test =
-          setTransferFlag -> {
+          pair -> {
-            HttpEntity<String> auth =
+            HttpEntity<Object> auth =
                 new HttpEntity<>(
-                    setTransferFlag.toString(), RestHelper.generateHeadersForUser("admin"));
+                    new TransferTaskRepresentationModel(pair.getLeft(), pair.getRight(), null),
+                    RestHelper.generateHeadersForUser("admin"));
             ResponseEntity<TaskRepresentationModel> response =
                 TEMPLATE.exchange(url, HttpMethod.POST, auth, TASK_MODEL_TYPE);
 
             assertThat(response.getBody()).isNotNull();
             assertThat(response.getBody().getWorkbasketSummary().getWorkbasketId())
                 .isEqualTo("WBI:100000000000000000000000000000000006");
-            assertThat(response.getBody().isTransferred()).isEqualTo(setTransferFlag);
+            assertThat(response.getBody().isTransferred()).isEqualTo(pair.getLeft());
+            assertThat(response.getBody().getOwner()).isEqualTo(pair.getRight());
           };
 
       return DynamicTest.stream(iterator, c -> "for setTransferFlag: " + c, test);
     }
 
     @Test
-    void should_SetTransferFlagToTrue_When_TransferringWithoutRequestBody() {
+    void should_SetTransferFlagToTrueAndOwnerToNull_When_TransferringWithoutRequestBody() {
       String url =
           restHelper.toUrl(
               RestEndpoints.URL_TASKS_ID_TRANSFER_WORKBASKET_ID,
@@ -2036,6 +2044,54 @@ class TaskControllerIntTest {
       assertThat(response.getBody().getWorkbasketSummary().getWorkbasketId())
           .isEqualTo("WBI:100000000000000000000000000000000006");
       assertThat(response.getBody().isTransferred()).isTrue();
+      assertThat(response.getBody().getOwner()).isNull();
+    }
+
+    @TestFactory
+    Stream<DynamicTest>
+        should_ReturnFailedTasks_When_TransferringTasks() {
+
+      Iterator<Pair<Boolean, String>> iterator =
+          Arrays.asList(Pair.of(true, "user-1-1"), Pair.of(false, "user-1-2")).iterator();
+      String url =
+          restHelper.toUrl(
+              RestEndpoints.URL_TRANSFER_WORKBASKET_ID, "WBI:100000000000000000000000000000000006");
+
+      List<String> taskIds =
+          Arrays.asList(
+              "TKI:000000000000000000000000000000000003",
+              "TKI:000000000000000000000000000000000004",
+              "TKI:000000000000000000000000000000000039");
+
+      ThrowingConsumer<Pair<Boolean, String>> test =
+          pair -> {
+            HttpEntity<Object> auth =
+                new HttpEntity<>(
+                    new TransferTaskRepresentationModel(pair.getLeft(), pair.getRight(), taskIds),
+                    RestHelper.generateHeadersForUser("admin"));
+            ResponseEntity<Map<String, Object>> response =
+                TEMPLATE.exchange(url, HttpMethod.POST, auth, BULK_RESULT_TASKS_MODEL_TYPE);
+
+            assertThat(response.getBody()).isNotNull();
+            Map<String, LinkedHashMap> failedTasks =
+                (Map<String, LinkedHashMap>) response.getBody().get("tasksWithErrors");
+            assertThat(failedTasks).hasSize(1);
+            assertThat(failedTasks).containsKey("TKI:000000000000000000000000000000000039");
+            String errorName =
+                (String) failedTasks.get("TKI:000000000000000000000000000000000039").get("key");
+            assertThat(errorName).isEqualTo("TASK_INVALID_STATE");
+            LinkedHashMap messageVariables =
+                (LinkedHashMap)
+                    failedTasks
+                        .get("TKI:000000000000000000000000000000000039")
+                        .get("messageVariables");
+            assertThat((List) messageVariables.get("requiredTaskStates"))
+                .containsExactly("READY", "CLAIMED", "READY_FOR_REVIEW", "IN_REVIEW");
+            assertThat(messageVariables).containsEntry("taskState", "COMPLETED");
+            assertThat(messageVariables)
+                .containsEntry("taskId", "TKI:000000000000000000000000000000000039");
+          };
+      return DynamicTest.stream(iterator, c -> "for setTransferFlag and owner: " + c, test);
     }
   }
 

rest/taskana-rest-spring/src/test/java/pro/taskana/task/rest/TaskControllerRestDocTest.java

@@ -5,6 +5,7 @@ import static org.springframework.restdocs.mockmvc.RestDocumentationRequestBuild
 import static org.springframework.restdocs.mockmvc.RestDocumentationRequestBuilders.post;
 import static org.springframework.restdocs.mockmvc.RestDocumentationRequestBuilders.put;
 
+import java.util.List;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -18,6 +19,7 @@ import pro.taskana.task.internal.models.ObjectReferenceImpl;
 import pro.taskana.task.rest.assembler.TaskRepresentationModelAssembler;
 import pro.taskana.task.rest.models.IsReadRepresentationModel;
 import pro.taskana.task.rest.models.TaskRepresentationModel;
+import pro.taskana.task.rest.models.TransferTaskRepresentationModel;
 import pro.taskana.testapi.security.JaasExtension;
 import pro.taskana.testapi.security.WithAccessId;
 
@@ -221,12 +223,15 @@ class TaskControllerRestDocTest extends BaseRestDocTest {
 
   @Test
   void transferTaskDocTest() throws Exception {
+    TransferTaskRepresentationModel transferTaskRepresentationModel =
+        new TransferTaskRepresentationModel(true, "user-1-1", null);
     mockMvc
         .perform(
             post(
-                RestEndpoints.URL_TASKS_ID_TRANSFER_WORKBASKET_ID,
+                    RestEndpoints.URL_TASKS_ID_TRANSFER_WORKBASKET_ID,
-                "TKI:000000000000000000000000000000000004",
+                    "TKI:000000000000000000000000000000000004",
-                "WBI:100000000000000000000000000000000001"))
+                    "WBI:100000000000000000000000000000000001")
+                .content(objectMapper.writeValueAsString(transferTaskRepresentationModel)))
         .andExpect(MockMvcResultMatchers.status().isOk());
   }
 
@@ -244,4 +249,20 @@ class TaskControllerRestDocTest extends BaseRestDocTest {
                 .content(objectMapper.writeValueAsString(repModel)))
         .andExpect(MockMvcResultMatchers.status().isOk());
   }
+
+  @Test
+  void transferTasksDocTest() throws Exception {
+    List<String> taskIds =
+        List.of(
+            "TKI:000000000000000000000000000000000000", "TKI:000000000000000000000000000000000001");
+    TransferTaskRepresentationModel transferTaskRepresentationModel =
+        new TransferTaskRepresentationModel(true, "user-1-1", taskIds);
+    mockMvc
+        .perform(
+            post(
+                    RestEndpoints.URL_TRANSFER_WORKBASKET_ID,
+                    "WBI:100000000000000000000000000000000001")
+                .content(objectMapper.writeValueAsString(transferTaskRepresentationModel)))
+        .andExpect(MockMvcResultMatchers.status().isOk());
+  }
 }

rest/taskana-rest-spring/src/test/java/pro/taskana/task/rest/assembler/BulkOperationResultsRepresentationModelAssemblerTest.java

@@ -0,0 +1,63 @@
+package pro.taskana.task.rest.assembler;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import pro.taskana.common.api.BulkOperationResults;
+import pro.taskana.common.api.TaskanaEngine;
+import pro.taskana.common.api.exceptions.ErrorCode;
+import pro.taskana.common.api.exceptions.TaskanaException;
+import pro.taskana.common.internal.util.EnumUtil;
+import pro.taskana.rest.test.TaskanaSpringBootTest;
+import pro.taskana.task.api.TaskService;
+import pro.taskana.task.api.TaskState;
+import pro.taskana.task.api.exceptions.InvalidTaskStateException;
+import pro.taskana.task.rest.models.BulkOperationResultsRepresentationModel;
+
+@TaskanaSpringBootTest
+class BulkOperationResultsRepresentationModelAssemblerTest {
+
+  TaskanaEngine taskanaEngine;
+  TaskService taskService;
+  BulkOperationResultsRepresentationModelAssembler assembler;
+
+  @Autowired
+  BulkOperationResultsRepresentationModelAssemblerTest(
+      TaskanaEngine taskanaEngine,
+      TaskService taskService,
+      BulkOperationResultsRepresentationModelAssembler assembler) {
+    this.taskanaEngine = taskanaEngine;
+    this.taskService = taskService;
+    this.assembler = assembler;
+  }
+
+  @Test
+  void should_ReturnRepresentationModel_When_ConvertingEntityToRepresentationModel() {
+
+    BulkOperationResults<String, TaskanaException> result = new BulkOperationResults<>();
+    String taskId = "TKI:000000000000000000000000000000000003";
+    InvalidTaskStateException taskanaException =
+        new InvalidTaskStateException(
+            taskId, TaskState.COMPLETED, EnumUtil.allValuesExceptFor(TaskState.END_STATES));
+
+    result.addError(taskId, taskanaException);
+
+    BulkOperationResultsRepresentationModel repModel = assembler.toModel(result);
+
+    assertEquality(result, repModel);
+  }
+
+  private void assertEquality(
+      BulkOperationResults<String, TaskanaException> bulkOperationResults,
+      BulkOperationResultsRepresentationModel repModel) {
+    Map<String, ErrorCode> newErrorMap = new HashMap<>();
+    for (Map.Entry<String, TaskanaException> entry :
+        bulkOperationResults.getErrorMap().entrySet()) {
+      newErrorMap.put(entry.getKey(), entry.getValue().getErrorCode());
+    }
+    assertThat(newErrorMap).isEqualTo(repModel.getTasksWithErrors());
+  }
+}