Improve snapshot creation process

Author: carlesarnal

app/src/main/java/io/apicurio/registry/rest/v3/AdminResourceImpl.java

@@ -133,6 +133,12 @@ public List<ArtifactTypeInfo> listArtifactTypes() {
 
     }
 
+    @Override
+    @Authorized(style=AuthorizedStyle.None, level=AuthorizedLevel.Admin)
+    public void triggerSnapshot() {
+        storage.triggerSnapshotCreation();
+    }
+
     /**
      * @see io.apicurio.registry.rest.v3.AdminResource#listGlobalRules()
      */

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

@@ -896,10 +896,9 @@ ArtifactVersionMetaDataDto createArtifactWithMetadata(String groupId, String art
     /**
      *  Triggers a snapshot creation of the internal database.
      *
-     * @param snapshotLocation
      * @throws RegistryStorageException
      */
-    String triggerSnapshotCreation(String snapshotLocation) throws RegistryStorageException;
+    String triggerSnapshotCreation() throws RegistryStorageException;
 
     /**
      *  Creates the snapshot of the internal database based on configuration.

app/src/main/java/io/apicurio/registry/storage/decorator/ReadOnlyRegistryStorageDecorator.java

@@ -471,9 +471,9 @@ public void deleteArtifactBranch(GA ga, BranchId branchId) {
     }
 
     @Override
-    public String triggerSnapshotCreation(String snapshotLocation) throws RegistryStorageException {
+    public String triggerSnapshotCreation() throws RegistryStorageException {
         checkReadOnly();
-        return delegate.triggerSnapshotCreation(snapshotLocation);
+        return delegate.triggerSnapshotCreation();
     }
 
     @Override

app/src/main/java/io/apicurio/registry/storage/decorator/RegistryStorageDecoratorBase.java

@@ -384,8 +384,8 @@ public void deleteArtifactBranch(GA ga, BranchId branchId) {
     }
 
     @Override
-    public String triggerSnapshotCreation(String snapshotLocation) throws RegistryStorageException {
-        return delegate.triggerSnapshotCreation(snapshotLocation);
+    public String triggerSnapshotCreation() throws RegistryStorageException {
+        return delegate.triggerSnapshotCreation();
     }
 
     @Override

app/src/main/java/io/apicurio/registry/storage/impl/gitops/GitOpsRegistryStorage.java

@@ -500,8 +500,8 @@ public List<GAV> getArtifactBranch(GA ga, BranchId branchId, ArtifactRetrievalBe
     }
 
     @Override
-    public String triggerSnapshotCreation(String snapshotLocation) throws RegistryStorageException {
-        return proxy((storage -> storage.triggerSnapshotCreation(snapshotLocation)));
+    public String triggerSnapshotCreation() throws RegistryStorageException {
+        return proxy((RegistryStorage::triggerSnapshotCreation));
     }
 
     @Override

app/src/main/java/io/apicurio/registry/storage/impl/kafkasql/KafkaSqlRegistryStorage.java

@@ -7,11 +7,14 @@
 import java.util.stream.Collectors;
 
 import io.apicurio.registry.storage.impl.kafkasql.messages.*;
+import io.apicurio.registry.utils.kafka.ProducerActions;
 import jakarta.inject.Named;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.errors.TopicExistsException;
 import org.slf4j.Logger;
 
@@ -106,6 +109,10 @@ public class KafkaSqlRegistryStorage extends RegistryStorageDecoratorReadOnlyBas
     @Named("KafkaSqlSnapshotsConsumer")
     KafkaConsumer<String, String> snapshotsConsumer;
 
+    @Inject
+    @Named("KafkaSqlSnapshotsProducer")
+    ProducerActions<String, String> snapshotsProducer;
+
     @Inject
     KafkaSqlSubmitter submitter;
 
@@ -251,14 +258,15 @@ private void startConsumerThread(final KafkaConsumer<KafkaSqlMessageKey, KafkaSq
                         log.debug("Consuming {} journal records.", records.count());
 
                         if (null != snapshotMessageKey && !snapshotProcessed) {
-                            //If there is a snapshot key present, we process (and discard) all the messages until we found the snapshot marker that corresponds to the snapshot key.
+                            //If there is a snapshot key present, we process (and discard) all the messages until we find the snapshot marker that corresponds to the snapshot key.
                             Iterator<ConsumerRecord<KafkaSqlMessageKey, KafkaSqlMessage>> it = records.iterator();
                             while (it.hasNext() && !snapshotProcessed) {
                                 ConsumerRecord<KafkaSqlMessageKey, KafkaSqlMessage> record = it.next();
                                 if (processSnapshot(snapshotMessageKey, record)) {
-                                    log.info("Subscribing to {}", configuration.topic());
                                     snapshotProcessed = true;
                                     break;
+                                } else {
+                                    log.info("Discarding message with key {} as it was sent before a snapshot was created", record.key());
                                 }
                             }
 
@@ -857,10 +865,21 @@ public void deleteArtifactBranch(GA ga, BranchId branchId) {
     }
 
     @Override
-    public String triggerSnapshotCreation(String snapshotLocation) throws RegistryStorageException {
-        var message = new CreateSnapshot1Message(snapshotLocation);
+    public String triggerSnapshotCreation() throws RegistryStorageException {
+        //First we generate an identifier for the snapshot, then we send a snapshot marker to the journal topic.
+        String snapshotId = UUID.randomUUID().toString();
+        Path path = Path.of(configuration.snapshotLocation(), snapshotId + ".sql");
+        var message = new CreateSnapshot1Message(path.toString());
         var uuid = ConcurrentUtil.get(submitter.submitMessage(message));
-        return (String) coordinator.waitForResponse(uuid);
+        String snapshotLocation = (String) coordinator.waitForResponse(uuid);
+
+        //Then we send a new message to the snapshots topic, using the marker uuid as the keyof the snapshot message.
+        ProducerRecord<String, String> record = new ProducerRecord<>(configuration.snapshotsTopic(), 0, uuid.toString(), snapshotLocation,
+                Collections.emptyList());
+
+        RecordMetadata recordMetadata = ConcurrentUtil.get(snapshotsProducer.apply(record));
+
+        return recordMetadata.toString();
     }
 
     @Override

app/src/main/java/io/apicurio/registry/storage/impl/kafkasql/KafkaSqlSnapshotManager.java

@@ -3486,14 +3486,8 @@ public void importArtifactBranch(ArtifactBranchEntity entity) {
     }
 
     @Override
-    public String triggerSnapshotCreation(String location) throws RegistryStorageException {
-        //Snapshot creation is only supported in h2 and kafkasql. In this case, when this is invoked directly, it just creates the snapshot.
-        if ("h2".equals(sqlStatements.dbType())) {
-            return createSnapshot(location);
-        }
-        else {
-            return null;
-        }
+    public String triggerSnapshotCreation() throws RegistryStorageException {
+        throw new RegistryStorageException("Directly triggering the snapshot creation is not supported for sql storages.");
     }
 
     @Override

app/src/main/java/io/apicurio/registry/storage/impl/sql/AbstractSqlRegistryStorage.java

@@ -2801,6 +2801,28 @@
                 "description": "Gets a list of all the configured artifact types.\n\nThis operation can fail for the following reasons:\n\n* A server error occurred (HTTP error `500`)\n"
             }
         },
+        "/admin/config/triggerSnapshot": {
+            "summary": "Triggers a snapshot of the Registry storage. Only supported in KafkaSQL storage",
+            "get": {
+                "tags": [
+                    "KafkaSQL",
+                    "Admin",
+                    "Snapshot"
+                ],
+                "responses": {
+                    "200": {
+                        "content": {},
+                        "description": "Empty content. A 200 means that the snapshot has been successfully triggered."
+                    },
+                    "500": {
+                        "$ref": "#/components/responses/ServerError"
+                    }
+                },
+                "operationId": "triggerSnapshot",
+                "summary": "Trigger storage snapshot",
+                "description": "Triggers the creation of a snapshot of the internal database for compatible storages.\n\nThis operation can fail for the following reasons:\n\n* A server error occurred (HTTP error `500`)\n"
+            }
+        },
         "/groups/{groupId}/artifacts/{artifactId}": {
             "summary": "Manage a single artifact.",
             "get": {

app/src/main/resources-unfiltered/META-INF/resources/api-specifications/registry/v3/openapi.json

@@ -2,6 +2,8 @@
 
 import io.apicurio.registry.AbstractResourceTestBase;
 import io.apicurio.registry.rest.client.models.ArtifactContent;
+import io.apicurio.registry.rest.client.models.Rule;
+import io.apicurio.registry.rest.client.models.RuleType;
 import io.apicurio.registry.utils.kafka.KafkaUtil;
 import io.apicurio.registry.utils.tests.KafkasqlSnapshotTestProfile;
 import io.quarkus.test.common.QuarkusTestResource;
@@ -10,18 +12,20 @@
 import io.quarkus.test.junit.TestProfile;
 import lombok.SneakyThrows;
 import org.apache.kafka.clients.CommonClientConfigs;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.internals.RecordHeader;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
 import java.net.URISyntaxException;
 import java.net.URL;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Paths;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
@@ -37,9 +41,9 @@ public class KafkaSqlSnapshotTest extends AbstractResourceTestBase {
 
     @BeforeAll
     public void init() {
-        //Create a bunch of artifacts so they're added on top of the snapshot
+        //Create a bunch of artifacts and rules, so they're added on top of the snapshot.
         String simpleAvro = resourceToString("avro.json");
-/*
+
         for (int idx = 0; idx < 1000; idx++) {
             System.out.println("Iteration: " + idx);
             String artifactId = UUID.randomUUID().toString();
@@ -48,15 +52,18 @@ public void init() {
             clientV3.groups().byGroupId(NEW_ARTIFACTS_SNAPSHOT_TEST_GROUP_ID).artifacts().post(content, config -> {
                 config.headers.add("X-Registry-ArtifactId", artifactId);
             });
-            clientV3.groups().byGroupId(NEW_ARTIFACTS_SNAPSHOT_TEST_GROUP_ID).artifacts().byArtifactId(artifactId).delete();
-        }
-        */
-
+            Rule rule = new Rule();
+            rule.setType(RuleType.VALIDITY);
+            rule.setConfig("SYNTAX_ONLY");
+            clientV3.groups().byGroupId(NEW_ARTIFACTS_SNAPSHOT_TEST_GROUP_ID).artifacts().byArtifactId(artifactId).rules().post(rule);        }
     }
 
     @Test
     public void testRecoverFromSnapshot() throws InterruptedException {
+        //We expect 4001 artifacts coming from the snapshot
         Assertions.assertEquals(4001, clientV3.groups().byGroupId("default").artifacts().get().getCount());
+        //We expect another 1000 artifacts coming added on top of the snapshot
+        Assertions.assertEquals(1000, clientV3.groups().byGroupId(NEW_ARTIFACTS_SNAPSHOT_TEST_GROUP_ID).artifacts().get().getCount());
     }
 
     protected static class KafkaSqlSnapshotTestInitializer implements QuarkusTestResourceLifecycleManager {
@@ -73,7 +80,6 @@ public int order() {
         @SneakyThrows
         public Map<String, String> start() {
             Properties props = connectionProperties();
-            props.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
 
             KafkaUtil.createTopics(props, Set.of("kafkasql-snapshots", "kafkasql-journal"), Collections.emptyMap());
 
@@ -89,6 +95,7 @@ private void prepareSnapshotMarker(Properties props) throws ExecutionException,
 
             //Create the data producer to send a snapshot marker
             KafkaProducer<String, String> dataProducer = new KafkaProducer<>(props, keySerializer, valueSerializer);
+            RecordHeader messageTypeHeader = new RecordHeader("mt", "CreateSnapshot1Message".getBytes(StandardCharsets.UTF_8));
             ProducerRecord<String, String> snapshotMarkerRecord = new ProducerRecord<>("kafkasql-journal", 0,
                     "{\"uuid\":\"1302b402-c707-457e-af76-10c1045e68e8\",\"messageType\":\"CreateSnapshot1Message\",\"partitionKey\":\"__GLOBAL_PARTITION__\"}", "{\n"
                     + "                \"snapshotLocation\": \"/io/apicurio/registry/storage/impl/kafkasql/1302b402-c707-457e-af76-10c1045e68e8.sql\",\n"
@@ -97,9 +104,10 @@ private void prepareSnapshotMarker(Properties props) throws ExecutionException,
                     + "                        \"messageType\": \"CreateSnapshot1Message\",\n"
                     + "                        \"partitionKey\": \"__GLOBAL_PARTITION__\"\n"
                     + "            }\n"
-                    + "            }", Collections.emptyList());
+                    + "            }", List.of(messageTypeHeader));
 
             //Send snapshot marker
+
             dataProducer.send(snapshotMarkerRecord).get();
         }
 
@@ -129,7 +137,7 @@ private void prepareSnapshotMessages(Properties props) throws URISyntaxException
 
         public Properties connectionProperties() {
             Properties properties = new Properties();
-            properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, System.getProperty("bootstrap.servers.external"));
+            properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
             properties.put(CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG, 10000);
             properties.put(CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG, 5000);
             return properties;

app/src/test/java/io/apicurio/registry/storage/impl/kafkasql/KafkaSqlSnapshotTest.java

@@ -147,7 +147,7 @@ public class ReadOnlyRegistryStorageTest {
                 entry("updateGlobalRule2", new State(true, s -> s.updateGlobalRule(null, null))),
                 entry("updateGroupMetaData2", new State(true, s -> s.updateGroupMetaData(null, null))),
                 entry("updateRoleMapping2", new State(true, s -> s.updateRoleMapping(null, null))),
-                entry("triggerSnapshotCreation1", new State(true, s -> s.triggerSnapshotCreation(null))),
+                entry("triggerSnapshotCreation1", new State(true, s -> s.triggerSnapshotCreation())),
                 entry("createSnapshot1", new State(true, s -> s.createSnapshot(null)))
         );
 

app/src/test/java/io/apicurio/registry/storage/impl/readonly/ReadOnlyRegistryStorageTest.java

@@ -2737,6 +2737,28 @@
                 "description": "Gets a list of all the configured artifact types.\n\nThis operation can fail for the following reasons:\n\n* A server error occurred (HTTP error `500`)\n"
             }
         },
+        "/admin/config/triggerSnapshot": {
+            "summary": "Triggers a snapshot of the Registry storage. Only supported in KafkaSQL storage",
+            "get": {
+                "tags": [
+                    "KafkaSQL",
+                    "Admin",
+                    "Snapshot"
+                ],
+                "responses": {
+                    "200": {
+                        "content": {},
+                        "description": "Empty content. A 200 means that the snapshot has been successfully triggered."
+                    },
+                    "500": {
+                        "$ref": "#/components/responses/ServerError"
+                    }
+                },
+                "operationId": "triggerSnapshot",
+                "summary": "Trigger storage snapshot",
+                "description": "Triggers the creation of a snapshot of the internal database for compatible storages.\n\nThis operation can fail for the following reasons:\n\n* A server error occurred (HTTP error `500`)\n"
+            }
+        },
         "/groups/{groupId}/artifacts/{artifactId}": {
             "summary": "Manage a single artifact.",
             "get": {

common/src/main/resources/META-INF/openapi.json

@@ -38,9 +38,9 @@ public Map<String, String> start() {
             System.setProperty("bootstrap.servers.external", externalBootstrapServers);
 
             return Map.of(
-                    "bootstrap.servers", externalBootstrapServers,
-                    "apicurio.events.kafka.config.bootstrap.servers", externalBootstrapServers,
-                    "apicurio.kafkasql.bootstrap.servers", externalBootstrapServers);
+                    "bootstrap.servers", "localhost:9092",
+                    "apicurio.events.kafka.config.bootstrap.servers", "localhost:9092",
+                    "apicurio.kafkasql.bootstrap.servers", "localhost:9092");
         } else {
             return Collections.emptyMap();
         }