KAFKA-6115: TaskManager should be type aware #4129

mjsax · 2017-10-24T23:42:58Z

remove type specific methods from Task interface
add generics to preserve task type
add sub classes for different task types

mjsax · 2017-10-24T23:43:14Z

Call for review @guozhangwang @dguy @bbejeck

guozhangwang · 2017-10-25T17:32:25Z

Jenkins failures are relevant?

mjsax · 2017-10-25T21:33:00Z

Yes. final methods can't be mocked... removed final.

Updated this.

guozhangwang · 2017-10-26T16:55:52Z

retest this please

guozhangwang · 2017-10-27T02:33:59Z

retest this please

- remove type specific methods from Task interface - add generics to preserve task type - add sub classes for different task types

bbejeck · 2017-10-30T16:20:43Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/AssignedStreamTasksTest.java

@@ -38,21 +38,21 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;

-public class AssignedTasksTest {
+public class AssignedStreamTasksTest {


nit: change test name to AssignedStreamsTasksTest otherwise code coverage not picked up.

bbejeck · 2017-10-30T16:28:52Z

@mjsax thanks for the patch. One minor comment above.

Also from test coverage it looks like line 89 from AssignedStreamsTasks is not hit (I realize this probably has been this way).

mjsax · 2017-10-31T08:16:14Z

I think L89 in AssignedStreamsTasks will be hit each time there is no error -- or do I miss anything?

mjsax · 2017-10-31T08:17:02Z

Updated.

bbejeck · 2017-10-31T14:31:19Z

I understand that is the case, I wanted to add a test hitting the success scenario, but I wasn't clear about my intention.

guozhangwang

One nit question, otherwise LGTM.

Will merge after addressed @bbejeck 's comment

guozhangwang · 2017-10-31T22:14:46Z

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

-    private Map<TopicPartition, Task> runningByPartition = new HashMap<>();
-    private Map<TopicPartition, Task> restoringByPartition = new HashMap<>();
-    private int committed = 0;
+    Map<TaskId, T> running = new ConcurrentHashMap<>();


Can this be protected?

protected is weaker than "package private" (or am I wrong?)

guozhangwang · 2017-10-31T22:29:46Z

@dguy Maybe you can take a look at this PR as well?

dguy

Thanks @mjsax, left a few comments. Is a shame we can't just use the Task interface everywhere, but i understand the motivation.

dguy · 2017-11-01T10:47:47Z

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

+    AssignedStandbyTasks(final LogContext logContext) {
+        super(logContext, "standby task");
+
+        this.log = logContext.logger(getClass());


Why do we need to do this? log is private and is not used

dguy · 2017-11-01T10:48:53Z

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

+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+class AssignedStandbyTasks extends AssignedTasks<StandbyTask> {


Seems that this class is a bit redundant, i.e, we could just construct an AssignedTasks with the logContext and "standby task"

TaskManager uses it as explicit return type in some methods, thus, we need it as explicit type

dguy · 2017-11-01T10:51:30Z

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

+    Map<TaskId, T> running = new ConcurrentHashMap<>();
+    private Map<TopicPartition, T> runningByPartition = new HashMap<>();
+    private Map<TopicPartition, T> restoringByPartition = new HashMap<>();
+    protected int committed = 0;


This isn't used in this class. Probably should move to AssignedStreamsTasks

dguy · 2017-11-01T10:55:12Z

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

@@ -279,7 +261,7 @@ boolean maybeResumeSuspendedTask(final TaskId taskId, final Set<TopicPartition>
        return false;
    }

-    private void addToRestoring(final Task task) {
+    private void addToRestoring(final T task) {


As task restoration is only done for active tasks should we move all methods and fields to do with restoring to the AssignedStreamsTask class ?

Hmmm... even if addToRestoring is only called for active tasks, we use it within initializeNewTasks that is used for both (active and standby). Not sure if we could do this in an elegant way...

Looked into this part as well. I think extracting initializeNewTasks for active tasks and standby tasks is a bit overkill than just letting StandbyTask.initialize() return true.

dguy · 2017-11-01T10:56:07Z

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

@@ -18,6 +18,6 @@

 import org.apache.kafka.common.TopicPartition;

-public interface RestoringTasks {
-    Task restoringTaskFor(final TopicPartition partition);
+public interface RestoringTasks<T extends Task> {


Only StreamsTask will ever be restored. Perhaps we can remove the generics here?

mjsax · 2017-11-06T12:35:13Z

@bbejeck Thanks for clarification. I looked into it, but it seems to be rather hard to do a test -- need to create/mock a lot of input parameters. Also, this case is covered by TaskManagerTest#shouldProcessActiveTasks() -- do you think that's sufficient? If you insist, I can still a test.

mjsax · 2017-11-06T13:10:44Z

Updated this.

bbejeck · 2017-11-06T14:21:54Z

@mjsax having the TaskManagerTest#shouldProcessActiveTasks() test is fine. Thanks for looking into it.

asfgit · 2017-11-06T14:22:31Z

FAILURE
No test results found.
--none--

asfgit · 2017-11-06T15:25:08Z

SUCCESS
8050 tests run, 5 skipped, 0 failed.
--none--

asfgit · 2017-11-06T15:51:08Z

SUCCESS
8050 tests run, 5 skipped, 0 failed.
--none--

guozhangwang · 2017-11-06T18:30:56Z

Merged to trunk.

mjsax added 2 commits October 27, 2017 17:58

KAFKA-6115: TaskManager should be type aware

78b6f89

- remove type specific methods from Task interface - add generics to preserve task type - add sub classes for different task types

final methods can't be mocked -- removed final

6c3a24f

mjsax force-pushed the kafka-6115-taskManager-should-be-type-aware branch from 984de0e to 6c3a24f Compare October 27, 2017 15:59

bbejeck reviewed Oct 30, 2017

View reviewed changes

Github comments

49541be

guozhangwang approved these changes Oct 31, 2017

View reviewed changes

dguy reviewed Nov 1, 2017

View reviewed changes

Github comments

e722e1d

asfgit closed this in d637ad0 Nov 6, 2017

mjsax deleted the kafka-6115-taskManager-should-be-type-aware branch November 6, 2017 19:18

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-6115: TaskManager should be type aware #4129

KAFKA-6115: TaskManager should be type aware #4129

mjsax commented Oct 24, 2017

mjsax commented Oct 24, 2017

guozhangwang commented Oct 25, 2017

mjsax commented Oct 25, 2017

guozhangwang commented Oct 26, 2017

guozhangwang commented Oct 27, 2017

bbejeck Oct 30, 2017

bbejeck commented Oct 30, 2017

mjsax commented Oct 31, 2017

mjsax commented Oct 31, 2017

bbejeck commented Oct 31, 2017

guozhangwang left a comment

guozhangwang Oct 31, 2017

mjsax Nov 6, 2017

guozhangwang commented Oct 31, 2017

dguy left a comment

dguy Nov 1, 2017

dguy Nov 1, 2017

mjsax Nov 6, 2017

dguy Nov 1, 2017

dguy Nov 1, 2017

mjsax Nov 6, 2017

guozhangwang Nov 6, 2017

dguy Nov 1, 2017

mjsax commented Nov 6, 2017

mjsax commented Nov 6, 2017

bbejeck commented Nov 6, 2017 •

edited

Loading

asfgit commented Nov 6, 2017

asfgit commented Nov 6, 2017

asfgit commented Nov 6, 2017

guozhangwang commented Nov 6, 2017

KAFKA-6115: TaskManager should be type aware #4129

KAFKA-6115: TaskManager should be type aware #4129

Conversation

mjsax commented Oct 24, 2017

mjsax commented Oct 24, 2017

guozhangwang commented Oct 25, 2017

mjsax commented Oct 25, 2017

guozhangwang commented Oct 26, 2017

guozhangwang commented Oct 27, 2017

Choose a reason for hiding this comment

bbejeck commented Oct 30, 2017

mjsax commented Oct 31, 2017

mjsax commented Oct 31, 2017

bbejeck commented Oct 31, 2017

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

guozhangwang commented Oct 31, 2017

dguy 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

mjsax commented Nov 6, 2017

mjsax commented Nov 6, 2017

bbejeck commented Nov 6, 2017 • edited Loading

asfgit commented Nov 6, 2017

asfgit commented Nov 6, 2017

asfgit commented Nov 6, 2017

guozhangwang commented Nov 6, 2017

bbejeck commented Nov 6, 2017 •

edited

Loading