Finish coordination mechanism for snapshot creation and add snapshot creation test

Author: carlesarnal

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

@@ -139,15 +139,15 @@ public void initialize() {
         }
 
         //Try to restore the internal database from a snapshot
-        String snapshotMessageKey = consumeSnapshotsTopic(snapshotsConsumer);
+        String snapshotId = consumeSnapshotsTopic(snapshotsConsumer);
 
         //Once the topics are created, and the snapshots processed, initialize the internal SQL Storage.
         sqlStore.initialize();
         setDelegate(sqlStore);
 
         //Once the SQL storage has been initialized, start the Kafka consumer thread.
         log.info("SQL store initialized, starting consumer thread.");
-        startConsumerThread(journalConsumer, snapshotMessageKey);
+        startConsumerThread(journalConsumer, snapshotId);
     }
 
     @Override
@@ -233,7 +233,7 @@ private String consumeSnapshotsTopic(KafkaConsumer<String, String> snapshotsCons
      * consuming JournalRecord entries found on that topic, and applying those journal entries to
      * the internal data model.
      */
-    private void startConsumerThread(final KafkaConsumer<KafkaSqlMessageKey, KafkaSqlMessage> consumer, String snapshotMessageKey) {
+    private void startConsumerThread(final KafkaConsumer<KafkaSqlMessageKey, KafkaSqlMessage> consumer, String snapshotId) {
         log.info("Starting KSQL consumer thread on topic: {}", configuration.topic());
         log.info("Bootstrap servers: {}", configuration.bootstrapServers());
 
@@ -257,23 +257,26 @@ private void startConsumerThread(final KafkaConsumer<KafkaSqlMessageKey, KafkaSq
                     if (records != null && !records.isEmpty()) {
                         log.debug("Consuming {} journal records.", records.count());
 
-                        if (null != snapshotMessageKey && !snapshotProcessed) {
+                        if (null != snapshotId && !snapshotProcessed) {
                             //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)) {
+                                if (processSnapshot(snapshotId, record)) {
                                     snapshotProcessed = true;
                                     break;
                                 } else {
-                                    log.info("Discarding message with key {} as it was sent before a snapshot was created", record.key());
+                                    log.debug("Discarding message with key {} as it was sent before a newer snapshot was created", record.key());
                                 }
                             }
 
-                            //Once the snapshot marker message has been found, we can process the rest of the messages as usual, applying the new changes on top of the existing ones in the snapshot.
-                            while (it.hasNext()) {
-                                ConsumerRecord<KafkaSqlMessageKey, KafkaSqlMessage> record = it.next();
-                                processRecord(record, bootstrapId, bootstrapStart);
+                            //If the snapshot marker has not been found, continue with message skipping until we find it.
+                            if (snapshotProcessed) {
+                                //Once the snapshot marker message has been found, we can process the rest of the messages as usual, applying the new changes on top of the existing ones in the snapshot.
+                                while (it.hasNext()) {
+                                    ConsumerRecord<KafkaSqlMessageKey, KafkaSqlMessage> record = it.next();
+                                    processRecord(record, bootstrapId, bootstrapStart);
+                                }
                             }
                         }
                         else {
@@ -291,8 +294,8 @@ private void startConsumerThread(final KafkaConsumer<KafkaSqlMessageKey, KafkaSq
         thread.start();
     }
 
-    private boolean processSnapshot(String snapshotKey, ConsumerRecord<KafkaSqlMessageKey, KafkaSqlMessage> record) {
-        return record.key() != null && snapshotKey.equals(record.key().getUuid());
+    private boolean processSnapshot(String snapshotId, ConsumerRecord<KafkaSqlMessageKey, KafkaSqlMessage> record) {
+        return record.value() instanceof CreateSnapshot1Message && snapshotId.equals(((CreateSnapshot1Message) record.value()).getSnapshotId());
     }
 
     private void processRecord(ConsumerRecord<KafkaSqlMessageKey, KafkaSqlMessage> record, String bootstrapId, long bootstrapStart) {
@@ -869,17 +872,17 @@ 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 message = new CreateSnapshot1Message(path.toString(), snapshotId);
         var uuid = ConcurrentUtil.get(submitter.submitMessage(message));
         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,
+        //Then we send a new message to the snapshots topic, using the snapshot id as the key of the snapshot message.
+        ProducerRecord<String, String> record = new ProducerRecord<>(configuration.snapshotsTopic(), 0, snapshotId, snapshotLocation,
                 Collections.emptyList());
 
         RecordMetadata recordMetadata = ConcurrentUtil.get(snapshotsProducer.apply(record));
 
-        return recordMetadata.toString();
+        return snapshotLocation;
     }
 
     @Override

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

@@ -20,6 +20,7 @@
 public class CreateSnapshot1Message extends AbstractMessage {
 
     private String snapshotLocation;
+    private String snapshotId;
 
     /**
      * @see io.apicurio.registry.storage.impl.kafkasql.KafkaSqlMessage#dispatchTo(io.apicurio.registry.storage.RegistryStorage)

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

@@ -10,6 +10,7 @@
 import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
 import io.quarkus.test.junit.QuarkusTest;
 import io.quarkus.test.junit.TestProfile;
+import jakarta.inject.Inject;
 import lombok.SneakyThrows;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.producer.KafkaProducer;
@@ -20,9 +21,12 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
+import java.io.IOException;
 import java.net.URISyntaxException;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.Collections;
 import java.util.List;
@@ -39,6 +43,9 @@ public class KafkaSqlSnapshotTest extends AbstractResourceTestBase {
 
     private static final String NEW_ARTIFACTS_SNAPSHOT_TEST_GROUP_ID = "SNAPSHOT_TEST_GROUP_ID";
 
+    @Inject
+    KafkaSqlRegistryStorage kafkaSqlRegistryStorage;
+
     @BeforeAll
     public void init() {
         //Create a bunch of artifacts and rules, so they're added on top of the snapshot.
@@ -58,6 +65,15 @@ public void init() {
             clientV3.groups().byGroupId(NEW_ARTIFACTS_SNAPSHOT_TEST_GROUP_ID).artifacts().byArtifactId(artifactId).rules().post(rule);        }
     }
 
+    @Test
+    public void testSnapshotCreation() throws IOException {
+        String snapshotLocation = kafkaSqlRegistryStorage.triggerSnapshotCreation();
+        Path path = Path.of(snapshotLocation);
+        Assertions.assertTrue(Files.exists(path));
+        Files.delete(path);
+    }
+
+
     @Test
     public void testRecoverFromSnapshot() throws InterruptedException {
         //We expect 4001 artifacts coming from the snapshot
@@ -97,8 +113,9 @@ private void prepareSnapshotMarker(Properties props) throws ExecutionException,
             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"
+                    "{\"uuid\":\"1345b402-c707-457e-af76-10c1045e68e8\",\"messageType\":\"CreateSnapshot1Message\",\"partitionKey\":\"__GLOBAL_PARTITION__\"}", "{\n"
                     + "                \"snapshotLocation\": \"/io/apicurio/registry/storage/impl/kafkasql/1302b402-c707-457e-af76-10c1045e68e8.sql\",\n"
+                    + "                \"snapshotId\": \"1302b402-c707-457e-af76-10c1045e68e8\",\n"
                     + "                    \"key\": {\n"
                     + "                \"uuid\": \"1302b402-c707-457e-af76-10c1045e68e8\",\n"
                     + "                        \"messageType\": \"CreateSnapshot1Message\",\n"
@@ -107,7 +124,6 @@ private void prepareSnapshotMarker(Properties props) throws ExecutionException,
                     + "            }", List.of(messageTypeHeader));
 
             //Send snapshot marker
-
             dataProducer.send(snapshotMarkerRecord).get();
         }
 
@@ -137,7 +153,7 @@ private void prepareSnapshotMessages(Properties props) throws URISyntaxException
 
         public Properties connectionProperties() {
             Properties properties = new Properties();
-            properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+            properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, System.getProperty("bootstrap.servers.external"));
             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/readonly/ReadOnlyRegistryStorageTest.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())),
+                entry("triggerSnapshotCreation0", new State(true, RegistryStorage::triggerSnapshotCreation)),
                 entry("createSnapshot1", new State(true, s -> s.createSnapshot(null)))
         );
 

utils/tests/src/main/java/io/apicurio/registry/utils/tests/KafkaTestContainerManager.java

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