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

Rename @PulsarTypeMapping -> @PulsarMessage

Browse Source
This commit is contained in:
Chris Bono
2024-02-07 15:27:56 -06:00
parent 6a9fb6c523
commit d8524d1db8
10 changed files with 154 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/schema-info/custom-schema-mapping.adoc
View File

@@ -36,16 +36,14 @@ public SchemaResolverCustomizer<DefaultSchemaResolver> schemaResolverCustomizer(
----
==== Type mapping annotation
Another option for specifying default schema information to use for a particular message type is to mark the message class with the `@PulsarTypeMapping` annotation.
Another option for specifying default schema information to use for a particular message type is to mark the message class with the `@PulsarMessage` annotation.
The schema info can be specified via the `schemaType` attribute on the annotation.
The following example configures the system to use JSON as the default schema when producing or consuming messages of type `Foo`:
[source,java,indent=0,subs="verbatim"]
----
@PulsarTypeMapping(schemaType = SchemaType.JSON)
@PulsarMessage(schemaType = SchemaType.JSON)
record Foo(String value) {
}
----
NOTE: The annotations are looked up on-demand and their result is cached. However, there is still a small performance hit on the first lookup. If you want to disable this feature you can invoke the `usePulsarTypeMappingAnnotations(false)` method on the `DefaultSchemaResolver`.

6
spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/topic-resolution.adoc
View File

@@ -36,20 +36,18 @@ WARNING: If the message (or the first message of a `Publisher` input) is `null`,
=== Specified via annotation
When no topic is passed into the API and there are no custom topic mappings configured, the system looks for a `@PulsarTypeMapping` annotation on the class of the message being produced or consumed.
When no topic is passed into the API and there are no custom topic mappings configured, the system looks for a `@PulsarMessage` annotation on the class of the message being produced or consumed.
The default topic can be specified via the `topic` attribute on the annotation.
The following example configures the default topic to use when producing or consuming messages of type `Foo`:
[source,java,indent=0,subs="verbatim"]
----
@PulsarTypeMapping(topic = "foo-topic")
@PulsarMessage(topic = "foo-topic")
record Foo(String value) {
}
----
NOTE: The annotations are looked up on-demand and their result is cached. However, there is still a small performance hit on the first lookup. If you want to disable this feature you can invoke the `usePulsarTypeMappingAnnotations(false)` method on the `DefaultTopicResolver`.
=== Custom topic resolver
The preferred method of adding mappings is via the property mentioned above.
However, if more control is needed you can replace the default resolver by proving your own implementation, for example:

2
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarTypeMapping.java → spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarMessage.java
View File

@@ -37,7 +37,7 @@ import org.apache.pulsar.common.schema.SchemaType;
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface PulsarTypeMapping {
public @interface PulsarMessage {
/**
* Default topic for the annotated message class.

21
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java
View File

@@ -41,7 +41,7 @@ import org.apache.pulsar.common.schema.SchemaType;
import org.springframework.core.ResolvableType;
import org.springframework.core.log.LogAccessor;
import org.springframework.lang.Nullable;
import org.springframework.pulsar.annotation.PulsarTypeMapping;
import org.springframework.pulsar.annotation.PulsarMessage;
import org.springframework.util.Assert;
/**
@@ -92,18 +92,17 @@ public class DefaultSchemaResolver implements SchemaResolver {
private final Map<Class<?>, Schema<?>> customSchemaMappings = new LinkedHashMap<>();
private final PulsarTypeMappingRegistry pulsarTypeMappingRegistry = new PulsarTypeMappingRegistry();
private final PulsarMessageAnnotationRegistry pulsarMessageAnnotationRegistry = new PulsarMessageAnnotationRegistry();
private boolean usePulsarTypeMappingAnnotations = true;
private boolean usePulsarMessageAnnotations = true;
/**
* Sets whether to inspect message classes for the
* {@link PulsarTypeMapping @PulsarTypeMapping} annotation during schema resolution.
* @param usePulsarTypeMappingAnnotations whether to inspect messages for the
* annotation
* {@link PulsarMessage @PulsarMessage} annotation during schema resolution.
* @param usePulsarMessageAnnotations whether to inspect messages for the annotation
*/
public void usePulsarTypeMappingAnnotations(boolean usePulsarTypeMappingAnnotations) {
this.usePulsarTypeMappingAnnotations = usePulsarTypeMappingAnnotations;
public void usePulsarMessageAnnotations(boolean usePulsarMessageAnnotations) {
this.usePulsarMessageAnnotations = usePulsarMessageAnnotations;
}
/**
@@ -157,8 +156,8 @@ public class DefaultSchemaResolver implements SchemaResolver {
// Check for custom schema mapping
Schema<?> schema = this.customSchemaMappings.get(messageClass);
// If no custom schema mapping found, look for @PulsarTypeMapping (if enabled)
if (this.usePulsarTypeMappingAnnotations && schema == null && messageClass != null) {
// If no custom schema mapping found, look for @PulsarMessage (if enabled)
if (this.usePulsarMessageAnnotations && schema == null && messageClass != null) {
schema = getAnnotatedSchemaType(messageClass);
if (schema != null) {
this.addCustomSchemaMapping(messageClass, schema);
@@ -182,7 +181,7 @@ public class DefaultSchemaResolver implements SchemaResolver {
// VisibleForTesting
Schema<?> getAnnotatedSchemaType(Class<?> messageClass) {
PulsarTypeMapping annotation = this.pulsarTypeMappingRegistry.getTypeMappingFor(messageClass).orElse(null);
PulsarMessage annotation = this.pulsarMessageAnnotationRegistry.getAnnotationFor(messageClass).orElse(null);
if (annotation == null || annotation.schemaType() == SchemaType.NONE) {
return null;
}

23
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultTopicResolver.java
View File

@@ -22,7 +22,7 @@ import java.util.Map;
import java.util.function.Supplier;
import org.springframework.lang.Nullable;
import org.springframework.pulsar.annotation.PulsarTypeMapping;
import org.springframework.pulsar.annotation.PulsarMessage;
import org.springframework.util.StringUtils;
/**
@@ -39,18 +39,17 @@ public class DefaultTopicResolver implements TopicResolver {
private final Map<Class<?>, String> customTopicMappings = new LinkedHashMap<>();
private final PulsarTypeMappingRegistry pulsarTypeMappingRegistry = new PulsarTypeMappingRegistry();
private final PulsarMessageAnnotationRegistry pulsarMessageAnnotationRegistry = new PulsarMessageAnnotationRegistry();
private boolean usePulsarTypeMappingAnnotations = true;
private boolean usePulsarMessageAnnotations = true;
/**
* Sets whether to inspect message classes for the
* {@link PulsarTypeMapping @PulsarTypeMapping} annotation during topic resolution.
* @param usePulsarTypeMappingAnnotations whether to inspect messages for the
* annotation
* {@link PulsarMessage @PulsarMessage} annotation during topic resolution.
* @param usePulsarMessageAnnotations whether to inspect messages for the annotation
*/
public void usePulsarTypeMappingAnnotations(boolean usePulsarTypeMappingAnnotations) {
this.usePulsarTypeMappingAnnotations = usePulsarTypeMappingAnnotations;
public void usePulsarMessageAnnotations(boolean usePulsarMessageAnnotations) {
this.usePulsarMessageAnnotations = usePulsarMessageAnnotations;
}
/**
@@ -119,8 +118,8 @@ public class DefaultTopicResolver implements TopicResolver {
// Check for custom topic mapping
String topic = this.customTopicMappings.get(messageType);
// If no custom topic mapping found, look for @PulsarTypeMapping (if enabled)
if (this.usePulsarTypeMappingAnnotations && topic == null) {
// If no custom topic mapping found, look for @PulsarMessage (if enabled)
if (this.usePulsarMessageAnnotations && topic == null) {
topic = getAnnotatedTopicInfo(messageType);
if (topic != null) {
this.addCustomTopicMapping(messageType, topic);
@@ -137,8 +136,8 @@ public class DefaultTopicResolver implements TopicResolver {
// VisibleForTesting
String getAnnotatedTopicInfo(Class<?> messageType) {
return this.pulsarTypeMappingRegistry.getTypeMappingFor(messageType)
.map(PulsarTypeMapping::topic)
return this.pulsarMessageAnnotationRegistry.getAnnotationFor(messageType)
.map(PulsarMessage::topic)
.filter(StringUtils::hasText)
.orElse(null);
}

44
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTypeMappingRegistry.java → spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarMessageAnnotationRegistry.java
View File

@@ -21,61 +21,61 @@ import java.util.concurrent.ConcurrentHashMap;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.log.LogAccessor;
import org.springframework.pulsar.annotation.PulsarTypeMapping;
import org.springframework.pulsar.annotation.PulsarMessage;
import org.springframework.util.Assert;
/**
* A registry that holds the {@link PulsarTypeMapping @PulsarTypeMapping} annotations and
* each associated class that is marked with the annotation.
* A registry that holds the {@link PulsarMessage @PulsarMessage} annotations and each
* associated class that is marked with the annotation.
* <p>
* The annotations are looked up on-demand and the result is cached.
* <p>
* Once the cache reaches a {@link #maxNumberOfMappingsCached certain size} (default of
* Once the cache reaches a {@link #maxNumberOfAnnotationsCached certain size} (default of
* {@link #DEFAULT_MAX_CACHE_SIZE}) it is cleared and the annotations will be looked up
* again the next time they are requested.
*
* @author Chris Bono
*/
class PulsarTypeMappingRegistry {
class PulsarMessageAnnotationRegistry {
private static final int DEFAULT_MAX_CACHE_SIZE = 1000;
private final int maxNumberOfMappingsCached;
private final int maxNumberOfAnnotationsCached;
private final LogAccessor logger = new LogAccessor(this.getClass());
private ConcurrentHashMap<Class<?>, Optional<PulsarTypeMapping>> typeMappingsByClass = new ConcurrentHashMap<>();
private ConcurrentHashMap<Class<?>, Optional<PulsarMessage>> annotationsByClass = new ConcurrentHashMap<>();
PulsarTypeMappingRegistry() {
PulsarMessageAnnotationRegistry() {
this(DEFAULT_MAX_CACHE_SIZE);
}
PulsarTypeMappingRegistry(int maxNumberOfMappingsCached) {
Assert.state(maxNumberOfMappingsCached > 0, "maxNumberOfMappingsCached must be > 0");
this.maxNumberOfMappingsCached = maxNumberOfMappingsCached;
PulsarMessageAnnotationRegistry(int maxNumberOfAnnotationsCached) {
Assert.state(maxNumberOfAnnotationsCached > 0, "maxNumberOfAnnotationsCached must be > 0");
this.maxNumberOfAnnotationsCached = maxNumberOfAnnotationsCached;
}
/**
* Gets the {@link PulsarTypeMapping @PulsarTypeMapping} on the specified class or
* empty if the class is not marked with the annotation.
* Gets the {@link PulsarMessage @PulsarMessage} on the specified class or empty if
* the class is not marked with the annotation.
* @param targetClass the class to check for the annotation
* @return an optional containing the annotation or empty if the class is not marked
* with the annotation.
*/
Optional<PulsarTypeMapping> getTypeMappingFor(Class<?> targetClass) {
var optionalTypeMapping = this.typeMappingsByClass.computeIfAbsent(targetClass, this::findTypeMappingOn);
if (this.typeMappingsByClass.size() > this.maxNumberOfMappingsCached) {
Optional<PulsarMessage> getAnnotationFor(Class<?> targetClass) {
var annotation = this.annotationsByClass.computeIfAbsent(targetClass, this::findAnnotationOn);
if (this.annotationsByClass.size() > this.maxNumberOfAnnotationsCached) {
this.logger
.info(() -> "Clearing cache - max entries exceeded (%d)".formatted(this.maxNumberOfMappingsCached));
this.typeMappingsByClass = new ConcurrentHashMap<>();
.info(() -> "Clearing cache - max entries exceeded (%d)".formatted(this.maxNumberOfAnnotationsCached));
this.annotationsByClass = new ConcurrentHashMap<>();
}
return optionalTypeMapping;
return annotation;
}
// VisibleForTesting
protected Optional<PulsarTypeMapping> findTypeMappingOn(Class<?> targetClass) {
this.logger.debug(() -> "Looking for @PulsarTypeMapping on " + targetClass);
PulsarTypeMapping annotation = AnnotationUtils.findAnnotation(targetClass, PulsarTypeMapping.class);
protected Optional<PulsarMessage> findAnnotationOn(Class<?> targetClass) {
this.logger.debug(() -> "Looking for @PulsarMessage on " + targetClass);
PulsarMessage annotation = AnnotationUtils.findAnnotation(targetClass, PulsarMessage.class);
return Optional.ofNullable(annotation);
}

16
spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java
View File

@@ -54,7 +54,7 @@ import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;
import org.springframework.core.ResolvableType;
import org.springframework.pulsar.annotation.PulsarTypeMapping;
import org.springframework.pulsar.annotation.PulsarMessage;
import org.springframework.pulsar.listener.Proto;
import org.springframework.pulsar.listener.Proto.Person;
@@ -431,7 +431,7 @@ class DefaultSchemaResolverTests {
@Test
void annotationMappingIgnoredWhenFeatureDisabled() {
resolver.usePulsarTypeMappingAnnotations(false);
resolver.usePulsarMessageAnnotations(false);
assertThatIllegalArgumentException()
.isThrownBy(() -> resolver.resolveSchema(JsonMsgType.class, false).orElseThrow())
.withMessage("Schema not specified and no schema found for " + JsonMsgType.class);
@@ -443,24 +443,24 @@ class DefaultSchemaResolverTests {
assertThat(resolver.resolveSchema(JsonMsgType.class, false).orElseThrow()).isEqualTo(Schema.STRING);
}
@PulsarTypeMapping(schemaType = SchemaType.JSON)
@PulsarMessage(schemaType = SchemaType.JSON)
record JsonMsgType(String value) {
}
@PulsarTypeMapping(schemaType = SchemaType.KEY_VALUE, messageKeyType = String.class,
@PulsarMessage(schemaType = SchemaType.KEY_VALUE, messageKeyType = String.class,
messageValueSchemaType = SchemaType.JSON)
record KeyValueMsgType(String key) {
}
@PulsarTypeMapping(schemaType = SchemaType.KEY_VALUE, messageValueSchemaType = SchemaType.JSON)
@PulsarMessage(schemaType = SchemaType.KEY_VALUE, messageValueSchemaType = SchemaType.JSON)
record KeyValueMsgTypeNoKeyInfo(String key) {
}
@PulsarTypeMapping(schemaType = SchemaType.KEY_VALUE, messageKeyType = String.class)
@PulsarMessage(schemaType = SchemaType.KEY_VALUE, messageKeyType = String.class)
record KeyValueMsgTypeNoValueInfo(String key) {
}
@PulsarTypeMapping(topic = "ignore-topic")
@PulsarMessage(topic = "ignore-topic")
record NoSchemaInfoMsgType(String value) {
}
@@ -475,7 +475,7 @@ class DefaultSchemaResolverTests {
record Zaa(String value) {
}
@PulsarTypeMapping(schemaType = SchemaType.STRING)
@PulsarMessage(schemaType = SchemaType.STRING)
record Zaz(String value) {
}

8
spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultTopicResolverTests.java
View File

@@ -34,7 +34,7 @@ import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.springframework.lang.Nullable;
import org.springframework.pulsar.annotation.PulsarTypeMapping;
import org.springframework.pulsar.annotation.PulsarMessage;
/**
* Unit tests for {@link DefaultTopicResolver}.
@@ -145,7 +145,7 @@ class DefaultTopicResolverTests {
@Test
void annotationMappingIgnoredWhenFeatureDisabled() {
resolver.usePulsarTypeMappingAnnotations(false);
resolver.usePulsarMessageAnnotations(false);
assertThat(resolver.resolveTopic(null, Baz.class, () -> defaultTopic).value().orElse(null))
.isEqualTo(defaultTopic);
}
@@ -211,11 +211,11 @@ class DefaultTopicResolverTests {
record Bar(String value) {
}
@PulsarTypeMapping(topic = bazTopic)
@PulsarMessage(topic = bazTopic)
record Baz(String value) {
}
@PulsarTypeMapping(schemaType = SchemaType.STRING)
@PulsarMessage(schemaType = SchemaType.STRING)
record BazNoTopicInfo(String value) {
}

94
spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarMessageAnnotationRegistryTests.java Normal file
View File

@@ -0,0 +1,94 @@
/*
* Copyright 2023-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.pulsar.core;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.springframework.pulsar.annotation.PulsarMessage;
/**
* Unit tests for {@link PulsarMessageAnnotationRegistry}.
*
* @author Chris Bono
*/
class PulsarMessageAnnotationRegistryTests {
@Test
void annotationFoundAndCached() {
PulsarMessageAnnotationRegistry registry = spy(new PulsarMessageAnnotationRegistry(2));
var annotation = registry.getAnnotationFor(Foo.class);
assertThat(annotation).map(PulsarMessage::topic).hasValue("foo-topic");
// subsequent calls are cached
assertThat(registry.getAnnotationFor(Foo.class)).isSameAs(annotation);
assertThat(registry.getAnnotationFor(Foo.class)).isSameAs(annotation);
verify(registry, times(1)).findAnnotationOn(Foo.class);
}
@Test
void annotationNotFoundAndCached() {
PulsarMessageAnnotationRegistry registry = spy(new PulsarMessageAnnotationRegistry(2));
var annotation = registry.getAnnotationFor(Bar.class);
assertThat(annotation).isEmpty();
// subsequent calls are cached
assertThat(registry.getAnnotationFor(Bar.class)).isSameAs(annotation);
assertThat(registry.getAnnotationFor(Bar.class)).isSameAs(annotation);
verify(registry, times(1)).findAnnotationOn(Bar.class);
}
@Test
void cacheIsClearedOnceMaxNumberReached() {
PulsarMessageAnnotationRegistry registry = spy(new PulsarMessageAnnotationRegistry(2));
registry.getAnnotationFor(Foo.class);
registry.getAnnotationFor(Bar.class);
registry.getAnnotationFor(Zaa.class);
// the 3rd request will force a cache clear - subsequent calls will do lookup
// again
registry.getAnnotationFor(Foo.class);
registry.getAnnotationFor(Bar.class);
// now values should be cached again
registry.getAnnotationFor(Foo.class);
registry.getAnnotationFor(Bar.class);
verify(registry, times(2)).findAnnotationOn(Foo.class);
verify(registry, times(2)).findAnnotationOn(Bar.class);
}
@ParameterizedTest
@ValueSource(ints = { -1, 0 })
void maxNumberOfAnnotationsMustBePositive(int maxAnnotations) {
assertThatIllegalStateException().isThrownBy(() -> new PulsarMessageAnnotationRegistry(maxAnnotations))
.withMessage("maxNumberOfAnnotationsCached must be > 0");
}
@PulsarMessage(topic = "foo-topic")
record Foo(String value) {
}
record Bar(String value) {
}
record Zaa(String value) {
}
}

94
spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTypeMappingRegistryTests.java
View File

@@ -1,94 +0,0 @@
/*
* Copyright 2023-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.pulsar.core;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.springframework.pulsar.annotation.PulsarTypeMapping;
/**
* Unit tests for {@link PulsarTypeMappingRegistry}.
*
* @author Chris Bono
*/
class PulsarTypeMappingRegistryTests {
@Test
void typeMappingFoundAndCached() {
PulsarTypeMappingRegistry registry = spy(new PulsarTypeMappingRegistry(2));
var typeMapping = registry.getTypeMappingFor(Foo.class);
assertThat(typeMapping).map(PulsarTypeMapping::topic).hasValue("foo-topic");
// subsequent calls are cached
assertThat(registry.getTypeMappingFor(Foo.class)).isSameAs(typeMapping);
assertThat(registry.getTypeMappingFor(Foo.class)).isSameAs(typeMapping);
verify(registry, times(1)).findTypeMappingOn(Foo.class);
}
@Test
void typeMappingNotFoundAndCached() {
PulsarTypeMappingRegistry registry = spy(new PulsarTypeMappingRegistry(2));
var typeMapping = registry.getTypeMappingFor(Bar.class);
assertThat(typeMapping).isEmpty();
// subsequent calls are cached
assertThat(registry.getTypeMappingFor(Bar.class)).isSameAs(typeMapping);
assertThat(registry.getTypeMappingFor(Bar.class)).isSameAs(typeMapping);
verify(registry, times(1)).findTypeMappingOn(Bar.class);
}
@Test
void cacheIsClearedOnceMaxNumberReached() {
PulsarTypeMappingRegistry registry = spy(new PulsarTypeMappingRegistry(2));
registry.getTypeMappingFor(Foo.class);
registry.getTypeMappingFor(Bar.class);
registry.getTypeMappingFor(Zaa.class);
// the 3rd request will force a cache clear - subsequent calls will do lookup
// again
registry.getTypeMappingFor(Foo.class);
registry.getTypeMappingFor(Bar.class);
// now values should be cached again
registry.getTypeMappingFor(Foo.class);
registry.getTypeMappingFor(Bar.class);
verify(registry, times(2)).findTypeMappingOn(Foo.class);
verify(registry, times(2)).findTypeMappingOn(Bar.class);
}
@ParameterizedTest
@ValueSource(ints = { -1, 0 })
void maxNumberOfMappingsMustBePositive(int maxNumberOfMappings) {
assertThatIllegalStateException().isThrownBy(() -> new PulsarTypeMappingRegistry(maxNumberOfMappings))
.withMessage("maxNumberOfMappingsCached must be > 0");
}
@PulsarTypeMapping(topic = "foo-topic")
record Foo(String value) {
}
record Bar(String value) {
}
record Zaa(String value) {
}
}