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

Simplify defaults type mappings (#377)

Browse Source 
* Remove redundant SchemaInfo.messageType as
  the key of the type mappings tells us that
  info
This commit is contained in:
Chris Bono
2023-03-21 20:19:01 -05:00
committed by GitHub
parent 0f84199138
commit 19cc0ab070
6 changed files with 26 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-pulsar-docs/src/main/asciidoc/schema-info/custom-schema-mapping.adoc
View File

@@ -10,11 +10,9 @@ spring:
- message-type: com.acme.User
schema-info:
schema-type: AVRO
message-type: com.acme.User
- message-type: com.acme.Address
schema-info:
schema-type: JSON
message-type: com.acme.Address
----
NOTE: The `message-type` is the fully-qualified name of the message class.

4
spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java
View File

@@ -117,8 +117,8 @@ public class PulsarAutoConfiguration {
if (pulsarProperties.getDefaults().getTypeMappings() != null) {
pulsarProperties.getDefaults().getTypeMappings().stream().filter((tm) -> tm.schemaInfo() != null)
.forEach((tm) -> {
var schema = schemaResolver.resolveSchema(tm.schemaInfo().schemaType(),
tm.schemaInfo().messageType(), tm.schemaInfo().messageKeyType()).orElseThrow();
var schema = schemaResolver.resolveSchema(tm.schemaInfo().schemaType(), tm.messageType(),
tm.schemaInfo().messageKeyType()).orElseThrow();
schemaResolver.addCustomSchemaMapping(tm.messageType(), schema);
});
}

7
spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java
View File

@@ -1365,16 +1365,17 @@ public class PulsarProperties {
* Represents a schema - holds enough information to construct an actual schema
* instance.
* @param schemaType schema type
* @param messageType message type (not required for primitive schema types or key
* value type)
* @param messageKeyType message key type (required for key value type)
*/
public record SchemaInfo(SchemaType schemaType, @Nullable Class<?> messageType, @Nullable Class<?> messageKeyType) {
public record SchemaInfo(SchemaType schemaType, @Nullable Class<?> messageKeyType) {
public SchemaInfo {
Objects.requireNonNull(schemaType, "schemaType must not be null");
if (schemaType == SchemaType.NONE) {
throw new IllegalArgumentException("schemaType NONE not supported");
}
if (schemaType != SchemaType.KEY_VALUE && messageKeyType != null) {
throw new IllegalArgumentException("messageKeyType can only be set when schemaType is KEY_VALUE");
}
}
}

6
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java
View File

@@ -321,9 +321,7 @@ class PulsarAutoConfigurationTests {
contextRunner
.withPropertyValues(
"spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(Foo.class.getName()),
"spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=JSON",
"spring.pulsar.defaults.type-mappings[0].schema-info.message-type=%s"
.formatted(Foo.class.getName()))
"spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=JSON")
.run((context -> assertThat(context).hasNotFailed().getBean(SchemaResolver.class)
.asInstanceOf(InstanceOfAssertFactories.type(DefaultSchemaResolver.class))
.extracting(DefaultSchemaResolver::getCustomSchemaMappings,
@@ -339,8 +337,6 @@ class PulsarAutoConfigurationTests {
"spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(Foo.class.getName()),
"spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=%s"
.formatted(SchemaType.KEY_VALUE.name()),
"spring.pulsar.defaults.type-mappings[0].schema-info.message-type=%s"
.formatted(Foo.class.getName()),
"spring.pulsar.defaults.type-mappings[0].schema-info.message-key-type=%s"
.formatted(String.class.getName()))
.run((context -> assertThat(context).hasNotFailed().getBean(SchemaResolver.class)

25
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java
View File

@@ -329,10 +329,9 @@ public class PulsarPropertiesTests {
Map<String, String> props = new HashMap<>();
props.put("spring.pulsar.defaults.type-mappings[0].message-type", Foo.class.getName());
props.put("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type", "JSON");
props.put("spring.pulsar.defaults.type-mappings[0].schema-info.message-type", Foo.class.getName());
bind(props);
assertThat(properties.getDefaults().getTypeMappings()).containsExactly(
new TypeMapping(Foo.class, null, new SchemaInfo(SchemaType.JSON, Foo.class, null)));
assertThat(properties.getDefaults().getTypeMappings())
.containsExactly(new TypeMapping(Foo.class, null, new SchemaInfo(SchemaType.JSON, null)));
}
@Test
@@ -341,10 +340,9 @@ public class PulsarPropertiesTests {
props.put("spring.pulsar.defaults.type-mappings[0].message-type", Foo.class.getName());
props.put("spring.pulsar.defaults.type-mappings[0].topic-name", "foo-topic");
props.put("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type", "JSON");
props.put("spring.pulsar.defaults.type-mappings[0].schema-info.message-type", Foo.class.getName());
bind(props);
assertThat(properties.getDefaults().getTypeMappings()).containsExactly(
new TypeMapping(Foo.class, "foo-topic", new SchemaInfo(SchemaType.JSON, Foo.class, null)));
assertThat(properties.getDefaults().getTypeMappings())
.containsExactly(new TypeMapping(Foo.class, "foo-topic", new SchemaInfo(SchemaType.JSON, null)));
}
@Test
@@ -352,18 +350,17 @@ public class PulsarPropertiesTests {
Map<String, String> props = new HashMap<>();
props.put("spring.pulsar.defaults.type-mappings[0].message-type", Foo.class.getName());
props.put("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type", "KEY_VALUE");
props.put("spring.pulsar.defaults.type-mappings[0].schema-info.message-type", Foo.class.getName());
props.put("spring.pulsar.defaults.type-mappings[0].schema-info.message-key-type", String.class.getName());
bind(props);
assertThat(properties.getDefaults().getTypeMappings()).containsExactly(
new TypeMapping(Foo.class, null, new SchemaInfo(SchemaType.KEY_VALUE, Foo.class, String.class)));
new TypeMapping(Foo.class, null, new SchemaInfo(SchemaType.KEY_VALUE, String.class)));
}
@Test
void schemaTypeRequired() {
Map<String, String> props = new HashMap<>();
props.put("spring.pulsar.defaults.type-mappings[0].message-type", Foo.class.getName());
props.put("spring.pulsar.defaults.type-mappings[0].schema-info.message-type", Foo.class.getName());
props.put("spring.pulsar.defaults.type-mappings[0].schema-info.message-key-type", String.class.getName());
assertThatExceptionOfType(BindException.class).isThrownBy(() -> bind(props)).havingRootCause()
.withMessageContaining("schemaType must not be null");
}
@@ -377,6 +374,16 @@ public class PulsarPropertiesTests {
.withMessageContaining("schemaType NONE not supported");
}
@Test
void messageKeyTypeOnlyAllowedForKeyValueSchemaType() {
Map<String, String> props = new HashMap<>();
props.put("spring.pulsar.defaults.type-mappings[0].message-type", Foo.class.getName());
props.put("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type", "JSON");
props.put("spring.pulsar.defaults.type-mappings[0].schema-info.message-key-type", String.class.getName());
assertThatExceptionOfType(BindException.class).isThrownBy(() -> bind(props)).havingRootCause()
.withMessageContaining("messageKeyType can only be set when schemaType is KEY_VALUE");
}
record Foo(String value) {
}

12
spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java
View File

@@ -258,9 +258,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
+ User.class.getName(),
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription-name=pbit-user-sub2",
"--spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(User.class.getName()),
"--spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=AVRO",
"--spring.pulsar.defaults.type-mappings[0].schema-info.message-type=%s"
.formatted(User.class.getName()))) {
"--spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=AVRO")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
.until(() -> output.toString().contains("Hello binder: User{name='user21', age=21}"));
}
@@ -312,9 +310,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
+ String.class.getName(),
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription-name=pbit-kv-sub1",
"--spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(User.class.getName()),
"--spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=AVRO",
"--spring.pulsar.defaults.type-mappings[0].schema-info.message-type=%s"
.formatted(User.class.getName()))) {
"--spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=AVRO")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
.until(() -> output.toString().contains("Hello binder: 21->User{name='user21', age=21}"));
}
@@ -342,9 +338,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
+ String.class.getName(),
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription-name=pbit-kv-sub2",
"--spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(User.class.getName()),
"--spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=AVRO",
"--spring.pulsar.defaults.type-mappings[0].schema-info.message-type=%s"
.formatted(User.class.getName()))) {
"--spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=AVRO")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
.until(() -> output.toString().contains("Hello binder: 21->User{name='user21', age=21}"));
}