Merge pull request #4002 from confluentinc/pr_merge_from_8_0_x_to_8_1_x

Merge Conflict Resolution (from 8.0.x to 8.1.x)

Author: karthikeyanas

client/src/main/java/io/confluent/kafka/schemaregistry/client/rest/RestService.java

@@ -1117,9 +1117,23 @@ public Mode deleteSubjectMode(String subject)
 
   public Mode deleteSubjectMode(Map<String, String> requestProperties, String subject)
       throws IOException, RestClientException {
+    return deleteSubjectMode(requestProperties, subject, false);
+  }
+
+  public Mode deleteSubjectMode(Map<String, String> requestProperties, String subject,
+                                boolean recursive)
+      throws IOException, RestClientException {
+    UriBuilder builder = subject != null
+            ? UriBuilder.fromPath("/mode/{subject}")
+            : UriBuilder.fromPath("/mode");
+
+    if (recursive) {
+      builder.queryParam("recursive", "true");
+    }
+
     String path = subject != null
-            ? UriBuilder.fromPath("/mode/{subject}").build(subject).toString()
-            : "/mode";
+            ? builder.build(subject).toString()
+            : builder.build().toString();
 
     Mode response = httpRequest(path, "DELETE", null, requestProperties,
         DELETE_SUBJECT_MODE_RESPONSE_TYPE);

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

@@ -361,6 +361,19 @@ public static String normalizeContext(String context) {
     return DEFAULT_CONTEXT.equals(context) ? "" : CONTEXT_DELIMITER + context + CONTEXT_DELIMITER;
   }
 
+  /**
+   * 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.
+   *
+   * @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();
+  }
+
   /**
    * Checks if the given qualified subject is the default context for the given tenant.
    * The default context is the context with name "." and an empty subject.
@@ -473,4 +486,3 @@ public int compareTo(QualifiedSubject that) {
     }
   }
 }
-

core/src/main/java/io/confluent/kafka/schemaregistry/rest/resources/ModeResource.java

@@ -256,10 +256,14 @@ public Mode getTopLevelMode(
   @DELETE
   @DocumentedName("deleteGlobalMode")
   @Operation(summary = "Delete global mode",
-      description = "Deletes the global mode and reverts to the default mode.",
+      description = "Deletes the global mode and reverts to the default mode. "
+        + "If recursive is true, also deletes mode for all subjects in the default context.",
       responses = {
           @ApiResponse(responseCode = "200",
-                      description = "Operation succeeded. Returns old mode.",
+                      description = "Operation succeeded. Returns old mode."
+                              + "If recursive is enabled and mode is not found, returns 200 "
+                              + "and returns null mode as it would delete mode for all subjects "
+                              + "under it",
                       content = @Content(schema = @Schema(implementation = Mode.class))),
           @ApiResponse(responseCode = "422",
                       description = "Unprocessable Entity. "
@@ -273,19 +277,26 @@ public Mode getTopLevelMode(
   @PerformanceMetric("mode.delete-global")
   public void deleteGlobalMode(
           final @Suspended AsyncResponse asyncResponse,
-          @Context HttpHeaders headers) {
-    log.info("Deleting global mode");
-    deleteSubjectMode(asyncResponse, headers, null);
+          @Context HttpHeaders headers,
+          @Parameter(description =
+              "Whether to recursively delete mode for all subjects in the default context")
+          @QueryParam("recursive") boolean recursive) {
+    log.info("Deleting global mode, recursive={}", recursive);
+    deleteSubjectMode(asyncResponse, headers, null, recursive);
   }
 
   @DELETE
   @Path("/{subject}")
   @DocumentedName("deleteSubjectMode")
   @Operation(summary = "Delete subject mode",
       description = "Deletes the specified subject-level mode and reverts to "
-        + "the global default.",
+        + "the global default. "
+        + "If the subject is a context and recursive is true, also deletes mode "
+        + "for all subjects under that context.",
       responses = {
-        @ApiResponse(responseCode = "200", description = "Operation succeeded. Returns old mode.",
+        @ApiResponse(responseCode = "200", description = "Operation succeeded. Returns old mode."
+                + "If recursive is enabled and mode is not found, returns 200 and returns null"
+                + "mode as it would delete mode for all subjects under it",
           content = @Content(schema = @Schema(implementation = Mode.class))),
         @ApiResponse(responseCode = "404",
           description = "Not Found. Error code 40401 indicates subject not found.",
@@ -300,8 +311,12 @@ public void deleteSubjectMode(
       final @Suspended AsyncResponse asyncResponse,
       @Context HttpHeaders headers,
       @Parameter(description = "Name of the subject", required = true)
-      @PathParam("subject") String subject) {
-    log.info("Deleting mode for subject {}", subject);
+      @PathParam("subject") String subject,
+      @Parameter(description =
+          "Whether to recursively delete mode for all subjects under the context, "
+            + "if the subject is a context")
+      @QueryParam("recursive") boolean recursive) {
+    log.info("Deleting mode for subject {}, recursive={}", subject, recursive);
 
     if (QualifiedSubject.isDefaultContext(schemaRegistry.tenant(), subject)) {
       subject = null;
@@ -313,14 +328,19 @@ public void deleteSubjectMode(
     Mode deleteModeResponse;
     try {
       deletedMode = schemaRegistry.getMode(subject);
-      if (deletedMode == null) {
+      if (deletedMode == null && !recursive) {
+        // If recursive is not enabled and mode is not found, return 404
+        // If recursive is enabled and mode is not found, continue to delete mode for all
+        // subjects under it
         throw Errors.subjectNotFoundException(subject);
       }
 
       Map<String, String> headerProperties = requestHeaderBuilder.buildRequestHeaders(
           headers, schemaRegistry.config().whitelistHeaders());
-      schemaRegistry.deleteSubjectModeOrForward(subject, headerProperties);
-      deleteModeResponse = new Mode(deletedMode.name());
+
+      // Delete mode for the context/subject itself (and recursively if requested)
+      schemaRegistry.deleteSubjectModeOrForward(subject, recursive, headerProperties);
+      deleteModeResponse = new Mode(deletedMode != null ? deletedMode.name() : null);
     } catch (OperationNotPermittedException e) {
       throw Errors.operationNotPermittedException(e.getMessage());
     } catch (SchemaRegistryStoreException e) {

core/src/main/java/io/confluent/kafka/schemaregistry/storage/KafkaSchemaRegistry.java

@@ -15,6 +15,8 @@
 
 package io.confluent.kafka.schemaregistry.storage;
 
+import static io.confluent.kafka.schemaregistry.utils.QualifiedSubject.CONTEXT_DELIMITER;
+import static io.confluent.kafka.schemaregistry.utils.QualifiedSubject.CONTEXT_PREFIX;
 import static io.confluent.kafka.schemaregistry.utils.QualifiedSubject.DEFAULT_CONTEXT;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -64,6 +66,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -994,13 +997,15 @@ private void forwardSetModeRequestToLeader(
 
   private void forwardDeleteSubjectModeRequestToLeader(
       String subject,
+      boolean recursive,
       Map<String, String> headerProperties)
       throws SchemaRegistryRequestForwardingException {
     UrlList baseUrl = leaderRestService.getBaseUrls();
 
-    log.debug("Forwarding delete subject mode request {} to {}", subject, baseUrl);
+    log.debug("Forwarding delete subject mode request {} to {} with recursive={}",
+        subject, baseUrl, recursive);
     try {
-      leaderRestService.deleteSubjectMode(headerProperties, subject);
+      leaderRestService.deleteSubjectMode(headerProperties, subject, recursive);
     } catch (IOException e) {
       throw new SchemaRegistryRequestForwardingException(
           String.format(
@@ -1248,31 +1253,49 @@ public void setModeOrForward(String subject, ModeUpdateRequest mode, boolean for
 
   public void deleteSubjectMode(String subject)
       throws SchemaRegistryStoreException, OperationNotPermittedException {
+    deleteSubjectMode(subject, false);
+  }
+
+  public void deleteSubjectMode(String subject, boolean recursive)
+      throws SchemaRegistryStoreException, OperationNotPermittedException {
     if (!allowModeChanges) {
       throw new OperationNotPermittedException("Mode changes are not allowed");
     }
     try {
       kafkaStore.waitUntilKafkaReaderReachesLastOffset(subject, kafkaStoreTimeoutMs);
       deleteMode(subject);
+
+      // If recursive and subject is a context, delete modes for all subjects under it
+      if (recursive && QualifiedSubject.isContext(tenant(), subject)) {
+        log.info("Recursively deleting mode for all subjects under context: {}", subject);
+        try {
+          deleteModesForSubjectsUnderContext(subject);
+        } catch (SchemaRegistryException e) {
+          throw new SchemaRegistryStoreException(
+              "Failed to recursively delete modes for subjects under context", e);
+        }
+      }
     } catch (StoreException e) {
       throw new SchemaRegistryStoreException("Failed to delete subject config value from store",
           e);
     }
   }
 
-  public void deleteSubjectModeOrForward(String subject, Map<String, String> headerProperties)
+  public void deleteSubjectModeOrForward(String subject, boolean recursive,
+                                         Map<String, String> headerProperties)
       throws SchemaRegistryStoreException, SchemaRegistryRequestForwardingException,
       OperationNotPermittedException, UnknownLeaderException {
     kafkaStore.lockFor(subject).lock();
     try {
       if (isLeader()) {
-        deleteSubjectMode(subject);
+        // Delete the subject/context mode itself (and recursively if requested)
+        deleteSubjectMode(subject, recursive);
       } else {
-        // forward delete subject config request to the leader
+        // forward delete subject mode request to the leader (with recursive flag)
         if (leaderIdentity != null) {
-          forwardDeleteSubjectModeRequestToLeader(subject, headerProperties);
+          forwardDeleteSubjectModeRequestToLeader(subject, recursive, headerProperties);
         } else {
-          throw new UnknownLeaderException("Delete config request failed since leader is "
+          throw new UnknownLeaderException("Delete mode request failed since leader is "
               + "unknown");
         }
       }
@@ -1281,6 +1304,68 @@ public void deleteSubjectModeOrForward(String subject, Map<String, String> heade
     }
   }
 
+  /**
+   * List all subjects that have a mode configured under the given prefix.
+   * This is different from listSubjectsWithPrefix which only returns subjects with schemas.
+   *
+   * @param prefix The subject prefix to match (e.g., ":.context:" for a context)
+   * @return Set of subject names that have modes configured under the prefix
+   * @throws SchemaRegistryException if there's an error accessing the store
+   */
+  private Set<String> listSubjectsWithModePrefix(String prefix)
+      throws SchemaRegistryException {
+    try {
+      ModeKey startKey = new ModeKey(prefix + Character.MIN_VALUE);
+      ModeKey endKey = new ModeKey(prefix + Character.MAX_VALUE);
+
+      Set<String> subjects = new LinkedHashSet<>();
+      try (CloseableIterator<SchemaRegistryValue> iterator =
+               kafkaStore.getAll(startKey, endKey)) {
+        while (iterator.hasNext()) {
+          SchemaRegistryValue value = iterator.next();
+          if (value instanceof ModeValue) {
+            ModeValue modeValue = (ModeValue) value;
+            String subject = modeValue.getSubject();
+            // Only include subjects that match our prefix
+            if (subject != null && subject.startsWith(prefix)) {
+              subjects.add(subject);
+            }
+          }
+        }
+      }
+      return subjects;
+    } catch (StoreException e) {
+      throw new SchemaRegistryStoreException(
+          "Error while retrieving subjects with modes under prefix: " + prefix, e);
+    }
+  }
+
+  private void deleteModesForSubjectsUnderContext(String context)
+      throws SchemaRegistryException {
+    // Context is already normalized and includes the trailing delimiter
+    // For default context: context would be like ":tenant:"
+    // For named context: context would be like ":.production:"
+    String subjectPrefix = context != null ? context :
+            QualifiedSubject.normalize(tenant(), CONTEXT_PREFIX + CONTEXT_DELIMITER);
+
+    // Get all subjects with modes under this context
+    // This includes subjects that have modes set but no schemas registered
+    Set<String> subjects = listSubjectsWithModePrefix(subjectPrefix);
+
+    log.info("Found {} subjects with modes under context '{}' for recursive mode deletion with"
+                    +  " subjectPrefix={}", subjects.size(), context, subjectPrefix);
+
+    // Delete mode for each subject (locally on leader)
+    int successCount = 0;
+    for (String subjectName : subjects) {
+      log.debug("Deleting mode for subject: {}", subjectName);
+      deleteSubjectMode(subjectName);
+      successCount++;
+    }
+
+    log.info("Recursive mode deletion completed successfully for {} subjects", successCount);
+  }
+
   @Override
   public KafkaStore<SchemaRegistryKey, SchemaRegistryValue> getKafkaStore() {
     return this.kafkaStore;

core/src/main/java/io/confluent/kafka/schemaregistry/storage/SchemaRegistry.java

@@ -263,6 +263,9 @@ Set<String> subjects(String subject, boolean lookupDeletedSubjects) throws
   void deleteSubjectMode(String subject) throws SchemaRegistryStoreException,
           OperationNotPermittedException;
 
+  void deleteSubjectMode(String subject, boolean recursive) throws SchemaRegistryStoreException,
+          OperationNotPermittedException;
+
   RuleSetHandler getRuleSetHandler();
 
   void setRuleSetHandler(RuleSetHandler ruleSetHandler);
@@ -297,7 +300,8 @@ default String getKafkaClusterId() {
     return null;
   }
 
-  default void deleteSubjectModeOrForward(String subject, Map<String, String> headerProperties)
+  default void deleteSubjectModeOrForward(String subject, boolean recursive,
+                                          Map<String, String> headerProperties)
           throws SchemaRegistryStoreException, SchemaRegistryRequestForwardingException,
           OperationNotPermittedException, UnknownLeaderException {}