Add apicurio json schema serialization support

.github/native-tests.json

@@ -57,7 +57,7 @@
         {
             "category": "Messaging1",
             "timeout": 115,
-            "test-modules": "kafka, kafka-ssl, kafka-sasl, kafka-avro-apicurio2, kafka-snappy, kafka-streams, reactive-messaging-kafka, kafka-oauth-keycloak",
+            "test-modules": "kafka, kafka-ssl, kafka-sasl, kafka-avro-apicurio2, kafka-json-schema-apicurio2, kafka-snappy, kafka-streams, reactive-messaging-kafka, kafka-oauth-keycloak",
             "os-name": "ubuntu-latest"
         },
         {

bom/application/pom.xml

@@ -207,7 +207,7 @@
         <log4j2-api.version>2.22.0</log4j2-api.version>
         <log4j-jboss-logmanager.version>1.3.0.Final</log4j-jboss-logmanager.version>
         <avro.version>1.11.3</avro.version>
-        <apicurio-registry.version>2.5.3.Final</apicurio-registry.version>
+        <apicurio-registry.version>2.5.7.Final</apicurio-registry.version>
         <apicurio-common-rest-client.version>0.1.18.Final</apicurio-common-rest-client.version> <!-- must be the version Apicurio Registry uses -->
         <testcontainers.version>1.19.3</testcontainers.version> <!-- Make sure to also update docker-java.version to match its needs -->
         <docker-java.version>3.3.4</docker-java.version> <!-- must be the version Testcontainers use -->
@@ -1379,6 +1379,16 @@
                 <artifactId>quarkus-apicurio-registry-avro-deployment</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>io.quarkus</groupId>
+                <artifactId>quarkus-apicurio-registry-json-schema</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>io.quarkus</groupId>
+                <artifactId>quarkus-apicurio-registry-json-schema-deployment</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>io.quarkus</groupId>
                 <artifactId>quarkus-confluent-registry-common</artifactId>
@@ -3379,6 +3389,11 @@
                 <artifactId>apicurio-registry-serdes-avro-serde</artifactId>
                 <version>${apicurio-registry.version}</version>
             </dependency>
+            <dependency>
+                <groupId>io.apicurio</groupId>
+                <artifactId>apicurio-registry-serdes-jsonschema-serde</artifactId>
+                <version>${apicurio-registry.version}</version>
+            </dependency>
             <dependency>
                 <groupId>io.apicurio</groupId>
                 <artifactId>apicurio-common-rest-client-vertx</artifactId>

core/deployment/src/main/java/io/quarkus/deployment/Capability.java

@@ -130,6 +130,7 @@ public interface Capability {
 
     String APICURIO_REGISTRY = QUARKUS_PREFIX + ".apicurio.registry";
     String APICURIO_REGISTRY_AVRO = APICURIO_REGISTRY + ".avro";
+    String APICURIO_REGISTRY_JSON_SCHEMA = APICURIO_REGISTRY + ".json";
 
     String CONFLUENT_REGISTRY = QUARKUS_PREFIX + ".confluent.registry";
     String CONFLUENT_REGISTRY_AVRO = CONFLUENT_REGISTRY + ".avro";

core/deployment/src/main/java/io/quarkus/deployment/Feature.java

@@ -13,6 +13,7 @@ public enum Feature {
     AMAZON_LAMBDA,
     AZURE_FUNCTIONS,
     APICURIO_REGISTRY_AVRO,
+    APICURIO_REGISTRY_JSON_SCHEMA,
     AWT,
     CACHE,
     CDI,

devtools/bom-descriptor-json/pom.xml

@@ -200,6 +200,19 @@
                         </exclusion>
                     </exclusions>
                 </dependency>
+                <dependency>
+                    <groupId>io.quarkus</groupId>
+                    <artifactId>quarkus-apicurio-registry-json-schema</artifactId>
+                    <version>${project.version}</version>
+                    <type>pom</type>
+                    <scope>test</scope>
+                    <exclusions>
+                        <exclusion>
+                            <groupId>*</groupId>
+                            <artifactId>*</artifactId>
+                        </exclusion>
+                    </exclusions>
+                </dependency>
                 <dependency>
                     <groupId>io.quarkus</groupId>
                     <artifactId>quarkus-arc</artifactId>

docs/pom.xml

@@ -216,6 +216,19 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-apicurio-registry-json-schema-deployment</artifactId>
+            <version>${project.version}</version>
+            <type>pom</type>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>*</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-arc-deployment</artifactId>

extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java

@@ -356,12 +356,19 @@ private void handleAvro(BuildProducer<ReflectiveClassBuildItem> reflectiveClass,
                     "java.lang.AutoCloseable"));
         }
 
-        // --- Apicurio Registry 2.x ---
+        // --- Apicurio Registry 2.x Avro ---
         if (QuarkusClassLoader.isClassPresentAtRuntime("io.apicurio.registry.serde.avro.AvroKafkaDeserializer")
                 && !capabilities.isPresent(Capability.APICURIO_REGISTRY_AVRO)) {
             throw new RuntimeException(
                     "Apicurio Registry 2.x Avro classes detected, please use the quarkus-apicurio-registry-avro extension");
         }
+
+        // --- Apicurio Registry 2.x Json Schema ---
+        if (QuarkusClassLoader.isClassPresentAtRuntime("io.apicurio.registry.serde.avro.JsonKafkaDeserializer")
+                && !capabilities.isPresent(Capability.APICURIO_REGISTRY_JSON_SCHEMA)) {
+            throw new RuntimeException(
+                    "Apicurio Registry 2.x Json classes detected, please use the quarkus-apicurio-registry-json extension");
+        }
     }
 
     @BuildStep

extensions/schema-registry/apicurio/json-schema/deployment/pom.xml

@@ -0,0 +1,45 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>io.quarkus</groupId>
+        <artifactId>quarkus-apicurio-registry-json-schema-parent</artifactId>
+        <version>999-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>quarkus-apicurio-registry-json-schema-deployment</artifactId>
+    <name>Quarkus - Apicurio Registry - Json Schema - Deployment</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-apicurio-registry-json-schema</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-apicurio-registry-common-deployment</artifactId>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <annotationProcessorPaths>
+                        <path>
+                            <groupId>io.quarkus</groupId>
+                            <artifactId>quarkus-extension-processor</artifactId>
+                            <version>${project.version}</version>
+                        </path>
+                    </annotationProcessorPaths>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

extensions/schema-registry/apicurio/json-schema/deployment/src/main/java/io/quarkus/apicurio/registry/jsonschema/ApicurioRegistryJsonSchemaProcessor.java

@@ -0,0 +1,51 @@
+package io.quarkus.apicurio.registry.jsonschema;
+
+import io.quarkus.bootstrap.classloading.QuarkusClassLoader;
+import io.quarkus.deployment.Feature;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.builditem.ExtensionSslNativeSupportBuildItem;
+import io.quarkus.deployment.builditem.FeatureBuildItem;
+import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
+
+public class ApicurioRegistryJsonSchemaProcessor {
+    @BuildStep
+    FeatureBuildItem feature() {
+        return new FeatureBuildItem(Feature.APICURIO_REGISTRY_JSON_SCHEMA);
+    }
+
+    @BuildStep
+    public void apicurioRegistryJsonSchema(BuildProducer<ReflectiveClassBuildItem> reflectiveClass,
+            BuildProducer<ExtensionSslNativeSupportBuildItem> sslNativeSupport) {
+
+        reflectiveClass
+                .produce(ReflectiveClassBuildItem.builder("io.apicurio.registry.serde.jsonschema.JsonSchemaKafkaDeserializer",
+                        "io.apicurio.registry.serde.jsonschema.JsonSchemaKafkaSerializer").methods().build());
+
+        reflectiveClass.produce(ReflectiveClassBuildItem.builder("io.apicurio.registry.serde.strategy.SimpleTopicIdStrategy",
+                "io.apicurio.registry.serde.strategy.TopicIdStrategy",
+                "io.apicurio.registry.serde.strategy.QualifiedRecordIdStrategy",
+                "io.apicurio.registry.serde.strategy.RecordIdStrategy",
+                "io.apicurio.registry.serde.jsonschema.strategy.TopicRecordIdStrategy").methods().fields()
+                .build());
+
+        reflectiveClass.produce(ReflectiveClassBuildItem.builder("io.apicurio.registry.serde.DefaultIdHandler",
+                "io.apicurio.registry.serde.Legacy4ByteIdHandler",
+                "io.apicurio.registry.serde.fallback.DefaultFallbackArtifactProvider",
+                "io.apicurio.registry.serde.headers.DefaultHeadersHandler").methods().fields()
+                .build());
+
+        String defaultSchemaResolver = "io.apicurio.registry.serde.DefaultSchemaResolver";
+        if (QuarkusClassLoader.isClassPresentAtRuntime(defaultSchemaResolver)) {
+            // Class not present after 2.2.0.Final
+            reflectiveClass.produce(ReflectiveClassBuildItem.builder(defaultSchemaResolver).methods()
+                    .fields().build());
+        }
+    }
+
+    @BuildStep
+    ExtensionSslNativeSupportBuildItem enableSslInNative() {
+        return new ExtensionSslNativeSupportBuildItem(Feature.APICURIO_REGISTRY_JSON_SCHEMA);
+    }
+
+}

extensions/schema-registry/apicurio/json-schema/pom.xml

@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>quarkus-apicurio-registry-parent</artifactId>
+        <groupId>io.quarkus</groupId>
+        <version>999-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>quarkus-apicurio-registry-json-schema-parent</artifactId>
+    <name>Quarkus - Apicurio Registry - Json Schema</name>
+    <packaging>pom</packaging>
+
+    <modules>
+        <module>deployment</module>
+        <module>runtime</module>
+    </modules>
+</project>

extensions/schema-registry/apicurio/json-schema/runtime/pom.xml

@@ -0,0 +1,61 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>io.quarkus</groupId>
+        <artifactId>quarkus-apicurio-registry-json-schema-parent</artifactId>
+        <version>999-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>quarkus-apicurio-registry-json-schema</artifactId>
+    <name>Quarkus - Apicurio Registry - Json Schema - Runtime</name>
+    <description>Use Apicurio as Json schema registry</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.apicurio</groupId>
+            <artifactId>apicurio-registry-serdes-jsonschema-serde</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.apicurio</groupId>
+                    <artifactId>apicurio-common-rest-client-jdk</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-apicurio-registry-common</artifactId>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <!-- Mark this as a runtime dependency, so to make sure it's included on the final classpath during native-image -->
+        <plugins>
+            <plugin>
+                <groupId>io.quarkus</groupId>
+                <artifactId>quarkus-extension-maven-plugin</artifactId>
+                <configuration>
+                    <capabilities>
+                        <provides>io.quarkus.apicurio.registry.json</provides>
+                    </capabilities>
+                </configuration>
+            </plugin>
+            <plugin>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <annotationProcessorPaths>
+                        <path>
+                            <groupId>io.quarkus</groupId>
+                            <artifactId>quarkus-extension-processor</artifactId>
+                            <version>${project.version}</version>
+                        </path>
+                    </annotationProcessorPaths>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+</project>

extensions/schema-registry/apicurio/json-schema/runtime/src/main/resources/META-INF/quarkus-extension.yaml

@@ -0,0 +1,11 @@
+---
+artifact: ${project.groupId}:${project.artifactId}:${project.version}
+name: "Apicurio Registry - Json Schema"
+metadata:
+  keywords:
+  - "apicurio"
+  - "json-schema"
+  guide: ""
+  categories:
+  - "serialization"
+  status: "draft"

extensions/schema-registry/apicurio/pom.xml

@@ -17,5 +17,6 @@
    <modules>
       <module>common</module>
       <module>avro</module>
+      <module>json-schema</module>
    </modules>
 </project>

extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeDiscoveryState.java

@@ -36,7 +36,7 @@ class DefaultSerdeDiscoveryState {
 
     private Boolean hasConfluent;
     private Boolean hasApicurio1;
-    private Boolean hasApicurio2;
+    private Boolean hasApicurio2Avro;
     private Boolean hasJsonb;
 
     DefaultSerdeDiscoveryState(IndexView index) {
@@ -155,18 +155,18 @@ boolean hasApicurio1() {
         return hasApicurio1;
     }
 
-    boolean hasApicurio2() {
-        if (hasApicurio2 == null) {
+    boolean hasApicurio2Avro() {
+        if (hasApicurio2Avro == null) {
             try {
                 Class.forName("io.apicurio.registry.serde.avro.AvroKafkaDeserializer", false,
                         Thread.currentThread().getContextClassLoader());
-                hasApicurio2 = true;
+                hasApicurio2Avro = true;
             } catch (ClassNotFoundException e) {
-                hasApicurio2 = false;
+                hasApicurio2Avro = false;
             }
         }
 
-        return hasApicurio2;
+        return hasApicurio2Avro;
     }
 
     boolean hasJsonb() {

extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/SmallRyeReactiveMessagingKafkaProcessor.java

@@ -880,7 +880,7 @@ private Result serializerDeserializerFor(DefaultSerdeDiscoveryState discovery, T
             int avroLibraries = 0;
             avroLibraries += discovery.hasConfluent() ? 1 : 0;
             avroLibraries += discovery.hasApicurio1() ? 1 : 0;
-            avroLibraries += discovery.hasApicurio2() ? 1 : 0;
+            avroLibraries += discovery.hasApicurio2Avro() ? 1 : 0;
             if (avroLibraries > 1) {
                 LOGGER.debugf("Skipping Avro serde autodetection for %s, because multiple Avro serde libraries are present",
                         typeName);
@@ -897,7 +897,7 @@ private Result serializerDeserializerFor(DefaultSerdeDiscoveryState discovery, T
                         ? Result.of("io.apicurio.registry.utils.serde.AvroKafkaSerializer")
                         : Result.of("io.apicurio.registry.utils.serde.AvroKafkaDeserializer")
                                 .with(isAvroGenerated, "apicurio.registry.use-specific-avro-reader", "true");
-            } else if (discovery.hasApicurio2()) {
+            } else if (discovery.hasApicurio2Avro()) {
                 return serializer
                         ? Result.of("io.apicurio.registry.serde.avro.AvroKafkaSerializer")
                         : Result.of("io.apicurio.registry.serde.avro.AvroKafkaDeserializer")
@@ -908,6 +908,8 @@ private Result serializerDeserializerFor(DefaultSerdeDiscoveryState discovery, T
             }
         }
 
+        //TODO autodiscovery of json serdes
+
         // Jackson-based serializer/deserializer
         // note that Jackson is always present with Kafka, so no need to check
         {