brandboat commented on code in PR #16360:
URL: https://github.com/apache/kafka/pull/16360#discussion_r1642064678
##########
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:
Discussed offline with Chia-Ping, since we test deprecated behavior in kafka
test, there is no need to worry about the scope of
`@SuppressWarnings("deprecation")`. Add `@SuppressWarnings("deprecation")` to
class level should be fine.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
