Support Kafka supervisor adopting running tasks between versions #7212

justinborromeo · 2019-03-08T02:03:37Z

The logic change occurs solely in SeekableStreamSupervisor. The rest of the changes are Serde tests to test backwards compatibility of DataSchema and TuningConfig and modifications of the existing tests in KinesisSupervisorTest and KafkaSupervisorTest.

clintropolis · 2019-03-15T00:23:27Z

.../main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java

@@ -1640,19 +1648,31 @@ private boolean isTaskCurrent(int taskGroupId, String taskId)
    SeekableStreamIndexTask<PartitionIdType, SequenceOffsetType> task = (SeekableStreamIndexTask<PartitionIdType, SequenceOffsetType>) taskOptional
        .get();

-    String taskSequenceName = task.getIOConfig().getBaseSequenceName();
+    String taskSequenceName = generateSequenceName(


Can you add a comment explaining what is going on here and why this is doing this?

gianm · 2019-03-15T21:06:06Z

.../main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java

@@ -1640,19 +1648,31 @@ private boolean isTaskCurrent(int taskGroupId, String taskId)
    SeekableStreamIndexTask<PartitionIdType, SequenceOffsetType> task = (SeekableStreamIndexTask<PartitionIdType, SequenceOffsetType>) taskOptional
        .get();

-    String taskSequenceName = task.getIOConfig().getBaseSequenceName();


Every task in a task group needs to have the same sequence name (so replicas segment allocations match up), meaning you do need to remember task.getIOConfig().getBaseSequenceName(), and should probably assign it as the sequence name for the relevant task group before starting to create new tasks.

gianm · 2019-03-27T16:02:36Z

@justinborromeo, now that changes to SS stuff seem to have settled down, could you please resolve the conflicts?

justinborromeo · 2019-03-27T21:49:24Z

@gianm done

jon-wei · 2019-04-04T20:33:16Z

.../main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java

@@ -426,6 +452,7 @@ boolean isValidTaskGroup(int taskGroupId, @Nullable TaskGroup taskGroup)
  // Map<{group RandomIdUtils}, {actively reading task group}>; see documentation for TaskGroup class
  private final ConcurrentHashMap<Integer, TaskGroup> activelyReadingTaskGroups = new ConcurrentHashMap<>();

+  // After telling a taskGroup to stop reading and begin publishing a segment, it is moved from [activelyReadingTaskGroups] to here so


Looks like a duplicate of the comment below

jon-wei · 2019-04-05T23:12:21Z

.../main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java

-          .checkNotNull(activelyReadingTaskGroups.get(taskGroupId), "null taskGroup for taskId[%s]", taskGroupId)
-          .baseSequenceName
-          .equals(taskSequenceName);
+      TaskGroup taskGroup = activelyReadingTaskGroups.get(taskGroupId);


I think this needs a null check

oh nm, I see it was redundant

jon-wei · 2019-04-06T01:02:15Z

...ng-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java

-  }
-
-  @Test
-  public void testKillBadPartitionAssignment() throws Exception


I don't think this test needs to be changed, it runs fine as-is if I add it back in

jon-wei · 2019-04-06T01:02:47Z

...ng-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java

-        null
-    );
-
-    // different datasource (don't kill)


Can you make sure you have equivalent cases in the replacement tests that you added?

jon-wei · 2019-04-09T04:09:32Z

...ervice/src/test/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorTest.java

-  }
-
-  @Test
-  public void testKillBadPartitionAssignment() throws Exception


Hm, I did some investigation here, and the Kinesis version fails while Kafka passes because the tests had different behavior.

In the Kafka version of the test, id3-id5 are checking for partition consistency within the partition set (e.g., 0 and 1 can't be in the same set with a taskCount of 2).

In the Kinesis version, there are only two partitions instead of 3, so the case above is tested in id3 only. id4 and id5 (which are duplicates) in kinesis are testing a mismatch of the group ID and the partitions within the group which hits the new check you added.

For the kinesis test, I think you could add it back in, remove the redundant "id5" case, and adjust the test to catch the failure on "id4" (might be cool to have the same groupID + partition mismatch test in the kafka version too)

It's definitely out of scope for this PR, but at some point I do want to look into why the kinesis test only uses 2 shards and possibly adjusting that, I think it would be simpler if it followed the structure of the equivalent kafka tests more closely.

jon-wei · 2019-04-10T22:14:30Z

...ervice/src/test/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorTest.java


    TreeMap<Integer, Map<String, String>> checkpoints1 = new TreeMap<>();
    checkpoints1.put(0, ImmutableMap.of(shardId1, "0"));
    TreeMap<Integer, Map<String, String>> checkpoints2 = new TreeMap<>();
    checkpoints2.put(0, ImmutableMap.of(shardId0, "0"));
+    TreeMap<Integer, Map<String, String>> checkpoints4 = new TreeMap<>();
+    checkpoints2.put(0, ImmutableMap.of(shardId0, "0"));


checkpoints2 here should be checkpoints4

jon-wei · 2019-04-10T22:59:21Z

...ng-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java

+  }
+
+  @Test
+  public void testIsTaskCurrent()


Can you add a case in this test (and kinesis) that uses TestModifiedKafkaIndexTaskTuningConfig with the same properties to see that isTaskCurrent will accept the tuningConfig with extra properties?

Ah, nevermind, the existing serde and isTaskCurrent tests are enough. I was thinking of the wrong place for inserting the modified config (the taskStorage on the overlord should always return tasks with the new class definition)

* Recompute hash in isTaskCurrent() and added tests * Fixed checkstyle stuff * Fixed failing tests * Make TestableKafkaSupervisorWithCustomIsTaskCurrent static * Add doc * baseSequenceName change * Added comment * WIP * Fixed imports * Undid lambda change for diff sake * Cleanup * Added comment * Reinsert Kafka tests * Readded kinesis test * Readd bad partition assignment in kinesis supervisor test * Nit * Misnamed var

…che#7212) * Recompute hash in isTaskCurrent() and added tests * Fixed checkstyle stuff * Fixed failing tests * Make TestableKafkaSupervisorWithCustomIsTaskCurrent static * Add doc * baseSequenceName change * Added comment * WIP * Fixed imports * Undid lambda change for diff sake * Cleanup * Added comment * Reinsert Kafka tests * Readded kinesis test * Readd bad partition assignment in kinesis supervisor test * Nit * Misnamed var

justinborromeo added 4 commits March 7, 2019 18:01

Recompute hash in isTaskCurrent() and added tests

372dca4

Fixed checkstyle stuff

44338fd

Fixed failing tests

8ec514b

Make TestableKafkaSupervisorWithCustomIsTaskCurrent static

88fe807

fjy added this to the 0.15.0 milestone Mar 11, 2019

clintropolis added the Area - Streaming Ingestion label Mar 12, 2019

jihoonson changed the title ~~6958 Support Kafka supervisor adopting running tasks between versions~~ Support Kafka supervisor adopting running tasks between versions Mar 14, 2019

clintropolis reviewed Mar 15, 2019

View reviewed changes

gianm reviewed Mar 15, 2019

View reviewed changes

Add doc

7febbef

gianm assigned dclim Mar 15, 2019

justinborromeo added 8 commits March 15, 2019 16:09

baseSequenceName change

c2efb25

Added comment

076d7da

WIP

684ba80

Fixed imports

3b88163

Undid lambda change for diff sake

06a8b0b

Cleanup

3a77c8c

Added comment

2cab52f

Merge branch 'master' into 6958-V3

5afdf08

Merge branch 'master' into 6958-V3

3cb0b92

jon-wei reviewed Apr 5, 2019

View reviewed changes

jon-wei reviewed Apr 6, 2019

View reviewed changes

justinborromeo added 2 commits April 8, 2019 11:37

Reinsert Kafka tests

fc1fcec

Readded kinesis test

fbd2873

jon-wei reviewed Apr 9, 2019

View reviewed changes

justinborromeo added 3 commits April 9, 2019 16:54

Readd bad partition assignment in kinesis supervisor test

d0e253a

Merge branch 'master' into 6958-V3

6183a0a

Nit

991fbea

jon-wei reviewed Apr 10, 2019

View reviewed changes

Misnamed var

814fc52

jon-wei reviewed Apr 10, 2019

View reviewed changes

jon-wei approved these changes Apr 11, 2019

View reviewed changes

jon-wei merged commit 2771ed5 into apache:master Apr 11, 2019

clintropolis modified the milestones: 0.15.0, 0.14.1 Apr 24, 2019

clintropolis mentioned this pull request Apr 25, 2019

0.14.1-incubating release notes #7553

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support Kafka supervisor adopting running tasks between versions #7212

Support Kafka supervisor adopting running tasks between versions #7212

justinborromeo commented Mar 8, 2019 •

edited

Loading

clintropolis Mar 15, 2019

justinborromeo Mar 15, 2019

gianm Mar 15, 2019

gianm commented Mar 27, 2019

justinborromeo commented Mar 27, 2019

jon-wei Apr 4, 2019

jon-wei Apr 5, 2019

jon-wei Apr 5, 2019

jon-wei Apr 6, 2019 •

edited

Loading

jon-wei Apr 6, 2019

jon-wei Apr 9, 2019

jon-wei Apr 9, 2019

jon-wei Apr 10, 2019

jon-wei Apr 10, 2019

jon-wei Apr 11, 2019 •

edited

Loading

Support Kafka supervisor adopting running tasks between versions #7212

Support Kafka supervisor adopting running tasks between versions #7212

Conversation

justinborromeo commented Mar 8, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gianm commented Mar 27, 2019

justinborromeo commented Mar 27, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jon-wei Apr 6, 2019 • edited Loading

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

jon-wei Apr 11, 2019 • edited Loading

Choose a reason for hiding this comment

justinborromeo commented Mar 8, 2019 •

edited

Loading

jon-wei Apr 6, 2019 •

edited

Loading

jon-wei Apr 11, 2019 •

edited

Loading