From 5d57f8de6bc725b3b19c60b5366e7adb3354814d Mon Sep 17 00:00:00 2001
From: Robert Yokota <rayokota@gmail.com>
Date: Tue, 24 Oct 2023 15:11:26 -0700
Subject: [PATCH] DGS-8908 Ensure tenant prefix is used for aliases (#2806)

* DGS-8908 Ensure tenant prefix is used for aliases

* Fix test
---
 .../schemaregistry/rest/filters/AliasFilter.java   | 14 ++++++++++++--
 .../rest/filters/AliasFilterTest.java              |  2 ++
 2 files changed, 14 insertions(+), 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
index 2c2278870e9..6a4ebbf60b4 100644
--- 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
index 6db19922a05..e50a28f7f79 100644
--- 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();