fix: multitenancy must be handled when testing KafkaSqlUpgraderManager

Author: jsenko

storage/kafkasql/src/main/java/io/apicurio/registry/storage/impl/kafkasql/upgrade/KafkaSqlUpgrader.java

@@ -7,6 +7,12 @@
 /**
  * Implementors of this interface provide upgrade functionality for KafkaSQL storage.
  * Please read the javadoc of {@link KafkaSqlUpgraderManager} as well.
+ * <p>
+ * Multitenancy considerations:
+ * Methods on this class are not called with request context active.
+ * Therefore, upgraders MUST NOT rely on multitenancy features, unless handled manually
+ * (e.g. use @ActivateRequestContext and load a tenant manually).
+ * As a result, they MUST be tenant agnostic.
  */
 public interface KafkaSqlUpgrader extends AutoCloseable {
 

storage/kafkasql/src/main/java/io/apicurio/registry/storage/impl/kafkasql/upgrade/KafkaSqlUpgraderManager.java

@@ -17,6 +17,8 @@
 package io.apicurio.registry.storage.impl.kafkasql.upgrade;
 
 import io.apicurio.common.apps.config.Info;
+import io.apicurio.common.apps.multitenancy.TenantContext;
+import io.apicurio.common.apps.multitenancy.TenantContextLoader;
 import io.apicurio.registry.exception.RuntimeAssertionFailedException;
 import io.apicurio.registry.exception.UnreachableCodeException;
 import io.apicurio.registry.storage.impl.kafkasql.KafkaSqlSubmitter;
@@ -26,7 +28,9 @@
 import io.apicurio.registry.storage.impl.kafkasql.values.ActionType;
 import io.apicurio.registry.storage.impl.kafkasql.values.MessageValue;
 import io.apicurio.registry.storage.impl.kafkasql.values.UpgraderValue;
+import io.apicurio.registry.utils.impexp.ContentEntity;
 import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.context.control.ActivateRequestContext;
 import jakarta.enterprise.inject.Instance;
 import jakarta.inject.Inject;
 import lombok.Getter;
@@ -103,6 +107,12 @@ public class KafkaSqlUpgraderManager {
     @Inject
     ThreadContext threadContext;
 
+    @Inject
+    TenantContextLoader tcl;
+
+    @Inject
+    TenantContext tctx;
+
     /**
      * Unique ID of this upgrader, generated on each Registry start.
      */
@@ -336,10 +346,11 @@ public synchronized void read(Instant currentTimestamp, MessageKey key, MessageV
                                     lockHeartbeat.heartbeat(); // Initialize heartbeat
                                     if (testMode) {
                                         try {
-                                            var c = sqlStore.getContentEntityByContentId(2);
+                                            var c = getContentEntityByContentIdForTest(2);
                                             log.debug("Content hash before: {}", c.contentHash);
                                             log.debug("Canonical content hash before: {}", c.canonicalHash);
-                                        } catch (Exception ignored) {
+                                        } catch (Exception ex) {
+                                            log.error("Suppressing exception in TEST MODE", ex);
                                         }
                                     }
                                     for (KafkaSqlUpgrader u : activeUpgraders) {
@@ -357,10 +368,11 @@ public synchronized void read(Instant currentTimestamp, MessageKey key, MessageV
                                     }
                                     if (testMode) {
                                         try {
-                                            var c = sqlStore.getContentEntityByContentId(2);
+                                            var c = getContentEntityByContentIdForTest(2);
                                             log.debug("Content hash after: {}", c.contentHash);
                                             log.debug("Canonical content hash after: {}", c.canonicalHash);
-                                        } catch (Exception ignored) {
+                                        } catch (Exception ex) {
+                                            log.error("Suppressing exception in TEST MODE", ex);
                                         }
                                     }
                                 }
@@ -527,6 +539,15 @@ private void updateLockMap(Instant timestamp, UpgraderValue value) {
     }
 
 
+    @ActivateRequestContext
+    synchronized ContentEntity getContentEntityByContentIdForTest(int contentId) {
+        tctx.setContext(tcl.loadBatchJobContext(TenantContext.DEFAULT_TENANT_ID));
+        var res = sqlStore.getContentEntityByContentId(contentId);
+        tctx.clearContext();
+        return res;
+    }
+
+
     private enum State {
         WAIT_FOR_BOOTSTRAP,
         WAIT,