KAFKA-4540: Suspended tasks that are not assigned to the StreamThread need to be closed before new active and standby tasks are created #2266

dguy · 2016-12-16T08:56:23Z

During onPartitionsAssigned first close, and remove, any suspended StandbyTasks that are no longer assigned to this consumer.

dguy · 2016-12-16T08:57:00Z

@mjsax @guozhangwang @enothereska - hopefully this is the last in the series of PRs to do with StandbyTask handling!

asfbot · 2016-12-16T09:38:43Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/186/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2016-12-16T09:39:43Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/184/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2016-12-16T12:03:25Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/185/
Test FAILed (JDK 8 and Scala 2.12).

mjsax · 2016-12-16T17:52:51Z

All three builds failed with a compile error...

mjsax · 2016-12-16T18:21:10Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -233,6 +233,9 @@ public void onPartitionsAssigned(Collection<TopicPartition> assignment) {
                    StreamThread.this.getName(), assignment);

                setStateWhenNotInPendingShutdown(State.ASSIGNING_PARTITIONS);
+                // do this first as we may have suspended standby tasks that
+                // will become active
+                closeNonAssignedSuspendedStandbyTasks();


Should we move closing non-assigned tasks into this method too? Would be a cleaner workflow -- close all not reassigned tasks (regular and stand-by), than reuse suspended tasks and create new tasks.

mjsax · 2016-12-16T18:21:59Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -965,34 +985,14 @@ private void removeSuspendedTasks() {
            // Close task and state manager
            for (final AbstractTask task : suspendedTasks.values()) {
                task.close();
-                task.flushState();


because it is already done in suspendTasksAndState so there is no need to flush again.

mjsax · 2016-12-16T18:22:05Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

                task.closeStateManager();
-                // flush out any extra data sent during close
-                producer.flush();


Because it is already done in suspendTasksAndState. So this is not required.

… standby tasks

asfbot · 2016-12-19T12:41:22Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/234/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2016-12-19T13:28:19Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/235/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2016-12-19T13:31:33Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/233/
Test FAILed (JDK 7 and Scala 2.10).

guozhangwang

A few comments.

guozhangwang · 2016-12-20T00:38:33Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

+                // do this first as we may have suspended standby tasks that
+                // will become active
+                closeNonAssignedSuspendedTasks();
+//                closeNonAssignedSuspendedStandbyTasks();


Is this intentional?

guozhangwang · 2016-12-20T03:05:11Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -848,7 +881,7 @@ private void addStreamTasks(Collection<TopicPartition> assignment) {
        }

        // destroy any remaining suspended tasks
-        removeSuspendedTasks();
+//        removeSuspendedTasks();


Ditto above.

Also we need to remove the comments.

guozhangwang · 2016-12-20T03:09:14Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

+        final Set<TaskId> currentActiveTaskIds = partitionAssignor.activeTasks().keySet();
+        final Iterator<Map.Entry<TaskId, StreamTask>> activeTaskIterator = suspendedActiveTasks.entrySet().iterator();
+        while (activeTaskIterator.hasNext()) {
+            closeAndRemoveIfNotAssigned(currentActiveTaskIds, activeTaskIterator.next(), activeTaskIterator);


Could we return a boolean indicating if this task is removed or not? Then in line 964 below we can instead check that this map should then be empty since we have removed both recycled tasks or non-recycled ones (in line 863).

I'm not sure how returning a boolean helps with anything. It would just indicate that a single task has been removed. If we really want to make sure we've removed all the suspended tasks we can just check if the map is empty in addStreamTasks after the for loop.

guozhangwang · 2016-12-20T03:11:19Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -960,34 +990,14 @@ private void removeStandbyTasks() {
    }

    private void removeSuspendedTasks() {
-        log.info("{} Removing all suspended tasks [{}]", logPrefix, suspendedTasks.keySet());
+        log.info("{} Removing all suspended tasks [{}]", logPrefix, suspendedActiveTasks.keySet());


See my comments above. If that case should we be guaranteed that suspendedActiveTasks is always an empty map here?

This method is no longer used and should've been removed

guozhangwang · 2016-12-20T03:12:56Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

-                        } else {
-                            throw e;
-                        }
+                    } catch (final LockException e) {


Did we change the thrown exceptions in ProcessorStateManager? I thought we are still throwing ProcessorStateException which embed the LockException?

No it just throws a LockException - i saw this when i was trying to test it.

asfbot · 2016-12-20T11:00:19Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/262/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2016-12-20T11:59:20Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/260/
Test FAILed (JDK 7 and Scala 2.10).

dguy · 2016-12-20T12:03:35Z

Updated this to take into account the partition assignment when finding StreamTasks to close. For a given TaskId the partition assignment can change, i.e., regex subscriptions.
I don't believe this matters for a StandbyTask as any StateStore will only have a single source topic, so even if a regex subscription changed this wouldn't change the partitions that are consumed to update the StateStore.

asfbot · 2016-12-20T12:12:58Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/261/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2016-12-20T12:33:14Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/265/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2016-12-20T12:33:22Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/264/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2016-12-20T13:57:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/263/
Test PASSed (JDK 7 and Scala 2.10).

guozhangwang · 2016-12-20T18:36:39Z

Merged to trunk.

… need to be closed before new active and standby tasks are created During `onPartitionsAssigned` first close, and remove, any suspended `StandbyTasks` that are no longer assigned to this consumer. Author: Damian Guy <damian.guy@gmail.com> Reviewers: Guozhang Wang <wangguoz@gmail.com> Closes apache#2266 from dguy/kafka-4540

dguy added 3 commits December 15, 2016 16:15

fix standby task suspension

347ef4c

merge trunk

4f5ab9d

remove metrics etc from TestStreamTask

3def030

mjsax reviewed Dec 16, 2016

View reviewed changes

dguy added 2 commits December 19, 2016 11:12

fix compilation

cefcd6c

remove all non-assigned suspended tasks before creating any active or…

bd74298

… standby tasks

guozhangwang reviewed Dec 20, 2016

View reviewed changes

removed commented out code etc

31ddcee

take into account partitions when closing suspended StreamTaks

26bd026

asfgit closed this in 56c6174 Dec 20, 2016

dguy deleted the kafka-4540 branch January 13, 2017 08:39

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-4540: Suspended tasks that are not assigned to the StreamThread need to be closed before new active and standby tasks are created #2266

KAFKA-4540: Suspended tasks that are not assigned to the StreamThread need to be closed before new active and standby tasks are created #2266

dguy commented Dec 16, 2016

dguy commented Dec 16, 2016

asfbot commented Dec 16, 2016

asfbot commented Dec 16, 2016

asfbot commented Dec 16, 2016

mjsax commented Dec 16, 2016

mjsax Dec 16, 2016

mjsax Dec 16, 2016

dguy Dec 19, 2016

mjsax Dec 16, 2016

dguy Dec 19, 2016

asfbot commented Dec 19, 2016

asfbot commented Dec 19, 2016

asfbot commented Dec 19, 2016

guozhangwang left a comment

guozhangwang Dec 20, 2016

dguy Dec 20, 2016

guozhangwang Dec 20, 2016

guozhangwang Dec 20, 2016

guozhangwang Dec 20, 2016

dguy Dec 20, 2016

guozhangwang Dec 20, 2016

dguy Dec 20, 2016

guozhangwang Dec 20, 2016

dguy Dec 20, 2016

asfbot commented Dec 20, 2016

asfbot commented Dec 20, 2016

dguy commented Dec 20, 2016

asfbot commented Dec 20, 2016

asfbot commented Dec 20, 2016

asfbot commented Dec 20, 2016

asfbot commented Dec 20, 2016

guozhangwang commented Dec 20, 2016

KAFKA-4540: Suspended tasks that are not assigned to the StreamThread need to be closed before new active and standby tasks are created #2266

KAFKA-4540: Suspended tasks that are not assigned to the StreamThread need to be closed before new active and standby tasks are created #2266

Conversation

dguy commented Dec 16, 2016

dguy commented Dec 16, 2016

asfbot commented Dec 16, 2016

asfbot commented Dec 16, 2016

asfbot commented Dec 16, 2016

mjsax commented Dec 16, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Dec 19, 2016

asfbot commented Dec 19, 2016

asfbot commented Dec 19, 2016

guozhangwang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Dec 20, 2016

asfbot commented Dec 20, 2016

dguy commented Dec 20, 2016

asfbot commented Dec 20, 2016

asfbot commented Dec 20, 2016

asfbot commented Dec 20, 2016

asfbot commented Dec 20, 2016

guozhangwang commented Dec 20, 2016