diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java index e5abfafc3..dbc9358a7 100644 --- a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java @@ -170,19 +170,11 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag for (Resource schemaLocation : this.schemaLocations) { try { Schema schema = parseSchema(schemaLocation); - if (this.logger.isInfoEnabled()) { - this.logger.info("Resource " + schemaLocation.getFilename() - + " parsed into schema " + schema.getNamespace() + "." - + schema.getName()); + if (schema.getType().equals(Schema.Type.UNION)) { + schema.getTypes().forEach(innerSchema -> registerSchema(schemaLocation, innerSchema)); + } else { + registerSchema(schemaLocation, schema); } - this.schemaRegistryClient.register(toSubject(schema), AVRO_FORMAT, - schema.toString()); - if (this.logger.isInfoEnabled()) { - this.logger.info("Schema " + schema.getName() - + " registered with id " + schema); - } - this.cacheManager.getCache(REFLECTION_CACHE_NAME) - .put(schema.getNamespace() + "." + schema.getName(), schema); } catch (IOException e) { if (this.logger.isWarnEnabled()) { @@ -200,6 +192,22 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag } } + private void registerSchema(Resource schemaLocation, Schema schema) { + if (this.logger.isInfoEnabled()) { + this.logger.info("Resource " + schemaLocation.getFilename() + + " parsed into schema " + schema.getNamespace() + "." + + schema.getName()); + } + this.schemaRegistryClient.register(toSubject(schema), AVRO_FORMAT, + schema.toString()); + if (this.logger.isInfoEnabled()) { + this.logger.info("Schema " + schema.getName() + + " registered with id " + schema); + } + this.cacheManager.getCache(REFLECTION_CACHE_NAME) + .put(schema.getNamespace() + "." + schema.getName(), schema); + } + protected String toSubject(Schema schema) { return subjectNamingStrategy.toSubject(schema); }