Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use test objects from spring-pulsar-test (#676)

Browse Source 
Removes duplicated test objects `User/User2` and replaces 
with `UserPojo/UserRecord` from `spring-pulsar-test`.
This commit is contained in:
Chris Bono
2024-05-11 11:16:09 -05:00
committed by GitHub
parent caea61801c
commit f5361f0916
2 changed files with 63 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files

180
spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java
View File

@@ -23,7 +23,6 @@ import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
@@ -70,12 +69,13 @@ import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCu
import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory;
import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.BasicListenersTestCases.BasicListenersTestCasesConfig;
import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.PulsarHeadersTest.PulsarListenerWithHeadersConfig;
import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SchemaCustomMappingsTestCases.SchemaCustomMappingsTestConfig.User2;
import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.StreamingListenerTestCases.StreamingListenerTestCasesConfig;
import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SubscriptionTypeTests.WithDefaultType.WithDefaultTypeConfig;
import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SubscriptionTypeTests.WithSpecificTypes.WithSpecificTypesConfig;
import org.springframework.pulsar.reactive.support.MessageUtils;
import org.springframework.pulsar.support.PulsarHeaders;
import org.springframework.pulsar.test.support.model.UserPojo;
import org.springframework.pulsar.test.support.model.UserRecord;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.util.ReflectionTestUtils;
import org.springframework.util.ObjectUtils;
@@ -283,20 +283,20 @@ class ReactivePulsarListenerTests extends ReactivePulsarListenerTestsBase {
@Test
void jsonSchema() throws Exception {
PulsarProducerFactory<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
PulsarTemplate<User> template = new PulsarTemplate<>(pulsarProducerFactory);
PulsarProducerFactory<UserPojo> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
PulsarTemplate<UserPojo> template = new PulsarTemplate<>(pulsarProducerFactory);
for (int i = 0; i < 3; i++) {
template.send("json-topic", new User("Jason", i), JSONSchema.of(User.class));
template.send("json-topic", new UserPojo("Jason", i), JSONSchema.of(UserPojo.class));
}
assertThat(jsonLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void avroSchema() throws Exception {
PulsarProducerFactory<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
PulsarTemplate<User> template = new PulsarTemplate<>(pulsarProducerFactory);
PulsarProducerFactory<UserPojo> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
PulsarTemplate<UserPojo> template = new PulsarTemplate<>(pulsarProducerFactory);
for (int i = 0; i < 3; i++) {
template.send("avro-topic", new User("Avi", i), AvroSchema.of(User.class));
template.send("avro-topic", new UserPojo("Avi", i), AvroSchema.of(UserPojo.class));
}
assertThat(avroLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@@ -332,14 +332,14 @@ class ReactivePulsarListenerTests extends ReactivePulsarListenerTestsBase {
@ReactivePulsarListener(id = "jsonListener", topics = "json-topic", schemaType = SchemaType.JSON,
consumerCustomizer = "subscriptionInitialPositionEarliest")
Mono<Void> listenJson(User ignored) {
Mono<Void> listenJson(UserPojo ignored) {
jsonLatch.countDown();
return Mono.empty();
}
@ReactivePulsarListener(id = "avroListener", topics = "avro-topic", schemaType = SchemaType.AVRO,
consumerCustomizer = "subscriptionInitialPositionEarliest")
Mono<Void> listenAvro(User ignored) {
Mono<Void> listenAvro(UserPojo ignored) {
avroLatch.countDown();
return Mono.empty();
}
@@ -360,61 +360,6 @@ class ReactivePulsarListenerTests extends ReactivePulsarListenerTestsBase {
}
static class User {
private String name;
private int age;
User() {
}
User(String name, int age) {
this.name = name;
this.age = age;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
User user = (User) o;
return age == user.age && Objects.equals(name, user.name);
}
@Override
public int hashCode() {
return Objects.hash(name, age);
}
@Override
public String toString() {
return "User{" + "name='" + name + '\'' + ", age=" + age + '}';
}
}
}
@Nested
@@ -428,33 +373,33 @@ class ReactivePulsarListenerTests extends ReactivePulsarListenerTestsBase {
@Test
void jsonSchema() throws Exception {
PulsarProducerFactory<User2> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
PulsarTemplate<User2> template = new PulsarTemplate<>(pulsarProducerFactory);
PulsarProducerFactory<UserRecord> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
PulsarTemplate<UserRecord> template = new PulsarTemplate<>(pulsarProducerFactory);
for (int i = 0; i < 3; i++) {
template.send("json-custom-schema-topic", new User2("Jason", i), JSONSchema.of(User2.class));
template.send("json-custom-schema-topic", new UserRecord("Jason", i), JSONSchema.of(UserRecord.class));
}
assertThat(jsonLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void avroSchema() throws Exception {
PulsarProducerFactory<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
PulsarTemplate<User> template = new PulsarTemplate<>(pulsarProducerFactory);
PulsarProducerFactory<UserPojo> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
PulsarTemplate<UserPojo> template = new PulsarTemplate<>(pulsarProducerFactory);
for (int i = 0; i < 3; i++) {
template.send("avro-custom-schema-topic", new User("Avi", i), AvroSchema.of(User.class));
template.send("avro-custom-schema-topic", new UserPojo("Avi", i), AvroSchema.of(UserPojo.class));
}
assertThat(avroLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void keyvalueSchema() throws Exception {
PulsarProducerFactory<KeyValue<String, User2>> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
PulsarProducerFactory<KeyValue<String, UserRecord>> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
pulsarClient);
PulsarTemplate<KeyValue<String, User2>> template = new PulsarTemplate<>(pulsarProducerFactory);
Schema<KeyValue<String, User2>> kvSchema = Schema.KeyValue(Schema.STRING, Schema.JSON(User2.class),
KeyValueEncodingType.INLINE);
PulsarTemplate<KeyValue<String, UserRecord>> template = new PulsarTemplate<>(pulsarProducerFactory);
Schema<KeyValue<String, UserRecord>> kvSchema = Schema.KeyValue(Schema.STRING,
Schema.JSON(UserRecord.class), KeyValueEncodingType.INLINE);
for (int i = 0; i < 3; i++) {
template.send("keyvalue-custom-schema-topic", new KeyValue<>("Kevin", new User2("Kevin", 5150)),
template.send("keyvalue-custom-schema-topic", new KeyValue<>("Kevin", new UserRecord("Kevin", 5150)),
kvSchema);
}
assertThat(keyvalueLatch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -480,8 +425,8 @@ class ReactivePulsarListenerTests extends ReactivePulsarListenerTestsBase {
@Bean
SchemaResolver customSchemaResolver() {
DefaultSchemaResolver resolver = new DefaultSchemaResolver();
resolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class));
resolver.addCustomSchemaMapping(User2.class, Schema.JSON(User2.class));
resolver.addCustomSchemaMapping(UserPojo.class, Schema.AVRO(UserPojo.class));
resolver.addCustomSchemaMapping(UserRecord.class, Schema.JSON(UserRecord.class));
resolver.addCustomSchemaMapping(Proto.Person.class, Schema.PROTOBUF(Proto.Person.class));
return resolver;
}
@@ -496,21 +441,21 @@ class ReactivePulsarListenerTests extends ReactivePulsarListenerTestsBase {
@ReactivePulsarListener(id = "jsonListener", topics = "json-custom-schema-topic",
consumerCustomizer = "subscriptionInitialPositionEarliest")
Mono<Void> listenJson(User2 ignored) {
Mono<Void> listenJson(UserRecord ignored) {
jsonLatch.countDown();
return Mono.empty();
}
@ReactivePulsarListener(id = "avroListener", topics = "avro-custom-schema-topic",
consumerCustomizer = "subscriptionInitialPositionEarliest")
Mono<Void> listenAvro(User ignored) {
Mono<Void> listenAvro(UserPojo ignored) {
avroLatch.countDown();
return Mono.empty();
}
@ReactivePulsarListener(id = "keyvalueListener", topics = "keyvalue-custom-schema-topic",
consumerCustomizer = "subscriptionInitialPositionEarliest")
Mono<Void> listenKeyvalue(KeyValue<String, User2> ignored) {
Mono<Void> listenKeyvalue(KeyValue<String, UserRecord> ignored) {
keyvalueLatch.countDown();
return Mono.empty();
}
@@ -522,67 +467,6 @@ class ReactivePulsarListenerTests extends ReactivePulsarListenerTestsBase {
return Mono.empty();
}
record User2(String name, int age) {
}
}
/**
* Do not convert this to a Record as Avro does not seem to work well w/ records.
*/
static class User {
private String name;
private int age;
User() {
}
User(String name, int age) {
this.name = name;
this.age = age;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
User user = (User) o;
return age == user.age && Objects.equals(name, user.name);
}
@Override
public int hashCode() {
return Objects.hash(name, age);
}
@Override
public String toString() {
return "User{" + "name='" + name + '\'' + ", age=" + age + '}';
}
}
}
@@ -596,11 +480,11 @@ class ReactivePulsarListenerTests extends ReactivePulsarListenerTestsBase {
@Test
void complexMessageTypeTopicMapping() throws Exception {
PulsarProducerFactory<User2> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
PulsarTemplate<User2> template = new PulsarTemplate<>(pulsarProducerFactory);
Schema<User2> schema = Schema.JSON(User2.class);
PulsarProducerFactory<UserRecord> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
PulsarTemplate<UserRecord> template = new PulsarTemplate<>(pulsarProducerFactory);
Schema<UserRecord> schema = Schema.JSON(UserRecord.class);
for (int i = 0; i < 3; i++) {
template.send("rplt-topicMapping-user-topic", new User2("Jason", i), schema);
template.send("rplt-topicMapping-user-topic", new UserRecord("Jason", i), schema);
}
assertThat(userLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@@ -622,7 +506,7 @@ class ReactivePulsarListenerTests extends ReactivePulsarListenerTestsBase {
@Bean
TopicResolver topicResolver() {
DefaultTopicResolver resolver = new DefaultTopicResolver();
resolver.addCustomTopicMapping(User2.class, "rplt-topicMapping-user-topic");
resolver.addCustomTopicMapping(UserRecord.class, "rplt-topicMapping-user-topic");
resolver.addCustomTopicMapping(String.class, "rplt-topicMapping-string-topic");
return resolver;
}
@@ -637,7 +521,7 @@ class ReactivePulsarListenerTests extends ReactivePulsarListenerTestsBase {
@ReactivePulsarListener(id = "userListener", schemaType = SchemaType.JSON,
consumerCustomizer = "subscriptionInitialPositionEarliest")
Mono<Void> listenUser(User2 ignored) {
Mono<Void> listenUser(UserRecord ignored) {
userLatch.countDown();
return Mono.empty();
}

121
spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java
View File

@@ -23,7 +23,6 @@ import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -73,6 +72,8 @@ import org.springframework.pulsar.core.TopicResolver;
import org.springframework.pulsar.listener.PulsarListenerTests.SubscriptionTypeTests.WithDefaultType.WithDefaultTypeConfig;
import org.springframework.pulsar.listener.PulsarListenerTests.SubscriptionTypeTests.WithSpecificTypes.WithSpecificTypesConfig;
import org.springframework.pulsar.support.PulsarHeaders;
import org.springframework.pulsar.test.support.model.UserPojo;
import org.springframework.pulsar.test.support.model.UserRecord;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.util.backoff.FixedBackOff;
@@ -400,11 +401,11 @@ class PulsarListenerTests extends PulsarListenerTestsBase {
@Test
void jsonSchema() throws Exception {
var pulsarProducerFactory = new DefaultPulsarProducerFactory<User>(pulsarClient);
var pulsarProducerFactory = new DefaultPulsarProducerFactory<UserPojo>(pulsarClient);
var template = new PulsarTemplate<>(pulsarProducerFactory);
var schema = JSONSchema.of(User.class);
var schema = JSONSchema.of(UserPojo.class);
for (int i = 0; i < 3; i++) {
template.send("json-topic", new User("Jason", i), schema);
template.send("json-topic", new UserPojo("Jason", i), schema);
}
assertThat(jsonLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(jsonBatchLatch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -412,11 +413,11 @@ class PulsarListenerTests extends PulsarListenerTestsBase {
@Test
void avroSchema() throws Exception {
var pulsarProducerFactory = new DefaultPulsarProducerFactory<User>(pulsarClient);
var pulsarProducerFactory = new DefaultPulsarProducerFactory<UserPojo>(pulsarClient);
var template = new PulsarTemplate<>(pulsarProducerFactory);
var schema = AvroSchema.of(User.class);
var schema = AvroSchema.of(UserPojo.class);
for (int i = 0; i < 3; i++) {
template.send("avro-topic", new User("Avi", i), schema);
template.send("avro-topic", new UserPojo("Avi", i), schema);
}
assertThat(avroLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(avroBatchLatch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -452,25 +453,25 @@ class PulsarListenerTests extends PulsarListenerTestsBase {
@PulsarListener(id = "jsonListener", topics = "json-topic", subscriptionName = "subscription-4",
schemaType = SchemaType.JSON, properties = { "subscriptionInitialPosition=Earliest" })
void listenJson(User ignored) {
void listenJson(UserPojo ignored) {
jsonLatch.countDown();
}
@PulsarListener(id = "jsonBatchListener", topics = "json-topic", subscriptionName = "subscription-5",
schemaType = SchemaType.JSON, batch = true, properties = { "subscriptionInitialPosition=Earliest" })
void listenJsonBatch(List<User> messages) {
void listenJsonBatch(List<UserPojo> messages) {
messages.forEach(m -> jsonBatchLatch.countDown());
}
@PulsarListener(id = "avroListener", topics = "avro-topic", subscriptionName = "subscription-6",
schemaType = SchemaType.AVRO, properties = { "subscriptionInitialPosition=Earliest" })
void listenAvro(User ignored) {
void listenAvro(UserPojo ignored) {
avroLatch.countDown();
}
@PulsarListener(id = "avroBatchListener", topics = "avro-topic", subscriptionName = "subscription-7",
schemaType = SchemaType.AVRO, batch = true, properties = { "subscriptionInitialPosition=Earliest" })
void listenAvroBatch(Messages<User> messages) {
void listenAvroBatch(Messages<UserPojo> messages) {
messages.forEach(m -> avroBatchLatch.countDown());
}
@@ -504,66 +505,6 @@ class PulsarListenerTests extends PulsarListenerTestsBase {
}
/**
* Do not convert this to a Record as Avro does not seem to work well w/ records.
*/
static class User {
private String name;
private int age;
User() {
}
User(String name, int age) {
this.name = name;
this.age = age;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
User user = (User) o;
return age == user.age && Objects.equals(name, user.name);
}
@Override
public int hashCode() {
return Objects.hash(name, age);
}
@Override
public String toString() {
return "User{" + "name='" + name + '\'' + ", age=" + age + '}';
}
}
record User2(String name, int age) {
}
@Nested
@ContextConfiguration(classes = SchemaCustomMappingsTestCases.SchemaCustomMappingsTestConfig.class)
class SchemaCustomMappingsTestCases {
@@ -575,33 +516,33 @@ class PulsarListenerTests extends PulsarListenerTestsBase {
@Test
void jsonSchema() throws Exception {
var pulsarProducerFactory = new DefaultPulsarProducerFactory<User2>(pulsarClient);
var pulsarProducerFactory = new DefaultPulsarProducerFactory<UserRecord>(pulsarClient);
var template = new PulsarTemplate<>(pulsarProducerFactory);
var schema = Schema.JSON(User2.class);
var schema = Schema.JSON(UserRecord.class);
for (int i = 0; i < 3; i++) {
template.send("json-custom-mappings-topic", new User2("Jason", i), schema);
template.send("json-custom-mappings-topic", new UserRecord("Jason", i), schema);
}
assertThat(jsonLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void avroSchema() throws Exception {
var pulsarProducerFactory = new DefaultPulsarProducerFactory<User>(pulsarClient);
var pulsarProducerFactory = new DefaultPulsarProducerFactory<UserPojo>(pulsarClient);
var template = new PulsarTemplate<>(pulsarProducerFactory);
var schema = AvroSchema.of(User.class);
var schema = AvroSchema.of(UserPojo.class);
for (int i = 0; i < 3; i++) {
template.send("avro-custom-mappings-topic", new User("Avi", i), schema);
template.send("avro-custom-mappings-topic", new UserPojo("Avi", i), schema);
}
assertThat(avroLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void keyvalueSchema() throws Exception {
var pulsarProducerFactory = new DefaultPulsarProducerFactory<KeyValue<String, User2>>(pulsarClient);
var pulsarProducerFactory = new DefaultPulsarProducerFactory<KeyValue<String, UserRecord>>(pulsarClient);
var template = new PulsarTemplate<>(pulsarProducerFactory);
var kvSchema = Schema.KeyValue(Schema.STRING, Schema.JSON(User2.class), KeyValueEncodingType.INLINE);
var kvSchema = Schema.KeyValue(Schema.STRING, Schema.JSON(UserRecord.class), KeyValueEncodingType.INLINE);
for (int i = 0; i < 3; i++) {
template.send("keyvalue-custom-mappings-topic", new KeyValue<>("Kevin", new User2("Kevin", 5150)),
template.send("keyvalue-custom-mappings-topic", new KeyValue<>("Kevin", new UserRecord("Kevin", 5150)),
kvSchema);
}
assertThat(keyvalueLatch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -626,8 +567,8 @@ class PulsarListenerTests extends PulsarListenerTestsBase {
@Bean
SchemaResolver customSchemaResolver() {
DefaultSchemaResolver resolver = new DefaultSchemaResolver();
resolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class));
resolver.addCustomSchemaMapping(User2.class, Schema.JSON(User2.class));
resolver.addCustomSchemaMapping(UserPojo.class, Schema.AVRO(UserPojo.class));
resolver.addCustomSchemaMapping(UserRecord.class, Schema.JSON(UserRecord.class));
resolver.addCustomSchemaMapping(Proto.Person.class, Schema.PROTOBUF(Proto.Person.class));
return resolver;
}
@@ -644,19 +585,19 @@ class PulsarListenerTests extends PulsarListenerTestsBase {
@PulsarListener(id = "jsonListener", topics = "json-custom-mappings-topic",
subscriptionName = "subscription-4", properties = { "subscriptionInitialPosition=Earliest" })
void listenJson(User2 ignored) {
void listenJson(UserRecord ignored) {
jsonLatch.countDown();
}
@PulsarListener(id = "avroListener", topics = "avro-custom-mappings-topic",
subscriptionName = "subscription-6", properties = { "subscriptionInitialPosition=Earliest" })
void listenAvro(User ignored) {
void listenAvro(UserPojo ignored) {
avroLatch.countDown();
}
@PulsarListener(id = "keyvalueListener", topics = "keyvalue-custom-mappings-topic",
subscriptionName = "subscription-8", properties = { "subscriptionInitialPosition=Earliest" })
void listenKeyvalue(KeyValue<String, User2> ignored) {
void listenKeyvalue(KeyValue<String, UserRecord> ignored) {
keyvalueLatch.countDown();
}
@@ -679,11 +620,11 @@ class PulsarListenerTests extends PulsarListenerTestsBase {
@Test
void complexMessageTypeTopicMapping() throws Exception {
var pulsarProducerFactory = new DefaultPulsarProducerFactory<User2>(pulsarClient);
var pulsarProducerFactory = new DefaultPulsarProducerFactory<UserRecord>(pulsarClient);
var template = new PulsarTemplate<>(pulsarProducerFactory);
var schema = Schema.JSON(User2.class);
var schema = Schema.JSON(UserRecord.class);
for (int i = 0; i < 3; i++) {
template.send("plt-topicMapping-user-topic", new User2("Jason", i), schema);
template.send("plt-topicMapping-user-topic", new UserRecord("Jason", i), schema);
}
assertThat(userLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@@ -705,7 +646,7 @@ class PulsarListenerTests extends PulsarListenerTestsBase {
@Bean
TopicResolver topicResolver() {
DefaultTopicResolver resolver = new DefaultTopicResolver();
resolver.addCustomTopicMapping(User2.class, "plt-topicMapping-user-topic");
resolver.addCustomTopicMapping(UserRecord.class, "plt-topicMapping-user-topic");
resolver.addCustomTopicMapping(String.class, "plt-topicMapping-string-topic");
return resolver;
}
@@ -722,7 +663,7 @@ class PulsarListenerTests extends PulsarListenerTestsBase {
@PulsarListener(id = "userListener", schemaType = SchemaType.JSON, subscriptionName = "sub1",
properties = { "subscriptionInitialPosition=Earliest" })
void listenUser(User2 ignored) {
void listenUser(UserRecord ignored) {
userLatch.countDown();
}