Merge branch '8.1.x' into master by karthikeyanas

Author: semaphore-agent-production[bot]

client/src/main/java/io/confluent/kafka/schemaregistry/utils/QualifiedSubject.java

@@ -364,14 +364,19 @@ public static String normalizeContext(String context) {
   /**
    * Checks if the given qualified subject represents a context (i.e., has no subject part).
    * A context has the format ":.context:" with an empty subject part, or null for default context.
+   * Also returns true for wildcard subject ("*") when tenant is not the default tenant.
    *
    * @param tenant the tenant
    * @param qualifiedSubject the subject with a tenant prefix
    * @return true if the qualified subject is a context, false otherwise
    */
   public static boolean isContext(String tenant, String qualifiedSubject) {
     QualifiedSubject qs = QualifiedSubject.create(tenant, qualifiedSubject);
-    return qs == null || qs.getSubject().isEmpty();
+    if (qs == null || qs.getSubject().isEmpty()) {
+      return true;
+    }
+    // Check if it's the wildcard subject in non default tenant
+    return tenant != null && !DEFAULT_TENANT.equals(tenant) && WILDCARD.equals(qs.getSubject());
   }
 
   /**

client/src/test/java/io/confluent/kafka/schemaregistry/utils/QualifiedSubjectTest.java

@@ -240,4 +240,70 @@ public void testIsGlobalContext() {
     assertTrue(QualifiedSubject.isGlobalContext("default", ":" + GLOBAL_CONTEXT_NAME + ":"));
     assertTrue(QualifiedSubject.isGlobalContext("default", ":" + GLOBAL_CONTEXT_NAME));
   }
+
+  @Test
+  public void testIsContext() {
+    // Test null qualified subject - should return true (default context)
+    assertTrue(QualifiedSubject.isContext("default", null));
+    assertTrue(QualifiedSubject.isContext("tenant1", null));
+
+    // Test empty string qualified subject - should return true (default context)
+    assertTrue(QualifiedSubject.isContext("default", ""));
+    assertTrue(QualifiedSubject.isContext("tenant1", "tenant1_"));
+
+    // Test context with empty subject - should return true
+    assertTrue(QualifiedSubject.isContext("default", ":.ctx1:"));
+    assertTrue(QualifiedSubject.isContext("tenant1", "tenant1_:.ctx1:"));
+    assertTrue(QualifiedSubject.isContext("default", ":.prod:"));
+    assertTrue(QualifiedSubject.isContext("tenant1", "tenant1_:.prod:"));
+
+    // Test context with missing last colon but empty subject - should return true
+    assertTrue(QualifiedSubject.isContext("default", ":.ctx1"));
+    assertTrue(QualifiedSubject.isContext("tenant1", "tenant1_:.ctx1"));
+
+    // Test wildcard subject in non-default tenant - should return true
+    assertTrue(QualifiedSubject.isContext("tenant1", "tenant1_*"));
+    assertTrue(QualifiedSubject.isContext("tenant2", "tenant2_*"));
+
+    // Test wildcard subject in default tenant - should return false
+    assertFalse(QualifiedSubject.isContext("default", "*"));
+    assertFalse(QualifiedSubject.isContext(DEFAULT_TENANT, "*"));
+
+    // Test regular subjects - should return false
+    assertFalse(QualifiedSubject.isContext("default", "subject1"));
+    assertFalse(QualifiedSubject.isContext("tenant1", "tenant1_subject1"));
+    assertFalse(QualifiedSubject.isContext("default", ":.ctx1:subject1"));
+    assertFalse(QualifiedSubject.isContext("tenant1", "tenant1_:.ctx1:subject1"));
+
+    // Test context wildcard (special context) - should return true (empty subject)
+    assertTrue(QualifiedSubject.isContext("tenant1", "tenant1_:*:"));
+
+    // Test invalid context formats with subjects - should return false
+    assertFalse(QualifiedSubject.isContext("default", ":ctx1:subject1"));
+    assertFalse(QualifiedSubject.isContext("tenant1", "tenant1_:ctx1:subject1"));
+  }
+
+  @Test
+  public void testIsContextEdgeCases() {
+    // Test with null tenant
+    assertTrue(QualifiedSubject.isContext(null, null));
+    assertTrue(QualifiedSubject.isContext(null, ""));
+    assertTrue(QualifiedSubject.isContext(null, ":.ctx:"));
+
+    // Wildcard with null tenant (treated as default tenant)
+    assertFalse(QualifiedSubject.isContext(null, "*"));
+
+    // Test with empty context (default context)
+    assertTrue(QualifiedSubject.isContext("default", ":.:"));
+    assertTrue(QualifiedSubject.isContext("tenant1", "tenant1_:.:"));
+
+    // Test subjects that look like contexts but have content after
+    assertFalse(QualifiedSubject.isContext("default", ":.ctx:foo"));
+    assertFalse(QualifiedSubject.isContext("tenant1", "tenant1_:.ctx:foo"));
+
+    // Test with special subjects that are not wildcards
+    assertFalse(QualifiedSubject.isContext("tenant1", "tenant1_**"));
+    assertFalse(QualifiedSubject.isContext("tenant1", "tenant1_*subject"));
+    assertFalse(QualifiedSubject.isContext("default", "**"));
+  }
 }