mjsax commented on a change in pull request #8856: URL: https://github.com/apache/kafka/pull/8856#discussion_r439695107
########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java ########## @@ -56,21 +56,21 @@ * | | | | * | v | | * | +------+--------+ | | - * | | Suspended (3) | <---+ | //TODO Suspended(3) could be removed after we've stable on KIP-429 - * | +------+--------+ | - * | | | - * | v | - * | +-----+-------+ | - * +----> | Closed (4) | -----------+ + * +---->| Suspended (3) | ----+ | //TODO Suspended(3) could be removed after we've stable on KIP-429 + * +------+--------+ | + * | | + * v | + * +-----+-------+ | + * | Closed (4) | -----------+ * +-------------+ * </pre> */ enum State { - CREATED(1, 4), // 0 - RESTORING(2, 3, 4), // 1 - RUNNING(3), // 2 - SUSPENDED(1, 4), // 3 - CLOSED(0); // 4, we allow CLOSED to transit to CREATED to handle corrupted tasks + CREATED(1, 3), // 0 + RESTORING(2, 3), // 1 + RUNNING(3), // 2 + SUSPENDED(1, 3, 4), // 3 Review comment: So far, we did not allow idempotent state transitions in the state machine itself, but handle it caller side. -- It seem inconsistent to allow `SUSPENDED -> SUSPEND` but not `CREATE -> CREATED` etc. I would recommend to keep the current pattern and avoid calling `transiteState()` if the task is already in the target state. -- I would also be happy to change it, but for this case, we should change it for _all_ cases. However, this would enlarge the scope of this PR and I think it better _not_ to do it in this PR. ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java ########## @@ -250,14 +250,10 @@ public void completeRestoration() { public void suspend() { switch (state()) { case CREATED: - case SUSPENDED: Review comment: IMHO, we should keep the `SUSPENDED` case for consistency reasons. Only merge `CREATED` and `RESTORING` (cf. my other comment on `Task.java`) ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java ########## @@ -112,10 +112,19 @@ public void completeRestoration() { @Override public void suspend() { log.trace("No-op suspend with state {}", state()); - if (state() == State.RUNNING) { - transitionTo(State.SUSPENDED); - } else if (state() == State.RESTORING) { - throw new IllegalStateException("Illegal state " + state() + " while suspending standby task " + id); + switch (state()) { + case CREATED: + case RUNNING: + case SUSPENDED: Review comment: The `SUSPEND` case should be no-op IMHO and not call `transiteTo()` (compare my other comment in `Task.java`). ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java ########## @@ -474,20 +470,17 @@ public void update(final Set<TopicPartition> topicPartitions, final Map<String, @Override public void closeAndRecycleState() { - suspend(); - prepareCommit(); - writeCheckpointIfNeed(); - + // Stream tasks should have already been suspended and their consumed offsets committed before recycling Review comment: Is the comment necessary? Seem the code is self-explaining? ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java ########## @@ -215,91 +215,54 @@ public void handleAssignment(final Map<TaskId, Set<TopicPartition>> activeTasks, "\tExisting standby tasks: {}", activeTasks.keySet(), standbyTasks.keySet(), activeTaskIds(), standbyTaskIds()); - final Map<TaskId, Set<TopicPartition>> activeTasksToCreate = new HashMap<>(activeTasks); - final Map<TaskId, Set<TopicPartition>> standbyTasksToCreate = new HashMap<>(standbyTasks); - final Set<Task> tasksToRecycle = new HashSet<>(); - builder.addSubscribedTopicsFromAssignment( activeTasks.values().stream().flatMap(Collection::stream).collect(Collectors.toList()), logPrefix ); - // first rectify all existing tasks final LinkedHashMap<TaskId, RuntimeException> taskCloseExceptions = new LinkedHashMap<>(); - final Set<Task> tasksToClose = new HashSet<>(); - final Map<TaskId, Map<TopicPartition, OffsetAndMetadata>> consumedOffsetsAndMetadataPerTask = new HashMap<>(); - final Set<Task> additionalTasksForCommitting = new HashSet<>(); + final Map<TaskId, Set<TopicPartition>> activeTasksToCreate = new HashMap<>(activeTasks); + final Map<TaskId, Set<TopicPartition>> standbyTasksToCreate = new HashMap<>(standbyTasks); + final LinkedList<Task> tasksToClose = new LinkedList<>(); Review comment: Nit: Why `LinkedList<Task> tasksToClose`? Should we only declare it as `List`? ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java ########## @@ -56,21 +56,21 @@ * | | | | * | v | | * | +------+--------+ | | - * | | Suspended (3) | <---+ | //TODO Suspended(3) could be removed after we've stable on KIP-429 - * | +------+--------+ | - * | | | - * | v | - * | +-----+-------+ | - * +----> | Closed (4) | -----------+ + * +---->| Suspended (3) | ----+ | //TODO Suspended(3) could be removed after we've stable on KIP-429 Review comment: Why remove the `<` arrow? We can still transit from `RESTORING` to `SUSPENDED`. Super-nit: `+---->| Suspended` -> `+---> | Suspended` ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java ########## @@ -714,13 +696,20 @@ void shutdown(final boolean clean) { } } - if (clean && !consumedOffsetsAndMetadataPerTask.isEmpty()) { - commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask); + try { + if (clean && !consumedOffsetsAndMetadataPerTask.isEmpty()) { + commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask); + } + for (final TaskId taskId : consumedOffsetsAndMetadataPerTask.keySet()) { + final Task task = tasks.get(taskId); + task.postCommit(); + } + } catch (final RuntimeException e) { + firstException.compareAndSet(null, e); Review comment: Why do we do the `try-catch` as outer-layer? In an exception occurs, we should stop looping through the tasks to call `postCommit()` -- is this intended? If yes, why? ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java ########## @@ -215,91 +215,54 @@ public void handleAssignment(final Map<TaskId, Set<TopicPartition>> activeTasks, "\tExisting standby tasks: {}", activeTasks.keySet(), standbyTasks.keySet(), activeTaskIds(), standbyTaskIds()); - final Map<TaskId, Set<TopicPartition>> activeTasksToCreate = new HashMap<>(activeTasks); - final Map<TaskId, Set<TopicPartition>> standbyTasksToCreate = new HashMap<>(standbyTasks); - final Set<Task> tasksToRecycle = new HashSet<>(); - builder.addSubscribedTopicsFromAssignment( activeTasks.values().stream().flatMap(Collection::stream).collect(Collectors.toList()), logPrefix ); - // first rectify all existing tasks final LinkedHashMap<TaskId, RuntimeException> taskCloseExceptions = new LinkedHashMap<>(); - final Set<Task> tasksToClose = new HashSet<>(); - final Map<TaskId, Map<TopicPartition, OffsetAndMetadata>> consumedOffsetsAndMetadataPerTask = new HashMap<>(); - final Set<Task> additionalTasksForCommitting = new HashSet<>(); + final Map<TaskId, Set<TopicPartition>> activeTasksToCreate = new HashMap<>(activeTasks); + final Map<TaskId, Set<TopicPartition>> standbyTasksToCreate = new HashMap<>(standbyTasks); + final LinkedList<Task> tasksToClose = new LinkedList<>(); + final Set<Task> tasksToRecycle = new HashSet<>(); final Set<Task> dirtyTasks = new HashSet<>(); + // first rectify all existing tasks for (final Task task : tasks.values()) { if (activeTasks.containsKey(task.id()) && task.isActive()) { updateInputPartitionsAndResume(task, activeTasks.get(task.id())); - if (task.commitNeeded()) { - additionalTasksForCommitting.add(task); - } activeTasksToCreate.remove(task.id()); } else if (standbyTasks.containsKey(task.id()) && !task.isActive()) { updateInputPartitionsAndResume(task, standbyTasks.get(task.id())); standbyTasksToCreate.remove(task.id()); - // check for tasks that were owned previously but have changed active/standby status } else if (activeTasks.containsKey(task.id()) || standbyTasks.containsKey(task.id())) { + // check for tasks that were owned previously but have changed active/standby status tasksToRecycle.add(task); } else { - try { - task.suspend(); - final Map<TopicPartition, OffsetAndMetadata> committableOffsets = task.prepareCommit(); - - tasksToClose.add(task); - if (!committableOffsets.isEmpty()) { - consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets); - } - } catch (final RuntimeException e) { - final String uncleanMessage = String.format( - "Failed to close task %s cleanly. Attempting to close remaining tasks before re-throwing:", - task.id()); - log.error(uncleanMessage, e); - taskCloseExceptions.put(task.id(), e); - // We've already recorded the exception (which is the point of clean). - // Now, we should go ahead and complete the close because a half-closed task is no good to anyone. - dirtyTasks.add(task); - } + tasksToClose.add(task); } } - if (!consumedOffsetsAndMetadataPerTask.isEmpty()) { + for (final Task task : tasksToClose) { try { - for (final Task task : additionalTasksForCommitting) { - final Map<TopicPartition, OffsetAndMetadata> committableOffsets = task.prepareCommit(); - if (!committableOffsets.isEmpty()) { - consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets); + task.suspend(); // Should be a no-op for active tasks, unless we hit an exception during handleRevocation Review comment: If we hit an exception in `handleRevocation` why would we continue here? Are we still in a "clean enough" state to actually continue? Below we call `completeTaskCloseClean(task)` what seem incorrect for this case as it might close clean task even if we did not successfully commit before. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org