int process(final int maxNumRecords, final Time time) {
        int totalProcessed = 0;
        Task lastProcessed = null;

        for (final Task task : tasks.activeTasks()) {
            final long now = time.milliseconds();
            try {
                if (executionMetadata.canProcessTask(task, now)) {
                    lastProcessed = task;
                    totalProcessed += processTask(task, maxNumRecords, now, time);
                }
            } catch (final Throwable t) {
                executionMetadata.registerTaskError(task, t, now);
                executionMetadata.removeTaskFromSuccessfullyProcessedBeforeClosing(lastProcessed);
                commitSuccessfullyProcessedTasks();
                throw t;
            }
        }

        return totalProcessed;
    }
@throws TaskMigratedException if the task producer got fenced (EOS only)
@throws StreamsException      if any task threw an exception while processing
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutor.java