brandboat commented on code in PR #16360:
URL: https://github.com/apache/kafka/pull/16360#discussion_r1642056129
##########
streams/src/test/java/org/apache/kafka/streams/integration/EOSUncleanShutdownIntegrationTest.java:
##########
@@ -59,59 +56,52 @@
import static org.apache.kafka.common.utils.Utils.mkProperties;
import static
org.apache.kafka.streams.integration.utils.IntegrationTestUtils.cleanStateBeforeTest;
import static
org.apache.kafka.streams.integration.utils.IntegrationTestUtils.quietlyCleanStateAfterTest;
-import static org.junit.Assert.assertTrue;
-
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Test the unclean shutdown behavior around state store cleanup.
*/
-@RunWith(Parameterized.class)
-@Category(IntegrationTest.class)
+@Tag("integration")
+@Timeout(600)
public class EOSUncleanShutdownIntegrationTest {
- @Rule
- public Timeout globalTimeout = Timeout.seconds(600);
@SuppressWarnings("deprecation")
- @Parameterized.Parameters(name = "{0}")
- public static Collection<String[]> data() {
- return Arrays.asList(new String[][] {
- {StreamsConfig.EXACTLY_ONCE},
- {StreamsConfig.EXACTLY_ONCE_V2}
- });
+ public static Stream<Arguments> data() {
+ return Stream.of(
+ Arguments.of(StreamsConfig.EXACTLY_ONCE),
+ Arguments.of(StreamsConfig.EXACTLY_ONCE_V2));
}
- @Parameterized.Parameter
- public String eosConfig;
-
public static final EmbeddedKafkaCluster CLUSTER = new
EmbeddedKafkaCluster(3);
+ private static File testFolder;
- @BeforeClass
+ @BeforeAll
public static void startCluster() throws IOException {
+ testFolder = TestUtils.tempDirectory();
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());
STREAMS_CONFIG.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
Serdes.String().getClass());
STREAMS_CONFIG.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG,
COMMIT_INTERVAL);
- STREAMS_CONFIG.put(StreamsConfig.STATE_DIR_CONFIG,
TEST_FOLDER.getRoot().getPath());
+ STREAMS_CONFIG.put(StreamsConfig.STATE_DIR_CONFIG,
testFolder.getPath());
}
- @AfterClass
- public static void closeCluster() {
+ @AfterAll
+ public static void closeCluster() throws IOException {
CLUSTER.stop();
+ Utils.delete(testFolder);
}
- @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
- public void shouldWorkWithUncleanShutdownWipeOutStateStore() throws
InterruptedException {
+ @ParameterizedTest
+ @MethodSource("data")
Review Comment:
I used `@MethodSource` here since I want scope the
`@SuppressWarnings("deprecation")` in `public static Stream<Arguments> data()
{` instead of adding it before every test method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
