Skip to content

HOTFIX: fix broken Streams upgrade system test #13654

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mjsax merged 3 commits into from
May 8, 2023
Merged

HOTFIX: fix broken Streams upgrade system test #13654

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1804982
HOTFIX: fix broken Streams upgrade system test
mjsax May 1, 2023
a8f3172
fix dev version
mjsax May 3, 2023
4c082f4
fix test setup
mjsax May 4, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions ...rade-system-tests-24/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
import static org.apache.kafka.streams.tests.SmokeTestUtil.intSerde;
import static org.apache.kafka.streams.tests.SmokeTestUtil.stringSerde;

import java.util.Random;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
Expand Down Expand Up @@ -71,7 +70,7 @@ public static void main(final String[] args) throws Exception {
final Properties config = new Properties();
config.setProperty(
StreamsConfig.APPLICATION_ID_CONFIG,
"StreamsUpgradeTest-" + new Random().nextLong());
"StreamsUpgradeTest");
config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000L);
config.putAll(streamsProperties);

Expand Down
3 changes: 1 addition & 2 deletions ...rade-system-tests-25/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
import static org.apache.kafka.streams.tests.SmokeTestUtil.intSerde;
import static org.apache.kafka.streams.tests.SmokeTestUtil.stringSerde;

import java.util.Random;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
Expand Down Expand Up @@ -71,7 +70,7 @@ public static void main(final String[] args) throws Exception {
final Properties config = new Properties();
config.setProperty(
StreamsConfig.APPLICATION_ID_CONFIG,
"StreamsUpgradeTest-" + new Random().nextLong());
"StreamsUpgradeTest");
config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000L);
config.putAll(streamsProperties);

Expand Down
3 changes: 1 addition & 2 deletions ...rade-system-tests-26/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
import static org.apache.kafka.streams.tests.SmokeTestUtil.intSerde;
import static org.apache.kafka.streams.tests.SmokeTestUtil.stringSerde;

import java.util.Random;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
Expand Down Expand Up @@ -71,7 +70,7 @@ public static void main(final String[] args) throws Exception {
final Properties config = new Properties();
config.setProperty(
StreamsConfig.APPLICATION_ID_CONFIG,
"StreamsUpgradeTest-" + new Random().nextLong());
"StreamsUpgradeTest");
config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000L);
config.putAll(streamsProperties);

Expand Down
3 changes: 1 addition & 2 deletions ...rade-system-tests-27/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
import static org.apache.kafka.streams.tests.SmokeTestUtil.intSerde;
import static org.apache.kafka.streams.tests.SmokeTestUtil.stringSerde;

import java.util.Random;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
Expand Down Expand Up @@ -71,7 +70,7 @@ public static void main(final String[] args) throws Exception {
final Properties config = new Properties();
config.setProperty(
StreamsConfig.APPLICATION_ID_CONFIG,
"StreamsUpgradeTest-" + new Random().nextLong());
"StreamsUpgradeTest");
config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000L);
config.putAll(streamsProperties);

Expand Down
3 changes: 1 addition & 2 deletions ...rade-system-tests-28/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
import static org.apache.kafka.streams.tests.SmokeTestUtil.intSerde;
import static org.apache.kafka.streams.tests.SmokeTestUtil.stringSerde;

import java.util.Random;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
Expand Down Expand Up @@ -71,7 +70,7 @@ public static void main(final String[] args) throws Exception {
final Properties config = new Properties();
config.setProperty(
StreamsConfig.APPLICATION_ID_CONFIG,
"StreamsUpgradeTest-" + new Random().nextLong());
"StreamsUpgradeTest");
config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
config.putAll(streamsProperties);

Expand Down
3 changes: 1 addition & 2 deletions ...rade-system-tests-30/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
import static org.apache.kafka.streams.tests.SmokeTestUtil.intSerde;
import static org.apache.kafka.streams.tests.SmokeTestUtil.stringSerde;

import java.util.Random;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
Expand Down Expand Up @@ -73,7 +72,7 @@ public static void main(final String[] args) throws Exception {
final Properties config = new Properties();
config.setProperty(
StreamsConfig.APPLICATION_ID_CONFIG,
"StreamsUpgradeTest-" + new Random().nextLong());
"StreamsUpgradeTest");
config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
config.putAll(streamsProperties);

Expand Down
3 changes: 1 addition & 2 deletions ...rade-system-tests-31/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
import static org.apache.kafka.streams.tests.SmokeTestUtil.intSerde;
import static org.apache.kafka.streams.tests.SmokeTestUtil.stringSerde;

import java.util.Random;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
Expand Down Expand Up @@ -73,7 +72,7 @@ public static void main(final String[] args) throws Exception {
final Properties config = new Properties();
config.setProperty(
StreamsConfig.APPLICATION_ID_CONFIG,
"StreamsUpgradeTest-" + new Random().nextLong());
"StreamsUpgradeTest");
config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
config.putAll(streamsProperties);

Expand Down
3 changes: 1 addition & 2 deletions ...rade-system-tests-32/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@
import org.apache.kafka.streams.processor.api.Record;

import java.util.Properties;
import java.util.Random;

import static org.apache.kafka.streams.tests.SmokeTestUtil.intSerde;
import static org.apache.kafka.streams.tests.SmokeTestUtil.stringSerde;
Expand Down Expand Up @@ -73,7 +72,7 @@ public static void main(final String[] args) throws Exception {
final Properties config = new Properties();
config.setProperty(
StreamsConfig.APPLICATION_ID_CONFIG,
"StreamsUpgradeTest-" + new Random().nextLong());
"StreamsUpgradeTest");
config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
config.putAll(streamsProperties);

Expand Down
3 changes: 1 addition & 2 deletions ...rade-system-tests-33/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@
import org.apache.kafka.streams.processor.api.Record;

import java.util.Properties;
import java.util.Random;

import static org.apache.kafka.streams.tests.SmokeTestUtil.intSerde;
import static org.apache.kafka.streams.tests.SmokeTestUtil.stringSerde;
Expand Down Expand Up @@ -73,7 +72,7 @@ public static void main(final String[] args) throws Exception {
final Properties config = new Properties();
config.setProperty(
StreamsConfig.APPLICATION_ID_CONFIG,
"StreamsUpgradeTest-" + new Random().nextLong());
"StreamsUpgradeTest");
config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
config.putAll(streamsProperties);

Expand Down
10 changes: 7 additions & 3 deletions tests/kafkatest/tests/streams/streams_upgrade_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,9 +38,13 @@
str(DEV_BRANCH)]

metadata_1_versions = [str(LATEST_0_10_0)]
metadata_2_versions = [str(LATEST_0_10_1), str(LATEST_0_10_2), str(LATEST_0_11_0), str(LATEST_1_0), str(LATEST_1_1)]
fk_join_versions = [str(LATEST_2_4), str(LATEST_2_5), str(LATEST_2_6), str(LATEST_2_7), str(LATEST_2_8),
str(LATEST_3_0), str(LATEST_3_1), str(LATEST_3_2), str(LATEST_3_3)]
metadata_2_versions = [str(LATEST_0_10_1), str(LATEST_0_10_2), str(LATEST_0_11_0), str(LATEST_1_0), str(LATEST_1_1),
str(LATEST_2_4), str(LATEST_2_5), str(LATEST_2_6), str(LATEST_2_7), str(LATEST_2_8),
str(LATEST_3_0)]
# upgrading from version (2.4...3.0) is broken and only fixed later in 3.1
# we cannot test two bounce rolling upgrade because we know it's broken
# instead we add version 2.4...3.0 to the `metadata_2_versions` upgrade list
fk_join_versions = [str(LATEST_3_1), str(LATEST_3_2), str(LATEST_3_3)]

"""
After each release one should first check that the released version has been uploaded to
Expand Down
9 changes: 3 additions & 6 deletions tests/kafkatest/version.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,8 @@ def supports_topic_ids_when_using_zk(self):
return self >= V_2_8_0

def supports_fk_joins(self):
return hasattr(self, "version") and self >= V_2_4_0
# while we support FK joins since 2.4, rolling upgrade is broken in older versions and only fixed in 3.1
return hasattr(self, "version") and self >= V_3_1_2

def get_version(node=None):
"""Return the version attached to the given node.
Expand All @@ -119,7 +120,7 @@ def get_version(node=None):
return DEV_BRANCH

DEV_BRANCH = KafkaVersion("dev")
DEV_VERSION = KafkaVersion("3.5.0-SNAPSHOT")
DEV_VERSION = KafkaVersion("3.6.0-SNAPSHOT")

LATEST_METADATA_VERSION = "3.3"

Expand Down Expand Up @@ -249,7 +250,3 @@ def get_version(node=None):
# 3.5.x versions
V_3_5_0 = KafkaVersion("3.5.0")
LATEST_3_5 = V_3_5_0

# 3.6.x versions
V_3_6_0 = KafkaVersion("3.6.0")

@jolshan jolshan Jul 20, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here -- did we mean to remove?

@ahuang98 ahuang98 Jul 20, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

guessing it was removed because there were no references to it, I'll add it back when I raise a PR for more comprehensive upgrade tests

@mjsax mjsax Jul 21, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes -- 3.6.0 is not released yet. We should only need it after 3.6.0 is out.

If you want to refer to current version it would be DEV_VERSION (ie, 3.6.0-SNAPSHOT).

@jolshan jolshan Jul 21, 2023 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm -- if you look at all the snapshot commits, we've added this cutting the release branch for the previous version. (Ie, cutting the branch for the new trunk as per the commit comments) Seems like the top commit I linked should not have been reverted unless all of these were wrong.
dc1ede8
c1a5467
6ace67b

@jolshan jolshan Jul 21, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the issue in David's commit was that he didn't set the dev version to match the new version defined?

@mjsax mjsax Jul 21, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could be -- I mean, it does not really hurt to add it early (it's just a definition of a var), but it's also useless to have, a nobody should use it, but use DEV_VERSION instead.

LATEST_3_6 = V_3_6_0