private void maybeLockTasks(final Set<TaskId> ids) {
        if (schedulingTaskManager != null && !ids.isEmpty()) {
            if (log.isDebugEnabled()) {
                log.debug("Locking tasks {}", ids.stream().map(TaskId::toString).collect(Collectors.joining(", ")));
            }
            boolean locked = false;
            while (!locked) {
                try {
                    schedulingTaskManager.lockTasks(ids).get();
                    locked = true;
                } catch (final InterruptedException e) {
                    log.warn("Interrupted while waiting for tasks {} to be locked",
                        ids.stream().map(TaskId::toString).collect(Collectors.joining(",")));
                } catch (final ExecutionException e) {
                    log.info("Failed to lock tasks");
                    throw new RuntimeException(e);
                }
            }
        }
    }
Take records and add them to each respective task

@param records Records, can be null
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java