DGS-22736 Add support for DELETE /mode

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

@@ -196,7 +196,7 @@
   private static final TypeReference<AssociationResponse> ASSOCIATION_RESPONSE_TYPE =
       new TypeReference<AssociationResponse>() {
       };
   private static final TypeReference<List<Association>> ASSOCIATIONS_RESPONSE_TYPE =
       new TypeReference<List<Association>>() {
       };
   private static final TypeReference<AssociationBatchResponse> ASSOCIATION_BATCH_RESPONSE_TYPE =
@@ -571,7 +571,7 @@
             request.setEntity(new ByteArrayEntity(requestBodyData, null));
           }
           break;
         case "DELETE":
           request = new HttpDelete(requestUrl);
           break;
         default:
@@ -1076,9 +1076,9 @@
                                    boolean force)
       throws IOException, RestClientException {
     String path = subject != null
         ? UriBuilder.fromPath("/mode/{subject}")
         .queryParam("force", force).build(subject).toString()
         : UriBuilder.fromPath("/mode")
             .queryParam("force", force).build().toString();
 
     ModeUpdateRequest response =
@@ -1117,8 +1117,9 @@
 
   public Mode deleteSubjectMode(Map<String, String> requestProperties, String subject)
       throws IOException, RestClientException {
-    UriBuilder builder = UriBuilder.fromPath("/mode/{subject}");
-    String path = builder.build(subject).toString();
+    String path = subject != null
+            ? UriBuilder.fromPath("/mode/{subject}").build(subject).toString()
+            : "/mode";
 
     Mode response = httpRequest(path, "DELETE", null, requestProperties,
         DELETE_SUBJECT_MODE_RESPONSE_TYPE);
@@ -1863,7 +1864,7 @@
       builder.queryParam("dryRun", dryRun);
     }
 
     AssociationResponse response = httpRequest(path, "POST",
         request.toJson().getBytes(StandardCharsets.UTF_8),
         requestProperties, ASSOCIATION_RESPONSE_TYPE);
     return response;
@@ -1883,7 +1884,7 @@
       builder.queryParam("dryRun", dryRun);
     }
 
     AssociationBatchResponse response = httpRequest(path, "POST",
         request.toJson().getBytes(StandardCharsets.UTF_8),
         requestProperties, ASSOCIATION_BATCH_RESPONSE_TYPE);
     return response;

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

@@ -132,7 +132,7 @@ public ModeUpdateRequest updateMode(
     }
 
     if (io.confluent.kafka.schemaregistry.storage.Mode.FORWARD.toString()
-            .equals(request.getMode()) 
+            .equals(request.getMode())
         && !QualifiedSubject.isGlobalContext(schemaRegistry.tenant(), subject)) {
       throw new RestInvalidModeException("Forward mode only supported on global level");
     }
@@ -253,6 +253,30 @@ public Mode getTopLevelMode(
     return getMode(null, defaultToGlobal);
   }
 
+
+  @DELETE
+  @DocumentedName("deleteGlobalMode")
+  @Operation(summary = "Delete global mode",
+      description = "Deletes the global mode and reverts to the default mode.",
+      responses = {
+        @ApiResponse(responseCode = "200", description = "Operation succeeded. Returns old mode.",
+          content = @Content(schema = @Schema(implementation = Mode.class))),
+        @ApiResponse(responseCode = "422",
+          description = "Unprocessable Entity. Error code 42205 indicates operation not permitted.",
+          content = @Content(schema = @Schema(implementation = ErrorMessage.class))),
+        @ApiResponse(responseCode = "500",
+          description = "Internal Server Error. "
+                  + "Error code 50001 indicates a failure in the backend data store.",
+          content = @Content(schema = @Schema(implementation = ErrorMessage.class)))})
+  @Tags(@Tag(name = apiTag))
+  @PerformanceMetric("mode.delete-global")
+  public void deleteGlobalMode(
+      final @Suspended AsyncResponse asyncResponse,
+      @Context HttpHeaders headers) {
+    log.info("Deleting global mode");
+    deleteSubjectMode(asyncResponse, headers, null);
+  }
+
   @DELETE
   @Path("/{subject}")
   @DocumentedName("deleteSubjectMode")
@@ -276,14 +300,14 @@ public void deleteSubjectMode(
       @Context HttpHeaders headers,
       @Parameter(description = "Name of the subject", required = true)
       @PathParam("subject") String subject) {
-    log.debug("Deleting mode for subject {}", subject);
+    log.info("Deleting mode for subject {}", subject);
 
     if (QualifiedSubject.isDefaultContext(schemaRegistry.tenant(), subject)) {
-      throw Errors.invalidSubjectException(subject);
+      subject = null;
+    } else {
+      subject = QualifiedSubject.normalize(schemaRegistry.tenant(), subject);
     }
 
-    subject = QualifiedSubject.normalize(schemaRegistry.tenant(), subject);
-
     io.confluent.kafka.schemaregistry.storage.Mode deletedMode;
     Mode deleteModeResponse;
     try {

core/src/test/java/io/confluent/kafka/schemaregistry/rest/RestApiModeTest.java

@@ -111,8 +111,8 @@
 
   @Test
   public void testReadOnlyMode() throws Exception {
     String subject = "testSubject";
     String mode = "READONLY";
 
     // set mode to read only
     assertEquals(
@@ -136,7 +136,7 @@
   @Test
   public void testReadWriteMode() throws Exception {
     String subject = "testSubject";
     String mode = "READWRITE";
 
     // set mode to read write
     assertEquals(
@@ -699,4 +699,66 @@
       assertEquals("Forward mode only supported on global level; error code: 42204", e.getMessage());
     }
   }
+
+  @Test
+  public void testDeleteGlobalMode() throws Exception {
+    String mode = "READONLY";
+
+    // set global mode to read only
+    assertEquals(
+        mode,
+        restApp.restClient.setMode(mode).getMode());
+
+    // verify mode is set
+    assertEquals(
+        mode,
+        restApp.restClient.getMode().getMode());
+
+    // delete global mode - should succeed and return the old mode
+    Mode deletedMode = restApp.restClient.deleteSubjectMode(null);
+    assertEquals(
+        mode,
+        deletedMode.getMode(),
+        "Deleted mode should return the old global mode");
+
+    // verify global mode is now reset to default (READWRITE)
+    assertEquals(
+        "READWRITE",
+        restApp.restClient.getMode().getMode(),
+        "Global mode should revert to default READWRITE");
+  }
+
+  @Test
+  public void testDeleteSubjectModeAfterGlobalMode() throws Exception {
+    String subject = "testSubject";
+    String globalMode = "READONLY";
+    String subjectMode = "READWRITE";
+
+    // set global mode to read only
+    assertEquals(
+        globalMode,
+        restApp.restClient.setMode(globalMode).getMode());
+
+    // set subject mode to read write
+    assertEquals(
+        subjectMode,
+        restApp.restClient.setMode(subjectMode, subject).getMode());
+
+    // verify subject mode is set
+    assertEquals(
+        subjectMode,
+        restApp.restClient.getMode(subject, false).getMode());
+
+    // delete subject mode
+    Mode deletedMode = restApp.restClient.deleteSubjectMode(subject);
+    assertEquals(
+        subjectMode,
+        deletedMode.getMode(),
+        "Deleted mode should return the old mode");
+
+    // verify subject mode falls back to global mode
+    assertEquals(
+        globalMode,
+        restApp.restClient.getMode(subject, true).getMode());
+  }
 }

core/src/test/java/io/confluent/kafka/schemaregistry/rest/RestApiTest.java

@@ -21,6 +21,7 @@
 import static io.confluent.kafka.schemaregistry.CompatibilityLevel.NONE;
 import static io.confluent.kafka.schemaregistry.storage.Mode.IMPORT;
 import static io.confluent.kafka.schemaregistry.storage.Mode.READONLY;
+import static io.confluent.kafka.schemaregistry.storage.Mode.READWRITE;
 import static io.confluent.kafka.schemaregistry.utils.QualifiedSubject.DEFAULT_CONTEXT;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -38,18 +39,7 @@
 import io.confluent.kafka.schemaregistry.avro.AvroSchemaUtils;
 import io.confluent.kafka.schemaregistry.avro.AvroUtils;
 import io.confluent.kafka.schemaregistry.client.rest.RestService;
-import io.confluent.kafka.schemaregistry.client.rest.entities.ContextId;
-import io.confluent.kafka.schemaregistry.client.rest.entities.Metadata;
-import io.confluent.kafka.schemaregistry.client.rest.entities.Rule;
-import io.confluent.kafka.schemaregistry.client.rest.entities.RuleMode;
-import io.confluent.kafka.schemaregistry.client.rest.entities.RuleSet;
-import io.confluent.kafka.schemaregistry.client.rest.entities.Schema;
-import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaReference;
-import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaRegistryDeployment;
-import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaRegistryServerVersion;
-import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaString;
-import io.confluent.kafka.schemaregistry.client.rest.entities.ServerClusterId;
-import io.confluent.kafka.schemaregistry.client.rest.entities.SubjectVersion;
+import io.confluent.kafka.schemaregistry.client.rest.entities.*;
 import io.confluent.kafka.schemaregistry.client.rest.entities.requests.ConfigUpdateRequest;
 import io.confluent.kafka.schemaregistry.client.rest.entities.requests.ModeUpdateRequest;
 import io.confluent.kafka.schemaregistry.client.rest.entities.requests.RegisterSchemaRequest;
@@ -831,10 +821,20 @@ public void testDefaultContextConfigAndMode() throws Exception {
             .getMode()
     );
 
-    // We don't support deleting the mode for the default context
-    assertThrows(
-        RestClientException.class,
-        () -> restApp.restClient.deleteSubjectMode(defaultContext)
+    // Delete the mode for the default context - should succeed and revert to global mode
+    Mode deletedMode = restApp.restClient.deleteSubjectMode(defaultContext);
+    assertEquals(
+        READONLY.name(),
+        deletedMode.getMode(),
+        "Deleted mode should return the old mode");
+
+    // Verify mode reverts to global mode (IMPORT)
+    assertEquals(
+        READWRITE.name(),
+        restApp.restClient
+            .getMode(null, true)
+            .getMode(),
+        "Mode should revert to global mode after deleting default context mode"
     );
   }
 
@@ -2630,7 +2630,7 @@ public void testGlobalMode() throws Exception {
     } catch (RestClientException rce) {
       assertEquals(
           Errors.OPERATION_NOT_PERMITTED_ERROR_CODE, rce.getErrorCode(),
-          String.format("Subject %s in context" 
+          String.format("Subject %s in context"
           +" %s is in read-only mode", "testSubject2", context)
       );
     }