feat: Remove avro fields from schema (#503)

* feat: Remove avro fields from schema

* refactor(core): extract SchemasPostProcessor

* chore(kafka): better avro topic name

* chore(kafka): use kafka kraft in example

No need for zookeeper anymore

* chore(cloud-stream): use kafka kraft in example

No need for zookeeper anymore

* chore(kafka): add akhq as kafka example development container

* chore(kafka): remove zookeeper_jaas.conf as well

* feat(kafka): add avro publication support to example

* chore(core): reduce ExampleJsonGenerator log level

* chore(kafka): fix asyncapi.json

* chore(kafka): rename example kafka container internal listener

* test(cloud-stream): add sasl to test setup

* chore(kafka): extract avro listener to own class

* fix(kafka): kafka example can publish again

* docs(kafka): kafka-schema-registry is not started by default

build.gradle

@@ -44,6 +44,8 @@ allprojects {
         }
 
         mavenCentral()
+
+        maven { url "https://packages.confluent.io/maven/" }
     }
 
     spotless {

dependencies.gradle

@@ -29,12 +29,16 @@ ext {
 
     asyncapiCoreVersion = '1.0.0-EAP-2'
 
+    avroVersion = '1.11.3'
+
     awaitilityVersion = '4.2.0'
 
     commonsLang3Version = '3.14.0'
 
     jsr305Version = '3.0.2'
 
+    kafkaAvroSerializerVersion = '7.5.1'
+
     kafkaClientsVersion = '3.6.1'
     kafkaStreamsVersion = '3.6.1'
 

springwolf-core/src/main/java/io/github/stavshamir/springwolf/SpringwolfAutoConfiguration.java

@@ -17,12 +17,17 @@
 import io.github.stavshamir.springwolf.schemas.SchemasService;
 import io.github.stavshamir.springwolf.schemas.example.ExampleGenerator;
 import io.github.stavshamir.springwolf.schemas.example.ExampleJsonGenerator;
+import io.github.stavshamir.springwolf.schemas.postprocessor.AvroSchemaPostProcessor;
+import io.github.stavshamir.springwolf.schemas.postprocessor.ExampleGeneratorPostProcessor;
+import io.github.stavshamir.springwolf.schemas.postprocessor.SchemasPostProcessor;
+import io.github.stavshamir.springwolf.schemas.postprocessor.SwaggerSchemaPostProcessor;
 import io.swagger.v3.core.converter.ModelConverter;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
+import org.springframework.core.annotation.Order;
 
 import java.util.List;
 
@@ -68,9 +73,9 @@ public ChannelsService channelsService(List<? extends ChannelsScanner> channelsS
     @ConditionalOnMissingBean
     public SchemasService schemasService(
             List<ModelConverter> modelConverters,
-            ExampleGenerator exampleGenerator,
+            List<SchemasPostProcessor> schemaPostProcessors,
             SpringwolfConfigProperties springwolfConfigProperties) {
-        return new DefaultSchemasService(modelConverters, exampleGenerator, springwolfConfigProperties);
+        return new DefaultSchemasService(modelConverters, schemaPostProcessors, springwolfConfigProperties);
     }
 
     @Bean
@@ -79,6 +84,27 @@ public AsyncApiDocketService asyncApiDocketService(SpringwolfConfigProperties sp
         return new DefaultAsyncApiDocketService(springwolfConfigProperties);
     }
 
+    @Bean
+    @ConditionalOnMissingBean
+    @Order(0)
+    public AvroSchemaPostProcessor avroSchemaPostProcessor() {
+        return new AvroSchemaPostProcessor();
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    @Order(10)
+    public ExampleGeneratorPostProcessor exampleGeneratorPostProcessor(ExampleGenerator exampleGenerator) {
+        return new ExampleGeneratorPostProcessor(exampleGenerator);
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    @Order(100)
+    public SwaggerSchemaPostProcessor swaggerSchemaPostProcessor() {
+        return new SwaggerSchemaPostProcessor();
+    }
+
     @Bean
     @ConditionalOnMissingBean
     public ExampleGenerator exampleGenerator() {

springwolf-core/src/main/java/io/github/stavshamir/springwolf/schemas/DefaultSchemasService.java

@@ -3,7 +3,7 @@
 
 import io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.header.AsyncHeaders;
 import io.github.stavshamir.springwolf.configuration.properties.SpringwolfConfigProperties;
-import io.github.stavshamir.springwolf.schemas.example.ExampleGenerator;
+import io.github.stavshamir.springwolf.schemas.postprocessor.SchemasPostProcessor;
 import io.swagger.v3.core.converter.ModelConverter;
 import io.swagger.v3.core.converter.ModelConverters;
 import io.swagger.v3.core.jackson.TypeNameResolver;
@@ -23,18 +23,18 @@
 public class DefaultSchemasService implements SchemasService {
 
     private final ModelConverters converter = ModelConverters.getInstance();
-    private final ExampleGenerator exampleGenerator;
+    private final List<SchemasPostProcessor> schemaPostProcessors;
     private final SpringwolfConfigProperties properties;
 
     private final Map<String, Schema> definitions = new HashMap<>();
 
     public DefaultSchemasService(
             List<ModelConverter> externalModelConverters,
-            ExampleGenerator exampleGenerator,
+            List<SchemasPostProcessor> schemaPostProcessors,
             SpringwolfConfigProperties properties) {
 
         externalModelConverters.forEach(converter::addConverter);
-        this.exampleGenerator = exampleGenerator;
+        this.schemaPostProcessors = schemaPostProcessors;
         this.properties = properties;
     }
 
@@ -107,25 +107,6 @@ private <R> R runWithFqnSetting(Function<Void, R> callable) {
     }
 
     private void postProcessSchema(Schema schema) {
-        generateExampleWhenMissing(schema);
-        removeSwaggerSchemaFields(schema);
-    }
-
-    private void removeSwaggerSchemaFields(Schema schema) {
-        schema.setAdditionalProperties(null);
-
-        Map<String, Schema> properties = schema.getProperties();
-        if (properties != null) {
-            properties.values().forEach(this::removeSwaggerSchemaFields);
-        }
-    }
-
-    private void generateExampleWhenMissing(Schema schema) {
-        if (schema.getExample() == null) {
-            log.debug("Generate example for {}", schema.getName());
-
-            Object example = exampleGenerator.fromSchema(schema, definitions);
-            schema.setExample(example);
-        }
+        schemaPostProcessors.forEach(processor -> processor.process(schema, definitions));
     }
 }

springwolf-core/src/main/java/io/github/stavshamir/springwolf/schemas/example/ExampleJsonGenerator.java

@@ -59,7 +59,7 @@ public Object fromSchema(Schema schema, Map<String, Schema> definitions) {
             String exampleString = buildSchema(schema, definitions);
             return objectMapper.readValue(exampleString, Object.class);
         } catch (JsonProcessingException | ExampleGeneratingException ex) {
-            log.warn("Failed to build json example for schema {}", schema.getName());
+            log.info("Failed to build json example for schema {}", schema.getName(), ex);
         }
         return null;
     }

springwolf-core/src/main/java/io/github/stavshamir/springwolf/schemas/postprocessor/AvroSchemaPostProcessor.java

@@ -0,0 +1,46 @@
+// SPDX-License-Identifier: Apache-2.0
+package io.github.stavshamir.springwolf.schemas.postprocessor;
+
+import io.swagger.v3.oas.models.media.Schema;
+import org.springframework.util.StringUtils;
+
+import java.util.Map;
+
+/**
+ * Removes internal avro fields and classes from the schema.
+ * <br/>
+ * For now, this class is located in springwolf-core as it provides value for the cloud-stream and kafka plugin.
+ * To avoid to many (one-class) add-ons, this class was not moved to yet another artifact - as also no new dependencies are required.
+ * This may change in the future.
+ */
+public class AvroSchemaPostProcessor implements SchemasPostProcessor {
+    private static final String SCHEMA_PROPERTY = "schema";
+    private static final String SPECIFIC_DATA_PROPERTY = "specificData";
+    private static final String SCHEMA_REF = "org.apache.avro.Schema";
+    private static final String SPECIFIC_DAT_REF = "org.apache.avro.specific.SpecificData";
+
+    @Override
+    public void process(Schema schema, Map<String, Schema> definitions) {
+        removeAvroSchemas(definitions);
+        removeAvroProperties(schema);
+    }
+
+    private void removeAvroProperties(Schema schema) {
+        Map<String, Schema> properties = schema.getProperties();
+        if (properties != null) {
+            Schema schemaPropertySchema = properties.getOrDefault(SCHEMA_PROPERTY, null);
+            Schema specificDataPropertySchema = properties.getOrDefault(SPECIFIC_DATA_PROPERTY, null);
+            if (schemaPropertySchema != null && specificDataPropertySchema != null) {
+                if (StringUtils.endsWithIgnoreCase(schemaPropertySchema.get$ref(), SCHEMA_REF)
+                        && StringUtils.endsWithIgnoreCase(specificDataPropertySchema.get$ref(), SPECIFIC_DAT_REF)) {
+                    properties.remove(SCHEMA_PROPERTY);
+                    properties.remove(SPECIFIC_DATA_PROPERTY);
+                }
+            }
+        }
+    }
+
+    private void removeAvroSchemas(Map<String, Schema> definitions) {
+        definitions.entrySet().removeIf(entry -> StringUtils.startsWithIgnoreCase(entry.getKey(), "org.apache.avro"));
+    }
+}

springwolf-core/src/main/java/io/github/stavshamir/springwolf/schemas/postprocessor/ExampleGeneratorPostProcessor.java

@@ -0,0 +1,25 @@
+// SPDX-License-Identifier: Apache-2.0
+package io.github.stavshamir.springwolf.schemas.postprocessor;
+
+import io.github.stavshamir.springwolf.schemas.example.ExampleGenerator;
+import io.swagger.v3.oas.models.media.Schema;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Map;
+
+@RequiredArgsConstructor
+@Slf4j
+public class ExampleGeneratorPostProcessor implements SchemasPostProcessor {
+    private final ExampleGenerator exampleGenerator;
+
+    @Override
+    public void process(Schema schema, Map<String, Schema> definitions) {
+        if (schema.getExample() == null) {
+            log.debug("Generate example for {}", schema.getName());
+
+            Object example = exampleGenerator.fromSchema(schema, definitions);
+            schema.setExample(example);
+        }
+    }
+}

springwolf-core/src/main/java/io/github/stavshamir/springwolf/schemas/postprocessor/SchemasPostProcessor.java

@@ -0,0 +1,15 @@
+// SPDX-License-Identifier: Apache-2.0
+package io.github.stavshamir.springwolf.schemas.postprocessor;
+
+import io.swagger.v3.oas.models.media.Schema;
+
+import java.util.Map;
+
+/**
+ * Internal interface to allow post-processing of a new schema (and their definition) after detection.
+ * <br/>
+ * It is closely coupled with the data structure of the SchemaService.
+ */
+public interface SchemasPostProcessor {
+    void process(Schema schema, Map<String, Schema> definitions);
+}

springwolf-core/src/main/java/io/github/stavshamir/springwolf/schemas/postprocessor/SwaggerSchemaPostProcessor.java

@@ -0,0 +1,22 @@
+// SPDX-License-Identifier: Apache-2.0
+package io.github.stavshamir.springwolf.schemas.postprocessor;
+
+import io.swagger.v3.oas.models.media.Schema;
+
+import java.util.Map;
+
+public class SwaggerSchemaPostProcessor implements SchemasPostProcessor {
+    @Override
+    public void process(Schema schema, Map<String, Schema> definitions) {
+        removeAdditionalProperties(schema);
+    }
+
+    private void removeAdditionalProperties(Schema schema) {
+        schema.setAdditionalProperties(null);
+
+        Map<String, Schema> properties = schema.getProperties();
+        if (properties != null) {
+            properties.values().forEach((property) -> removeAdditionalProperties(property));
+        }
+    }
+}

springwolf-core/src/test/java/io/github/stavshamir/springwolf/asyncapi/scanners/channels/annotation/AsyncAnnotationChannelsScannerTest.java

@@ -23,7 +23,6 @@
 import io.github.stavshamir.springwolf.configuration.properties.SpringwolfConfigProperties;
 import io.github.stavshamir.springwolf.schemas.DefaultSchemasService;
 import io.github.stavshamir.springwolf.schemas.SchemasService;
-import io.github.stavshamir.springwolf.schemas.example.ExampleJsonGenerator;
 import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.Data;
 import lombok.NoArgsConstructor;
@@ -73,8 +72,7 @@ public OperationData.OperationType getOperationType() {
             };
     private final SpringwolfConfigProperties properties = new SpringwolfConfigProperties();
     private final ClassScanner classScanner = mock(ClassScanner.class);
-    private final SchemasService schemasService =
-            new DefaultSchemasService(emptyList(), new ExampleJsonGenerator(), properties);
+    private final SchemasService schemasService = new DefaultSchemasService(emptyList(), emptyList(), properties);
     private final AsyncApiDocketService asyncApiDocketService = mock(AsyncApiDocketService.class);
     private final PayloadClassExtractor payloadClassExtractor = new PayloadClassExtractor(properties);
 

springwolf-core/src/test/java/io/github/stavshamir/springwolf/schemas/DefaultSchemasServiceTest.java

@@ -7,15 +7,18 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
 import io.github.stavshamir.springwolf.configuration.properties.SpringwolfConfigProperties;
-import io.github.stavshamir.springwolf.schemas.example.ExampleGenerator;
 import io.github.stavshamir.springwolf.schemas.example.ExampleJsonGenerator;
+import io.github.stavshamir.springwolf.schemas.postprocessor.ExampleGeneratorPostProcessor;
+import io.github.stavshamir.springwolf.schemas.postprocessor.SchemasPostProcessor;
+import io.github.stavshamir.springwolf.schemas.postprocessor.SwaggerSchemaPostProcessor;
 import io.swagger.v3.core.util.Json;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.annotation.Nullable;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -30,12 +33,18 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
 
 class DefaultSchemasServiceTest {
-
-    private final ExampleGenerator exampleGenerator = new ExampleJsonGenerator();
-    private final SchemasService schemasService =
-            new DefaultSchemasService(List.of(), exampleGenerator, new SpringwolfConfigProperties());
+    private final SchemasPostProcessor schemasPostProcessor = Mockito.mock(SchemasPostProcessor.class);
+    private final SchemasService schemasService = new DefaultSchemasService(
+            List.of(),
+            List.of(
+                    new ExampleGeneratorPostProcessor(new ExampleJsonGenerator()),
+                    schemasPostProcessor,
+                    new SwaggerSchemaPostProcessor()),
+            new SpringwolfConfigProperties());
 
     private static final ObjectMapper objectMapper =
             Json.mapper().enable(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS);
@@ -111,7 +120,7 @@ void getDefinitionWithFqnClassName() throws IOException {
         SpringwolfConfigProperties properties = new SpringwolfConfigProperties();
         properties.setUseFqn(true);
 
-        SchemasService schemasServiceWithFqn = new DefaultSchemasService(List.of(), exampleGenerator, properties);
+        SchemasService schemasServiceWithFqn = new DefaultSchemasService(List.of(), List.of(), properties);
 
         // when
         Class<?> clazz =
@@ -127,6 +136,13 @@ void getDefinitionWithFqnClassName() throws IOException {
         assertThat(fqnClassName.length()).isGreaterThan(clazz.getSimpleName().length());
     }
 
+    @Test
+    void postProcessorsAreCalled() {
+        schemasService.register(FooWithEnum.class);
+
+        verify(schemasPostProcessor).process(any(), any());
+    }
+
     private String jsonResource(String path) throws IOException {
         InputStream s = this.getClass().getResourceAsStream(path);
         return new String(s.readAllBytes(), StandardCharsets.UTF_8);

springwolf-core/src/test/java/io/github/stavshamir/springwolf/schemas/postprocessor/AvroSchemaPostProcessorTest.java

@@ -0,0 +1,40 @@
+// SPDX-License-Identifier: Apache-2.0
+package io.github.stavshamir.springwolf.schemas.postprocessor;
+
+import io.swagger.v3.oas.models.media.StringSchema;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class AvroSchemaPostProcessorTest {
+    SchemasPostProcessor processor = new AvroSchemaPostProcessor();
+
+    @Test
+    void avroSchemasAreRemovedTest() {
+        // given
+        var avroSchema = new io.swagger.v3.oas.models.media.Schema();
+        avroSchema.set$ref("#/components/schemas/org.apache.avro.Schema");
+
+        var avroSpecificData = new io.swagger.v3.oas.models.media.Schema();
+        avroSpecificData.set$ref("#/components/schemas/org.apache.avro.specific.SpecificData");
+
+        var schema = new io.swagger.v3.oas.models.media.Schema();
+        schema.setProperties(new HashMap<>(
+                Map.of("foo", new StringSchema(), "schema", avroSchema, "specificData", avroSpecificData)));
+
+        var definitions = new HashMap<String, io.swagger.v3.oas.models.media.Schema>();
+        definitions.put("customClassRefUnusedInThisTest", new StringSchema());
+        definitions.put("org.apache.avro.Schema", new io.swagger.v3.oas.models.media.Schema());
+        definitions.put("org.apache.avro.ConversionJava.lang.Object", new io.swagger.v3.oas.models.media.Schema());
+
+        // when
+        processor.process(schema, definitions);
+
+        // then
+        assertThat(schema.getProperties()).isEqualTo(Map.of("foo", new StringSchema()));
+        assertThat(definitions).isEqualTo(Map.of("customClassRefUnusedInThisTest", new StringSchema()));
+    }
+}