You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2022/05/17 07:09:18 UTC
[flink] branch master updated: [FLINK-27567][aws][tests] Migrate connector-aws-kinesis-streams to JUnit5
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new a10fd231426 [FLINK-27567][aws][tests] Migrate connector-aws-kinesis-streams to JUnit5
a10fd231426 is described below
commit a10fd2314269e0686d9fbbd3eacb470a03432ef3
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Tue May 17 09:09:12 2022 +0200
[FLINK-27567][aws][tests] Migrate connector-aws-kinesis-streams to JUnit5
---
.../sink/KinesisStreamsSinkBuilderTest.java | 14 ++--
.../kinesis/sink/KinesisStreamsSinkITCase.java | 77 +++++++++--------
.../sink/KinesisStreamsStateSerializerTest.java | 6 +-
.../table/KinesisDynamicTableSinkFactoryTest.java | 19 ++---
...DataFieldsKinesisPartitionKeyGeneratorTest.java | 96 ++++++++++------------
.../util/KinesisProducerOptionsMapperTest.java | 11 ++-
.../org.junit.jupiter.api.extension.Extension | 16 ++++
7 files changed, 123 insertions(+), 116 deletions(-)
diff --git a/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkBuilderTest.java b/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkBuilderTest.java
index a6827621e2b..ce88c63d5e3 100644
--- a/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkBuilderTest.java
+++ b/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkBuilderTest.java
@@ -21,17 +21,17 @@ import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.assertj.core.api.Assertions;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
/** Covers construction, defaults and sanity checking of KinesisStreamsSinkBuilder. */
-public class KinesisStreamsSinkBuilderTest {
+class KinesisStreamsSinkBuilderTest {
private static final SerializationSchema<String> SERIALIZATION_SCHEMA =
new SimpleStringSchema();
private static final PartitionKeyGenerator<String> PARTITION_KEY_GENERATOR =
element -> String.valueOf(element.hashCode());
@Test
- public void elementConverterOfSinkMustBeSetWhenBuilt() {
+ void elementConverterOfSinkMustBeSetWhenBuilt() {
Assertions.assertThatExceptionOfType(NullPointerException.class)
.isThrownBy(() -> KinesisStreamsSink.builder().setStreamName("stream").build())
.withMessageContaining(
@@ -39,7 +39,7 @@ public class KinesisStreamsSinkBuilderTest {
}
@Test
- public void streamNameOfSinkMustBeSetWhenBuilt() {
+ void streamNameOfSinkMustBeSetWhenBuilt() {
Assertions.assertThatExceptionOfType(NullPointerException.class)
.isThrownBy(
() ->
@@ -52,7 +52,7 @@ public class KinesisStreamsSinkBuilderTest {
}
@Test
- public void streamNameOfSinkMustBeSetToNonEmptyWhenBuilt() {
+ void streamNameOfSinkMustBeSetToNonEmptyWhenBuilt() {
Assertions.assertThatExceptionOfType(IllegalArgumentException.class)
.isThrownBy(
() ->
@@ -66,7 +66,7 @@ public class KinesisStreamsSinkBuilderTest {
}
@Test
- public void serializationSchemaMustBeSetWhenSinkIsBuilt() {
+ void serializationSchemaMustBeSetWhenSinkIsBuilt() {
Assertions.assertThatExceptionOfType(NullPointerException.class)
.isThrownBy(
() ->
@@ -79,7 +79,7 @@ public class KinesisStreamsSinkBuilderTest {
}
@Test
- public void partitionKeyGeneratorMustBeSetWhenSinkIsBuilt() {
+ void partitionKeyGeneratorMustBeSetWhenSinkIsBuilt() {
Assertions.assertThatExceptionOfType(NullPointerException.class)
.isThrownBy(
() ->
diff --git a/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkITCase.java b/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkITCase.java
index d93d899e6c4..c4930055525 100644
--- a/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkITCase.java
+++ b/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkITCase.java
@@ -29,16 +29,16 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.datagen.DataGeneratorSource;
import org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator;
import org.apache.flink.util.DockerImageVersions;
-import org.apache.flink.util.TestLogger;
import org.assertj.core.api.Assertions;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import org.rnorth.ducttape.ratelimits.RateLimiter;
import org.rnorth.ducttape.ratelimits.RateLimiterBuilder;
import org.testcontainers.containers.Network;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;
import software.amazon.awssdk.core.SdkSystemSetting;
import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
@@ -63,7 +63,8 @@ import static org.apache.flink.connector.aws.config.AWSConfigConstants.HTTP_PROT
import static org.apache.flink.connector.aws.config.AWSConfigConstants.TRUST_ALL_CERTIFICATES;
/** IT cases for using Kinesis Data Streams Sink based on Kinesalite. */
-public class KinesisStreamsSinkITCase extends TestLogger {
+@Testcontainers
+class KinesisStreamsSinkITCase {
private static final String DEFAULT_FIRST_SHARD_NAME = "shardId-000000000000";
@@ -72,8 +73,8 @@ public class KinesisStreamsSinkITCase extends TestLogger {
element -> String.valueOf(element.hashCode());
private final PartitionKeyGenerator<String> longPartitionKeyGenerator = element -> element;
- @ClassRule
- public static final KinesaliteContainer KINESALITE =
+ @Container
+ private static final KinesaliteContainer KINESALITE =
new KinesaliteContainer(DockerImageName.parse(DockerImageVersions.KINESALITE))
.withNetwork(Network.newNetwork())
.withNetworkAliases("kinesalite");
@@ -82,8 +83,8 @@ public class KinesisStreamsSinkITCase extends TestLogger {
private SdkAsyncHttpClient httpClient;
private KinesisAsyncClient kinesisClient;
- @Before
- public void setUp() throws Exception {
+ @BeforeEach
+ void setUp() throws Exception {
System.setProperty(SdkSystemSetting.CBOR_ENABLED.property(), "false");
env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -93,15 +94,14 @@ public class KinesisStreamsSinkITCase extends TestLogger {
kinesisClient = KINESALITE.createHostClient(httpClient);
}
- @After
- public void teardown() {
+ @AfterEach
+ void teardown() {
System.clearProperty(SdkSystemSetting.CBOR_ENABLED.property());
AWSGeneralUtil.closeResources(httpClient, kinesisClient);
}
@Test
- public void elementsMaybeWrittenSuccessfullyToLocalInstanceWhenBatchSizeIsReached()
- throws Exception {
+ void elementsMaybeWrittenSuccessfullyToLocalInstanceWhenBatchSizeIsReached() throws Exception {
new Scenario()
.withKinesaliteStreamName("test-stream-name-1")
@@ -110,7 +110,7 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void elementsBufferedAndTriggeredByTimeBasedFlushShouldBeFlushedIfSourcedIsKeptAlive()
+ void elementsBufferedAndTriggeredByTimeBasedFlushShouldBeFlushedIfSourcedIsKeptAlive()
throws Exception {
new Scenario()
@@ -123,7 +123,7 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void veryLargeMessagesSucceedInBeingPersisted() throws Exception {
+ void veryLargeMessagesSucceedInBeingPersisted() throws Exception {
new Scenario()
.withNumberOfElementsToSend(5)
@@ -136,8 +136,7 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void multipleInFlightRequestsResultsInCorrectNumberOfElementsPersisted()
- throws Exception {
+ void multipleInFlightRequestsResultsInCorrectNumberOfElementsPersisted() throws Exception {
new Scenario()
.withNumberOfElementsToSend(150)
@@ -152,12 +151,12 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void nonExistentStreamNameShouldResultInFailureInFailOnErrorIsOn() {
+ void nonExistentStreamNameShouldResultInFailureInFailOnErrorIsOn() {
testJobFatalFailureTerminatesCorrectlyWithFailOnErrorFlagSetTo(true, "test-stream-name-5");
}
@Test
- public void nonExistentStreamNameShouldResultInFailureInFailOnErrorIsOff() {
+ void nonExistentStreamNameShouldResultInFailureInFailOnErrorIsOff() {
testJobFatalFailureTerminatesCorrectlyWithFailOnErrorFlagSetTo(false, "test-stream-name-6");
}
@@ -177,7 +176,7 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void veryLargeMessagesFailGracefullyWithBrokenElementConverter() {
+ void veryLargeMessagesFailGracefullyWithBrokenElementConverter() {
Assertions.assertThatExceptionOfType(JobExecutionException.class)
.isThrownBy(
() ->
@@ -197,12 +196,12 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void badRegionShouldResultInFailureWhenInFailOnErrorIsOn() {
+ void badRegionShouldResultInFailureWhenInFailOnErrorIsOn() {
badRegionShouldResultInFailureWhenInFailOnErrorIs(true);
}
@Test
- public void badRegionShouldResultInFailureWhenInFailOnErrorIsOff() {
+ void badRegionShouldResultInFailureWhenInFailOnErrorIsOff() {
badRegionShouldResultInFailureWhenInFailOnErrorIs(false);
}
@@ -215,12 +214,12 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void missingRegionShouldResultInFailureWhenInFailOnErrorIsOn() {
+ void missingRegionShouldResultInFailureWhenInFailOnErrorIsOn() {
missingRegionShouldResultInFailureWhenInFailOnErrorIs(true);
}
@Test
- public void missingRegionShouldResultInFailureWhenInFailOnErrorIsOff() {
+ void missingRegionShouldResultInFailureWhenInFailOnErrorIsOff() {
missingRegionShouldResultInFailureWhenInFailOnErrorIs(false);
}
@@ -232,12 +231,12 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void noURIEndpointShouldResultInFailureWhenInFailOnErrorIsOn() {
+ void noURIEndpointShouldResultInFailureWhenInFailOnErrorIsOn() {
noURIEndpointShouldResultInFailureWhenInFailOnErrorIs(true);
}
@Test
- public void noURIEndpointShouldResultInFailureWhenInFailOnErrorIsOff() {
+ void noURIEndpointShouldResultInFailureWhenInFailOnErrorIsOff() {
noURIEndpointShouldResultInFailureWhenInFailOnErrorIs(false);
}
@@ -249,12 +248,12 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void badEndpointShouldResultInFailureWhenInFailOnErrorIsOn() {
+ void badEndpointShouldResultInFailureWhenInFailOnErrorIsOn() {
badEndpointShouldResultInFailureWhenInFailOnErrorIs(true);
}
@Test
- public void badEndpointShouldResultInFailureWhenInFailOnErrorIsOff() {
+ void badEndpointShouldResultInFailureWhenInFailOnErrorIsOff() {
badEndpointShouldResultInFailureWhenInFailOnErrorIs(false);
}
@@ -268,7 +267,7 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void envVarWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOn() {
+ void envVarWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOn() {
noCredentialsProvidedAndCredentialsProviderSpecifiedShouldResultInFailure(
true,
AWSConfigConstants.CredentialProvider.ENV_VAR.toString(),
@@ -276,7 +275,7 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void envVarWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOff() {
+ void envVarWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOff() {
noCredentialsProvidedAndCredentialsProviderSpecifiedShouldResultInFailure(
false,
AWSConfigConstants.CredentialProvider.ENV_VAR.toString(),
@@ -284,7 +283,7 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void sysPropWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOn() {
+ void sysPropWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOn() {
noCredentialsProvidedAndCredentialsProviderSpecifiedShouldResultInFailure(
true,
AWSConfigConstants.CredentialProvider.SYS_PROP.toString(),
@@ -292,7 +291,7 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void sysPropWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOff() {
+ void sysPropWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOff() {
noCredentialsProvidedAndCredentialsProviderSpecifiedShouldResultInFailure(
false,
AWSConfigConstants.CredentialProvider.SYS_PROP.toString(),
@@ -300,7 +299,7 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void basicWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOn() {
+ void basicWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOn() {
noCredentialsProvidedAndCredentialsProviderSpecifiedShouldResultInFailure(
true,
AWSConfigConstants.CredentialProvider.BASIC.toString(),
@@ -308,7 +307,7 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void basicWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOff() {
+ void basicWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOff() {
noCredentialsProvidedAndCredentialsProviderSpecifiedShouldResultInFailure(
false,
AWSConfigConstants.CredentialProvider.BASIC.toString(),
@@ -316,7 +315,7 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void webIdentityTokenWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOn() {
+ void webIdentityTokenWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOn() {
noCredentialsProvidedAndCredentialsProviderSpecifiedShouldResultInFailure(
true,
AWSConfigConstants.CredentialProvider.WEB_IDENTITY_TOKEN.toString(),
@@ -324,7 +323,7 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void webIdentityTokenWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOff() {
+ void webIdentityTokenWithNoCredentialsShouldResultInFailureWhenInFailOnErrorIsOff() {
noCredentialsProvidedAndCredentialsProviderSpecifiedShouldResultInFailure(
false,
AWSConfigConstants.CredentialProvider.WEB_IDENTITY_TOKEN.toString(),
@@ -332,13 +331,13 @@ public class KinesisStreamsSinkITCase extends TestLogger {
}
@Test
- public void wrongCredentialProviderNameShouldResultInFailureWhenInFailOnErrorIsOn() {
+ void wrongCredentialProviderNameShouldResultInFailureWhenInFailOnErrorIsOn() {
noCredentialsProvidedAndCredentialsProviderSpecifiedShouldResultInFailure(
true, "WRONG", "Invalid AWS Credential Provider Type");
}
@Test
- public void wrongCredentialProviderNameShouldResultInFailureWhenInFailOnErrorIsOff() {
+ void wrongCredentialProviderNameShouldResultInFailureWhenInFailOnErrorIsOff() {
noCredentialsProvidedAndCredentialsProviderSpecifiedShouldResultInFailure(
false, "WRONG", "Invalid AWS Credential Provider Type");
}
diff --git a/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsStateSerializerTest.java b/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsStateSerializerTest.java
index ce1bde5cd15..9c475206385 100644
--- a/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsStateSerializerTest.java
+++ b/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsStateSerializerTest.java
@@ -22,7 +22,7 @@ import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
import org.apache.flink.connector.base.sink.writer.ElementConverter;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import software.amazon.awssdk.services.kinesis.model.PutRecordsRequestEntry;
import java.io.IOException;
@@ -31,7 +31,7 @@ import static org.apache.flink.connector.base.sink.writer.AsyncSinkWriterTestUti
import static org.apache.flink.connector.base.sink.writer.AsyncSinkWriterTestUtils.getTestState;
/** Test class for {@link KinesisStreamsStateSerializer}. */
-public class KinesisStreamsStateSerializerTest {
+class KinesisStreamsStateSerializerTest {
private static final ElementConverter<String, PutRecordsRequestEntry> ELEMENT_CONVERTER =
KinesisStreamsSinkElementConverter.<String>builder()
@@ -40,7 +40,7 @@ public class KinesisStreamsStateSerializerTest {
.build();
@Test
- public void testSerializeAndDeserialize() throws IOException {
+ void testSerializeAndDeserialize() throws IOException {
BufferedRequestState<PutRecordsRequestEntry> expectedState =
getTestState(ELEMENT_CONVERTER, this::getRequestSize);
diff --git a/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/KinesisDynamicTableSinkFactoryTest.java b/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/KinesisDynamicTableSinkFactoryTest.java
index e50acdbfde6..ef939cd341c 100644
--- a/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/KinesisDynamicTableSinkFactoryTest.java
+++ b/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/KinesisDynamicTableSinkFactoryTest.java
@@ -32,10 +32,9 @@ import org.apache.flink.table.factories.TestFormatFactory;
import org.apache.flink.table.runtime.connector.sink.SinkRuntimeProviderContext;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
-import org.apache.flink.util.TestLogger;
import org.assertj.core.api.Assertions;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.List;
@@ -51,11 +50,11 @@ import static org.apache.flink.connector.kinesis.table.KinesisConnectorOptions.S
import static org.apache.flink.table.factories.utils.FactoryMocks.createTableSink;
/** Test for {@link KinesisDynamicSink} created by {@link KinesisDynamicTableSinkFactory}. */
-public class KinesisDynamicTableSinkFactoryTest extends TestLogger {
+class KinesisDynamicTableSinkFactoryTest {
private static final String STREAM_NAME = "myStream";
@Test
- public void testGoodTableSinkForPartitionedTable() {
+ void testGoodTableSinkForPartitionedTable() {
ResolvedSchema sinkSchema = defaultSinkSchema();
DataType physicalDataType = sinkSchema.toPhysicalRowDataType();
Map<String, String> sinkOptions = defaultTableOptions().build();
@@ -89,7 +88,7 @@ public class KinesisDynamicTableSinkFactoryTest extends TestLogger {
}
@Test
- public void testGoodTableSinkCopyForPartitionedTable() {
+ void testGoodTableSinkCopyForPartitionedTable() {
ResolvedSchema sinkSchema = defaultSinkSchema();
DataType physicalDataType = sinkSchema.toPhysicalRowDataType();
Map<String, String> sinkOptions = defaultTableOptions().build();
@@ -117,7 +116,7 @@ public class KinesisDynamicTableSinkFactoryTest extends TestLogger {
}
@Test
- public void testGoodTableSinkForNonPartitionedTable() {
+ void testGoodTableSinkForNonPartitionedTable() {
ResolvedSchema sinkSchema = defaultSinkSchema();
Map<String, String> sinkOptions = defaultTableOptions().build();
@@ -146,7 +145,7 @@ public class KinesisDynamicTableSinkFactoryTest extends TestLogger {
}
@Test
- public void testGoodTableSinkForNonPartitionedTableWithSinkOptions() {
+ void testGoodTableSinkForNonPartitionedTableWithSinkOptions() {
ResolvedSchema sinkSchema = defaultSinkSchema();
Map<String, String> sinkOptions = defaultTableOptionsWithSinkOptions().build();
@@ -175,7 +174,7 @@ public class KinesisDynamicTableSinkFactoryTest extends TestLogger {
}
@Test
- public void testGoodTableSinkForNonPartitionedTableWithProducerOptions() {
+ void testGoodTableSinkForNonPartitionedTableWithProducerOptions() {
ResolvedSchema sinkSchema = defaultSinkSchema();
Map<String, String> sinkOptions = defaultTableOptionsWithDeprecatedOptions().build();
@@ -209,7 +208,7 @@ public class KinesisDynamicTableSinkFactoryTest extends TestLogger {
}
@Test
- public void testBadTableSinkForCustomPartitionerForPartitionedTable() {
+ void testBadTableSinkForCustomPartitionerForPartitionedTable() {
ResolvedSchema sinkSchema = defaultSinkSchema();
Map<String, String> sinkOptions =
defaultTableOptions()
@@ -229,7 +228,7 @@ public class KinesisDynamicTableSinkFactoryTest extends TestLogger {
}
@Test
- public void testBadTableSinkForNonExistingPartitionerClass() {
+ void testBadTableSinkForNonExistingPartitionerClass() {
ResolvedSchema sinkSchema = defaultSinkSchema();
Map<String, String> sinkOptions =
defaultTableOptions()
diff --git a/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/RowDataFieldsKinesisPartitionKeyGeneratorTest.java b/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/RowDataFieldsKinesisPartitionKeyGeneratorTest.java
index 0b85c97f64f..73054362257 100644
--- a/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/RowDataFieldsKinesisPartitionKeyGeneratorTest.java
+++ b/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/RowDataFieldsKinesisPartitionKeyGeneratorTest.java
@@ -26,11 +26,8 @@ import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.factories.TableOptionsBuilder;
import org.apache.flink.table.factories.TestFormatFactory;
import org.apache.flink.table.types.logical.RowType;
-import org.apache.flink.util.TestLogger;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
+import org.junit.jupiter.api.Test;
import java.time.LocalDate;
import java.time.LocalDateTime;
@@ -42,12 +39,12 @@ import java.util.HashMap;
import java.util.List;
import static org.apache.flink.connector.kinesis.table.RowDataFieldsKinesisPartitionKeyGenerator.MAX_PARTITION_KEY_LENGTH;
-import static org.apache.flink.core.testutils.FlinkMatchers.containsCause;
import static org.apache.flink.table.utils.EncodingUtils.repeat;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link RowDataFieldsKinesisPartitionKeyGenerator}. */
-public class RowDataFieldsKinesisPartitionKeyGeneratorTest extends TestLogger {
+class RowDataFieldsKinesisPartitionKeyGeneratorTest {
/** Table name to use for the tests. */
private static final String TABLE_NAME = "click_stream";
@@ -91,14 +88,12 @@ public class RowDataFieldsKinesisPartitionKeyGeneratorTest extends TestLogger {
/** A default IP to use for sample {@link RowData} elements in the tests. */
private static final String IP = "255.255.255.255";
- @Rule public ExpectedException thrown = ExpectedException.none();
-
// --------------------------------------------------------------------------------------------
// Positive tests
// --------------------------------------------------------------------------------------------
@Test
- public void testGoodPartitioner() {
+ void testGoodPartitioner() {
for (String delimiter : FIELD_DELIMITERS) {
RowDataFieldsKinesisPartitionKeyGenerator partitioner =
new RowDataFieldsKinesisPartitionKeyGenerator(
@@ -108,13 +103,13 @@ public class RowDataFieldsKinesisPartitionKeyGeneratorTest extends TestLogger {
String expectedKey = String.join(delimiter, String.valueOf(days(time)), IP);
String actualKey = partitioner.apply(createElement(time, IP));
- assertEquals(expectedKey, actualKey);
+ assertThat(actualKey).isEqualTo(expectedKey);
}
}
}
@Test
- public void testGoodPartitionerExceedingMaxLength() {
+ void testGoodPartitionerExceedingMaxLength() {
RowDataFieldsKinesisPartitionKeyGenerator partitioner =
new RowDataFieldsKinesisPartitionKeyGenerator(ROW_TYPE, PARTITION_BY_ROUTE);
@@ -124,12 +119,12 @@ public class RowDataFieldsKinesisPartitionKeyGeneratorTest extends TestLogger {
for (LocalDateTime time : DATE_TIMES) {
String actualKey = partitioner.apply(createElement(time, ip, route));
- assertEquals(expectedKey, actualKey);
+ assertThat(actualKey).isEqualTo(expectedKey);
}
}
@Test
- public void testGoodPartitionerWithStaticPrefix() {
+ void testGoodPartitionerWithStaticPrefix() {
// fixed prefix
String year = String.valueOf(year(DATE_TIMES.get(0)));
String month = String.valueOf(monthOfYear(DATE_TIMES.get(0)));
@@ -152,13 +147,13 @@ public class RowDataFieldsKinesisPartitionKeyGeneratorTest extends TestLogger {
String expectedKey = String.join(delimiter, year, month, day);
String actualKey = partitioner.apply(createElement(time, IP));
- assertEquals(expectedKey, actualKey);
+ assertThat(actualKey).isEqualTo(expectedKey);
}
}
}
@Test
- public void testGoodPartitionerWithStaticSuffix() {
+ void testGoodPartitionerWithStaticSuffix() {
// fixed suffix
String month = String.valueOf(monthOfYear(DATE_TIMES.get(0)));
String day = String.valueOf(dayOfMonth(DATE_TIMES.get(0)));
@@ -181,13 +176,13 @@ public class RowDataFieldsKinesisPartitionKeyGeneratorTest extends TestLogger {
String expectedKey = String.join(delimiter, year, month, day);
String actualKey = partitioner.apply(createElement(time, IP));
- assertEquals(expectedKey, actualKey);
+ assertThat(actualKey).isEqualTo(expectedKey);
}
}
}
@Test
- public void testGoodPartitionerWithStaticInfix() {
+ void testGoodPartitionerWithStaticInfix() {
// fixed infix
String month = String.valueOf(monthOfYear(DATE_TIMES.get(0)));
@@ -209,7 +204,7 @@ public class RowDataFieldsKinesisPartitionKeyGeneratorTest extends TestLogger {
String expectedKey = String.join(delimiter, year, month, day);
String actualKey = partitioner.apply(createElement(time, IP));
- assertEquals(expectedKey, actualKey);
+ assertThat(actualKey).isEqualTo(expectedKey);
}
}
}
@@ -219,47 +214,46 @@ public class RowDataFieldsKinesisPartitionKeyGeneratorTest extends TestLogger {
// --------------------------------------------------------------------------------------------
@Test
- public void testBadPartitionerWithEmptyPrefix() {
- thrown.expect(IllegalArgumentException.class);
- thrown.expect(
- containsCause(
- new IllegalArgumentException(
- "Cannot create a RowDataFieldsKinesisPartitioner for a non-partitioned table")));
-
- new RowDataFieldsKinesisPartitionKeyGenerator(ROW_TYPE, Collections.emptyList());
+ void testBadPartitionerWithEmptyPrefix() {
+ assertThatThrownBy(
+ () ->
+ new RowDataFieldsKinesisPartitionKeyGenerator(
+ ROW_TYPE, Collections.emptyList()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ "Cannot create a RowDataFieldsKinesisPartitioner for a non-partitioned table");
}
@Test
- public void testBadPartitionerWithDuplicatePartitionKeys() {
- thrown.expect(IllegalArgumentException.class);
- thrown.expect(
- containsCause(
- new IllegalArgumentException(
- "The sequence of partition keys cannot contain duplicates")));
-
- new RowDataFieldsKinesisPartitionKeyGenerator(ROW_TYPE, Arrays.asList("ip", "ip"));
+ void testBadPartitionerWithDuplicatePartitionKeys() {
+ assertThatThrownBy(
+ () ->
+ new RowDataFieldsKinesisPartitionKeyGenerator(
+ ROW_TYPE, Arrays.asList("ip", "ip")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("The sequence of partition keys cannot contain duplicates");
}
@Test
- public void testBadPartitionerWithBadFieldFieldNames() {
- thrown.expect(IllegalArgumentException.class);
- thrown.expect(
- containsCause(
- new IllegalArgumentException(
- "The following partition keys are not present in the table: abc")));
-
- new RowDataFieldsKinesisPartitionKeyGenerator(ROW_TYPE, Arrays.asList("ip", "abc"));
+ void testBadPartitionerWithBadFieldFieldNames() {
+ assertThatThrownBy(
+ () ->
+ new RowDataFieldsKinesisPartitionKeyGenerator(
+ ROW_TYPE, Arrays.asList("ip", "abc")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ "The following partition keys are not present in the table: abc");
}
@Test
- public void testBadPartitionerWithBadFieldFieldTypes() {
- thrown.expect(IllegalArgumentException.class);
- thrown.expect(
- containsCause(
- new IllegalArgumentException(
- "The following partition keys have types that are not supported by Kinesis: time")));
-
- new RowDataFieldsKinesisPartitionKeyGenerator(ROW_TYPE, Arrays.asList("time", "ip"));
+ void testBadPartitionerWithBadFieldFieldTypes() {
+ assertThatThrownBy(
+ () ->
+ new RowDataFieldsKinesisPartitionKeyGenerator(
+ ROW_TYPE, Arrays.asList("time", "ip")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ "The following partition keys have types that are not supported by Kinesis: time");
}
// --------------------------------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/util/KinesisProducerOptionsMapperTest.java b/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/util/KinesisProducerOptionsMapperTest.java
index f4a295b33bf..141e87098a0 100644
--- a/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/util/KinesisProducerOptionsMapperTest.java
+++ b/flink-connectors/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/util/KinesisProducerOptionsMapperTest.java
@@ -20,19 +20,18 @@ package org.apache.flink.connector.kinesis.table.util;
import org.apache.flink.connector.aws.config.AWSConfigConstants;
import org.apache.flink.connector.kinesis.table.util.KinesisStreamsConnectorOptionsUtils.KinesisProducerOptionsMapper;
-import org.apache.flink.util.TestLogger;
import org.assertj.core.api.Assertions;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.HashMap;
import java.util.Map;
/** Unit tests for {@link KinesisProducerOptionsMapper}. */
-public class KinesisProducerOptionsMapperTest extends TestLogger {
+class KinesisProducerOptionsMapperTest {
@Test
- public void testProducerVerifyCertificateOptionsMapping() {
+ void testProducerVerifyCertificateOptionsMapping() {
Map<String, String> deprecatedOptions = new HashMap<>();
deprecatedOptions.put("sink.producer.verify-certificate", "false");
Map<String, String> expectedOptions = new HashMap<>();
@@ -48,7 +47,7 @@ public class KinesisProducerOptionsMapperTest extends TestLogger {
}
@Test
- public void testProducerEndpointExtraction() {
+ void testProducerEndpointExtraction() {
Map<String, String> deprecatedOptions = new HashMap<>();
Map<String, String> expectedOptions = new HashMap<>();
deprecatedOptions.put("sink.producer.kinesis-endpoint", "some-end-point.kinesis");
@@ -63,7 +62,7 @@ public class KinesisProducerOptionsMapperTest extends TestLogger {
}
@Test
- public void testProducerEndpointAndPortExtraction() {
+ void testProducerEndpointAndPortExtraction() {
Map<String, String> deprecatedOptions = new HashMap<>();
Map<String, String> expectedOptions = new HashMap<>();
deprecatedOptions.put("sink.producer.kinesis-endpoint", "some-end-point.kinesis");
diff --git a/flink-connectors/flink-connector-aws-kinesis-streams/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension b/flink-connectors/flink-connector-aws-kinesis-streams/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
new file mode 100644
index 00000000000..28999133c2b
--- /dev/null
+++ b/flink-connectors/flink-connector-aws-kinesis-streams/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+org.apache.flink.util.TestLoggerExtension
\ No newline at end of file