int commit(final Collection<Task> tasksToCommit) {
        int committed = 0;
        final Set<TaskId> ids =
            tasksToCommit.stream()
                .map(Task::id)
                .collect(Collectors.toSet());
        maybeLockTasks(ids);

        // We have to throw the first uncaught exception after locking the tasks, to not attempt to commit failure records.
        maybeThrowTaskExceptionsFromProcessingThreads();

        final Map<Task, Map<TopicPartition, OffsetAndMetadata>> consumedOffsetsAndMetadataPerTask = new HashMap<>();
        try {
            committed = commitTasksAndMaybeUpdateCommittableOffsets(tasksToCommit, consumedOffsetsAndMetadataPerTask);
        } catch (final TimeoutException timeoutException) {
            consumedOffsetsAndMetadataPerTask
                .keySet()
                .forEach(t -> t.maybeInitTaskTimeoutOrThrow(time.milliseconds(), timeoutException));
        }

        maybeUnlockTasks(ids);
        return committed;
    }
@Test
    public void shouldLockCommitableTasksOnCorruptionWithProcessingThreads() {
        final StreamTask activeTask1 = statefulTask(taskId00, taskId00ChangelogPartitions)
            .inState(State.RUNNING)
            .withInputPartitions(taskId00Partitions).build();
        final StreamTask activeTask2 = statefulTask(taskId01, taskId01ChangelogPartitions)
            .inState(State.RUNNING)
            .withInputPartitions(taskId01Partitions).build();
        final TasksRegistry tasks = mock(TasksRegistry.class);
        final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true, true);
        final KafkaFuture<Void> mockFuture = KafkaFuture.completedFuture(null);
        when(schedulingTaskManager.lockTasks(any())).thenReturn(mockFuture);

        taskManager.commit(mkSet(activeTask1, activeTask2));

        verify(schedulingTaskManager).lockTasks(mkSet(taskId00, taskId01));
        verify(schedulingTaskManager).unlockTasks(mkSet(taskId00, taskId01));
    }