KAFKA-6054: Update Kafka Streams metadata to version 3 #4880

mjsax · 2018-04-16T14:23:58Z

adds Streams upgrade tests for 1.1 release

- adds Streams upgrade tests for 1.1 release - introduces metadata version 3 and 'version probing'

mjsax · 2018-04-16T14:26:26Z

This PR does not yet add "version probing" -- it bumps rebalance metadata version to 3 and introduces a new field "supported version" into the metadata. Additionally, we add corresponding flag to downgrade to metadata 2 (for versions 0.10.1 to 1.1). System tests are changed accordingly.

mjsax · 2018-04-17T13:58:51Z

System tests passed: https://jenkins.confluent.io/job/system-test-kafka-branch-builder/1709/

vvcephei

LGTM! Just a couple of small questions...

vvcephei · 2018-04-17T15:55:37Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

@@ -512,7 +526,7 @@ public Subscription subscription(final Set<String> topics) {

        // construct the global partition assignment per host map
        final Map<HostInfo, Set<TopicPartition>> partitionsByHostState = new HashMap<>();
-        if (minUserMetadataVersion == 2) {
+        if (minUserMetadataVersion == 2 || minUserMetadataVersion == 3) {


do you think it's safe to say minUserMetadata >= 2 here so we don't have to add a new clause with every version? Or do you think the logic that follows may not apply to later metadata versions?

Maybe --- for know I think it's fine, but we should consider to do a "closed range" if we add version 4. I don't like open-ended. Not sure what others think.

I agree, closed range seems less risky.

vvcephei · 2018-04-17T15:59:45Z

...ms/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfo.java

@@ -161,6 +196,13 @@ private void writeTopicPartitions(final DataOutputStream out,
        }
    }

+    private void encodeVersionThree(final DataOutputStream out) throws IOException {
+        out.writeInt(usedVersion);


In the other two variants, we used a literal 1 and 2 here. Not sure if it matters at all, since the way we get here is that usedVersion == 3.

I think this is "left over" from "reverting" version probing -- we will need it there.

I do not understand this reply completely: could you elaborate a bit more?

Just double checked the code. We can change it to hard coded 3.

guozhangwang

Made a pass over it.

guozhangwang · 2018-04-17T18:31:27Z

...ms/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfo.java

@@ -161,6 +196,13 @@ private void writeTopicPartitions(final DataOutputStream out,
        }
    }

+    private void encodeVersionThree(final DataOutputStream out) throws IOException {
+        out.writeInt(usedVersion);


I do not understand this reply completely: could you elaborate a bit more?

guozhangwang · 2018-04-17T18:32:35Z

...ms/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfo.java

@@ -161,6 +196,13 @@ private void writeTopicPartitions(final DataOutputStream out,
        }
    }

+    private void encodeVersionThree(final DataOutputStream out) throws IOException {
+        out.writeInt(usedVersion);
+        out.writeInt(LATEST_SUPPORTED_VERSION);


Should we encode latestSupportedVersion here?

I think this and following usages around latestSupportedVersion are related to the upcoming version probing code.

It's a little mysterious to have a "latest supported version" always equal to the "current version" in this PR in isolation, but I don' think it's actually problematic.

Exactly. For version probing, the assignment must sent back the leaders supported version to tell already upgraded instances to which version they need to downgrade.

I though it's better to do the metadata change in this PR already.

guozhangwang · 2018-04-17T18:36:04Z

...ms/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfo.java

@@ -233,27 +293,39 @@ private static void decodeVersionTwoData(final AssignmentInfo assignmentInfo,
        return partitions;
    }

+    private static void decodeVersionThreeData(final AssignmentInfo assignmentInfo,
+                                               final DataInputStream in) throws IOException {
+        decodeActiveTasks(assignmentInfo, in);


Could we just call decodeVersionTwoData as we did for processVersionThreeAssignment?

It seems like these two approaches are fighting for dominance in this code.

Personally, I favor the "flat" design, where these methods don't call each other. It's just easier to read when each one in isolation lists all the operations it needs to do.

It's also more general, since a hypothetical future version N might add some new metadata in the middle, or even drop some previous metadata and so wouldn't be able to call decodeVersion{N-1} in its implementation.

Version three will change and be different to version two. Thus, I prefer to not just call encodeVersionTwo. Guess it's personal taste.

guozhangwang · 2018-04-17T18:38:59Z

.../src/main/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfo.java

@@ -66,6 +88,10 @@ public int version() {
        return usedVersion;
    }

+    public int latestSupportedVersion() {


Where is this field going to be used?

Version Probing. cf : #4636 (full PR -- this PR is just a "reduced" version with version probing reverted to simplify reviewing)

guozhangwang · 2018-04-17T18:42:23Z

.../src/main/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfo.java

+        final ByteBuffer buf = ByteBuffer.allocate(getVersionThreeByteLength(endPointBytes));
+
+        buf.putInt(3); // used version
+        buf.putInt(3); // supported version


If latestSupportedVersion is ever going be different, we should use that field than hardcoding it here. But personally I am not sure where is latestSupportedVersion ever going to be used. Although the KIP did include this in the proposed changes, it only talks about how the SupportedVersionNumber of AssignmentInfo will be used, but not SubscriptionInfo..

Ack. Will make it dynamic.

The field is unused atm, but we wanted to include it to be "future prove".

guozhangwang · 2018-04-17T18:45:37Z

.../src/main/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfo.java

-        decodeVersionOneData(subscriptionInfo, data);
+        decodeClientUUID(subscriptionInfo, data);
+
+        subscriptionInfo.prevTasks = new HashSet<>();


nit: could we just pass in subscriptionInfo into decodeTasks, and initialize its prevTasks and standbyTasks internally?

Not possible if we want to share code. We call decodeTasks that does not know if it's decoding prevTasks or standbyTasks -- alternative, we can do two methods. Let me know what you think

Personally I'd prefer to have decodeTasksData in which we hard-code the logic of doing both prevTasks and standbyTasks, we do not code-share for these two task sets but we share code of constructing the set for version two and version three. I guess we cannot get both code sharing, and since it is really a nit I'm fine either way :)

I'll clean this up in the next PR. Merging this now.

guozhangwang · 2018-04-17T18:56:26Z

.../test/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfoTest.java

-        SubscriptionInfo decoded = SubscriptionInfo.decode(original.encode());
-        assertEquals(original, decoded);
+    public void shouldEncodeAndDecodeVersion1() {
+        final SubscriptionInfo info = new SubscriptionInfo(1, processId, activeTasks, standbyTasks, IGNORED_USER_ENDPOINT);


To make it more rigid: we can pass a valid end point string, and then check that the field is still null below.

guozhangwang · 2018-04-17T19:07:26Z

tests/kafkatest/services/console_consumer.py

@@ -64,7 +64,7 @@
 """


-class ConsoleConsumer(KafkaPathResolverMixin, JmxMixin, BackgroundThreadService):
+class oConsoleConsumer(KafkaPathResolverMixin, JmxMixin, BackgroundThreadService):


I'm embarrassed that I missed this one.

guozhangwang · 2018-04-17T19:08:59Z

tests/kafkatest/services/streams.py


    def set_version(self, kafka_streams_version):
        self.KAFKA_STREAMS_VERSION = kafka_streams_version

    def set_upgrade_from(self, upgrade_from):
        self.UPGRADE_FROM = upgrade_from

+    def set_upgrade_to(self, upgrade_to):
+        self.UPGRADE_TO = upgrade_to


Where is UPGRADE_TO used?

Forgot to revert this -- needed for version probing system test

mjsax · 2018-04-17T20:52:02Z

Updated this. New system test run: https://jenkins.confluent.io/job/system-test-kafka-branch-builder/1712/

guozhangwang · 2018-04-17T21:21:37Z

LGTM assuming unit test and system test passed. Please feel free to merge afterwards.

- adds Streams upgrade tests for 1.1 release - introduces metadata version 3 Reviewers: John Roesler <[email protected]>, Guozhang Wang <[email protected]>

KAFKA-6054: Update Kafka Streams metadata to version 3

111de08

- adds Streams upgrade tests for 1.1 release - introduces metadata version 3 and 'version probing'

mjsax added the streams label Apr 16, 2018

mjsax requested review from dguy and guozhangwang April 16, 2018 14:23

Revert version probing

12becef

mjsax force-pushed the kafka-6054-fix-upgrade-trunk branch from ea73ba1 to 12becef Compare April 16, 2018 17:20

vvcephei approved these changes Apr 17, 2018

View reviewed changes

guozhangwang reviewed Apr 17, 2018

View reviewed changes

Review comments

0e762d6

mjsax merged commit cae4221 into apache:trunk Apr 18, 2018

mjsax deleted the kafka-6054-fix-upgrade-trunk branch June 5, 2018 23:47

KAFKA-6054: Update Kafka Streams metadata to version 3 #4880

KAFKA-6054: Update Kafka Streams metadata to version 3 #4880

Conversation

mjsax commented Apr 16, 2018

mjsax commented Apr 16, 2018

mjsax commented Apr 17, 2018

vvcephei 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

mjsax Apr 17, 2018 • edited Loading

Choose a reason for hiding this comment

guozhangwang Apr 17, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang left a comment

Choose a reason for hiding this comment

guozhangwang Apr 17, 2018 • 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

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

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 Apr 17, 2018

guozhangwang commented Apr 17, 2018

mjsax Apr 17, 2018 •

edited

Loading

guozhangwang Apr 17, 2018 •

edited

Loading

guozhangwang Apr 17, 2018 •

edited

Loading