Skip to content

KAFKA-12173 Migrate streams:streams-scala module to JUnit 5 #9858

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
chia7712 merged 8 commits into from
Mar 24, 2021
Merged

KAFKA-12173 Migrate streams:streams-scala module to JUnit 5 #9858

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
d2b3cf2
KAFKA-12173 Migrate streams:streams-scala module to JUnit 5
chia7712 Jan 11, 2021
bda5610
Merge branch 'trunk' into KAFKA-12173
chia7712 Jan 14, 2021
77d2b1e
Merge branch 'trunk' into KAFKA-12173
chia7712 Jan 18, 2021
f40695e
remove scalatest
chia7712 Jan 18, 2021
1bf002b
Merge branch 'trunk' into KAFKA-12173
chia7712 Jan 19, 2021
24317ad
Merge branch 'trunk' into KAFKA-12173
chia7712 Jan 21, 2021
83762f5
Merge branch 'trunk' into KAFKA-12173
chia7712 Mar 23, 2021
bda798f
fix failed tests
chia7712 Mar 23, 2021
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
6 changes: 2 additions & 4 deletions build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -249,7 +249,7 @@ subprojects {
}

// Remove the relevant project name once it's converted to JUnit 5
def shouldUseJUnit5 = !(["runtime", "streams-scala", "streams"].contains(it.project.name))
def shouldUseJUnit5 = !(["runtime", "streams"].contains(it.project.name))

def testLoggingEvents = ["passed", "skipped", "failed"]
def testShowStandardStreams = false
Expand Down Expand Up @@ -1498,9 +1498,7 @@ project(':streams:streams-scala') {
testCompile project(':clients').sourceSets.test.output
testCompile project(':streams:test-utils')

testCompile libs.junitJupiterApi
testCompile libs.junitVintageEngine
testCompile libs.scalatest
testCompile libs.junitJupiter
testCompile libs.easymock
testCompile libs.hamcrest

Expand Down
1 change: 0 additions & 1 deletion gradle/dependencies.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,6 @@ versions += [
scalaCollectionCompat: "2.3.0",
scalafmt: "1.5.1",
scalaJava8Compat : "0.9.1",
scalatest: "3.0.8",
scoverage: "1.4.1",
scoveragePlugin: "5.0.0",
shadowPlugin: "6.1.0",
Expand Down
15 changes: 13 additions & 2 deletions streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,8 +26,9 @@
import org.apache.kafka.test.IntegrationTest;
import org.apache.kafka.test.TestUtils;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
Expand Down Expand Up @@ -62,9 +63,19 @@
@Category(IntegrationTest.class)
public class AdjustStreamThreadCountTest {

@ClassRule
public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1);

@BeforeClass
public static void startCluster() throws IOException {
CLUSTER.start();
}

@AfterClass
public static void closeCluster() {
CLUSTER.stop();

@ableegoldman ableegoldman Aug 25, 2021

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.

Hey @chia7712 , sorry for never bothering to take a look at this PR until just now, but I had a question about this. I know it was necessary to remove the ExternalResource feature that used to be responsible for calling start() and stop() for us in the integration tests since @ClassRule was removed in JUnit5, but that was really quite a loss since this now leaves us vulnerable to

  1. resource leaks due to forgetting to clean up the EmbeddedKafkaCluster in an integration test, or doing so in an incorrect way (eg such that a test failure might skip the cleanup stage, a mistake that we've certainly encountered in our tests in the past)
  2. breaking compatibility of integration tests across older branches, so that if we ever need to backport a fix that includes an integration test -- which many will/should do -- we can easily break the build of older branches by accident. See for example #11257: aka the reason I started digging into this 🙂 . Even if we remember to fix this during the backport, it's just an extra hassle.

Now I'm certainly not an expert in all things JUnit, but a cursory glance suggests we can replicate the old behavior in which the EmbeddedKafkaCluster is automatically started/stopped without the need for this @Before/AfterClass boilerplate code in every integration test. I believe it's possible to do so using the @Extension/ExtendWith annotations. Can we try to port the EmbeddedKafkaCluster back to an automated lifecycle with these so we can clean up the Streams integration tests?

cc @ijuma @vvcephei who may be more familiar with these constructs and how/when/why to use them

@ableegoldman ableegoldman Aug 25, 2021

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.

cc also @showuon -- would you be interested in taking a shot at this?

@showuon showuon Aug 26, 2021

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'll check it! :)

@showuon showuon Aug 26, 2021

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.

https://issues.apache.org/jira/browse/KAFKA-13230 is created for this issue. Need more discussion to make sure what we really want.

}


@Rule
public TestName testName = new TestName();

Expand Down
29 changes: 18 additions & 11 deletions ...src/test/java/org/apache/kafka/streams/integration/EOSUncleanShutdownIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.test.IntegrationTest;
import org.apache.kafka.test.TestUtils;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
Expand All @@ -42,6 +43,7 @@
import org.junit.runners.Parameterized;

import java.io.File;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Optional;
Expand Down Expand Up @@ -76,20 +78,11 @@ public static Collection<String[]> data() {
@Parameterized.Parameter
public String eosConfig;

@ClassRule
public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(3);

@ClassRule
public static final TemporaryFolder TEST_FOLDER = new TemporaryFolder(TestUtils.tempDirectory());

private static final Properties STREAMS_CONFIG = new Properties();
private static final StringSerializer STRING_SERIALIZER = new StringSerializer();
private static final Long COMMIT_INTERVAL = 100L;

private static final int RECORD_TOTAL = 3;

@BeforeClass
public static void setupConfigsAndUtils() {
public static void startCluster() throws IOException {
CLUSTER.start();
STREAMS_CONFIG.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
STREAMS_CONFIG.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
STREAMS_CONFIG.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
Expand All @@ -98,6 +91,20 @@ public static void setupConfigsAndUtils() {
STREAMS_CONFIG.put(StreamsConfig.STATE_DIR_CONFIG, TEST_FOLDER.getRoot().getPath());
}

@AfterClass
public static void closeCluster() {
CLUSTER.stop();
}

@ClassRule
public static final TemporaryFolder TEST_FOLDER = new TemporaryFolder(TestUtils.tempDirectory());

private static final Properties STREAMS_CONFIG = new Properties();
private static final StringSerializer STRING_SERIALIZER = new StringSerializer();
private static final Long COMMIT_INTERVAL = 100L;

private static final int RECORD_TOTAL = 3;

@Test
@SuppressWarnings("deprecation")
public void shouldWorkWithUncleanShutdownWipeOutStateStore() throws InterruptedException {
Expand Down
16 changes: 14 additions & 2 deletions ...ams/src/test/java/org/apache/kafka/streams/integration/EosBetaUpgradeIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,14 +52,16 @@
import org.apache.kafka.test.IntegrationTest;
import org.apache.kafka.test.StreamsTestUtils;
import org.apache.kafka.test.TestUtils;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.io.File;
import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
Expand Down Expand Up @@ -124,12 +126,22 @@ public static Collection<Boolean[]> data() {
)
);

@ClassRule
public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(
NUM_BROKERS,
Utils.mkProperties(Collections.singletonMap("auto.create.topics.enable", "false"))
);


@BeforeClass
public static void startCluster() throws IOException {
CLUSTER.start();
}

@AfterClass
public static void closeCluster() {
CLUSTER.stop();
}

private static String applicationId;
private final static int NUM_TOPIC_PARTITIONS = 4;
private final static String CONSUMER_GROUP_ID = "readCommitted";
Expand Down
16 changes: 14 additions & 2 deletions streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,8 +49,9 @@
import org.apache.kafka.test.IntegrationTest;
import org.apache.kafka.test.StreamsTestUtils;
import org.apache.kafka.test.TestUtils;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
Expand All @@ -61,6 +62,7 @@
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
Expand Down Expand Up @@ -95,12 +97,22 @@ public class EosIntegrationTest {
private static final int MAX_POLL_INTERVAL_MS = 5 * 1000;
private static final int MAX_WAIT_TIME_MS = 60 * 1000;

@ClassRule
public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(
NUM_BROKERS,
Utils.mkProperties(Collections.singletonMap("auto.create.topics.enable", "false"))
);

@BeforeClass
public static void startCluster() throws IOException {
CLUSTER.start();
}

@AfterClass
public static void closeCluster() {
CLUSTER.stop();
}


private String applicationId;
private final static int NUM_TOPIC_PARTITIONS = 2;
private final static String CONSUMER_GROUP_ID = "readCommitted";
Expand Down
67 changes: 36 additions & 31 deletions ...c/test/java/org/apache/kafka/streams/integration/FineGrainedAutoResetIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,9 +41,9 @@
import org.apache.kafka.test.IntegrationTest;
import org.apache.kafka.test.StreamsTestUtils;
import org.apache.kafka.test.TestUtils;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.experimental.categories.Category;

Expand Down Expand Up @@ -71,8 +71,42 @@ public class FineGrainedAutoResetIntegrationTest {
private static final String OUTPUT_TOPIC_1 = "outputTopic_1";
private static final String OUTPUT_TOPIC_2 = "outputTopic_2";

@ClassRule
public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(NUM_BROKERS);

@BeforeClass
public static void startCluster() throws IOException, InterruptedException {
CLUSTER.start();
CLUSTER.createTopics(
TOPIC_1_0,
TOPIC_2_0,
TOPIC_A_0,
TOPIC_C_0,
TOPIC_Y_0,
TOPIC_Z_0,
TOPIC_1_1,
TOPIC_2_1,
TOPIC_A_1,
TOPIC_C_1,
TOPIC_Y_1,
TOPIC_Z_1,
TOPIC_1_2,
TOPIC_2_2,
TOPIC_A_2,
TOPIC_C_2,
TOPIC_Y_2,
TOPIC_Z_2,
NOOP,
DEFAULT_OUTPUT_TOPIC,
OUTPUT_TOPIC_0,
OUTPUT_TOPIC_1,
OUTPUT_TOPIC_2);
}

@AfterClass
public static void closeCluster() {
CLUSTER.stop();
}

private final MockTime mockTime = CLUSTER.time;

private static final String TOPIC_1_0 = "topic-1_0";
Expand Down Expand Up @@ -106,35 +140,6 @@ public class FineGrainedAutoResetIntegrationTest {
private final String topicYTestMessage = "topic-Y test";
private final String topicZTestMessage = "topic-Z test";


@BeforeClass
public static void startKafkaCluster() throws InterruptedException {
CLUSTER.createTopics(
TOPIC_1_0,
TOPIC_2_0,
TOPIC_A_0,
TOPIC_C_0,
TOPIC_Y_0,
TOPIC_Z_0,
TOPIC_1_1,
TOPIC_2_1,
TOPIC_A_1,
TOPIC_C_1,
TOPIC_Y_1,
TOPIC_Z_1,
TOPIC_1_2,
TOPIC_2_2,
TOPIC_A_2,
TOPIC_C_2,
TOPIC_Y_2,
TOPIC_Z_2,
NOOP,
DEFAULT_OUTPUT_TOPIC,
OUTPUT_TOPIC_0,
OUTPUT_TOPIC_1,
OUTPUT_TOPIC_2);
}

@Before
public void setUp() throws IOException {

Expand Down
16 changes: 14 additions & 2 deletions ...ms/src/test/java/org/apache/kafka/streams/integration/GlobalKTableEOSIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,15 +42,17 @@
import org.apache.kafka.test.IntegrationTest;
import org.apache.kafka.test.TestUtils;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.rules.TestName;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
Expand All @@ -72,10 +74,20 @@ public class GlobalKTableEOSIntegrationTest {
BROKER_CONFIG.put("transaction.state.log.min.isr", 1);
}

@ClassRule
public static final EmbeddedKafkaCluster CLUSTER =
new EmbeddedKafkaCluster(NUM_BROKERS, BROKER_CONFIG);

@BeforeClass
public static void startCluster() throws IOException {
CLUSTER.start();
}

@AfterClass
public static void closeCluster() {
CLUSTER.stop();
}


@Parameterized.Parameters(name = "{0}")
public static Collection<String[]> data() {
return Arrays.asList(new String[][] {
Expand Down
16 changes: 14 additions & 2 deletions streams/src/test/java/org/apache/kafka/streams/integration/GlobalKTableIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,13 +44,15 @@
import org.apache.kafka.test.MockProcessorSupplier;
import org.apache.kafka.test.TestUtils;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.rules.TestName;

import java.io.IOException;
import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
Expand All @@ -68,9 +70,19 @@
public class GlobalKTableIntegrationTest {
private static final int NUM_BROKERS = 1;

@ClassRule
public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(NUM_BROKERS);

@BeforeClass
public static void startCluster() throws IOException {
CLUSTER.start();
}

@AfterClass
public static void closeCluster() {
CLUSTER.stop();
}


private final MockTime mockTime = CLUSTER.time;
private final KeyValueMapper<String, Long, Long> keyMapper = (key, value) -> value;
private final ValueJoiner<Long, String, String> joiner = (value1, value2) -> value1 + "+" + value2;
Expand Down
Loading