SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1517 Avro Union Types import schema issue when we have multiple avro files

Browse Source
This commit is contained in:
Sercan Karaoglu
2018-10-29 22:04:24 +01:00
committed by Oleg Zhurakousky
parent 5d2c3ff89c
commit 829604ae32
11 changed files with 517 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-cloud-stream-schema/pom.xml
View File

@@ -68,8 +68,17 @@
</execution>
</executions>
<configuration>
<outputDirectory>${project.basedir}/target/generated-test-sources</outputDirectory>
<testOutputDirectory>${project.basedir}/target/generated-test-sources</testOutputDirectory>
<testSourceDirectory>${project.basedir}/src/test/resources/schemas</testSourceDirectory>
<testIncludes>
<testInclude>**/*.avsc</testInclude>
</testIncludes>
<imports>
<import>${project.basedir}/src/test/resources/schemas/imports/Email.avsc</import>
<import>${project.basedir}/src/test/resources/schemas/imports/Sms.avsc</import>
<import>${project.basedir}/src/test/resources/schemas/imports/PushNotification.avsc</import>
</imports>
</configuration>
</plugin>
</plugins>

22
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AbstractAvroMessageConverter.java
View File

@@ -43,15 +43,24 @@ import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.AbstractMessageConverter;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.util.MimeType;
import org.springframework.util.ObjectUtils;
/**
* Base class for Apache Avro
* {@link org.springframework.messaging.converter.MessageConverter} implementations.
* @author Marius Bogoevici
* @author Vinicius Carvalho
* @author Sercan Karaoglu
*/
public abstract class AbstractAvroMessageConverter extends AbstractMessageConverter {
/**
* common parser will let user to import external schemas.
*/
private Schema.Parser schemaParser = new Schema.Parser();
protected Resource[] schemaImports = new Resource[]{};
protected AbstractAvroMessageConverter(MimeType supportedMimeType) {
this(Collections.singletonList(supportedMimeType));
}
@@ -61,8 +70,17 @@ public abstract class AbstractAvroMessageConverter extends AbstractMessageConver
setContentTypeResolver(new OriginalContentTypeResolver());
}
protected static Schema parseSchema(Resource r) throws IOException {
return new Schema.Parser().parse(r.getInputStream());
protected Schema parseSchema(Resource r) throws IOException {
if (ObjectUtils.isEmpty(schemaImports)) {
return new Schema.Parser().parse(r.getInputStream());
}
else {
return schemaParser.parse(r.getInputStream());
}
}
protected void setSchemaImports(Resource[] imports) {
this.schemaImports = imports;
}
@Override

5
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java
View File

@@ -36,6 +36,7 @@ import org.springframework.util.ReflectionUtils;
/**
* @author Marius Bogoevici
* @author Vinicius Carvalho
* @author Sercan Karaoglu
*/
@Configuration
@ConditionalOnClass(name = "org.apache.avro.Schema")
@@ -64,6 +65,10 @@ public class AvroMessageConverterAutoConfiguration {
avroSchemaRegistryClientMessageConverter.setSchemaLocations(
this.avroMessageConverterProperties.getSchemaLocations());
}
if (!ObjectUtils.isEmpty(this.avroMessageConverterProperties.getSchemaImports())) {
avroSchemaRegistryClientMessageConverter.setSchemaImports(
this.avroMessageConverterProperties.getSchemaImports());
}
avroSchemaRegistryClientMessageConverter.setPrefix(this.avroMessageConverterProperties.getPrefix());
try {

24
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterProperties.java
View File

@@ -22,6 +22,7 @@ import org.springframework.util.Assert;
/**
* @author Vinicius Carvalho
* @author Sercan Karaoglu
*/
@ConfigurationProperties(prefix = "spring.cloud.stream.schema.avro")
public class AvroMessageConverterProperties {
@@ -30,8 +31,22 @@ public class AvroMessageConverterProperties {
private Resource readerSchema;
/**
* The source directory of Apache Avro schema. This schema is used by this
* converter. If this schema depends on other schemas consider defining those
* those dependent ones in the {@link #schemaImports}
* @parameter
*/
private Resource[] schemaLocations;
/**
* A list of files or directories that should be loaded first thus making
* them importable by subsequent schemas. Note that imported files
* should not reference each other.
* @parameter
*/
private Resource[] schemaImports;
private String prefix = "vnd";
private Class<? extends SubjectNamingStrategy> subjectNamingStrategy = DefaultSubjectNamingStrategy.class;
@@ -78,4 +93,13 @@ public class AvroMessageConverterProperties {
Assert.notNull(subjectNamingStrategy, "cannot be null");
this.subjectNamingStrategy = subjectNamingStrategy;
}
public Resource[] getSchemaImports() {
return schemaImports;
}
public void setSchemaImports(Resource[] schemaImports) {
this.schemaImports = schemaImports;
}
}

56
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java
View File

@@ -18,10 +18,12 @@ package org.springframework.cloud.stream.schema.avro;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Stream;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericContainer;
@@ -71,6 +73,7 @@ import org.springframework.util.ObjectUtils;
* @author Marius Bogoevici
* @author Vinicius Carvalho
* @author Oleg Zhurakousky
* @author Sercan Karaoglu
*/
public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessageConverter
implements InitializingBean {
@@ -147,6 +150,16 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
this.schemaLocations = schemaLocations;
}
/**
* A set of schema locations where should be imported first. Schemas provided at these
* locations will be reference, thus they should not reference each other.
*
* @param schemaImports
*/
public void setSchemaImports(Resource[] schemaImports) {
this.schemaImports = schemaImports;
}
/**
* Set the prefix to be used in the publised subtype. Default 'vnd'.
* @param prefix
@@ -162,28 +175,35 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
public void afterPropertiesSet() throws Exception {
this.versionedSchema = Pattern.compile("application/" + this.prefix
+ "\\.([\\p{Alnum}\\$\\.]+)\\.v(\\p{Digit}+)\\+"+AVRO_FORMAT);
if (!ObjectUtils.isEmpty(this.schemaLocations)) {
this.logger.info("Scanning avro schema resources on classpath");
if (this.logger.isInfoEnabled()) {
this.logger.info("Parsing" + this.schemaLocations.length);
}
for (Resource schemaLocation : this.schemaLocations) {
try {
Schema schema = parseSchema(schemaLocation);
if (schema.getType().equals(Schema.Type.UNION)) {
schema.getTypes().forEach(innerSchema -> registerSchema(schemaLocation, innerSchema));
} else {
registerSchema(schemaLocation, schema);
Stream.of(this.schemaImports, this.schemaLocations)
.filter(arr -> !ObjectUtils.isEmpty(arr))
.distinct()
.peek(resources -> {
this.logger.info("Scanning avro schema resources on classpath");
if (this.logger.isInfoEnabled()) {
this.logger.info("Parsing" + this.schemaImports.length);
}
}).flatMap(Arrays::stream).forEach(resource -> {
try {
Schema schema = parseSchema(resource);
if (schema.getType().equals(Schema.Type.UNION)) {
schema.getTypes().forEach(
innerSchema -> registerSchema(resource, innerSchema));
}
catch (IOException e) {
if (this.logger.isWarnEnabled()) {
this.logger.warn("Failed to parse schema at "
+ schemaLocation.getFilename(), e);
}
else {
registerSchema(resource, schema);
}
}
}
catch (IOException e) {
if (this.logger.isWarnEnabled()) {
this.logger.warn(
"Failed to parse schema at " + resource.getFilename(),
e);
}
}
});
if (this.cacheManager instanceof NoOpCacheManager) {
logger.warn("Schema caching is effectively disabled "
+ "since configured cache manager is a NoOpCacheManager. If this was not "

249
spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroMessageConverterSerializationTests.java
View File

@@ -21,6 +21,10 @@ import java.util.Collections;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import example.avro.Command;
import example.avro.Email;
import example.avro.PushNotification;
import example.avro.Sms;
import example.avro.User;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
@@ -55,101 +59,168 @@ import org.springframework.util.MimeTypeUtils;
/**
* @author Vinicius Carvalho
* @author Sercan Karaoglu
*/
public class AvroMessageConverterSerializationTests {
Pattern versionedSchema = Pattern.compile(
"application/" + "vnd" + "\\.([\\p{Alnum}\\$\\.]+)\\.v(\\p{Digit}+)\\+avro");
Pattern versionedSchema = Pattern.compile("application/" + "vnd"
+ "\\.([\\p{Alnum}\\$\\.]+)\\.v(\\p{Digit}+)\\+avro");
Log logger = LogFactory.getLog(getClass());
Log logger = LogFactory.getLog(getClass());
private ConfigurableApplicationContext schemaRegistryServerContext;
private ConfigurableApplicationContext schemaRegistryServerContext;
@Before
public void setup() {
schemaRegistryServerContext = SpringApplication
.run(SchemaRegistryServerApplication.class, "--spring.main.allow-bean-definition-overriding=true");
}
@After
public void tearDown() {
schemaRegistryServerContext.close();
}
@Test
public void sourceWriteSameVersion() throws Exception {
User specificRecord = new User();
specificRecord.setName("joe");
Schema v1 = new Schema.Parser().parse(AvroMessageConverterSerializationTests.class
.getClassLoader().getResourceAsStream("schemas/user.avsc"));
GenericRecord genericRecord = new GenericData.Record(v1);
genericRecord.put("name", "joe");
SchemaRegistryClient client = new DefaultSchemaRegistryClient();
AvroSchemaRegistryClientMessageConverter converter = new AvroSchemaRegistryClientMessageConverter(
client, new NoOpCacheManager());
converter.setSubjectNamingStrategy(new DefaultSubjectNamingStrategy());
converter.setDynamicSchemaGenerationEnabled(false);
converter.afterPropertiesSet();
Message specificMessage = converter.toMessage(specificRecord,
new MutableMessageHeaders(Collections.<String, Object> emptyMap()),
MimeTypeUtils.parseMimeType("application/*+avro"));
SchemaReference specificRef = extractSchemaReference(MimeTypeUtils.parseMimeType(
specificMessage.getHeaders().get("contentType").toString()));
Message genericMessage = converter.toMessage(genericRecord,
new MutableMessageHeaders(Collections.<String, Object> emptyMap()),
MimeTypeUtils.parseMimeType("application/*+avro"));
SchemaReference genericRef = extractSchemaReference(MimeTypeUtils.parseMimeType(
genericMessage.getHeaders().get("contentType").toString()));
Assert.assertEquals(genericRef, specificRef);
Assert.assertEquals(1, genericRef.getVersion());
}
@Test
public void testOriginalContentTypeHeaderOnly() throws Exception {
User specificRecord = new User();
specificRecord.setName("joe");
Schema v1 = new Schema.Parser().parse(AvroMessageConverterSerializationTests.class
.getClassLoader().getResourceAsStream("schemas/user.avsc"));
GenericRecord genericRecord = new GenericData.Record(v1);
genericRecord.put("name", "joe");
SchemaRegistryClient client = new DefaultSchemaRegistryClient();
client.register("user", "avro", v1.toString());
AvroSchemaRegistryClientMessageConverter converter = new AvroSchemaRegistryClientMessageConverter(
client, new NoOpCacheManager());
converter.setDynamicSchemaGenerationEnabled(false);
converter.afterPropertiesSet();
ByteArrayOutputStream baos = new ByteArrayOutputStream();
DatumWriter<User> writer = new SpecificDatumWriter<>(User.class);
Encoder encoder = EncoderFactory.get().binaryEncoder(baos, null);
writer.write(specificRecord, encoder);
encoder.flush();
Message source = MessageBuilder.withPayload(baos.toByteArray())
.setHeader(MessageHeaders.CONTENT_TYPE,
MimeTypeUtils.APPLICATION_OCTET_STREAM)
.setHeader(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE,
"application/vnd.user.v1+avro")
.build();
Object converted = converter.fromMessage(source, User.class);
Assert.assertNotNull(converted);
Assert.assertEquals(specificRecord.getName().toString(),
((User) converted).getName().toString());
}
private SchemaReference extractSchemaReference(MimeType mimeType) {
SchemaReference schemaReference = null;
Matcher schemaMatcher = this.versionedSchema.matcher(mimeType.toString());
if (schemaMatcher.find()) {
String subject = schemaMatcher.group(1);
Integer version = Integer.parseInt(schemaMatcher.group(2));
schemaReference = new SchemaReference(subject, version,
AvroSchemaRegistryClientMessageConverter.AVRO_FORMAT);
@Before
public void setup() {
schemaRegistryServerContext = SpringApplication
.run(SchemaRegistryServerApplication.class,
"--spring.main.allow-bean-definition-overriding=true");
}
@After
public void tearDown() {
schemaRegistryServerContext.close();
}
@Test
public void testSchemaImport() throws Exception {
SchemaRegistryClient client = new DefaultSchemaRegistryClient();
AvroSchemaRegistryClientMessageConverter converter = new AvroSchemaRegistryClientMessageConverter(
client, new NoOpCacheManager());
converter.setSubjectNamingStrategy(new DefaultSubjectNamingStrategy());
converter.setDynamicSchemaGenerationEnabled(false);
converter.setSchemaLocations(schemaRegistryServerContext
.getResources("classpath:schemas/Command.avsc"));
converter.setSchemaImports(schemaRegistryServerContext
.getResources("classpath:schemas/imports/*.avsc"));
converter.afterPropertiesSet();
Command notification = notification();
Message specificMessage = converter.toMessage(notification,
new MutableMessageHeaders(
Collections.<String, Object> emptyMap()));
Object o = converter.fromMessage(specificMessage, Command.class);
Assert.assertEquals("Serialization issue when use schema-imports", o, notification);
}
@Test
public void sourceWriteSameVersion() throws Exception {
User specificRecord = new User();
specificRecord.setName("joe");
Schema v1 = new Schema.Parser()
.parse(AvroMessageConverterSerializationTests.class
.getClassLoader()
.getResourceAsStream("schemas/user.avsc"));
GenericRecord genericRecord = new GenericData.Record(v1);
genericRecord.put("name", "joe");
SchemaRegistryClient client = new DefaultSchemaRegistryClient();
AvroSchemaRegistryClientMessageConverter converter = new AvroSchemaRegistryClientMessageConverter(
client, new NoOpCacheManager());
converter.setSubjectNamingStrategy(new DefaultSubjectNamingStrategy());
converter.setDynamicSchemaGenerationEnabled(false);
converter.afterPropertiesSet();
Message specificMessage = converter.toMessage(specificRecord,
new MutableMessageHeaders(
Collections.<String, Object> emptyMap()),
MimeTypeUtils.parseMimeType("application/*+avro"));
SchemaReference specificRef = extractSchemaReference(MimeTypeUtils
.parseMimeType(specificMessage.getHeaders().get("contentType")
.toString()));
Message genericMessage = converter.toMessage(genericRecord,
new MutableMessageHeaders(
Collections.<String, Object> emptyMap()),
MimeTypeUtils.parseMimeType("application/*+avro"));
SchemaReference genericRef = extractSchemaReference(MimeTypeUtils
.parseMimeType(genericMessage.getHeaders().get("contentType")
.toString()));
Assert.assertEquals(genericRef, specificRef);
Assert.assertEquals(1, genericRef.getVersion());
}
@Test
public void testOriginalContentTypeHeaderOnly() throws Exception {
User specificRecord = new User();
specificRecord.setName("joe");
Schema v1 = new Schema.Parser()
.parse(AvroMessageConverterSerializationTests.class
.getClassLoader()
.getResourceAsStream("schemas/user.avsc"));
GenericRecord genericRecord = new GenericData.Record(v1);
genericRecord.put("name", "joe");
SchemaRegistryClient client = new DefaultSchemaRegistryClient();
client.register("user", "avro", v1.toString());
AvroSchemaRegistryClientMessageConverter converter = new AvroSchemaRegistryClientMessageConverter(
client, new NoOpCacheManager());
converter.setDynamicSchemaGenerationEnabled(false);
converter.afterPropertiesSet();
ByteArrayOutputStream baos = new ByteArrayOutputStream();
DatumWriter<User> writer = new SpecificDatumWriter<>(User.class);
Encoder encoder = EncoderFactory.get().binaryEncoder(baos, null);
writer.write(specificRecord, encoder);
encoder.flush();
Message source = MessageBuilder.withPayload(baos.toByteArray())
.setHeader(MessageHeaders.CONTENT_TYPE,
MimeTypeUtils.APPLICATION_OCTET_STREAM)
.setHeader(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE,
"application/vnd.user.v1+avro").build();
Object converted = converter.fromMessage(source, User.class);
Assert.assertNotNull(converted);
Assert.assertEquals(specificRecord.getName().toString(),
((User) converted).getName().toString());
}
private SchemaReference extractSchemaReference(MimeType mimeType) {
SchemaReference schemaReference = null;
Matcher schemaMatcher = this.versionedSchema.matcher(mimeType.toString());
if (schemaMatcher.find()) {
String subject = schemaMatcher.group(1);
Integer version = Integer.parseInt(schemaMatcher.group(2));
schemaReference = new SchemaReference(subject, version,
AvroSchemaRegistryClientMessageConverter.AVRO_FORMAT);
}
return schemaReference;
}
public static Command notification() {
Command messageToSend = getCommandToSend();
messageToSend.setType("notification");
PushNotification pushNotification = new PushNotification();
pushNotification.setArn("google");
pushNotification.setText("hello");
messageToSend.setPayload(pushNotification);
return messageToSend;
}
public static Command sms() {
Command messageToSend = getCommandToSend();
messageToSend.setType("sms");
Sms sms = new Sms();
sms.setPhoneNumber("6141231212");
sms.setText("hello");
messageToSend.setPayload(sms);
return messageToSend;
}
public static Command email() {
Command messageToSend = getCommandToSend();
messageToSend.setType("email");
Email email = new Email();
email.setAddressTo("sercan");
email.setText("hello");
email.setTitle("hi");
messageToSend.setPayload(email);
return messageToSend;
}
public static Command getCommandToSend() {
Command messageToSend = new Command();
messageToSend.setCorrelationId("abc");
return messageToSend;
}
return schemaReference;
}
}

306
spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaRegistryClientMessageConverterTests.java
View File

@@ -21,6 +21,9 @@ import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import example.avro.Command;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
@@ -47,130 +50,209 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.cloud.schema.avro.AvroMessageConverterSerializationTests.notification;
/**
* @author Marius Bogoevici
* @author Oleg Zhurakousky
* @author Sercan Karaoglu
*/
public class AvroSchemaRegistryClientMessageConverterTests {
static SchemaRegistryClient stubSchemaRegistryClient = new StubSchemaRegistryClient();
static SchemaRegistryClient stubSchemaRegistryClient = new StubSchemaRegistryClient();
@Test
public void testSendMessage() throws Exception {
private ConfigurableApplicationContext schemaRegistryServerContext;
ConfigurableApplicationContext schemaRegistryServerContext = SpringApplication.run(
SchemaRegistryServerApplication.class, "--spring.main.allow-bean-definition-overriding=true");
ConfigurableApplicationContext sourceContext = SpringApplication.run(AvroSourceApplication.class,
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.output.contentType=application/*+avro",
"--spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled=true");
Source source = sourceContext.getBean(Source.class);
User1 firstOutboundFoo = new User1();
firstOutboundFoo.setFavoriteColor("foo" + UUID.randomUUID().toString());
firstOutboundFoo.setName("foo" + UUID.randomUUID().toString());
source.output().send(MessageBuilder.withPayload(firstOutboundFoo).build());
MessageCollector sourceMessageCollector = sourceContext.getBean(MessageCollector.class);
Message<?> outboundMessage = sourceMessageCollector.forChannel(source.output()).poll(1000,
TimeUnit.MILLISECONDS);
ConfigurableApplicationContext barSourceContext = SpringApplication.run(AvroSourceApplication.class,
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.output.contentType=application/vnd.user1.v1+avro",
"--spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled=true");
Source barSource = barSourceContext.getBean(Source.class);
User2 firstOutboundUser2 = new User2();
firstOutboundUser2.setFavoriteColor("foo" + UUID.randomUUID().toString());
firstOutboundUser2.setName("foo" + UUID.randomUUID().toString());
barSource.output().send(MessageBuilder.withPayload(firstOutboundUser2).build());
MessageCollector barSourceMessageCollector = barSourceContext.getBean(MessageCollector.class);
Message<?> barOutboundMessage = barSourceMessageCollector.forChannel(barSource.output()).poll(1000,
TimeUnit.MILLISECONDS);
assertThat(barOutboundMessage).isNotNull();
User2 secondBarOutboundPojo = new User2();
secondBarOutboundPojo.setFavoriteColor("foo" + UUID.randomUUID().toString());
secondBarOutboundPojo.setName("foo" + UUID.randomUUID().toString());
source.output().send(MessageBuilder.withPayload(secondBarOutboundPojo).build());
Message<?> secondBarOutboundMessage = sourceMessageCollector.forChannel(source.output()).poll(1000,
TimeUnit.MILLISECONDS);
ConfigurableApplicationContext sinkContext = SpringApplication.run(AvroSinkApplication.class,
"--server.port=0", "--spring.jmx.enabled=false");
Sink sink = sinkContext.getBean(Sink.class);
sink.input().send(outboundMessage);
sink.input().send(barOutboundMessage);
sink.input().send(secondBarOutboundMessage);
List<User2> receivedPojos = sinkContext.getBean(AvroSinkApplication.class).receivedPojos;
assertThat(receivedPojos).hasSize(3);
assertThat(receivedPojos.get(0)).isNotSameAs(firstOutboundFoo);
assertThat(receivedPojos.get(0).getFavoriteColor()).isEqualTo(firstOutboundFoo.getFavoriteColor());
assertThat(receivedPojos.get(0).getName()).isEqualTo(firstOutboundFoo.getName());
assertThat(receivedPojos.get(0).getFavoritePlace()).isEqualTo("NYC");
assertThat(receivedPojos.get(1)).isNotSameAs(firstOutboundUser2);
assertThat(receivedPojos.get(1).getFavoriteColor()).isEqualTo(firstOutboundUser2.getFavoriteColor());
assertThat(receivedPojos.get(1).getName()).isEqualTo(firstOutboundUser2.getName());
assertThat(receivedPojos.get(1).getFavoritePlace()).isEqualTo("Boston");
assertThat(receivedPojos.get(2)).isNotSameAs(secondBarOutboundPojo);
assertThat(receivedPojos.get(2).getFavoriteColor()).isEqualTo(secondBarOutboundPojo.getFavoriteColor());
assertThat(receivedPojos.get(2).getName()).isEqualTo(secondBarOutboundPojo.getName());
assertThat(receivedPojos.get(2).getFavoritePlace()).isEqualTo(secondBarOutboundPojo.getFavoritePlace());
sinkContext.close();
barSourceContext.close();
sourceContext.close();
schemaRegistryServerContext.close();
}
@Test
public void testNoCacheConfiguration() {
ConfigurableApplicationContext sourceContext = SpringApplication.run(NoCacheConfiguration.class,
"--spring.main.web-environment=false");
AvroSchemaRegistryClientMessageConverter converter = sourceContext
.getBean(AvroSchemaRegistryClientMessageConverter.class);
DirectFieldAccessor accessor = new DirectFieldAccessor(converter);
assertThat(accessor.getPropertyValue("cacheManager")).isInstanceOf(NoOpCacheManager.class);
}
@EnableBinding(Source.class)
@EnableAutoConfiguration
@EnableSchemaRegistryClient
public static class AvroSourceApplication {
}
@EnableBinding(Sink.class)
@EnableAutoConfiguration
@EnableSchemaRegistryClient
public static class AvroSinkApplication {
public List<User2> receivedPojos = new ArrayList<>();
@StreamListener(Sink.INPUT)
public void listen(User2 fooPojo) {
receivedPojos.add(fooPojo);
@Before
public void setup() {
schemaRegistryServerContext = SpringApplication
.run(SchemaRegistryServerApplication.class,
"--spring.main.allow-bean-definition-overriding=true");
}
}
@Configuration
public static class NoCacheConfiguration {
@Bean
@StreamMessageConverter
AvroSchemaRegistryClientMessageConverter avroSchemaRegistryClientMessageConverter() {
return new AvroSchemaRegistryClientMessageConverter(new DefaultSchemaRegistryClient(), new NoOpCacheManager());
@After
public void tearDown() {
schemaRegistryServerContext.close();
}
@Bean
ServletWebServerFactory servletWebServerFactory(){
return new TomcatServletWebServerFactory();
@Test
public void testSendMessage() throws Exception {
ConfigurableApplicationContext sourceContext = SpringApplication
.run(AvroSourceApplication.class, "--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.output.contentType=application/*+avro",
"--spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled=true");
Source source = sourceContext.getBean(Source.class);
User1 firstOutboundFoo = new User1();
firstOutboundFoo.setFavoriteColor("foo" + UUID.randomUUID().toString());
firstOutboundFoo.setName("foo" + UUID.randomUUID().toString());
source.output()
.send(MessageBuilder.withPayload(firstOutboundFoo).build());
MessageCollector sourceMessageCollector = sourceContext
.getBean(MessageCollector.class);
Message<?> outboundMessage = sourceMessageCollector
.forChannel(source.output()).poll(1000, TimeUnit.MILLISECONDS);
ConfigurableApplicationContext barSourceContext = SpringApplication
.run(AvroSourceApplication.class, "--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.output.contentType=application/vnd.user1.v1+avro",
"--spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled=true");
Source barSource = barSourceContext.getBean(Source.class);
User2 firstOutboundUser2 = new User2();
firstOutboundUser2.setFavoriteColor("foo" + UUID.randomUUID().toString());
firstOutboundUser2.setName("foo" + UUID.randomUUID().toString());
barSource.output()
.send(MessageBuilder.withPayload(firstOutboundUser2).build());
MessageCollector barSourceMessageCollector = barSourceContext
.getBean(MessageCollector.class);
Message<?> barOutboundMessage = barSourceMessageCollector
.forChannel(barSource.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(barOutboundMessage).isNotNull();
User2 secondBarOutboundPojo = new User2();
secondBarOutboundPojo
.setFavoriteColor("foo" + UUID.randomUUID().toString());
secondBarOutboundPojo.setName("foo" + UUID.randomUUID().toString());
source.output()
.send(MessageBuilder.withPayload(secondBarOutboundPojo).build());
Message<?> secondBarOutboundMessage = sourceMessageCollector
.forChannel(source.output()).poll(1000, TimeUnit.MILLISECONDS);
ConfigurableApplicationContext sinkContext = SpringApplication
.run(AvroSinkApplication.class, "--server.port=0",
"--spring.jmx.enabled=false");
Sink sink = sinkContext.getBean(Sink.class);
sink.input().send(outboundMessage);
sink.input().send(barOutboundMessage);
sink.input().send(secondBarOutboundMessage);
List<User2> receivedPojos = sinkContext
.getBean(AvroSinkApplication.class).receivedPojos;
assertThat(receivedPojos).hasSize(3);
assertThat(receivedPojos.get(0)).isNotSameAs(firstOutboundFoo);
assertThat(receivedPojos.get(0).getFavoriteColor())
.isEqualTo(firstOutboundFoo.getFavoriteColor());
assertThat(receivedPojos.get(0).getName())
.isEqualTo(firstOutboundFoo.getName());
assertThat(receivedPojos.get(0).getFavoritePlace()).isEqualTo("NYC");
assertThat(receivedPojos.get(1)).isNotSameAs(firstOutboundUser2);
assertThat(receivedPojos.get(1).getFavoriteColor())
.isEqualTo(firstOutboundUser2.getFavoriteColor());
assertThat(receivedPojos.get(1).getName())
.isEqualTo(firstOutboundUser2.getName());
assertThat(receivedPojos.get(1).getFavoritePlace()).isEqualTo("Boston");
assertThat(receivedPojos.get(2)).isNotSameAs(secondBarOutboundPojo);
assertThat(receivedPojos.get(2).getFavoriteColor())
.isEqualTo(secondBarOutboundPojo.getFavoriteColor());
assertThat(receivedPojos.get(2).getName())
.isEqualTo(secondBarOutboundPojo.getName());
assertThat(receivedPojos.get(2).getFavoritePlace())
.isEqualTo(secondBarOutboundPojo.getFavoritePlace());
sinkContext.close();
barSourceContext.close();
sourceContext.close();
schemaRegistryServerContext.close();
}
@Test
public void testSchemaImportConfiguration() throws Exception{
final String[] args = { "--server.port=0", "--spring.jmx.enabled=false",
"--spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled=true",
"--spring.cloud.stream.bindings.output.contentType=application/*+avro",
"--spring.cloud.stream.bindings.output.destination=test",
"--spring.cloud.stream.bindings.schema-registry-client.endpoint=http://localhost:8990",
"--spring.cloud.stream.schema.avro.schema-locations=classpath:schemas/Command.avsc",
"--spring.cloud.stream.schema.avro.schema-imports=classpath:schemas/imports/Sms.avsc, classpath:schemas/imports/Email.avsc, classpath:schemas/imports/PushNotification.avsc" };
final ConfigurableApplicationContext sourceContext = SpringApplication
.run(AvroSourceApplication.class, args);
final ConfigurableApplicationContext sinkContext = SpringApplication
.run(CommandSinkApplication.class, args);
final Source barSource = sourceContext.getBean(Source.class);
final Command notification = notification();
barSource.output()
.send(MessageBuilder.withPayload(notification).build());
final MessageCollector barSourceMessageCollector = sourceContext
.getBean(MessageCollector.class);
final Message<?> outboundMessage = barSourceMessageCollector
.forChannel(barSource.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(outboundMessage).isNotNull();
Sink sink = sinkContext.getBean(Sink.class);
sink.input().send(outboundMessage);
List<Command> receivedPojos = sinkContext
.getBean(CommandSinkApplication.class).receivedPojos;
assertThat(receivedPojos).hasSize(1);
assertThat(receivedPojos.get(0)).isEqualTo(notification);
}
@Test
public void testNoCacheConfiguration() {
ConfigurableApplicationContext sourceContext = SpringApplication
.run(NoCacheConfiguration.class,
"--spring.main.web-environment=false");
AvroSchemaRegistryClientMessageConverter converter = sourceContext
.getBean(AvroSchemaRegistryClientMessageConverter.class);
DirectFieldAccessor accessor = new DirectFieldAccessor(converter);
assertThat(accessor.getPropertyValue("cacheManager"))
.isInstanceOf(NoOpCacheManager.class);
}
@EnableBinding(Source.class)
@EnableAutoConfiguration
@EnableSchemaRegistryClient
public static class AvroSourceApplication {
}
@EnableBinding(Sink.class)
@EnableAutoConfiguration
@EnableSchemaRegistryClient
public static class AvroSinkApplication {
public List<User2> receivedPojos = new ArrayList<>();
@StreamListener(Sink.INPUT)
public void listen(User2 fooPojo) {
receivedPojos.add(fooPojo);
}
}
@EnableBinding(Sink.class)
@EnableAutoConfiguration
@EnableSchemaRegistryClient
public static class CommandSinkApplication {
public List<Command> receivedPojos = new ArrayList<>();
@StreamListener(Sink.INPUT)
public void listen(Command fooPojo) {
receivedPojos.add(fooPojo);
}
}
@Configuration
public static class NoCacheConfiguration {
@Bean
@StreamMessageConverter
AvroSchemaRegistryClientMessageConverter avroSchemaRegistryClientMessageConverter() {
return new AvroSchemaRegistryClientMessageConverter(
new DefaultSchemaRegistryClient(),
new NoOpCacheManager());
}
@Bean
ServletWebServerFactory servletWebServerFactory() {
return new TomcatServletWebServerFactory();
}
}
}
}

19
spring-cloud-stream-schema/src/test/resources/schemas/Command.avsc Normal file
View File

@@ -0,0 +1,19 @@
{
"namespace":"example.avro",
"name":"Command",
"type":"record",
"fields":[
{
"name":"type",
"type":"string"
},
{
"name":"correlationId",
"type":"string"
},
{
"name":"payload",
"type":["Sms", "Email", "PushNotification"]
}
]
}

19
spring-cloud-stream-schema/src/test/resources/schemas/imports/Email.avsc Normal file
View File

@@ -0,0 +1,19 @@
{
"namespace":"example.avro",
"name": "Email",
"type": "record",
"fields":[
{
"name":"addressTo",
"type":"string"
},
{
"name":"title",
"type":"string"
},
{
"name":"text",
"type":"string"
}
]
}

15
spring-cloud-stream-schema/src/test/resources/schemas/imports/PushNotification.avsc Normal file
View File

@@ -0,0 +1,15 @@
{
"namespace":"example.avro",
"name": "PushNotification",
"type": "record",
"fields":[
{
"name":"arn",
"type":"string"
},
{
"name":"text",
"type":"string"
}
]
}

14
spring-cloud-stream-schema/src/test/resources/schemas/imports/Sms.avsc Normal file
View File

@@ -0,0 +1,14 @@
{
"namespace":"example.avro",
"name": "Sms",
"type": "record",
"fields":[
{
"name":"phoneNumber",
"type":"string"
},{
"name":"text",
"type":"string"
}
]
}