DGS-8908 Ensure tenant prefix is used for aliases (#2806)

* DGS-8908 Ensure tenant prefix is used for aliases * Fix test
confluentinc · Oct 24, 2023 · 5d57f8d · 5d57f8d
1 parent 1c390a0
commit 5d57f8d
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 2 deletions.
diff --git a/core/src/main/java/io/confluent/kafka/schemaregistry/rest/filters/AliasFilter.java b/core/src/main/java/io/confluent/kafka/schemaregistry/rest/filters/AliasFilter.java
@@ -15,6 +15,9 @@
 
 package io.confluent.kafka.schemaregistry.rest.filters;
 
+import static io.confluent.kafka.schemaregistry.utils.QualifiedSubject.DEFAULT_TENANT;
+import static io.confluent.kafka.schemaregistry.utils.QualifiedSubject.TENANT_DELIMITER;
+
 import com.google.common.annotations.VisibleForTesting;
 import io.confluent.kafka.schemaregistry.client.rest.entities.Config;
 import io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry;
@@ -30,7 +33,7 @@
 import javax.ws.rs.core.UriBuilder;
 
 @PreMatching
-@Priority(Priorities.ENTITY_CODER + 100) // ensure runs after ContextFilter
+@Priority(Priorities.USER + 100) // ensure runs after ContextFilter and MultiTenantSubjectFilter
 public class AliasFilter implements ContainerRequestFilter {
   private final KafkaSchemaRegistry schemaRegistry;
 
@@ -127,6 +130,13 @@ private String replaceAlias(String subject) {
       return subject;
     }
     String alias = config.getAlias();
-    return alias != null && !alias.isEmpty() ? alias : subject;
+    if (alias != null && !alias.isEmpty()) {
+      if (!DEFAULT_TENANT.equals(schemaRegistry.tenant())) {
+        alias = schemaRegistry.tenant() + TENANT_DELIMITER + alias;
+      }
+      return alias;
+    } else {
+      return subject;
+    }
   }
 }
diff --git a/core/src/test/java/io/confluent/kafka/schemaregistry/rest/filters/AliasFilterTest.java b/core/src/test/java/io/confluent/kafka/schemaregistry/rest/filters/AliasFilterTest.java
@@ -4,6 +4,7 @@
 
 package io.confluent.kafka.schemaregistry.rest.filters;
 
+import static io.confluent.kafka.schemaregistry.utils.QualifiedSubject.DEFAULT_TENANT;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -26,6 +27,7 @@ public class AliasFilterTest {
   @Before
   public void setUp() throws Exception {
     KafkaSchemaRegistry schemaRegistry = mock(KafkaSchemaRegistry.class);
+    when(schemaRegistry.tenant()).thenReturn(DEFAULT_TENANT);
     aliasFilter = new AliasFilter(schemaRegistry);
 
     Config config = new Config();