Updated Kafka + SQL storage variant, re-using existing kafka storage utils/data model

.github/workflows/integration-tests.yaml

@@ -39,13 +39,15 @@ jobs:
           restore-keys: |
             ${{ runner.os }}-maven-
       - name: Build All Variants
-        run: mvn clean install -Pprod -Psql -Pinfinispan -Pstreams -Pasyncmem -DskipTests -pl !tests
+        run: mvn clean install -Pprod -Psql -Pinfinispan -Pstreams -Pasyncmem -Pksql -DskipTests -pl !tests
       - name: Run Integration Tests - streams
         run: mvn verify -Pacceptance -Pstreams -pl tests -Dmaven.javadoc.skip=true --no-transfer-progress -PdisableSerdesTest
       - name: Run Integration Tests - sql
         run: mvn verify -Pacceptance -Psql -pl tests -Dmaven.javadoc.skip=true --no-transfer-progress -PdisableSerdesTest
       - name: Run Integration Tests - infinispan
         run: mvn verify -Pacceptance -Pinfinispan -pl tests -Dmaven.javadoc.skip=true --no-transfer-progress -PdisableSerdesTest
+      - name: Run Integration Tests - ksql
+        run: mvn verify -Pacceptance -Pksql -pl tests -Dmaven.javadoc.skip=true --no-transfer-progress -PdisableSerdesTest
       - name: Collect logs
         if: failure()
         run: ./.github/scripts/collect_logs.sh

.github/workflows/verify.yaml

@@ -38,7 +38,7 @@ jobs:
           restore-keys: |
             ${{ runner.os }}-maven-
       - name: Build All Variants
-        run: mvn clean install -Pprod -Psql -Pinfinispan -Pstreams -Pasyncmem -pl !tests
+        run: mvn clean install -Pprod -Psql -Pinfinispan -Pstreams -Pasyncmem -Pksql -pl !tests
       - name: Login to DockerHub Registry
         if: github.event_name == 'push' && github.ref == 'refs/heads/master'
         run: echo ${{ secrets.DOCKERHUB_PASSWORD }} | docker login -u ${{ secrets.DOCKERHUB_USERNAME }} --password-stdin
@@ -51,6 +51,7 @@ jobs:
           mvn package -Pprod -Pinfinispan -DskipTests -Ddocker -Ddocker.tag.name=latest-snapshot
           mvn package -Pprod -Pstreams -DskipTests -Ddocker -Ddocker.tag.name=latest-snapshot
           mvn package -Pprod -Pasyncmem -DskipTests -Ddocker -Ddocker.tag.name=latest-snapshot
+          mvn package -Pprod -Pksql -DskipTests -Ddocker -Ddocker.tag.name=latest-snapshot
       - name: Push The Tagged Docker Images
         if: github.event_name == 'push' && github.ref == 'refs/heads/master'
         run: |
@@ -59,6 +60,7 @@ jobs:
           docker push apicurio/apicurio-registry-infinispan:latest-snapshot
           docker push apicurio/apicurio-registry-streams:latest-snapshot
           docker push apicurio/apicurio-registry-asyncmem:latest-snapshot
+          docker push apicurio/apicurio-registry-ksql:latest-snapshot
 
   kubernetes-tests:
     name: Kubernetes Tests

README.md

@@ -47,9 +47,9 @@ The following parameters are available for executable files:
 
 |Option|Command argument|Env. variable|
 |---|---|---|
-|Data Source URL|`-Dquarkus.datasource.jdbc.url`|`QUARKUS_DATASOURCE_URL`|
-|DS Username|`-Dquarkus.datasource.username`|`QUARKUS_DATASOURCE_USERNAME`|
-|DS Password|`-Dquarkus.datasource.password`|`QUARKUS_DATASOURCE_PASSWORD`|
+|Data Source URL|`-Dquarkus.datasource.jdbc.url`|`REGISTRY_DATASOURCE_URL`|
+|DS Username|`-Dquarkus.datasource.username`|`REGISTRY_DATASOURCE_USERNAME`|
+|DS Password|`-Dquarkus.datasource.password`|`REGISTRY_DATASOURCE_PASSWORD`|
 
 To see additional options, visit:
  - [Data Source config](https://quarkus.io/guides/datasource) 
@@ -141,9 +141,9 @@ services:
     ports:
       - 8080:8080
     environment:
-      QUARKUS_DATASOURCE_URL: 'jdbc:postgresql://postgres/apicurio-registry'
-      QUARKUS_DATASOURCE_USERNAME: apicurio-registry
-      QUARKUS_DATASOURCE_PASSWORD: password
+      REGISTRY_DATASOURCE_URL: 'jdbc:postgresql://postgres/apicurio-registry'
+      REGISTRY_DATASOURCE_USERNAME: apicurio-registry
+      REGISTRY_DATASOURCE_PASSWORD: password
 ```
   - Run `docker-compose -f test.yml up`
 

app/.gitignore

@@ -0,0 +1,2 @@
+/.apt_generated/
+/.apt_generated_tests/

app/src/main/java/io/apicurio/registry/logging/LoggingInterceptor.java

@@ -51,12 +51,24 @@ public Object logMethodEntry(InvocationContext context) throws Exception {
         } catch (Throwable t) {
         }
 
-        logger.debug("ENTERING method [{}] with {} parameters", context.getMethod().getName(), context.getParameters().length);
+        logEnter(context, logger);
         Object rval = context.proceed();
-        logger.debug("LEAVING method [{}]", context.getMethod().getName());
+        logLeave(context, logger);
         return rval;
     }
 
+    private void logEnter(InvocationContext context, Logger logger) {
+        if (context != null && context.getMethod() != null && context.getMethod().getName() != null) {
+            logger.debug("ENTERING method [{}] with {} parameters", context.getMethod().getName(), context.getParameters().length);
+        }
+    }
+
+    private void logLeave(InvocationContext context, Logger logger) {
+        if (context != null && context.getMethod() != null && context.getMethod().getName() != null) {
+            logger.debug("LEAVING method [{}]", context.getMethod().getName());
+        }
+    }
+
     /**
      * Gets a logger for the given target class.
      * @param targetClass

app/src/main/java/io/apicurio/registry/storage/ArtifactNotFoundException.java

@@ -25,10 +25,6 @@ public class ArtifactNotFoundException extends NotFoundException {
 
     private String artifactId;
 
-    public ArtifactNotFoundException(Throwable cause) {
-        super(cause);
-    }
-
     /**
      * Constructor.
      */

app/src/main/java/io/apicurio/registry/storage/RegistryStorageProducer.java

@@ -40,25 +40,35 @@ public class RegistryStorageProducer {
     @Inject
     Instance<RegistryStorage> storages;
 
+    @Inject
+    Instance<RegistryStorageProvider> provider;
+
     @Inject
     EventsService eventsService;
 
     @Produces
     @ApplicationScoped
     @Current
     public RegistryStorage realImpl() {
-        List<RegistryStorage> list = storages.stream().collect(Collectors.toList());
+
         RegistryStorage impl = null;
-        if (list.size() == 1) {
-            impl = list.get(0);
+
+        if (provider.isResolvable()) {
+            impl= provider.get().storage();
         } else {
-            for (RegistryStorage rs : list) {
-                if (rs instanceof InMemoryRegistryStorage == false) {
-                    impl = rs;
-                    break;
+            List<RegistryStorage> list = storages.stream().collect(Collectors.toList());
+            if (list.size() == 1) {
+                impl = list.get(0);
+            } else {
+                for (RegistryStorage rs : list) {
+                    if (rs instanceof InMemoryRegistryStorage == false) {
+                        impl = rs;
+                        break;
+                    }
                 }
             }
         }
+
         if (impl != null) {
             log.info(String.format("Using RegistryStore: %s", impl.getClass().getName()));
             if (eventsService.isConfigured()) {
@@ -67,6 +77,7 @@ public RegistryStorage realImpl() {
                 return impl;
             }
         }
-        throw new IllegalStateException("Should not be here ... ?!");
+
+        throw new IllegalStateException("No RegistryStorage available on the classpath!");
     }
 }

app/src/main/java/io/apicurio/registry/storage/RegistryStorageProvider.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2020 Red Hat
+ *
+ * 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
+ *
+ *      http://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 io.apicurio.registry.storage;
+
+/**
+ * @author Fabian Martinez
+ */
+public interface RegistryStorageProvider {
+
+    RegistryStorage storage();
+
+}

app/src/main/resources/application.properties

@@ -5,54 +5,9 @@ quarkus.index-dependency.jaxrs.artifact-id=jboss-jaxrs-api_2.1_spec
 
 # === Dev profile - see README
 %dev.quarkus.http.port=${HTTP_PORT:8080}
-%dev.quarkus.log.level=${LOG_LEVEL:DEBUG}
+%dev.quarkus.log.level=${LOG_LEVEL:INFO}
+%dev.quarkus.log.category."io.apicurio".level=${REGISTRY_LOG_LEVEL:DEBUG}
 %dev.quarkus.log.console.enable=true
-%dev.quarkus.datasource.db-kind=postgresql
-%dev.quarkus.datasource.jdbc.url=${QUARKUS_DATASOURCE_URL:jdbc:postgresql://localhost:5432/registry}
-%dev.quarkus.datasource.username=${QUARKUS_DATASOURCE_USERNAME:postgres}
-%dev.quarkus.datasource.password=${QUARKUS_DATASOURCE_PASSWORD:postgres}
-%dev.quarkus.datasource.jdbc.initial-size=20
-%dev.quarkus.datasource.jdbc.min-size=20
-%dev.quarkus.datasource.jdbc.max-size=100
-%dev.registry.sql.init=true
-
-# Kafka -- when used
-%dev.registry.kafka.common.bootstrap.servers=${bootstrap.servers:localhost:9092}
-# Kafka - Registry producer
-%dev.registry.kafka.storage-producer.enable.idempotence=true
-#%dev.registry.kafka.storage-producer.max.in.flight.requests.per.connection=5
-%dev.registry.kafka.storage-producer.retries=3
-%dev.registry.kafka.storage-producer.acks=all
-# Kafka - Snapshot handling
-%dev.registry.kafka.snapshot-producer.enable.idempotence=true
-%dev.registry.kafka.snapshot-producer.retries=3
-%dev.registry.kafka.snapshot-producer.acks=all
-%dev.registry.kafka.snapshot.requests=1000
-
-# Streams -- when used
-%dev.registry.streams.common.bootstrap.servers=${bootstrap.servers:localhost:9092}
-%dev.registry.streams.topology.application.id=apicurio-registry
-%dev.registry.streams.topology.application.server=localhost:9000
-%dev.registry.streams.topology.num.stream.threads=2
-%dev.registry.streams.topology.num.standby.replicas=1
-%dev.registry.streams.topology.processing.guarantee=exactly_once
-%dev.registry.streams.topology.replication.factor=1
-%dev.registry.streams.topology.global.id.topic=global-id-topic
-%dev.registry.streams.topology.storage.topic=storage-topic
-%dev.registry.streams.storage-producer.enable.idempotence=true
-#%dev.registry.streams.storage-producer.max.in.flight.requests.per.connection=5
-%dev.registry.streams.storage-producer.retries=3
-%dev.registry.streams.storage-producer.acks=all
-%dev.registry.streams.events.enable.idempotence=true
-%dev.registry.streams.events.retries=3
-%dev.registry.streams.events.acks=all
-
-# Asyncmem -- when used
-%dev.registry.asyncmem.delays.create=${ASYNC_DELAYS_CREATE:1000}
-%dev.registry.asyncmem.delays.update=${ASYNC_DELAYS_UPDATE:1000}
-
-# Infinispan
-%dev.registry.infinispan.cluster.name=${INFINISPAN_CLUSTER_NAME:apicurio-registry}
 
 # === Prod profile - see README
 # HTTP Port
@@ -61,52 +16,7 @@ quarkus.index-dependency.jaxrs.artifact-id=jboss-jaxrs-api_2.1_spec
 # Log
 %prod.quarkus.log.level=${LOG_LEVEL:INFO}
 %prod.quarkus.log.console.enable=true
-
-# SQL
-%prod.quarkus.datasource.db-kind=postgresql
-%prod.quarkus.datasource.jdbc.url=${QUARKUS_DATASOURCE_URL}
-%prod.quarkus.datasource.username=${QUARKUS_DATASOURCE_USERNAME:postgres}
-%prod.quarkus.datasource.password=${QUARKUS_DATASOURCE_PASSWORD:postgres}
-%prod.quarkus.datasource.jdbc.initial-size=20
-%prod.quarkus.datasource.jdbc.min-size=20
-%prod.quarkus.datasource.jdbc.max-size=100
-%prod.registry.sql.init=true
-
-# Kafka
-%prod.registry.kafka.common.bootstrap.servers=${KAFKA_BOOTSTRAP_SERVERS}
-# Kafka - Registry producer
-%prod.registry.kafka.storage-producer.enable.idempotence=true
-#%prod.registry.kafka.storage-producer.max.in.flight.requests.per.connection=5
-%prod.registry.kafka.storage-producer.retries=3
-%prod.registry.kafka.storage-producer.acks=all
-
-# Kafka - Snapshot handling
-%prod.registry.kafka.snapshot-producer.enable.idempotence=true
-%prod.registry.kafka.snapshot-producer.retries=3
-%prod.registry.kafka.snapshot-producer.acks=all
-%prod.registry.kafka.snapshot.requests=1000
-
-# Streams -- when used
-%prod.registry.streams.common.bootstrap.servers=${KAFKA_BOOTSTRAP_SERVERS}
-%prod.registry.streams.topology.application.id=${APPLICATION_ID}
-%prod.registry.streams.topology.application.server=${APPLICATION_SERVER_HOST:localhost}:${APPLICATION_SERVER_PORT:9000}
-%prod.registry.streams.topology.num.stream.threads=2
-%prod.registry.streams.topology.num.standby.replicas=1
-%prod.registry.streams.topology.processing.guarantee=exactly_once
-%prod.registry.streams.topology.replication.factor=1
-%prod.registry.streams.topology.global.id.topic=global-id-topic
-%prod.registry.streams.topology.storage.topic=storage-topic
-%prod.registry.streams.storage-producer.enable.idempotence=true
-#%prod.registry.streams.storage-producer.max.in.flight.requests.per.connection=5
-%prod.registry.streams.storage-producer.retries=3
-%prod.registry.streams.storage-producer.acks=all
-
-# Asyncmem -- when used
-%prod.registry.asyncmem.delays.create=${ASYNC_DELAYS_CREATE:1000}
-%prod.registry.asyncmem.delays.update=${ASYNC_DELAYS_UPDATE:1000}
-
-# Infinispan
-%prod.registry.infinispan.cluster.name=${INFINISPAN_CLUSTER_NAME:apicurio-registry}
+%prod.quarkus.log.category."io.apicurio".level=${REGISTRY_LOG_LEVEL:INFO}
 
 # Search
 # This currently disables search --> noop, remove once search is enabled

app/src/test/java/io/apicurio/registry/RegistryClientTest.java

@@ -92,6 +92,7 @@ public void testAsyncCRUD(RegistryRestClient restClient) throws Exception {
 
     @RegistryRestClientTest
     void testSearchArtifact(RegistryRestClient restClient) throws Exception {
+        try {
         // warm-up
         restClient.listArtifacts();
 
@@ -106,11 +107,6 @@ void testSearchArtifact(RegistryRestClient restClient) throws Exception {
 
         this.waitForGlobalId(id);
 
-        retry(() -> {
-            ArtifactMetaData artifactMetaData = restClient.getArtifactMetaDataByGlobalId(id);
-            Assertions.assertNotNull(artifactMetaData);
-        });
-
         ArtifactSearchResults results = restClient.searchArtifacts(name, SearchOver.name, SortOrder.asc, 0, 2);
         Assertions.assertNotNull(results);
         Assertions.assertEquals(1, results.getCount());
@@ -121,6 +117,10 @@ void testSearchArtifact(RegistryRestClient restClient) throws Exception {
         results = restClient.searchArtifacts(null, null, null, null, null);
         Assertions.assertNotNull(results);
         Assertions.assertTrue(results.getCount() > 0);
+        } catch (Exception e) {
+            e.printStackTrace();
+            throw e;
+        }
     }
 
     @RegistryRestClientTest
@@ -293,13 +293,8 @@ public void testLabels(RegistryRestClient restClient) throws Exception {
             });
 
             retry((() -> {
-                System.out.println("==============================");
                 ArtifactSearchResults results = client
                         .searchArtifacts("registry-client-test-testLabels", SearchOver.labels, SortOrder.asc, 0, 2);
-                results.getArtifacts().forEach(arty -> {
-                    System.out.println(arty);
-                });
-                System.out.println("==============================");
                 Assertions.assertNotNull(results);
                 Assertions.assertEquals(1, results.getCount());
                 Assertions.assertEquals(1, results.getArtifacts().size());