Fix local test execution and wrong column being fetched

carlesarnal · carlesarnal · commit 4939aa2adcaa · 2024-02-20T17:37:31.000+01:00
diff --git a/integration-tests/src/test/java/io/apicurio/tests/dbupgrade/KafkaSqlLogCompactionIT.java b/integration-tests/src/test/java/io/apicurio/tests/dbupgrade/KafkaSqlLogCompactionIT.java
@@ -40,6 +40,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.Network;
 import org.testcontainers.containers.wait.strategy.Wait;
 
 import java.util.Collections;
@@ -106,24 +107,28 @@ public int order() {
         public Map<String, String> start() {
             if (!Boolean.parseBoolean(System.getProperty("cluster.tests"))) {
 
-                String bootstrapServers = System.getProperty("bootstrap.servers");
+                String externalBootstrapServers = System.getProperty("bootstrap.servers.external");
+                String internalBootstrapServers = System.getProperty("bootstrap.servers.internal");
 
-                genericContainer = new GenericContainer("quay.io/apicurio/apicurio-registry-kafkasql:2.1.2.Final")
-                        .withEnv(Map.of("KAFKA_BOOTSTRAP_SERVERS", bootstrapServers, "QUARKUS_HTTP_PORT", "8081"));
+                genericContainer = new GenericContainer<>("quay.io/apicurio/apicurio-registry-kafkasql:2.1.2.Final")
+                        .withEnv(Map.of("KAFKA_BOOTSTRAP_SERVERS", internalBootstrapServers, "QUARKUS_HTTP_PORT", "8081"))
+                        .withExposedPorts(8081)
+                        .withNetwork(Network.SHARED);
 
+                genericContainer.setPortBindings(List.of("8081:8081"));
+                genericContainer.waitingFor(Wait.forHttp("/apis/registry/v2/search/artifacts").forStatusCode(200));
+                genericContainer.start();
                 //create the topic with agressive log compaction
-                createTopic("kafkasql-journal", 1, bootstrapServers);
-
+                createTopic("kafkasql-journal", 1, externalBootstrapServers);
                 genericContainer.start();
-                genericContainer.waitingFor(Wait.forLogMessage(".*(KSQL Kafka Consumer Thread) KafkaSQL storage bootstrapped.*", 1));
 
                 var registryClient = RegistryClientFactory.create("http://localhost:8081");
 
                 try {
-                    RegistryWaitUtils.retry(registryClient, registryClient1 -> CustomTestsUtils.createArtifact(registryClient, ArtifactType.AVRO, ApicurioRegistryBaseIT.resourceToString("artifactTypes/" + "avro/multi-field_v1.json")));
+                    RegistryWaitUtils.retry(registryClient, registryClient1 -> CustomTestsUtils.createArtifact(registryClient, PREPARE_LOG_COMPACTION, ArtifactType.AVRO, ApicurioRegistryBaseIT.resourceToString("artifactTypes/" + "avro/multi-field_v1.json")));
 
-                    var artifactdata = CustomTestsUtils.createArtifact(registryClient, ArtifactType.JSON, ApicurioRegistryBaseIT.resourceToString("artifactTypes/" + "jsonSchema/person_v1.json"));
-                    CustomTestsUtils.createArtifact(registryClient, ArtifactType.PROTOBUF, ApicurioRegistryBaseIT.resourceToString("artifactTypes/" + "protobuf/tutorial_v1.proto"));
+                    var artifactdata = CustomTestsUtils.createArtifact(registryClient, PREPARE_LOG_COMPACTION, ArtifactType.JSON, ApicurioRegistryBaseIT.resourceToString("artifactTypes/" + "jsonSchema/person_v1.json"));
+                    CustomTestsUtils.createArtifact(registryClient, PREPARE_LOG_COMPACTION, ArtifactType.PROTOBUF, ApicurioRegistryBaseIT.resourceToString("artifactTypes/" + "protobuf/tutorial_v1.proto"));
 
                     assertEquals(3, registryClient.listArtifactsInGroup(PREPARE_LOG_COMPACTION).getCount());
 
@@ -165,15 +170,11 @@ public AdminClient adminClient(String bootstrapServers) {
         public void createTopic(String topic, int partitions, String bootstrapServers) {
             var journal = new NewTopic(topic, partitions, (short) 1);
 
-            /*journal.configs(Map.of(
+            journal.configs(Map.of(
                     "min.cleanable.dirty.ratio","0.000001",
                     "cleanup.policy","compact",
                     "segment.ms", "100",
                     "delete.retention.ms", "100"
-            ));*/
-
-            journal.configs(Map.of(
-                    "cleanup.policy", "compact"
             ));
 
             adminClient(bootstrapServers).createTopics(List.of(journal));
diff --git a/integration-tests/src/test/java/io/apicurio/tests/dbupgrade/KafkaSqlStorageUpgradeIT.java b/integration-tests/src/test/java/io/apicurio/tests/dbupgrade/KafkaSqlStorageUpgradeIT.java
@@ -129,7 +129,7 @@ public static class KafkaSqlStorageUpgradeInitializer implements QuarkusTestReso
         public Map<String, String> start() {
             if (!Boolean.parseBoolean(System.getProperty("cluster.tests"))) {
 
-                String bootstrapServers = System.getProperty("bootstrap.servers");
+                String bootstrapServers = System.getProperty("bootstrap.servers.internal");
                 startOldRegistryVersion("quay.io/apicurio/apicurio-registry-kafkasql:2.1.2.Final", bootstrapServers);
 
                 try {
@@ -159,12 +159,13 @@ public void stop() {
         private void startOldRegistryVersion(String registryImage, String bootstrapServers) {
             genericContainer = new GenericContainer<>(registryImage)
                     .withEnv(Map.of("KAFKA_BOOTSTRAP_SERVERS", bootstrapServers, "QUARKUS_HTTP_PORT", "8081"))
+                    .withExposedPorts(8081)
                     .withNetwork(Network.SHARED);
 
             genericContainer.setPortBindings(List.of("8081:8081"));
 
+            genericContainer.waitingFor(Wait.forHttp("/apis/registry/v2/search/artifacts").forStatusCode(200));
             genericContainer.start();
-            genericContainer.waitingFor(Wait.forLogMessage(".*(KSQL Kafka Consumer Thread) KafkaSQL storage bootstrapped.*", 1));
         }
     }
 }
diff --git a/storage/kafkasql/src/main/java/io/apicurio/registry/storage/impl/kafkasql/KafkaSqlAvroCanonicalizerUpgrader.java b/storage/kafkasql/src/main/java/io/apicurio/registry/storage/impl/kafkasql/KafkaSqlAvroCanonicalizerUpgrader.java
@@ -34,6 +34,7 @@
 import jakarta.inject.Inject;
 import org.apache.commons.codec.digest.DigestUtils;
 import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -63,7 +64,7 @@ public void upgrade(Handle dbHandle) throws Exception {
 
     @Override
     public void upgrade(RegistryStorage registryStorage, Handle dbHandle) throws Exception {
-        String sql = "SELECT c.contentId, c.content, c.canonicalHash, c.contentHash, c.artifactreferences, a.type "
+        String sql = "SELECT c.contentId, c.content, c.canonicalHash, c.contentHash, c.artifactreferences, v.tenantId "
                 + "FROM versions v "
                 + "JOIN content c on c.contentId = v.contentId "
                 + "JOIN artifacts a ON v.tenantId = a.tenantId AND v.groupId = a.groupId AND v.artifactId = a.artifactId "
@@ -83,8 +84,12 @@ protected void updateCanonicalHash(TenantContentEntity contentEntity) {
         try {
 
             String canonicalContentHash;
-            byte[] referencesBytes = contentEntity.contentEntity.serializedReferences.getBytes(StandardCharsets.UTF_8);
-            canonicalContentHash = DigestUtils.sha256Hex(concatContentAndReferences(this.canonicalizeContent(contentEntity.contentEntity, contentEntity.contentEntity.artifactType).bytes(), referencesBytes));
+            if (contentEntity.contentEntity.serializedReferences != null) {
+                byte[] referencesBytes = contentEntity.contentEntity.serializedReferences.getBytes(StandardCharsets.UTF_8);
+                canonicalContentHash = DigestUtils.sha256Hex(concatContentAndReferences(this.canonicalizeContent(contentEntity.contentEntity, ArtifactType.AVRO).bytes(), referencesBytes));
+            } else {
+                canonicalContentHash = DigestUtils.sha256Hex(this.canonicalizeContent(contentEntity.contentEntity, contentEntity.contentEntity.artifactType).bytes());
+            }
 
             if (canonicalContentHash.equals(contentEntity.contentEntity.canonicalHash)) {
                 logger.debug("Skipping content because the canonical hash is up to date, updating contentId {}", contentEntity.contentEntity.contentId);
@@ -121,6 +126,7 @@ public static class TenantContentEntity {
     }
 
     public static class TenantContentEntityRowMapper implements RowMapper<TenantContentEntity> {
+
         @Override
         public TenantContentEntity map(ResultSet rs) throws SQLException {
             TenantContentEntity e = new TenantContentEntity();
diff --git a/utils/kafka/src/test/java/io/apicurio/registry/test/utils/KafkaTestContainerManager.java b/utils/kafka/src/test/java/io/apicurio/registry/test/utils/KafkaTestContainerManager.java
@@ -50,13 +50,16 @@ public Map<String, String> start() {
             kafka.withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("kafka-testcontainer")));
             kafka.start();
 
+            String externalBootstrapServers = kafka.getBootstrapServers();
+
             String bootstrapServers = "redpanda:19092";
-            System.setProperty("bootstrap.servers", bootstrapServers);
+            System.setProperty("bootstrap.servers.internal", bootstrapServers);
+            System.setProperty("bootstrap.servers.external", externalBootstrapServers);
 
             return Map.of(
-                    "bootstrap.servers", bootstrapServers,
-                    "registry.events.kafka.config.bootstrap.servers", bootstrapServers,
-                    "registry.kafkasql.bootstrap.servers", bootstrapServers);
+                    "bootstrap.servers", externalBootstrapServers,
+                    "registry.events.kafka.config.bootstrap.servers", externalBootstrapServers,
+                    "registry.kafkasql.bootstrap.servers", externalBootstrapServers);
         } else {
             return Collections.emptyMap();
         }

Original file line number	Diff line number	Diff line change
`@@ -129,7 +129,7 @@ public static class KafkaSqlStorageUpgradeInitializer implements QuarkusTestReso`
`129`	`129`	`public Map<String, String> start() {`
`130`	`130`	`if (!Boolean.parseBoolean(System.getProperty("cluster.tests"))) {`
`131`	`131`
`132`		`- String bootstrapServers = System.getProperty("bootstrap.servers");`
	`132`	`+ String bootstrapServers = System.getProperty("bootstrap.servers.internal");`
`133`	`133`	`startOldRegistryVersion("quay.io/apicurio/apicurio-registry-kafkasql:2.1.2.Final", bootstrapServers);`
`134`	`134`
`135`	`135`	`try {`
`@@ -159,12 +159,13 @@ public void stop() {`
`159`	`159`	`private void startOldRegistryVersion(String registryImage, String bootstrapServers) {`
`160`	`160`	`genericContainer = new GenericContainer<>(registryImage)`
`161`	`161`	`.withEnv(Map.of("KAFKA_BOOTSTRAP_SERVERS", bootstrapServers, "QUARKUS_HTTP_PORT", "8081"))`
	`162`	`+ .withExposedPorts(8081)`
`162`	`163`	`.withNetwork(Network.SHARED);`
`163`	`164`
`164`	`165`	`genericContainer.setPortBindings(List.of("8081:8081"));`
`165`	`166`
	`167`	`+ genericContainer.waitingFor(Wait.forHttp("/apis/registry/v2/search/artifacts").forStatusCode(200));`
`166`	`168`	`genericContainer.start();`
`167`		`- genericContainer.waitingFor(Wait.forLogMessage(".(KSQL Kafka Consumer Thread) KafkaSQL storage bootstrapped.", 1));`
`168`	`169`	`}`
`169`	`170`	`}`
`170`	`171`	`}`