You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2023/02/16 14:17:46 UTC
[flink-connector-pulsar] branch main updated: [FLINK-31101][Test] Drop the use of JUnit 5 assertions and mockito.
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
The following commit(s) were added to refs/heads/main by this push:
new 2c43292 [FLINK-31101][Test] Drop the use of JUnit 5 assertions and mockito.
2c43292 is described below
commit 2c43292cc4af2bfedcfc801d0d53576cdbed92a9
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Thu Feb 16 20:21:23 2023 +0800
[FLINK-31101][Test] Drop the use of JUnit 5 assertions and mockito.
This closes #32
---
.../common/config/PulsarConfigBuilderTest.java | 25 ++++---
.../common/config/PulsarConfigValidatorTest.java | 16 +++--
.../common/config/PulsarConfigurationTest.java | 5 +-
.../handler/PulsarAdminInvocationHandlerTest.java | 6 +-
.../pulsar/common/schema/PulsarSchemaTest.java | 83 ++++++++++++----------
.../schema/PulsarSchemaTypeInformationTest.java | 15 ++--
.../schema/PulsarSchemaTypeSerializerTest.java | 11 ++-
.../common/schema/PulsarSchemaUtilsTest.java | 53 +++++++-------
.../schema/factories/AvroSchemaFactoryTest.java | 20 +++---
.../schema/factories/JSONSchemaFactoryTest.java | 4 +-
.../factories/KeyValueSchemaFactoryTest.java | 4 +-
.../factories/ProtobufNativeSchemaFactoryTest.java | 7 +-
.../factories/ProtobufSchemaFactoryTest.java | 10 ++-
.../pulsar/sink/PulsarSinkBuilderTest.java | 46 ++++++------
.../committer/PulsarCommittableSerializerTest.java | 4 +-
.../pulsar/sink/writer/PulsarWriterTest.java | 3 +-
.../sink/writer/router/KeyHashTopicRouterTest.java | 31 +++-----
.../writer/router/RoundRobinTopicRouterTest.java | 25 +++----
.../sink/writer/topic/MetadataListenerTest.java | 7 +-
.../sink/writer/topic/ProducerRegisterTest.java | 10 ++-
.../pulsar/source/PulsarSourceBuilderTest.java | 24 +++----
.../PulsarSourceEnumStateSerializerTest.java | 13 ++--
.../enumerator/assigner/SplitAssignerImplTest.java | 10 ++-
.../subscriber/PulsarSubscriberTest.java | 7 +-
.../enumerator/topic/TopicNameUtilsTest.java | 29 ++++----
.../enumerator/topic/TopicPartitionTest.java | 9 +--
.../source/enumerator/topic/TopicRangeTest.java | 18 ++---
.../topic/range/TopicRangeUtilsTest.java | 26 +++----
.../source/reader/PulsarSourceReaderTest.java | 13 ++--
.../GenericRecordDeserializationSchemaTest.java | 3 +-
.../PulsarDeserializationSchemaTest.java | 30 ++++----
.../split/PulsarPartitionSplitSerializerTest.java | 12 ++--
32 files changed, 279 insertions(+), 300 deletions(-)
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigBuilderTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigBuilderTest.java
index 9b3528f..8ae3097 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigBuilderTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigBuilderTest.java
@@ -26,9 +26,9 @@ import org.junit.jupiter.api.Test;
import java.util.Properties;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for {@link PulsarConfigBuilder}. */
class PulsarConfigBuilderTest {
@@ -39,8 +39,10 @@ class PulsarConfigBuilderTest {
PulsarConfigBuilder builder = new PulsarConfigBuilder();
builder.set(option, "value1");
- assertDoesNotThrow(() -> builder.set(option, "value1"));
- assertThrows(IllegalArgumentException.class, () -> builder.set(option, "value2"));
+ assertThatCode(() -> builder.set(option, "value1")).doesNotThrowAnyException();
+
+ assertThatThrownBy(() -> builder.set(option, "value2"))
+ .isInstanceOf(IllegalArgumentException.class);
}
@Test
@@ -52,10 +54,11 @@ class PulsarConfigBuilderTest {
configuration.set(option, "value1");
builder.set(option, "value1");
- assertDoesNotThrow(() -> builder.set(configuration));
+ assertThatCode(() -> builder.set(configuration)).doesNotThrowAnyException();
configuration.set(option, "value2");
- assertThrows(IllegalArgumentException.class, () -> builder.set(configuration));
+ assertThatThrownBy(() -> builder.set(configuration))
+ .isInstanceOf(IllegalArgumentException.class);
}
@Test
@@ -65,12 +68,14 @@ class PulsarConfigBuilderTest {
Properties properties = new Properties();
properties.put("int.type", "6");
- assertDoesNotThrow(() -> builder.set(properties));
+
+ assertThatCode(() -> builder.set(properties)).doesNotThrowAnyException();
properties.put("int.type", "1");
- assertThrows(IllegalArgumentException.class, () -> builder.set(properties));
+ assertThatThrownBy(() -> builder.set(properties))
+ .isInstanceOf(IllegalArgumentException.class);
Integer value = builder.get(option);
- assertEquals(value, 6);
+ assertThat(value).isEqualTo(6);
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigValidatorTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigValidatorTest.java
index 8f71286..f482eef 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigValidatorTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigValidatorTest.java
@@ -24,8 +24,8 @@ import org.apache.flink.configuration.Configuration;
import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for {@link PulsarConfigValidator}. */
class PulsarConfigValidatorTest {
@@ -44,14 +44,18 @@ class PulsarConfigValidatorTest {
Configuration configuration = new Configuration();
// Required options
- assertThrows(IllegalArgumentException.class, () -> validator.validate(configuration));
+ assertThatThrownBy(() -> validator.validate(configuration))
+ .isInstanceOf(IllegalArgumentException.class);
+
configuration.set(required, "required");
- assertDoesNotThrow(() -> validator.validate(configuration));
+ assertThatCode(() -> validator.validate(configuration)).doesNotThrowAnyException();
// Conflict options
configuration.set(c1, "c1");
- assertDoesNotThrow(() -> validator.validate(configuration));
+ assertThatCode(() -> validator.validate(configuration)).doesNotThrowAnyException();
+
configuration.set(c2, "c2");
- assertThrows(IllegalArgumentException.class, () -> validator.validate(configuration));
+ assertThatThrownBy(() -> validator.validate(configuration))
+ .isInstanceOf(IllegalArgumentException.class);
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigurationTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigurationTest.java
index cf6f757..ec57b00 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigurationTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigurationTest.java
@@ -29,7 +29,7 @@ import java.util.Properties;
import static java.util.Collections.emptyMap;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
-import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Unit tests for {@link PulsarConfiguration}. */
class PulsarConfigurationTest {
@@ -52,7 +52,8 @@ class PulsarConfigurationTest {
TestConfiguration configuration1 = new TestConfiguration(configuration);
Map<String, String> properties = configuration1.getProperties(PROP_OP);
- assertEquals(properties, expectProp);
+
+ assertThat(properties).isEqualTo(expectProp);
}
private static final class TestConfiguration extends PulsarConfiguration {
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/handler/PulsarAdminInvocationHandlerTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/handler/PulsarAdminInvocationHandlerTest.java
index c723a31..3c76edb 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/handler/PulsarAdminInvocationHandlerTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/handler/PulsarAdminInvocationHandlerTest.java
@@ -58,8 +58,8 @@ import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_REQUEST_RATES;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_REQUEST_RETRIES;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_REQUEST_WAIT_MILLIS;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.junit.jupiter.api.Assertions.assertEquals;
/** Unit test of {@link PulsarAdminInvocationHandler}. */
class PulsarAdminInvocationHandlerTest {
@@ -90,7 +90,7 @@ class PulsarAdminInvocationHandlerTest {
assertThatThrownBy(() -> proxyAdmin.lookups().lookupPartitionedTopic("aa"))
.isInstanceOf(PulsarAdminException.class)
.hasMessage(errorMsg);
- assertEquals(retryTimes, admin.lookup.callingTimes());
+ assertThat(admin.lookup.callingTimes()).isEqualTo(retryTimes);
}
@Test
@@ -110,7 +110,7 @@ class PulsarAdminInvocationHandlerTest {
assertThatThrownBy(() -> proxyAdmin.lookups().lookupTopic("some"))
.isInstanceOf(NotFoundException.class)
.hasMessage("not found");
- assertEquals(1, admin.lookup.callingTimes());
+ assertThat(admin.lookup.callingTimes()).isEqualTo(1);
}
/** Test implementation for PulsarAdmin. */
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTest.java
index 9547812..6d17bd4 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTest.java
@@ -38,10 +38,9 @@ import org.junit.jupiter.api.Test;
import java.io.Serializable;
-import static org.junit.jupiter.api.Assertions.assertAll;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for {@link PulsarSchema}. */
class PulsarSchemaTest {
@@ -55,44 +54,52 @@ class PulsarSchemaTest {
KeyValueSchemaImpl.of(Foo.class, FA.class, SchemaType.JSON);
@Test
- void pulsarSchemaCreation() {
- assertAll(
- "Primitive schemas creation",
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.BYTES)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.STRING)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.INT8)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.INT16)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.INT32)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.INT64)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.BOOL)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.FLOAT)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.DOUBLE)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.DATE)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.TIME)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.TIMESTAMP)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.INSTANT)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.LOCAL_DATE)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.LOCAL_TIME)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(Schema.LOCAL_DATE_TIME)));
+ void primitivePulsarSchemaCreation() {
+ assertThatCode(() -> new PulsarSchema<>(Schema.BYTES)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(Schema.STRING)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(Schema.INT8)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(Schema.INT16)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(Schema.INT32)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(Schema.INT64)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(Schema.BOOL)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(Schema.FLOAT)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(Schema.DOUBLE)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(Schema.DATE)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(Schema.TIME)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(Schema.TIMESTAMP)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(Schema.INSTANT)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(Schema.LOCAL_DATE)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(Schema.LOCAL_TIME)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(Schema.LOCAL_DATE_TIME)).doesNotThrowAnyException();
+ }
- assertAll(
- "Struct & KeyValue schema creation",
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(JSON, FL.class)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(AVRO, Bar.class)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(PROTO, TestMessage.class)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(PROTO_N, SubMessage.class)),
- () -> assertDoesNotThrow(() -> new PulsarSchema<>(KV, Foo.class, FA.class)));
+ @Test
+ void structAndKeyValuePulsarSchemaCreation() {
+ assertThatCode(() -> new PulsarSchema<>(JSON, FL.class)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(AVRO, Bar.class)).doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(PROTO, TestMessage.class))
+ .doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(PROTO_N, SubMessage.class))
+ .doesNotThrowAnyException();
+ assertThatCode(() -> new PulsarSchema<>(KV, Foo.class, FA.class))
+ .doesNotThrowAnyException();
}
@Test
@SuppressWarnings("unchecked")
void invalidPulsarSchemaCreationWithoutClassType() {
- assertThrows(IllegalArgumentException.class, () -> new PulsarSchema<>(AVRO));
- assertThrows(IllegalArgumentException.class, () -> new PulsarSchema<>(JSON));
- assertThrows(IllegalArgumentException.class, () -> new PulsarSchema<>(PROTO));
- assertThrows(IllegalArgumentException.class, () -> new PulsarSchema<>(PROTO_N));
- assertThrows(IllegalArgumentException.class, () -> new PulsarSchema<>(KV));
- assertThrows(IllegalArgumentException.class, () -> new PulsarSchema(KV, KeyValue.class));
+ assertThatThrownBy(() -> new PulsarSchema<>(AVRO))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> new PulsarSchema<>(JSON))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> new PulsarSchema<>(PROTO))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> new PulsarSchema<>(PROTO_N))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> new PulsarSchema<>(KV))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> new PulsarSchema(KV, KeyValue.class))
+ .isInstanceOf(IllegalArgumentException.class);
}
@Test
@@ -129,8 +136,8 @@ class PulsarSchemaTest {
private <T> void assertPulsarSchemaIsSerializable(PulsarSchema<T> schema) throws Exception {
PulsarSchema<T> clonedSchema = InstantiationUtil.clone(schema);
- assertEquals(clonedSchema.getSchemaInfo(), schema.getSchemaInfo());
- assertEquals(clonedSchema.getRecordClass(), schema.getRecordClass());
+ assertThat(clonedSchema.getSchemaInfo()).isEqualTo(schema.getSchemaInfo());
+ assertThat(clonedSchema.getRecordClass()).isEqualTo(schema.getRecordClass());
}
/** A POJO Class which would generate a large schema by Avro. */
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeInformationTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeInformationTest.java
index 48caa59..2f56f90 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeInformationTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeInformationTest.java
@@ -25,9 +25,8 @@ import org.apache.flink.util.InstantiationUtil;
import org.apache.pulsar.client.api.Schema;
import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
/** Unit tests for {@link PulsarSchemaTypeInformation}. */
class PulsarSchemaTypeInformationTest {
@@ -36,14 +35,14 @@ class PulsarSchemaTypeInformationTest {
void pulsarTypeInfoSerializationAndCreation() throws Exception {
PulsarSchema<Bar> schema = new PulsarSchema<>(Schema.AVRO(Bar.class), Bar.class);
PulsarSchemaTypeInformation<Bar> info = new PulsarSchemaTypeInformation<>(schema);
- assertDoesNotThrow(() -> InstantiationUtil.clone(info));
+ assertThatCode(() -> InstantiationUtil.clone(info)).doesNotThrowAnyException();
PulsarSchemaTypeInformation<Bar> clonedInfo = InstantiationUtil.clone(info);
- assertEquals(info, clonedInfo);
- assertNotSame(info, clonedInfo);
+ assertThat(clonedInfo).isEqualTo(info).isNotSameAs(info);
- assertDoesNotThrow(() -> info.createSerializer(new ExecutionConfig()));
+ assertThatCode(() -> info.createSerializer(new ExecutionConfig()))
+ .doesNotThrowAnyException();
- assertEquals(info.getTypeClass(), clonedInfo.getTypeClass());
+ assertThat(clonedInfo.getTypeClass()).isEqualTo(info.getTypeClass());
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeSerializerTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeSerializerTest.java
index 0bb0a32..afb75af 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeSerializerTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeSerializerTest.java
@@ -32,8 +32,7 @@ import java.util.concurrent.ThreadLocalRandom;
import static java.util.Collections.nCopies;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.apache.pulsar.client.api.Schema.PROTOBUF_NATIVE;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.assertj.core.api.Assertions.assertThat;
/** Unit tests for {@link PulsarSchemaTypeSerializer}. */
class PulsarSchemaTypeSerializerTest {
@@ -54,7 +53,7 @@ class PulsarSchemaTypeSerializerTest {
@Test
void createDeserializeRecordInstance() {
// Protobuf class creation
- assertNotNull(serializer.createInstance());
+ assertThat(serializer.createInstance()).isNotNull();
}
@Test
@@ -65,7 +64,7 @@ class PulsarSchemaTypeSerializerTest {
TestInputView input = output.getInputView();
TestMessage message1 = serializer.deserialize(input);
- assertEquals(message, message1);
+ assertThat(message1).isEqualTo(message);
}
@Test
@@ -80,7 +79,7 @@ class PulsarSchemaTypeSerializerTest {
TestInputView input2 = output2.getInputView();
TestMessage message1 = serializer.deserialize(input2);
- assertEquals(message, message1);
+ assertThat(message1).isEqualTo(message);
}
@Test
@@ -98,6 +97,6 @@ class PulsarSchemaTypeSerializerTest {
PulsarSchemaTypeSerializerTest.class.getClassLoader());
TypeSerializer<TestMessage> serializer1 = snapshot.restoreSerializer();
- assertEquals(serializer, serializer1);
+ assertThat(serializer1).isEqualTo(serializer);
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaUtilsTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaUtilsTest.java
index d4bd1fc..eb30971 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaUtilsTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaUtilsTest.java
@@ -30,30 +30,28 @@ import org.apache.pulsar.common.schema.KeyValue;
import org.apache.pulsar.common.schema.SchemaInfo;
import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNotEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.apache.flink.connector.pulsar.common.schema.PulsarSchemaUtils.isProtobufTypeClass;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for {@link PulsarSchemaUtils}. */
class PulsarSchemaUtilsTest {
@Test
void haveProtobufShouldReturnTrueIfWeProvidedIt() {
- assertTrue(PulsarSchemaUtils.haveProtobuf());
+ assertThat(PulsarSchemaUtils.haveProtobuf()).isTrue();
}
@Test
void protobufClassValidation() {
- assertTrue(PulsarSchemaUtils.isProtobufTypeClass(TestMessage.class));
- assertTrue(PulsarSchemaUtils.isProtobufTypeClass(SubMessage.class));
- assertTrue(PulsarSchemaUtils.isProtobufTypeClass(NestedMessage.class));
+ assertThat(isProtobufTypeClass(TestMessage.class)).isTrue();
+ assertThat(isProtobufTypeClass(SubMessage.class)).isTrue();
+ assertThat(isProtobufTypeClass(NestedMessage.class)).isTrue();
- assertFalse(PulsarSchemaUtils.isProtobufTypeClass(Bar.class));
- assertFalse(PulsarSchemaUtils.isProtobufTypeClass(FL.class));
- assertFalse(PulsarSchemaUtils.isProtobufTypeClass(Foo.class));
+ assertThat(isProtobufTypeClass(Bar.class)).isFalse();
+ assertThat(isProtobufTypeClass(FL.class)).isFalse();
+ assertThat(isProtobufTypeClass(Foo.class)).isFalse();
}
@Test
@@ -62,25 +60,31 @@ class PulsarSchemaUtilsTest {
Schema<Foo> avro1 = Schema.AVRO(Foo.class);
PulsarSchema<Foo> avro2 = new PulsarSchema<>(avro1, Foo.class);
SchemaInfo info1 = avro1.getSchemaInfo();
- assertThrows(NullPointerException.class, () -> PulsarSchemaUtils.createSchema(info1));
+ assertThatThrownBy(() -> PulsarSchemaUtils.createSchema(info1))
+ .isInstanceOf(NullPointerException.class);
- Schema<Foo> schema = PulsarSchemaUtils.createSchema(avro2.getSchemaInfo());
- assertNotEquals(schema.getSchemaInfo(), avro1.getSchemaInfo());
- assertEquals(schema.getSchemaInfo(), avro2.getSchemaInfo());
+ Schema<Foo> avro3 = PulsarSchemaUtils.createSchema(avro2.getSchemaInfo());
+ assertThat(avro3.getSchemaInfo())
+ .isNotEqualTo(avro1.getSchemaInfo())
+ .isEqualTo(avro2.getSchemaInfo());
// JSON
Schema<FL> json1 = Schema.JSON(FL.class);
PulsarSchema<FL> json2 = new PulsarSchema<>(json1, FL.class);
Schema<FL> json3 = PulsarSchemaUtils.createSchema(json2.getSchemaInfo());
- assertNotEquals(json3.getSchemaInfo(), json1.getSchemaInfo());
- assertEquals(json3.getSchemaInfo(), json2.getSchemaInfo());
+
+ assertThat(json3.getSchemaInfo())
+ .isNotEqualTo(json1.getSchemaInfo())
+ .isEqualTo(json2.getSchemaInfo());
// Protobuf Native
Schema<TestMessage> proto1 = Schema.PROTOBUF_NATIVE(TestMessage.class);
PulsarSchema<TestMessage> proto2 = new PulsarSchema<>(proto1, TestMessage.class);
Schema<TestMessage> proto3 = PulsarSchemaUtils.createSchema(proto2.getSchemaInfo());
- assertNotEquals(proto3.getSchemaInfo(), proto1.getSchemaInfo());
- assertEquals(proto3.getSchemaInfo(), proto2.getSchemaInfo());
+
+ assertThat(proto3.getSchemaInfo())
+ .isNotEqualTo(proto1.getSchemaInfo())
+ .isEqualTo(proto2.getSchemaInfo());
// KeyValue
Schema<KeyValue<byte[], byte[]>> kvBytes1 = Schema.KV_BYTES();
@@ -88,7 +92,8 @@ class PulsarSchemaUtilsTest {
new PulsarSchema<>(kvBytes1, byte[].class, byte[].class);
Schema<KeyValue<byte[], byte[]>> kvBytes3 =
PulsarSchemaUtils.createSchema(kvBytes2.getSchemaInfo());
- assertNotEquals(kvBytes3.getSchemaInfo(), kvBytes1.getSchemaInfo());
+
+ assertThat(kvBytes3.getSchemaInfo()).isNotEqualTo(kvBytes1.getSchemaInfo());
}
@Test
@@ -96,9 +101,9 @@ class PulsarSchemaUtilsTest {
Schema<Foo> schema = Schema.AVRO(Foo.class);
SchemaInfo info = schema.getSchemaInfo();
SchemaInfo newInfo = PulsarSchemaUtils.encodeClassInfo(info, Foo.class);
- assertDoesNotThrow(() -> PulsarSchemaUtils.decodeClassInfo(newInfo));
+ assertThatCode(() -> PulsarSchemaUtils.decodeClassInfo(newInfo)).doesNotThrowAnyException();
Class<Foo> clazz = PulsarSchemaUtils.decodeClassInfo(newInfo);
- assertEquals(clazz, Foo.class);
+ assertThat(clazz).isEqualTo(Foo.class);
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/AvroSchemaFactoryTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/AvroSchemaFactoryTest.java
index 8812ab0..14d907d 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/AvroSchemaFactoryTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/AvroSchemaFactoryTest.java
@@ -38,24 +38,20 @@ import java.util.concurrent.ThreadLocalRandom;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for {@link AvroSchemaFactory}. */
class AvroSchemaFactoryTest {
- private final AvroSchemaFactory<?> factory = new AvroSchemaFactory<>();
-
@Test
void createAvroSchemaFromSchemaInfo() {
AvroSchema<DefaultStruct> schema1 = AvroSchema.of(DefaultStruct.class);
// AvroSchema should provide type class
- assertThrows(
- IllegalArgumentException.class,
- () -> new PulsarSchema<>(schema1),
- "Avro Schema should provide the type class");
+ assertThatThrownBy(() -> new PulsarSchema<>(schema1))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Avro Schema should provide the type class");
PulsarSchema<DefaultStruct> pulsarSchema = new PulsarSchema<>(schema1, DefaultStruct.class);
AvroSchemaFactory<DefaultStruct> factory = new AvroSchemaFactory<>();
@@ -74,7 +70,7 @@ class AvroSchemaFactoryTest {
byte[] bytes = schema1.encode(struct1);
DefaultStruct struct2 = schema2.decode(bytes);
- assertEquals(struct1, struct2);
+ assertThat(struct2).isEqualTo(struct1);
}
@Test
@@ -97,8 +93,8 @@ class AvroSchemaFactoryTest {
TypeSerializer<StructWithAnnotations> serializer =
information.createSerializer(new ExecutionConfig());
// TypeInformation serialization.
- assertDoesNotThrow(() -> InstantiationUtil.clone(information));
- assertDoesNotThrow(() -> InstantiationUtil.clone(serializer));
+ assertThatCode(() -> InstantiationUtil.clone(information)).doesNotThrowAnyException();
+ assertThatCode(() -> InstantiationUtil.clone(serializer)).doesNotThrowAnyException();
TestOutputView output = new TestOutputView();
serializer.serialize(struct1, output);
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/JSONSchemaFactoryTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/JSONSchemaFactoryTest.java
index d67b64c..59b3c9b 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/JSONSchemaFactoryTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/JSONSchemaFactoryTest.java
@@ -31,7 +31,7 @@ import org.apache.pulsar.client.impl.schema.JSONSchema;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.assertj.core.api.Assertions.assertThatCode;
/** Unit tests for {@link JSONSchemaFactory}. */
class JSONSchemaFactoryTest {
@@ -66,6 +66,6 @@ class JSONSchemaFactoryTest {
.hasFieldOrPropertyWithValue("typeClass", FL.class);
// TypeInformation serialization.
- assertDoesNotThrow(() -> InstantiationUtil.clone(typeInfo));
+ assertThatCode(() -> InstantiationUtil.clone(typeInfo)).doesNotThrowAnyException();
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/KeyValueSchemaFactoryTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/KeyValueSchemaFactoryTest.java
index 425ec4f..0f11b9c 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/KeyValueSchemaFactoryTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/KeyValueSchemaFactoryTest.java
@@ -34,7 +34,7 @@ import org.apache.pulsar.common.schema.SchemaType;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.assertj.core.api.Assertions.assertThatCode;
/** Unit tests for {@link KeyValueSchemaFactory}. */
class KeyValueSchemaFactoryTest {
@@ -68,7 +68,7 @@ class KeyValueSchemaFactoryTest {
TypeInformation<KeyValue<Bar, FA>> info =
factory.createTypeInfo(pulsarSchema.getSchemaInfo());
// TypeInformation serialization.
- assertDoesNotThrow(() -> InstantiationUtil.clone(info));
+ assertThatCode(() -> InstantiationUtil.clone(info)).doesNotThrowAnyException();
assertThat(InstantiationUtil.clone(info))
.isInstanceOf(PulsarSchemaTypeInformation.class)
.hasFieldOrPropertyWithValue("typeClass", KeyValue.class);
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/ProtobufNativeSchemaFactoryTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/ProtobufNativeSchemaFactoryTest.java
index 0513959..26e68a3 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/ProtobufNativeSchemaFactoryTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/ProtobufNativeSchemaFactoryTest.java
@@ -35,8 +35,7 @@ import java.util.concurrent.ThreadLocalRandom;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.assertj.core.api.Assertions.assertThatCode;
/** Unit tests for {@link ProtobufNativeSchemaFactory}. */
class ProtobufNativeSchemaFactoryTest {
@@ -69,7 +68,7 @@ class ProtobufNativeSchemaFactoryTest {
byte[] bytes = schema1.encode(message);
TestMessage message1 = schema2.decode(bytes);
- assertEquals(message, message1);
+ assertThat(message1).isEqualTo(message);
}
@Test
@@ -80,7 +79,7 @@ class ProtobufNativeSchemaFactoryTest {
TypeInformation<NestedMessage> typeInfo =
factory.createTypeInfo(pulsarSchema.getSchemaInfo());
- assertDoesNotThrow(() -> InstantiationUtil.clone(typeInfo));
+ assertThatCode(() -> InstantiationUtil.clone(typeInfo)).doesNotThrowAnyException();
assertThat(typeInfo)
.isInstanceOf(PulsarSchemaTypeInformation.class)
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/ProtobufSchemaFactoryTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/ProtobufSchemaFactoryTest.java
index 3cbbe74..b95fd06 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/ProtobufSchemaFactoryTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/schema/factories/ProtobufSchemaFactoryTest.java
@@ -31,9 +31,7 @@ import org.junit.jupiter.api.Test;
import static org.apache.flink.connector.pulsar.common.schema.PulsarSchemaUtils.decodeClassInfo;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.assertj.core.api.Assertions.assertThatCode;
/** Unit tests for {@link ProtobufSchemaFactory}. */
class ProtobufSchemaFactoryTest {
@@ -49,8 +47,8 @@ class ProtobufSchemaFactoryTest {
.hasFieldOrPropertyWithValue("schemaInfo", pulsarSchema.getSchemaInfo())
.isInstanceOf(ProtobufSchema.class);
- assertEquals(decodeClassInfo(schema2.getSchemaInfo()), SubMessage.class);
- assertNotSame(schema, schema2);
+ assertThat(decodeClassInfo(schema2.getSchemaInfo())).isEqualTo(SubMessage.class);
+ assertThat(schema2).isNotSameAs(schema);
}
@Test
@@ -61,7 +59,7 @@ class ProtobufSchemaFactoryTest {
TypeInformation<TestMessage> typeInfo =
factory.createTypeInfo(pulsarSchema.getSchemaInfo());
- assertDoesNotThrow(() -> InstantiationUtil.clone(typeInfo));
+ assertThatCode(() -> InstantiationUtil.clone(typeInfo)).doesNotThrowAnyException();
assertThat(typeInfo)
.isInstanceOf(PulsarSchemaTypeInformation.class)
.hasFieldOrPropertyWithValue("typeClass", TestMessage.class);
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilderTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilderTest.java
index 6de87b9..e7bbd78 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilderTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilderTest.java
@@ -29,9 +29,8 @@ import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_SE
import static org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode.CUSTOM;
import static org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode.MESSAGE_KEY_HASH;
import static org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode.ROUND_ROBIN;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for {@link PulsarSinkBuilder}. */
class PulsarSinkBuilderTest {
@@ -41,16 +40,19 @@ class PulsarSinkBuilderTest {
PulsarSinkBuilder<String> builder = PulsarSink.builder();
builder.setTopics("a", "b");
- assertThrows(IllegalStateException.class, () -> builder.setTopics("c"));
+ assertThatThrownBy(() -> builder.setTopics("c")).isInstanceOf(IllegalStateException.class);
}
@Test
void topicRoutingModeCouldNotBeCustom() {
PulsarSinkBuilder<String> builder = PulsarSink.builder();
- assertDoesNotThrow(() -> builder.setTopicRoutingMode(ROUND_ROBIN));
- assertDoesNotThrow(() -> builder.setTopicRoutingMode(MESSAGE_KEY_HASH));
- assertThrows(IllegalArgumentException.class, () -> builder.setTopicRoutingMode(CUSTOM));
+ assertThatCode(() -> builder.setTopicRoutingMode(ROUND_ROBIN)).doesNotThrowAnyException();
+ assertThatCode(() -> builder.setTopicRoutingMode(MESSAGE_KEY_HASH))
+ .doesNotThrowAnyException();
+
+ assertThatThrownBy(() -> builder.setTopicRoutingMode(CUSTOM))
+ .isInstanceOf(IllegalArgumentException.class);
}
@Test
@@ -58,27 +60,30 @@ class PulsarSinkBuilderTest {
PulsarSinkBuilder<String> builder = PulsarSink.builder();
builder.setConfig(PULSAR_SEND_TIMEOUT_MS, 1L);
- assertDoesNotThrow(() -> builder.setConfig(PULSAR_SEND_TIMEOUT_MS, 1L));
+ assertThatCode(() -> builder.setConfig(PULSAR_SEND_TIMEOUT_MS, 1L))
+ .doesNotThrowAnyException();
- assertThrows(
- IllegalArgumentException.class,
- () -> builder.setConfig(PULSAR_SEND_TIMEOUT_MS, 2L));
+ assertThatThrownBy(() -> builder.setConfig(PULSAR_SEND_TIMEOUT_MS, 2L))
+ .isInstanceOf(IllegalArgumentException.class);
Configuration configuration = new Configuration();
configuration.set(PULSAR_SEND_TIMEOUT_MS, 3L);
- assertThrows(IllegalArgumentException.class, () -> builder.setConfig(configuration));
+ assertThatThrownBy(() -> builder.setConfig(configuration))
+ .isInstanceOf(IllegalArgumentException.class);
Properties properties = new Properties();
properties.put(PULSAR_SEND_TIMEOUT_MS.key(), 4L);
- assertThrows(IllegalArgumentException.class, () -> builder.setProperties(properties));
+ assertThatThrownBy(() -> builder.setProperties(properties))
+ .isInstanceOf(IllegalArgumentException.class);
}
@Test
void serializationSchemaIsRequired() {
PulsarSinkBuilder<String> builder = PulsarSink.builder();
- NullPointerException exception = assertThrows(NullPointerException.class, builder::build);
- assertThat(exception).hasMessage("serializationSchema must be set.");
+ assertThatThrownBy(builder::build)
+ .isInstanceOf(NullPointerException.class)
+ .hasMessage("serializationSchema must be set.");
}
@Test
@@ -86,8 +91,9 @@ class PulsarSinkBuilderTest {
PulsarSinkBuilder<String> builder = PulsarSink.builder();
builder.setSerializationSchema(new SimpleStringSchema());
- NullPointerException exception = assertThrows(NullPointerException.class, builder::build);
- assertThat(exception).hasMessage("No topic names or custom topic router are provided.");
+ assertThatThrownBy(builder::build)
+ .isInstanceOf(NullPointerException.class)
+ .hasMessage("No topic names or custom topic router are provided.");
}
@Test
@@ -95,12 +101,12 @@ class PulsarSinkBuilderTest {
PulsarSinkBuilder<String> builder = PulsarSink.builder();
builder.setSerializationSchema(new SimpleStringSchema());
builder.setTopics("a", "b");
- assertThrows(IllegalArgumentException.class, builder::build);
+ assertThatThrownBy(builder::build).isInstanceOf(IllegalArgumentException.class);
builder.setServiceUrl("pulsar://127.0.0.1:8888");
- assertThrows(IllegalArgumentException.class, builder::build);
+ assertThatThrownBy(builder::build).isInstanceOf(IllegalArgumentException.class);
builder.setAdminUrl("http://127.0.0.1:9999");
- assertDoesNotThrow(builder::build);
+ assertThatCode(builder::build).doesNotThrowAnyException();
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializerTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializerTest.java
index 393485b..f13a4bf 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializerTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializerTest.java
@@ -25,7 +25,7 @@ import java.io.IOException;
import java.util.concurrent.ThreadLocalRandom;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
-import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for serializing and deserializing {@link PulsarCommittable} with {@link
@@ -48,6 +48,6 @@ class PulsarCommittableSerializerTest {
byte[] bytes = INSTANCE.serialize(committable);
PulsarCommittable committable1 = INSTANCE.deserialize(INSTANCE.getVersion(), bytes);
- assertEquals(committable1, committable);
+ assertThat(committable1).isEqualTo(committable);
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
index 422d5e3..eba7b18 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
@@ -63,7 +63,6 @@ import static org.apache.flink.connector.base.DeliveryGuarantee.EXACTLY_ONCE;
import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_WRITE_SCHEMA_EVOLUTION;
import static org.apache.pulsar.client.api.Schema.STRING;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertEquals;
/** Unit tests for {@link PulsarWriter}. */
class PulsarWriterTest extends PulsarTestSuiteBase {
@@ -127,7 +126,7 @@ class PulsarWriterTest extends PulsarTestSuiteBase {
}
String consumedMessage = operator().receiveMessage(topic, STRING).getValue();
- assertEquals(consumedMessage, message);
+ assertThat(consumedMessage).isEqualTo(message);
}
private SinkConfiguration sinkConfiguration(DeliveryGuarantee deliveryGuarantee) {
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/KeyHashTopicRouterTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/KeyHashTopicRouterTest.java
index dac073e..8396a80 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/KeyHashTopicRouterTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/KeyHashTopicRouterTest.java
@@ -20,7 +20,6 @@ package org.apache.flink.connector.pulsar.sink.writer.router;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
-import org.apache.flink.connector.pulsar.sink.writer.context.PulsarSinkContext;
import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
import org.apache.pulsar.client.impl.Hash;
@@ -36,9 +35,8 @@ import static java.util.stream.Collectors.toList;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_MESSAGE_KEY_HASH;
import static org.apache.pulsar.client.util.MathUtils.signSafeMod;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.mockito.Mockito.mock;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for {@link KeyHashTopicRouter}. */
class KeyHashTopicRouterTest {
@@ -52,11 +50,9 @@ class KeyHashTopicRouterTest {
String message = randomAlphanumeric(10);
String key = randomAlphanumeric(10);
List<TopicPartition> emptyPartitions = emptyList();
- PulsarSinkContext sinkContext = mock(PulsarSinkContext.class);
- assertThrows(
- IllegalArgumentException.class,
- () -> router.route(message, key, emptyPartitions, sinkContext));
+ assertThatThrownBy(() -> router.route(message, key, emptyPartitions, null))
+ .isInstanceOf(IllegalArgumentException.class);
}
@ParameterizedTest
@@ -67,18 +63,12 @@ class KeyHashTopicRouterTest {
KeyHashTopicRouter<String> router1 = new KeyHashTopicRouter<>(configuration);
TopicPartition partition1 =
- router1.route(
- randomAlphanumeric(10),
- randomAlphanumeric(10),
- partitions,
- mock(PulsarSinkContext.class));
- assertEquals(partition1, partitions.get(0));
+ router1.route(randomAlphanumeric(10), randomAlphanumeric(10), partitions, null);
+ assertThat(partition1).isEqualTo(partitions.get(0));
KeyHashTopicRouter<String> router2 = new KeyHashTopicRouter<>(configuration);
- TopicPartition partition2 =
- router2.route(
- randomAlphanumeric(10), null, partitions, mock(PulsarSinkContext.class));
- assertEquals(partition2, partitions.get(0));
+ TopicPartition partition2 = router2.route(randomAlphanumeric(10), null, partitions, null);
+ assertThat(partition2).isEqualTo(partitions.get(0));
}
@ParameterizedTest
@@ -99,10 +89,9 @@ class KeyHashTopicRouterTest {
TopicPartition desiredPartitions = partitions.get(index);
String message = randomAlphanumeric(10);
- TopicPartition topic =
- router.route(message, messageKey, partitions, mock(PulsarSinkContext.class));
+ TopicPartition topic = router.route(message, messageKey, partitions, null);
- assertEquals(topic, desiredPartitions);
+ assertThat(topic).isEqualTo(desiredPartitions);
}
private SinkConfiguration sinkConfiguration(MessageKeyHash hash) {
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/RoundRobinTopicRouterTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/RoundRobinTopicRouterTest.java
index e64401a..046da30 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/RoundRobinTopicRouterTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/RoundRobinTopicRouterTest.java
@@ -20,7 +20,6 @@ package org.apache.flink.connector.pulsar.sink.writer.router;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
-import org.apache.flink.connector.pulsar.sink.writer.context.PulsarSinkContext;
import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
@@ -33,9 +32,8 @@ import java.util.concurrent.ThreadLocalRandom;
import static java.util.Collections.emptyList;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_BATCHING_MAX_MESSAGES;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.mockito.Mockito.mock;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for {@link RoundRobinTopicRouter}. */
class RoundRobinTopicRouterTest {
@@ -47,11 +45,9 @@ class RoundRobinTopicRouterTest {
String message = randomAlphabetic(10);
List<TopicPartition> partitions = emptyList();
- PulsarSinkContext context = mock(PulsarSinkContext.class);
- assertThrows(
- IllegalArgumentException.class,
- () -> router.route(message, null, partitions, context));
+ assertThatThrownBy(() -> router.route(message, null, partitions, null))
+ .isInstanceOf(IllegalArgumentException.class);
}
@Test
@@ -62,23 +58,22 @@ class RoundRobinTopicRouterTest {
List<TopicPartition> topics =
ImmutableList.of(new TopicPartition("topic1"), new TopicPartition("topic2"));
- PulsarSinkContext context = mock(PulsarSinkContext.class);
for (int i = 0; i < batchSize; i++) {
String message = randomAlphabetic(10);
- TopicPartition partition = router.route(message, null, topics, context);
- assertEquals(partition, topics.get(0));
+ TopicPartition partition = router.route(message, null, topics, null);
+ assertThat(partition).isEqualTo(topics.get(0));
}
for (int i = 0; i < batchSize; i++) {
String message = randomAlphabetic(10);
- TopicPartition partition = router.route(message, null, topics, context);
- assertEquals(partition, topics.get(1));
+ TopicPartition partition = router.route(message, null, topics, null);
+ assertThat(partition).isEqualTo(topics.get(1));
}
String message = randomAlphabetic(10);
- TopicPartition partition = router.route(message, null, topics, context);
- assertEquals(partition, topics.get(0));
+ TopicPartition partition = router.route(message, null, topics, null);
+ assertThat(partition).isEqualTo(topics.get(0));
}
private SinkConfiguration sinkConfiguration(int switchSize) {
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/MetadataListenerTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/MetadataListenerTest.java
index 8485fae..46c787a 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/MetadataListenerTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/MetadataListenerTest.java
@@ -36,7 +36,6 @@ import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_TOPIC_METADATA_REFRESH_INTERVAL;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicName;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertEquals;
/** Unit tests for {@link MetadataListener}. */
class MetadataListenerTest extends PulsarTestSuiteBase {
@@ -70,13 +69,13 @@ class MetadataListenerTest extends PulsarTestSuiteBase {
listener.open(configuration, timeService);
List<TopicPartition> partitions = listener.availablePartitions();
- assertEquals(desiredPartitions, partitions);
+ assertThat(partitions).isEqualTo(desiredPartitions);
operator().increaseTopicPartitions(topic, 12);
listener.refreshTopicMetadata(topic);
timeService.advance(interval);
partitions = listener.availablePartitions();
- assertEquals(desiredPartitions, partitions);
+ assertThat(partitions).isEqualTo(desiredPartitions);
}
@Test
@@ -140,7 +139,7 @@ class MetadataListenerTest extends PulsarTestSuiteBase {
listener.open(configuration, timeService);
List<TopicPartition> partitions = listener.availablePartitions();
- assertEquals(partitions, nonPartitionTopic);
+ assertThat(partitions).isEqualTo(nonPartitionTopic);
}
private List<TopicPartition> topicPartitions(String topic, int partitionSize) {
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegisterTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegisterTest.java
index 8329e8b..9e0dabf 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegisterTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegisterTest.java
@@ -43,8 +43,7 @@ import static org.apache.flink.connector.base.DeliveryGuarantee.EXACTLY_ONCE;
import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_VALIDATE_SINK_MESSAGE_BYTES;
import static org.apache.flink.metrics.groups.UnregisteredMetricsGroup.createSinkWriterMetricGroup;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for {@link ProducerRegister}. */
class ProducerRegisterTest extends PulsarTestSuiteBase {
@@ -75,7 +74,7 @@ class ProducerRegisterTest extends PulsarTestSuiteBase {
}
Message<String> receiveMessage = operator().receiveMessage(topic, Schema.STRING);
- assertEquals(receiveMessage.getValue(), message);
+ assertThat(receiveMessage.getValue()).isEqualTo(message);
}
@ParameterizedTest
@@ -116,8 +115,7 @@ class ProducerRegisterTest extends PulsarTestSuiteBase {
long message = ThreadLocalRandom.current().nextLong();
TypedMessageBuilder<byte[]> builder = register.createMessageBuilder(topic, Schema.BYTES);
- assertThrows(
- SchemaSerializationException.class,
- () -> builder.value(Schema.INT64.encode(message)));
+ assertThatThrownBy(() -> builder.value(Schema.INT64.encode(message)))
+ .isInstanceOf(SchemaSerializationException.class);
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilderTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilderTest.java
index 390ad07..043ae27 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilderTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilderTest.java
@@ -22,8 +22,6 @@ import org.apache.pulsar.client.api.Schema;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.junit.jupiter.api.Assertions.assertAll;
-import static org.junit.jupiter.api.Assertions.assertThrows;
/** Unit tests for {@link PulsarSourceBuilder}. */
class PulsarSourceBuilderTest {
@@ -32,19 +30,15 @@ class PulsarSourceBuilderTest {
void someSetterMethodCouldOnlyBeCalledOnce() {
PulsarSourceBuilder<String> builder = new PulsarSourceBuilder<>();
fillRequiredFields(builder);
- assertAll(
- () ->
- assertThrows(
- IllegalArgumentException.class,
- () -> builder.setAdminUrl("admin-url2")),
- () ->
- assertThrows(
- IllegalArgumentException.class,
- () -> builder.setServiceUrl("service-url2")),
- () ->
- assertThrows(
- IllegalArgumentException.class,
- () -> builder.setSubscriptionName("set_subscription_name2")));
+
+ assertThatThrownBy(() -> builder.setAdminUrl("admin-url2"))
+ .isInstanceOf(IllegalArgumentException.class);
+
+ assertThatThrownBy(() -> builder.setServiceUrl("service-url2"))
+ .isInstanceOf(IllegalArgumentException.class);
+
+ assertThatThrownBy(() -> builder.setSubscriptionName("set_subscription_name2"))
+ .isInstanceOf(IllegalArgumentException.class);
}
@Test
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumStateSerializerTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumStateSerializerTest.java
index dff0153..ba4ce2d 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumStateSerializerTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumStateSerializerTest.java
@@ -37,8 +37,7 @@ import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.apache.commons.lang3.RandomStringUtils.randomNumeric;
import static org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumStateSerializer.INSTANCE;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange.createFullRange;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.assertj.core.api.Assertions.assertThat;
/** Unit tests for {@link PulsarSourceEnumStateSerializer}. */
class PulsarSourceEnumStateSerializerTest {
@@ -57,8 +56,8 @@ class PulsarSourceEnumStateSerializerTest {
byte[] bytes = INSTANCE.serialize(state);
PulsarSourceEnumState state1 = INSTANCE.deserialize(3, bytes);
- assertEquals(state.getAppendedPartitions(), state1.getAppendedPartitions());
- assertNotSame(state, state1);
+ assertThat(state1.getAppendedPartitions()).isEqualTo(state.getAppendedPartitions());
+ assertThat(state1).isNotSameAs(state);
}
@Test
@@ -88,7 +87,7 @@ class PulsarSourceEnumStateSerializerTest {
new TopicPartition(
"topic33", 2, singletonList(new TopicRange(1222, 22233))));
- assertEquals(partitions, expectedPartitions);
+ assertThat(partitions).isEqualTo(expectedPartitions);
}
@Test
@@ -113,7 +112,7 @@ class PulsarSourceEnumStateSerializerTest {
new TopicPartition("topic1", 0, singletonList(new TopicRange(300, 4000))),
new TopicPartition("topic2", 4, singletonList(new TopicRange(600, 8000))));
- assertEquals(partitions, expectedPartitions);
+ assertThat(partitions).isEqualTo(expectedPartitions);
}
@Test
@@ -147,7 +146,7 @@ class PulsarSourceEnumStateSerializerTest {
new TopicPartition("topic3", 5, singletonList(new TopicRange(600, 2000))),
new TopicPartition("topic4", 8, singletonList(new TopicRange(300, 1000))));
- assertEquals(partitions, expectedPartitions);
+ assertThat(partitions).isEqualTo(expectedPartitions);
}
private void serializeVersion0SplitSet(DataOutputSerializer serializer) throws IOException {
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerImplTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerImplTest.java
index 3f11c41..c6ba5cc 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerImplTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerImplTest.java
@@ -44,8 +44,6 @@ import static org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEn
import static org.apache.flink.connector.pulsar.source.enumerator.assigner.SplitAssignerImpl.calculatePartitionOwner;
import static org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor.defaultStopCursor;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertTrue;
/** Unit tests for {@link SplitAssignerImpl}. */
class SplitAssignerImplTest {
@@ -99,19 +97,19 @@ class SplitAssignerImplTest {
@Test
void noMoreSplits() {
SplitAssigner assigner = splitAssigner(true, 4);
- assertFalse(assigner.noMoreSplits(3));
+ assertThat(assigner.noMoreSplits(3)).isFalse();
assigner = splitAssigner(false, 4);
- assertFalse(assigner.noMoreSplits(3));
+ assertThat(assigner.noMoreSplits(3)).isFalse();
Set<TopicPartition> partitions = createPartitions("persistent://public/default/f", 8);
int owner = calculatePartitionOwner("persistent://public/default/f", 8, 4);
assigner.registerTopicPartitions(partitions);
- assertFalse(assigner.noMoreSplits(owner));
+ assertThat(assigner.noMoreSplits(owner)).isFalse();
assigner.createAssignment(singletonList(owner));
- assertTrue(assigner.noMoreSplits(owner));
+ assertThat(assigner.noMoreSplits(owner)).isTrue();
}
@Test
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java
index 2bf6f8f..fb1ff32 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java
@@ -40,7 +40,6 @@ import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNam
import static org.apache.pulsar.client.api.RegexSubscriptionMode.AllTopics;
import static org.apache.pulsar.common.partition.PartitionedTopicMetadata.NON_PARTITIONED;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertEquals;
/** Unit tests for {@link PulsarSubscriber}. */
class PulsarSubscriberTest extends PulsarTestSuiteBase {
@@ -89,7 +88,7 @@ class PulsarSubscriberTest extends PulsarTestSuiteBase {
expectedPartitions.add(new TopicPartition(topic2, i));
}
- assertEquals(expectedPartitions, topicPartitions);
+ assertThat(topicPartitions).isEqualTo(expectedPartitions);
}
@Test
@@ -135,7 +134,7 @@ class PulsarSubscriberTest extends PulsarTestSuiteBase {
expectedPartitions.add(new TopicPartition(topic4, -1));
expectedPartitions.add(new TopicPartition(topic5, -1));
- assertEquals(expectedPartitions, topicPartitions);
+ assertThat(topicPartitions).isEqualTo(expectedPartitions);
}
@Test
@@ -156,6 +155,6 @@ class PulsarSubscriberTest extends PulsarTestSuiteBase {
expectedPartitions.add(new TopicPartition(topic3, i));
}
- assertEquals(expectedPartitions, topicPartitions);
+ assertThat(topicPartitions).isEqualTo(expectedPartitions);
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicNameUtilsTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicNameUtilsTest.java
index 8f6b3ff..60d349d 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicNameUtilsTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicNameUtilsTest.java
@@ -24,9 +24,7 @@ import java.util.Arrays;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for {@link TopicNameUtils}. */
class TopicNameUtilsTest {
@@ -40,38 +38,37 @@ class TopicNameUtilsTest {
@Test
void topicNameWouldReturnACleanTopicNameWithTenant() {
String name1 = TopicNameUtils.topicName(fullTopicName + "-partition-1");
- assertEquals(name1, fullTopicName);
+ assertThat(name1).isEqualTo(fullTopicName);
String name2 = TopicNameUtils.topicName(topicNameWithLocal);
- assertEquals(name2, topicNameWithLocal);
+ assertThat(name2).isEqualTo(topicNameWithLocal);
String name3 = TopicNameUtils.topicName(shortTopicName + "-partition-1");
- assertEquals(name3, "persistent://public/default/short-topic");
+ assertThat(name3).isEqualTo("persistent://public/default/short-topic");
String name4 = TopicNameUtils.topicName(shortTopicName);
- assertEquals(name4, "persistent://public/default/short-topic");
+ assertThat(name4).isEqualTo("persistent://public/default/short-topic");
String name5 = TopicNameUtils.topicName(topicNameWithoutCluster + "-partition-1");
- assertEquals(name5, topicNameWithoutCluster);
+ assertThat(name5).isEqualTo(topicNameWithoutCluster);
}
@Test
void topicNameWithPartitionInfo() {
- assertThrows(
- IllegalArgumentException.class,
- () -> TopicNameUtils.topicNameWithPartition(shortTopicName, -3));
+ assertThatThrownBy(() -> TopicNameUtils.topicNameWithPartition(shortTopicName, -3))
+ .isInstanceOf(IllegalArgumentException.class);
String name1 = TopicNameUtils.topicNameWithPartition(fullTopicName, 4);
- assertEquals(name1, fullTopicName + "-partition-4");
+ assertThat(name1).isEqualTo(fullTopicName + "-partition-4");
String name2 = TopicNameUtils.topicNameWithPartition(topicNameWithLocal, 3);
- assertEquals(name2, topicNameWithLocal + "-partition-3");
+ assertThat(name2).isEqualTo(topicNameWithLocal + "-partition-3");
String name3 = TopicNameUtils.topicNameWithPartition(shortTopicName, 5);
- assertEquals(name3, "persistent://public/default/short-topic-partition-5");
+ assertThat(name3).isEqualTo("persistent://public/default/short-topic-partition-5");
String name4 = TopicNameUtils.topicNameWithPartition(topicNameWithoutCluster, 8);
- assertEquals(name4, topicNameWithoutCluster + "-partition-8");
+ assertThat(name4).isEqualTo(topicNameWithoutCluster + "-partition-8");
}
@Test
@@ -91,6 +88,6 @@ class TopicNameUtilsTest {
boolean internal =
TopicNameUtils.isInternal(
"persistent://public/default/topic__transaction_pending_ack");
- assertTrue(internal);
+ assertThat(internal).isTrue();
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicPartitionTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicPartitionTest.java
index 7e5f171..f431e8e 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicPartitionTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicPartitionTest.java
@@ -20,7 +20,7 @@ package org.apache.flink.connector.pulsar.source.enumerator.topic;
import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Unit tests for {@link TopicPartition}. */
class TopicPartitionTest {
@@ -29,11 +29,12 @@ class TopicPartitionTest {
void topicNameForPartitionedAndNonPartitionedTopic() {
// For partitioned topic
TopicPartition partition = new TopicPartition("test-name", 12);
- assertEquals(
- partition.getFullTopicName(), "persistent://public/default/test-name-partition-12");
+ assertThat(partition.getFullTopicName())
+ .isEqualTo("persistent://public/default/test-name-partition-12");
// For non-partitioned topic
TopicPartition partition1 = new TopicPartition("test-topic", -1);
- assertEquals(partition1.getFullTopicName(), "persistent://public/default/test-topic");
+ assertThat(partition1.getFullTopicName())
+ .isEqualTo("persistent://public/default/test-topic");
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicRangeTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicRangeTest.java
index f5665e7..864b481 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicRangeTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicRangeTest.java
@@ -23,9 +23,9 @@ import org.apache.flink.util.InstantiationUtil;
import org.junit.jupiter.api.Test;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange.MAX_RANGE;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for {@link TopicRange}. */
class TopicRangeTest {
@@ -34,26 +34,28 @@ class TopicRangeTest {
void topicRangeIsSerializable() throws Exception {
final TopicRange range = new TopicRange(1, 5);
final TopicRange cloneRange = InstantiationUtil.clone(range);
- assertEquals(range, cloneRange);
+ assertThat(cloneRange).isEqualTo(range);
}
@Test
void negativeStart() {
- assertThrows(IllegalArgumentException.class, () -> new TopicRange(-1, 1));
+ assertThatThrownBy(() -> new TopicRange(-1, 1))
+ .isInstanceOf(IllegalArgumentException.class);
}
@Test
void endBelowTheMaximum() {
- assertDoesNotThrow(() -> new TopicRange(1, MAX_RANGE - 1));
+ assertThatCode(() -> new TopicRange(1, MAX_RANGE - 1)).doesNotThrowAnyException();
}
@Test
void endOnTheMaximum() {
- assertDoesNotThrow(() -> new TopicRange(1, MAX_RANGE));
+ assertThatCode(() -> new TopicRange(1, MAX_RANGE)).doesNotThrowAnyException();
}
@Test
void endAboveTheMaximum() {
- assertThrows(IllegalArgumentException.class, () -> new TopicRange(1, MAX_RANGE + 1));
+ assertThatThrownBy(() -> new TopicRange(1, MAX_RANGE + 1))
+ .isInstanceOf(IllegalArgumentException.class);
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/range/TopicRangeUtilsTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/range/TopicRangeUtilsTest.java
index fd53770..1451bda 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/range/TopicRangeUtilsTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/topic/range/TopicRangeUtilsTest.java
@@ -28,10 +28,9 @@ import java.util.List;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.range.TopicRangeUtils.isFullTopicRanges;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.range.TopicRangeUtils.validateTopicRanges;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test class for {@link TopicRangeUtils}. */
class TopicRangeUtilsTest {
@@ -39,16 +38,19 @@ class TopicRangeUtilsTest {
@Test
void testValidateTopicRanges() {
List<TopicRange> ranges1 = Arrays.asList(new TopicRange(1, 2), new TopicRange(2, 3));
- assertThrows(IllegalArgumentException.class, () -> validateTopicRanges(ranges1));
+ assertThatThrownBy(() -> validateTopicRanges(ranges1))
+ .isInstanceOf(IllegalArgumentException.class);
List<TopicRange> ranges2 = Arrays.asList(new TopicRange(1, 14), new TopicRange(2, 5));
- assertThrows(IllegalArgumentException.class, () -> validateTopicRanges(ranges2));
+ assertThatThrownBy(() -> validateTopicRanges(ranges2))
+ .isInstanceOf(IllegalArgumentException.class);
List<TopicRange> ranges3 = Arrays.asList(new TopicRange(1, 14), new TopicRange(5, 30));
- assertThrows(IllegalArgumentException.class, () -> validateTopicRanges(ranges3));
+ assertThatThrownBy(() -> validateTopicRanges(ranges3))
+ .isInstanceOf(IllegalArgumentException.class);
List<TopicRange> ranges4 = Arrays.asList(new TopicRange(1, 14), new TopicRange(15, 30));
- assertDoesNotThrow(() -> validateTopicRanges(ranges4));
+ assertThatCode(() -> validateTopicRanges(ranges4)).doesNotThrowAnyException();
}
@Test
@@ -59,7 +61,7 @@ class TopicRangeUtilsTest {
new TopicRange(0, 16383),
new TopicRange(32768, 49151),
new TopicRange(49152, 65535));
- assertTrue(isFullTopicRanges(ranges1));
+ assertThat(isFullTopicRanges(ranges1)).isTrue();
List<TopicRange> ranges2 =
Arrays.asList(
@@ -67,7 +69,7 @@ class TopicRangeUtilsTest {
new TopicRange(0, 16383),
new TopicRange(16384, 32767),
new TopicRange(49152, 65531));
- assertFalse(isFullTopicRanges(ranges2));
+ assertThat(isFullTopicRanges(ranges2)).isFalse();
List<TopicRange> ranges3 =
Arrays.asList(
@@ -75,9 +77,9 @@ class TopicRangeUtilsTest {
new TopicRange(32768, 49151),
new TopicRange(16384, 32767),
new TopicRange(49152, 65535));
- assertFalse(isFullTopicRanges(ranges3));
+ assertThat(isFullTopicRanges(ranges3)).isFalse();
List<TopicRange> ranges4 = Collections.singletonList(TopicRange.createFullRange());
- assertTrue(isFullTopicRanges(ranges4));
+ assertThat(isFullTopicRanges(ranges4)).isTrue();
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderTest.java
index 3ad03b8..093423f 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderTest.java
@@ -73,8 +73,6 @@ import static org.apache.flink.shaded.guava30.com.google.common.util.concurrent.
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.fail;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.mockito.Mockito.mock;
/** Unit test for {@link PulsarSourceReader}. */
class PulsarSourceReaderTest extends PulsarTestSuiteBase {
@@ -204,7 +202,7 @@ class PulsarSourceReaderTest extends PulsarTestSuiteBase {
reader.pauseOrResumeSplits(singletonList(split.splitId()), emptyList());
InputStatus status = reader.pollNext(output);
- assertEquals(InputStatus.NOTHING_AVAILABLE, status);
+ assertThat(status).isEqualTo(InputStatus.NOTHING_AVAILABLE);
reader.pauseOrResumeSplits(emptyList(), Collections.singleton(split.splitId()));
@@ -213,7 +211,7 @@ class PulsarSourceReaderTest extends PulsarTestSuiteBase {
Thread.sleep(5);
} while (status != InputStatus.MORE_AVAILABLE);
- assertEquals(InputStatus.MORE_AVAILABLE, status);
+ assertThat(status).isEqualTo(InputStatus.MORE_AVAILABLE);
reader.close();
}
@@ -241,7 +239,7 @@ class PulsarSourceReaderTest extends PulsarTestSuiteBase {
deserializationSchema.open(
new PulsarDeserializationSchemaInitializationContext(
context, operator().client()),
- mock(SourceConfiguration.class));
+ new SourceConfiguration(new Configuration()));
} catch (Exception e) {
fail("Error while opening deserializationSchema");
}
@@ -294,9 +292,8 @@ class PulsarSourceReaderTest extends PulsarTestSuiteBase {
.subscriptionName("verify-message")
.topic(partitionName)
.subscribe()) {
- assertEquals(
- expectedMessages - 1,
- ((MessageIdImpl) consumer.getLastMessageId()).getEntryId());
+ assertThat(((MessageIdImpl) consumer.getLastMessageId()).getEntryId())
+ .isEqualTo(expectedMessages - 1);
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/deserializer/GenericRecordDeserializationSchemaTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/deserializer/GenericRecordDeserializationSchemaTest.java
index ca2bdb5..5c80d1c 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/deserializer/GenericRecordDeserializationSchemaTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/deserializer/GenericRecordDeserializationSchemaTest.java
@@ -49,7 +49,6 @@ import static java.util.stream.Collectors.toList;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
import static org.apache.flink.api.common.eventtime.WatermarkStrategy.noWatermarks;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertEquals;
/** Unit test of {@link GenericRecordDeserializationSchema}. */
class GenericRecordDeserializationSchemaTest extends PulsarTestSuiteBase {
@@ -125,7 +124,7 @@ class GenericRecordDeserializationSchemaTest extends PulsarTestSuiteBase {
@Override
public String deserialize(GenericRecord message) {
- assertEquals(SchemaType.AVRO, message.getSchemaType());
+ assertThat(message.getSchemaType()).isEqualTo(SchemaType.AVRO);
Object object = message.getNativeObject();
assertThat(object)
.isInstanceOf(
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarDeserializationSchemaTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarDeserializationSchemaTest.java
index d20272d..339ec99 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarDeserializationSchemaTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarDeserializationSchemaTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.connector.pulsar.source.reader.deserializer;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.pulsar.SampleMessage.TestMessage;
import org.apache.flink.connector.pulsar.source.PulsarSource;
import org.apache.flink.connector.pulsar.source.PulsarSourceOptions;
@@ -56,29 +57,26 @@ import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
import static org.apache.flink.util.Preconditions.checkState;
import static org.apache.pulsar.client.api.Schema.PROTOBUF_NATIVE;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.AssertionsForClassTypes.assertThatCode;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.mockito.Mockito.mock;
+import static org.assertj.core.api.Assertions.assertThatCode;
/** Unit tests for {@link PulsarDeserializationSchema}. */
class PulsarDeserializationSchemaTest extends PulsarTestSuiteBase {
+ private final SourceConfiguration sourceConfig = new SourceConfiguration(new Configuration());
+
@Test
void createFromFlinkDeserializationSchema() throws Exception {
PulsarDeserializationSchema<String> schema =
new PulsarDeserializationSchemaWrapper<>(new SimpleStringSchema());
- schema.open(new PulsarTestingDeserializationContext(), mock(SourceConfiguration.class));
- assertDoesNotThrow(() -> InstantiationUtil.clone(schema));
+ schema.open(new PulsarTestingDeserializationContext(), sourceConfig);
+ assertThatCode(() -> InstantiationUtil.clone(schema)).doesNotThrowAnyException();
String content = "some-sample-message-" + randomAlphabetic(10);
Message<byte[]> message = getMessage(content, String::getBytes);
SingleMessageCollector<String> collector = new SingleMessageCollector<>();
schema.deserialize(message, collector);
- assertNotNull(collector.result);
- assertEquals(content, collector.result);
+ assertThat(collector.result).isNotNull().isEqualTo(content);
}
@Test
@@ -86,8 +84,8 @@ class PulsarDeserializationSchemaTest extends PulsarTestSuiteBase {
Schema<TestMessage> schema1 = PROTOBUF_NATIVE(TestMessage.class);
PulsarDeserializationSchema<TestMessage> schema2 =
new PulsarSchemaWrapper<>(schema1, TestMessage.class);
- schema2.open(new PulsarTestingDeserializationContext(), mock(SourceConfiguration.class));
- assertDoesNotThrow(() -> InstantiationUtil.clone(schema2));
+ schema2.open(new PulsarTestingDeserializationContext(), sourceConfig);
+ assertThatCode(() -> InstantiationUtil.clone(schema2)).doesNotThrowAnyException();
TestMessage message1 =
TestMessage.newBuilder()
@@ -99,16 +97,15 @@ class PulsarDeserializationSchemaTest extends PulsarTestSuiteBase {
SingleMessageCollector<TestMessage> collector = new SingleMessageCollector<>();
schema2.deserialize(message2, collector);
- assertNotNull(collector.result);
- assertEquals(collector.result, message1);
+ assertThat(collector.result).isNotNull().isEqualTo(message1);
}
@Test
void createFromFlinkTypeInformation() throws Exception {
PulsarDeserializationSchema<String> schema =
new PulsarTypeInformationWrapper<>(Types.STRING, null);
- schema.open(new PulsarTestingDeserializationContext(), mock(SourceConfiguration.class));
- assertDoesNotThrow(() -> InstantiationUtil.clone(schema));
+ schema.open(new PulsarTestingDeserializationContext(), sourceConfig);
+ assertThatCode(() -> InstantiationUtil.clone(schema)).doesNotThrowAnyException();
String content = "test-content-" + randomAlphanumeric(10);
Message<byte[]> message =
@@ -122,8 +119,7 @@ class PulsarDeserializationSchemaTest extends PulsarTestSuiteBase {
SingleMessageCollector<String> collector = new SingleMessageCollector<>();
schema.deserialize(message, collector);
- assertNotNull(collector.result);
- assertEquals(content, collector.result);
+ assertThat(collector.result).isNotNull().isEqualTo(content);
}
@Test
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/split/PulsarPartitionSplitSerializerTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/split/PulsarPartitionSplitSerializerTest.java
index acf7462..3340f49 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/split/PulsarPartitionSplitSerializerTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/split/PulsarPartitionSplitSerializerTest.java
@@ -31,8 +31,7 @@ import org.junit.jupiter.api.Test;
import static java.util.Collections.singletonList;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitSerializer.INSTANCE;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.assertj.core.api.Assertions.assertThat;
/** Unit tests for {@link PulsarPartitionSplitSerializer}. */
class PulsarPartitionSplitSerializerTest {
@@ -48,8 +47,7 @@ class PulsarPartitionSplitSerializerTest {
byte[] bytes = INSTANCE.serialize(split);
PulsarPartitionSplit split1 = INSTANCE.deserialize(2, bytes);
- assertEquals(split, split1);
- assertNotSame(split, split1);
+ assertThat(split1).isEqualTo(split).isNotSameAs(split);
}
@Test
@@ -86,8 +84,7 @@ class PulsarPartitionSplitSerializerTest {
MessageId.latest,
new TxnID(1000, 2000));
- assertEquals(split, expectedSplit);
- assertNotSame(split, expectedSplit);
+ assertThat(split).isEqualTo(expectedSplit).isNotSameAs(expectedSplit);
}
@Test
@@ -118,7 +115,6 @@ class PulsarPartitionSplitSerializerTest {
MessageId.latest,
new TxnID(1000, 2000));
- assertEquals(split, expectedSplit);
- assertNotSame(split, expectedSplit);
+ assertThat(split).isEqualTo(expectedSplit).isNotSameAs(expectedSplit);
}
}