Move the data upgrade process to the v2 endpoint and improve references management

Author: carlesarnal

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

@@ -38,7 +38,7 @@
 import io.apicurio.registry.types.RuleType;
 import io.apicurio.registry.types.provider.ArtifactTypeUtilProviderFactory;
 import io.apicurio.registry.utils.impexp.Entity;
-import io.apicurio.registry.utils.impexp.v3.EntityReader;
+import io.apicurio.registry.utils.impexp.v2.EntityReader;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
 import jakarta.interceptor.Interceptors;
@@ -264,7 +264,7 @@ public void close() throws IOException {
                 zip.close();
             }
         };
-        this.storage.importData(stream, isNullOrTrue(xRegistryPreserveGlobalId),
+        this.storage.upgradeData(stream, isNullOrTrue(xRegistryPreserveGlobalId),
                 isNullOrTrue(xRegistryPreserveContentId));
     }
 

app/src/main/java/io/apicurio/registry/storage/importing/v2/SqlDataUpgrader.java

@@ -5,6 +5,7 @@
 import io.apicurio.registry.model.GAV;
 import io.apicurio.registry.storage.RegistryStorage;
 import io.apicurio.registry.storage.dto.ArtifactReferenceDto;
+import io.apicurio.registry.storage.dto.EditableArtifactMetaDataDto;
 import io.apicurio.registry.storage.error.InvalidArtifactTypeException;
 import io.apicurio.registry.storage.error.VersionAlreadyExistsException;
 import io.apicurio.registry.storage.impexp.EntityInputStream;
@@ -61,6 +62,12 @@ public class SqlDataUpgrader extends AbstractDataImporter {
     protected final Map<Long, Long> globalIdMapping = new HashMap<>();
     protected final Map<Long, Long> contentIdMapping = new HashMap<>();
 
+    // Collection of content waiting for required references. A given content cannot be imported unless the
+    // expected reference is present.
+    // TODO do a second round to this, since this currently means enforcing the integrity rule. (Maybe try a
+    // first round and, if there are no artifacts remaining, just import the orphaned content).
+    protected final Map<ContentEntity, Set<GAV>> waitingForReference = new HashMap<>();
+
     // To keep track of which versions have been imported
     private final Set<GAV> gavDone = new HashSet<>();
 
@@ -126,7 +133,7 @@ public void importArtifactVersion(ArtifactVersionEntity entity) {
                     .groupId(entity.groupId).build();
 
             // If the version being imported is the first one, we have to create the artifact first
-            if (entity.versionId == 1) {
+            if (!storage.isArtifactExists(entity.groupId, entity.artifactId)) {
                 ArtifactEntity artifactEntity = ArtifactEntity.builder().artifactId(entity.artifactId)
                         .artifactType(entity.artifactType).createdOn(entity.createdOn)
                         .description(entity.description).groupId(entity.groupId).labels(artifactVersionLabels)
@@ -135,6 +142,16 @@ public void importArtifactVersion(ArtifactVersionEntity entity) {
                 storage.importArtifact(artifactEntity);
             }
 
+            if (entity.isLatest) {
+                // If this version is the latest, update the artifact metadata with its metadata
+                EditableArtifactMetaDataDto editableArtifactMetaDataDto = EditableArtifactMetaDataDto
+                        .builder().name(newEntity.name).owner(newEntity.owner)
+                        .description(newEntity.description).labels(newEntity.labels).build();
+
+                storage.updateArtifactMetaData(newEntity.groupId, newEntity.artifactId,
+                        editableArtifactMetaDataDto);
+            }
+
             storage.importArtifactVersion(newEntity);
             log.debug("Artifact version imported successfully: {}", entity);
             globalIdMapping.put(oldGlobalId, entity.globalId);
@@ -149,6 +166,20 @@ public void importArtifactVersion(ArtifactVersionEntity entity) {
             }
             waitingForVersion.removeAll(commentsToImport);
 
+            // Once the artifact version is processed, check if there is some content waiting for this as it's
+            // reference
+            // For each content waiting for the version we just inserted, remove it from the list.
+            waitingForReference.values().forEach(waitingReferences -> waitingReferences.remove(gav));
+
+            // Finally, once the list of required deps is updated, if it was the last reference needed, import
+            // the content.
+            waitingForReference.keySet().stream()
+                    .filter(content -> waitingForReference.get(content).isEmpty())
+                    .forEach(contentToImport -> {
+                        if (!contentIdMapping.containsKey(contentToImport.contentId)) {
+                            importContent(contentToImport);
+                        }
+                    });
         } catch (VersionAlreadyExistsException ex) {
             if (ex.getGlobalId() != null) {
                 log.warn("Duplicate globalId {} detected, skipping import of artifact version: {}",
@@ -167,6 +198,27 @@ public void importContent(ContentEntity entity) {
             List<ArtifactReferenceDto> references = SqlUtil
                     .deserializeReferences(entity.serializedReferences);
 
+            Set<GAV> referencesGavs = references
+                    .stream().map(referenceDto -> new GAV(referenceDto.getGroupId(),
+                            referenceDto.getArtifactId(), referenceDto.getVersion()))
+                    .collect(Collectors.toSet());
+
+            Set<ArtifactReferenceDto> requiredReferences = new HashSet<>();
+
+            // If there are references and they've not been imported yet, add them to the waiting collection
+            if (!references.isEmpty() && !gavDone.containsAll(referencesGavs)) {
+                waitingForReference.put(entity, referencesGavs);
+
+                // For each artifact reference, if it has not been imported yet, add it to the waiting list
+                // for this content.
+                referencesGavs.stream()
+                        .filter(artifactReference -> !referencesGavs.contains(artifactReference))
+                        .forEach(artifactReference -> waitingForReference.get(entity).add(artifactReference));
+
+                // This content cannot be imported until all the references are imported.
+                return;
+            }
+
             TypedContent typedContent = TypedContent.create(ContentHandle.create(entity.contentBytes), null);
             Map<String, TypedContent> resolvedReferences = storage.resolveReferences(references);
             entity.artifactType = utils.determineArtifactType(typedContent, null, resolvedReferences);

app/src/main/java/io/apicurio/registry/upgrade/v2/DataUpgrader.java

@@ -3,40 +3,36 @@
 import io.apicurio.registry.model.GroupId;
 import io.apicurio.registry.rest.client.models.ArtifactReference;
 import io.apicurio.registry.utils.IoUtil;
-import io.apicurio.registry.utils.tests.ApicurioTestTags;
 import io.quarkus.test.junit.QuarkusTest;
-import io.quarkus.test.junit.TestProfile;
 import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
+import java.io.InputStream;
 import java.util.List;
 
+import static io.restassured.RestAssured.given;
+import static org.hamcrest.CoreMatchers.anything;
+
 @QuarkusTest
-@TestProfile(DataUpgradeTestProfile.class)
-@Tag(ApicurioTestTags.SLOW)
 public class DataUpgradeTest extends AbstractResourceTestBase {
 
     @Override
-    @BeforeEach
-    protected void beforeEach() throws Exception {
+    @BeforeAll
+    protected void beforeAll() throws Exception {
+        super.beforeAll();
+
         setupRestAssured();
-    }
 
-    @Test
-    public void testArtifactsCount() {
-        Assertions.assertEquals(26, clientV3.search().artifacts().get().getCount());
+        try (InputStream data = resourceToInputStream("./upgrade/v2_export.zip")) {
+            given().when().contentType("application/zip").body(data).post("/registry/v2/admin/import").then()
+                    .statusCode(204).body(anything());
+        }
     }
 
     @Test
-    public void testCheckGlobalRules() throws Exception {
-        // Global rules are enabled in the export file, they must be activated.
-        Assertions.assertEquals(3, clientV3.admin().rules().get().size());
-        Assertions.assertEquals("FULL", clientV3.admin().rules().byRuleType("VALIDITY").get().getConfig());
-        Assertions.assertEquals("BACKWARD",
-                clientV3.admin().rules().byRuleType("COMPATIBILITY").get().getConfig());
-        Assertions.assertEquals("FULL", clientV3.admin().rules().byRuleType("INTEGRITY").get().getConfig());
+    public void testArtifactsCount() {
+        Assertions.assertEquals(32, clientV3.search().artifacts().get().getCount());
     }
 
     @Test

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

@@ -350,7 +350,7 @@ void testExportForBrowser() throws Exception {
     }
 
     @Test
-    @Disabled // TODO: Disabled, so I can submit the import/export migration feature in a separate PR.
+    @Disabled// TODO: Disabled, so I can submit the import/export migration feature in a separate PR.
     void testImport() throws Exception {
         var result = clientV3.search().artifacts().get(config -> {
             config.queryParameters.offset = 0;

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

@@ -826,19 +826,3 @@ The following {registry} configuration options are available for each component
 |Navigation prefix for all UI paths
 |===
 
-== upgrade
-.upgrade configuration options
-[.table-expandable,width="100%",cols="6,3,2,3,5",options="header"]
-|===
-|Name
-|Type
-|Default
-|Available from
-|Description
-|`apicurio.upgrade.file.location`
-|`optional<url>`
-|
-|`3.0.0`
-|The URL to a v2 export file
-|===
-