KAFKA-8496: System test for KIP-429 upgrades and compatibility by bbejeck · Pull Request #7529 · apache/kafka

bbejeck · 2019-10-16T15:25:14Z

More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…me extra success checks

bbejeck · 2019-10-16T15:27:31Z

ping @guozhangwang @vvcephei @ableegoldman for reviews.

The current PR doesn't include updates from all versions ATM. That will come soon in another push but the logic won't change at all. It may involve some adjustments to the new test class in
the older version with parameters to run ect.

bbejeck · 2019-10-16T15:33:18Z

branch builder with 25 repeats https://jenkins.confluent.io/job/system-test-kafka-branch-builder/3181/

bbejeck · 2019-10-16T15:45:40Z

will fix checkstyle errors on next push

…n probing fix

…y just added entry

… add all versions to test revert version.py as 2.3.1 has not been released yet.

bbejeck · 2019-10-16T22:05:59Z

small changes and fixed checkstyle errors

bbejeck · 2019-10-16T22:07:03Z

ping @guozhangwang, @mjsax, @vvcephei, @ableegoldman for reviews

bbejeck · 2019-10-16T22:07:41Z

            }
        }
-
+        log.info("Cooperative rebalancing enabled now");


Added for test and thinking this could be useful for users

I'm +1 on this: actually I'm thinking it would be nice to make the log message above symmetric, ie something like
log.info("Eager rebalancing enabled now for upgrade from {}.x", upgradeFrom);
so we can just grep for "eager" or "cooperative" to figure out what we're on?

Agreed, but there's already this log statement https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignorConfiguration.java#L150

~~Did you mean to change the wording of this one?~~ NM I re-read your comment and I get it now. I'll update the log statement and the system test and push.

+1 on @ableegoldman 's suggestion.

bbejeck · 2019-10-16T22:35:48Z

I'll fix the checkstyle errors along with changes from comments above

ableegoldman

Thanks Bill! This looks great and tests exactly what we wanted. Left a few comments but no pressing issues that I see.
I did have the idea while reviewing to run each "processor" with more than one thread -- and then do all the usual verification for non-overlapping tasks between the consumers (rather than at instance level).
Don't really have a sense of whether that would be a nontrivial extension or a huge additional effort, so feel free to ignore or not. Let's call it a "beneficial, but not necessary" addition

ableegoldman · 2019-10-16T23:16:39Z

+        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "cooperative-rebalance-upgrade");
+        config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+        config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+        config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);


Just out of curiosity why do we set the commit interval to 1s?

ableegoldman · 2019-10-16T23:22:56Z

+            }
+
+            if (newState == State.REBALANCING) {
+                System.out.println(String.format("%sStarting a REBALANCE", upgradePhase));


Not saying you need to change this or anything, but just want to point out that once everyone is on cooperative this state change actually indicates the end of a rebalance.

ableegoldman · 2019-10-16T23:26:49Z

+
+public class StreamsUpgradeToCooperativeRebalanceTest {
+
+


nit: there's a few double linebreaks throughout the PR, not sure if intentional or not?

ableegoldman · 2019-10-16T23:29:28Z

+import java.util.Properties;
+import java.util.Set;
+
+public class StreamsUpgradeToCooperativeRebalanceTest {


I'm assuming these are all just the same code and I just need to look over one, is that fair? For my own education, why do we need all these copies?

This is needed for using older streams versions in system tests. All of these classes go in the corresponding upgrade-system-tests-XX modules.

ableegoldman · 2019-10-16T23:36:26Z

+            try:
+                del properties['upgrade.from']
+            except KeyError:
+                self.logger.info("Key 'upgrade.from' not there, better safe than sorry")


ableegoldman · 2019-10-16T23:39:13Z

+        for processor in processors:
+            self.verify_processing(processor, self.first_bounce_phase + self.processing_message)
+
+        # second rolling bounce without "upgrade.from" conifg


nit: conifg -> config (applies elsewhere too)

guozhangwang

Made a pass and left some comments. cc @ableegoldman could you take a look at my comments and lmk wdyt?

guozhangwang · 2019-10-17T00:30:15Z

            }
        }
-
+        log.info("Cooperative rebalancing enabled now");


+1 on @ableegoldman 's suggestion.

guozhangwang · 2019-10-17T00:40:58Z

+                + (args.length > 0 ? args[0] + " " : "")
+                + (args.length > 1 ? args[1] : ""));
+        }
+        final String kafka = args[0];


This parameter is not used?

guozhangwang · 2019-10-17T00:41:24Z

+        final Properties config = new Properties();
+
+        System.out.println("StreamsTest instance started (StreamsUpgradeToCooperativeRebalanceTest v0.10.1)");
+        System.out.println("kafka=" + kafka);


Ditto here.

guozhangwang · 2019-10-17T00:41:38Z

+            System.err.println("StreamsUpgradeTest requires three argument (kafka-url, properties-file) but only " + args.length + " provided: "
+                + (args.length > 0 ? args[0] + " " : ""));
+        }
+        final String kafka = args[0];


Ditto here.

guozhangwang · 2019-10-17T00:50:13Z

+    public static void main(final String[] args) throws Exception {
+        if (args.length < 2) {
+            System.err.println("StreamsUpgradeToCooperativeRebalanceTest requires two argument (kafka-url, properties-file) but only " + args.length + " provided: "
+                + (args.length > 0 ? args[0] : ""));


Here we do not read the args[0] at all, is that intentional?

Yes, as we only need the prop file configs. Maybe I can refactor these upgrade tests in a follow-on PR?

Is args[0] not the name of the program in Java? ...Huh

It's not C :)

guozhangwang · 2019-10-17T01:03:31Z

+
+    source_topic = "source"
+    sink_topic = "sink"
+    task_delimiter = "#"


Is it necessary to make TASK_DELIMITER configurable? I feel it is unnecessarily complicating the setup..

Ack, although the test uses this delimiter (or will in the near future) so I'll leave the definition here but remove from the configs passed.

Yeah my point here is that we can just hard-coded it than passing it through.

guozhangwang · 2019-10-17T01:06:57Z

+        processor.SINK_TOPIC = self.sink_topic
+        processor.TASK_DELIMITER = self.task_delimiter
+        processor.REPORT_INTERVAL = self.report_interval
+        processor.UPGRADE_FROM = upgrade_from


Since we now let the log4j to print the upgrade_from string already, do we still need this and to print within the state listener any more?

Not sure I follow here. We need the upgrade_from field for setting in the CooperativeRebalanceUpgradeService service to set the streams config.

My bad, I thought the upgradePhase-Starting a REBALANCE message is used somewhere in verification and thought with the augmented log4j it can now be replaced, but now I see it is only for debugging purposes.

guozhangwang · 2019-10-17T01:16:14Z

+                                          err_msg="Never saw '%s' message " % rebalance_msg + str(
+                                              processor.node.account))
+
+                # verify processing


Seems we are verifying processing twice: once here inside maybe_upgrade_rolling_bounce_and_verify, and once after maybe_upgrade_rolling_bounce_and_verify in verify_processing. Here we did the verify right after bouncing each node, and outside the maybe_upgrade_rolling_bounce_and_verify we just loop over them again and verify again. Is that intentional and necessary?

Yes, it's intentional. The check inside the maybe_upgrade_rolling_bounce_and_verify is for validating the individual node resumes processing after it gets bounced. The additional check outside maybe_upgrade_rolling_bounce_and_verify confirms that all nodes are continuing to process and that rebalancing has ceased. I can remove the outside check if you insist.

That makes sense, let's keep it in that sense.

EDIT: Actually, I'm wondering that if the monitor would always grep the same log4j entry in the outside verification or it always try to grep the new lines after the inner verification? If it's the first case, then the outside verification would always be redundant as we are doomed to just grep the same lines.

The external check uses a new monitor object which when it's created grabs an offset of the current number of bytes in the file. The monitor only checks in the file from that point forward. The issue you describe could happen if you were to re-use an existing monitor.

guozhangwang · 2019-10-17T01:19:58Z

+
+                # verify rebalanced into a running state
+                rebalance_msg = current_phase + self.running_state_msg
+                stdout_monitor.wait_until(rebalance_msg,


Hmm, on old versions, we do not have state listener and hence would not print current_phase + self.running_state_msg, how we would still pass here?

At this point in the test, we've upgraded to the new version so we have a state listener.

guozhangwang · 2019-10-17T01:24:32Z

+            self.get_tasks_for_processor(processor)
+            self.logger.info("Active tasks %s" % processor.active_tasks)
+
+        overlapping_tasks = processor1.active_tasks.intersection(processor2.active_tasks, processor3.active_tasks)


Why only testing intersect(1, 2, 3) and intersect(2, 3) here? Reading on https://www.geeksforgeeks.org/intersection-function-python/ I think if there's a task owned by both 1 and 2 but not 3, or by both 2 and 3 but not 1, then this would still pass?

Ack, updated the tests to be

processor1.instersection(processor2) processor1.instersection(processor3) processor2.instersection(processor3)

bbejeck · 2019-10-17T01:33:45Z

I did have the idea while reviewing to run each "processor" with more than one thread -- and then do all the usual verification for non-overlapping tasks between the consumers (rather than at instance level).

Actually I want to run another test with multiple threads and standby tasks. But I'm thinking this may be good enough for starters and I'll add another test in a follow-on PR

bbejeck · 2019-10-17T01:34:41Z

updated this for checkstyle fixes and some comments, I'll address the other comments in another push

bbejeck · 2019-10-17T02:39:34Z

@guozhangwang updated this

ableegoldman · 2019-10-17T03:46:04Z

Was reviewing mostly for correctness but gave it another pass (good catch on the intersection function @guozhangwang), I think it looks good as a test of the 429 upgrade path -- I realize those extensions (more threads, standbys, etc) would be good to cover but don't/shouldn't really be part of the upgrade test anyway. Gotta keep it modularized...
Thanks @bbejeck!

Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

mjsax · 2019-10-17T07:10:36Z


 # 2.3.x versions
 V_2_3_0 = KafkaVersion("2.3.0")
+V_2_3_1 = KafkaVersion("2.3.1")


2.3.1 is not released yet. Why do we add this here as we don't use it yet?

apache#7529)" This reverts commit f38d47b.

bbejeck added 7 commits October 16, 2019 11:22

KAFKA-8496: Initial Java class for under test

c486448

KAFKA-8496: First pass on system test

6e04e34

KAFKA-8496: remove self from non-instance method

df37735

KAFKA-8496: passing end-to-end system test

3fc8d88

KAFKA-8496: tagging messages for verification

4e1d50f

KAFKA-8496: added all code for previous versions of streams, added so…

dbb42bb

…me extra success checks

KAFKA-8496: refactor test code methods

8383577

bbejeck changed the title ~~KAFKA-8496: System test for kip 429 upgrades and compatibility[Don~~ KAFKA-8496: System test for kip 429 upgrades and compatibility[Don't merge yet] Oct 16, 2019

bbejeck added the streams label Oct 16, 2019

bbejeck added 4 commits October 16, 2019 13:34

KAFKA-8496: update version.py to include latest 2.3 -> 2.3.1

7461859

KAFKA-8496: revert test to only upgrade from 2.3 for verifying versio…

6eb1590

…n probing fix

KAFKA-8496: update version.py latest_2_3 to point to 2_3_1, previousl…

029d545

…y just added entry

KAFKA-8496: fixed args for pervious version tests, checkstyle errors,…

e8cfdf6

… add all versions to test revert version.py as 2.3.1 has not been released yet.

bbejeck changed the title ~~KAFKA-8496: System test for kip 429 upgrades and compatibility[Don't merge yet]~~ KAFKA-8496: System test for kip 429 upgrades and compatibility Oct 16, 2019

bbejeck changed the title ~~KAFKA-8496: System test for kip 429 upgrades and compatibility~~ KAFKA-8496: System test for KIP-429 upgrades and compatibility Oct 16, 2019

bbejeck commented Oct 16, 2019

View reviewed changes

bbejeck mentioned this pull request Oct 16, 2019

KAFKA-9053: AssignmentInfo#encode hardcodes the LATEST_SUPPORTED_VERSION #7537

Merged

ableegoldman approved these changes Oct 17, 2019

View reviewed changes

guozhangwang reviewed Oct 17, 2019

View reviewed changes

KAFKA-8496: fix checkstyle errors some comments addressed

40771e9

KAFKA-8496: add license, updates per comments

ce2e86b

guozhangwang merged commit b62f2a1 into apache:trunk Oct 17, 2019

guozhangwang pushed a commit that referenced this pull request Oct 17, 2019

KAFKA-8496: System test for KIP-429 upgrades and compatibility (#7529)

f38d47b

Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

mjsax reviewed Oct 17, 2019

View reviewed changes

ableegoldman added a commit to ableegoldman/kafka that referenced this pull request Nov 13, 2019

Revert "KAFKA-8496: System test for KIP-429 upgrades and compatibility (

c9194a1

apache#7529)" This reverts commit f38d47b.

bbejeck deleted the KAFKA-8496_system_test_for_kip_429_upgrades_and_compatibility branch July 10, 2024 13:56

Conversation

bbejeck commented Oct 16, 2019

Committer Checklist (excluded from commit message)

Uh oh!

bbejeck commented Oct 16, 2019

Uh oh!

bbejeck commented Oct 16, 2019

Uh oh!

bbejeck commented Oct 16, 2019

Uh oh!

bbejeck commented Oct 16, 2019

Uh oh!

bbejeck commented Oct 16, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bbejeck Oct 16, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bbejeck commented Oct 16, 2019

Uh oh!

ableegoldman left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bbejeck Oct 17, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bbejeck Oct 16, 2019 •

edited

Loading

bbejeck Oct 17, 2019 •

edited

Loading

guozhangwang Oct 17, 2019 •

edited

Loading

bbejeck Oct 17, 2019 •

edited

Loading