DGS-1328 Add DELETE API for subject config and subject mode

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

@@ -135,9 +135,12 @@ public class RestService implements Configurable {
   private static final TypeReference<? extends List<Integer>> DELETE_SUBJECT_RESPONSE_TYPE =
       new TypeReference<List<Integer>>() {
       };
-  private static final TypeReference<String> DELETE_MODE_RESPONSE_TYPE =
+  private static final TypeReference<String> DELETE_SUBJECT_MODE_RESPONSE_TYPE =
       new TypeReference<String>() {
       };
+  private static final TypeReference<Config> DELETE_SUBJECT_CONFIG_RESPONSE_TYPE =
+      new TypeReference<Config>() {
+      };
   private static final TypeReference<ServerClusterId> GET_CLUSTER_ID_RESPONSE_TYPE =
       new TypeReference<ServerClusterId>() {
       };
@@ -600,21 +603,44 @@ public ConfigUpdateRequest updateConfig(Map<String, String> requestProperties,
 
   public Config getConfig(String subject)
       throws IOException, RestClientException {
-    return getConfig(DEFAULT_REQUEST_PROPERTIES, subject);
+    return getConfig(DEFAULT_REQUEST_PROPERTIES, subject, false);
   }
 
   public Config getConfig(Map<String, String> requestProperties,
                           String subject)
       throws IOException, RestClientException {
+    return getConfig(requestProperties, subject, false);
+  }
+
+  public Config getConfig(Map<String, String> requestProperties,
+                          String subject,
+                          boolean defaultToGlobal)
+      throws IOException, RestClientException {
     String path = subject != null
-                  ? UriBuilder.fromPath("/config/{subject}").build(subject).toString()
-                  : "/config";
+        ? UriBuilder.fromPath("/config/{subject}")
+        .queryParam("defaultToGlobal", defaultToGlobal).build(subject).toString()
+        : "/config";
 
     Config config =
         httpRequest(path, "GET", null, requestProperties, GET_CONFIG_RESPONSE_TYPE);
     return config;
   }
 
+  public Config deleteSubjectConfig(String subject)
+      throws IOException, RestClientException {
+    return deleteSubjectConfig(DEFAULT_REQUEST_PROPERTIES, subject);
+  }
+
+  public Config deleteSubjectConfig(Map<String, String> requestProperties, String subject)
+      throws IOException, RestClientException {
+    UriBuilder builder = UriBuilder.fromPath("/config/{subject}");
+    String path = builder.build(subject).toString();
+
+    Config response = httpRequest(path, "DELETE", null, requestProperties,
+        DELETE_SUBJECT_CONFIG_RESPONSE_TYPE);
+    return response;
+  }
+
   public ModeUpdateRequest setMode(String mode)
       throws IOException, RestClientException {
     return setMode(mode, null);
@@ -666,6 +692,21 @@ public ModeGetResponse getMode(String subject, boolean defaultToGlobal)
     return mode;
   }
 
+  public String deleteSubjectMode(String subject)
+      throws IOException, RestClientException {
+    return deleteSubjectMode(DEFAULT_REQUEST_PROPERTIES, subject);
+  }
+
+  public String deleteSubjectMode(Map<String, String> requestProperties, String subject)
+      throws IOException, RestClientException {
+    UriBuilder builder = UriBuilder.fromPath("/mode/{subject}");
+    String path = builder.build(subject).toString();
+
+    String response = httpRequest(path, "DELETE", null, requestProperties,
+        DELETE_SUBJECT_MODE_RESPONSE_TYPE);
+    return response;
+  }
+
   public List<Schema> getSchemas(
       String subjectPrefix,
       boolean lookupDeletedSchema,

core/generated/swagger-ui/schema-registry-api-spec.yaml

@@ -232,6 +232,43 @@ paths:
         500:
           description: "Error code 50001 -- Error in the backend data store\nError\
             \ code 50003 -- Error while forwarding the request to the primary"
+    delete:
+      summary: "Deletes the specified subject-level compatibility level config and\
+        \ revert to the global default."
+      description: ""
+      operationId: "deleteSubjectConfig"
+      consumes:
+      - "application/vnd.schemaregistry.v1+json"
+      - "application/vnd.schemaregistry+json"
+      - "application/json"
+      - "application/octet-stream"
+      produces:
+      - "application/vnd.schemaregistry.v1+json"
+      - "application/vnd.schemaregistry+json; qs=0.9"
+      - "application/json; qs=0.5"
+      parameters:
+      - name: "subject"
+        in: "path"
+        description: "the name of the subject"
+        required: true
+        type: "string"
+      responses:
+        200:
+          description: "successful operation"
+          schema:
+            type: "string"
+            enum:
+            - "NONE"
+            - "BACKWARD"
+            - "BACKWARD_TRANSITIVE"
+            - "FORWARD"
+            - "FORWARD_TRANSITIVE"
+            - "FULL"
+            - "FULL_TRANSITIVE"
+        404:
+          description: "Error code 40401 -- Subject not found"
+        500:
+          description: "Error code 50001 -- Error in the backend datastore"
   /mode:
     get:
       summary: "Get global mode."
@@ -356,6 +393,40 @@ paths:
           description: "Error code 50001 -- Error in the backend data store\nError\
             \ code 50003 -- Error while forwarding the request to the primary\nError\
             \ code 50004 -- Unknown leader"
+    delete:
+      summary: "Deletes the specified subject-level mode and revert to the global\
+        \ default."
+      description: ""
+      operationId: "deleteSubjectMode"
+      consumes:
+      - "application/vnd.schemaregistry.v1+json"
+      - "application/vnd.schemaregistry+json"
+      - "application/json"
+      - "application/octet-stream"
+      produces:
+      - "application/vnd.schemaregistry.v1+json"
+      - "application/vnd.schemaregistry+json; qs=0.9"
+      - "application/json; qs=0.5"
+      parameters:
+      - name: "subject"
+        in: "path"
+        description: "the name of the subject"
+        required: true
+        type: "string"
+      responses:
+        200:
+          description: "successful operation"
+          schema:
+            type: "string"
+            enum:
+            - "READWRITE"
+            - "READONLY"
+            - "READONLY_OVERRIDE"
+            - "IMPORT"
+        404:
+          description: "Error code 40401 -- Subject not found"
+        500:
+          description: "Error code 50001 -- Error in the backend datastore"
   /schemas:
     get:
       summary: "Get the schemas."

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

@@ -26,12 +26,15 @@
 
 import javax.validation.constraints.NotNull;
 import javax.ws.rs.Consumes;
+import javax.ws.rs.DELETE;
 import javax.ws.rs.GET;
 import javax.ws.rs.PUT;
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
 import javax.ws.rs.Produces;
 import javax.ws.rs.QueryParam;
+import javax.ws.rs.container.AsyncResponse;
+import javax.ws.rs.container.Suspended;
 import javax.ws.rs.core.Context;
 import javax.ws.rs.core.HttpHeaders;
 
@@ -178,4 +181,42 @@ public Config getTopLevelConfig() {
     }
     return config;
   }
+
+  @DELETE
+  @Path("/{subject}")
+  @ApiOperation(value = "Deletes the specified subject-level compatibility level config and "
+      + "revert to the global default.", response = CompatibilityLevel.class)
+  @ApiResponses(value = {
+      @ApiResponse(code = 404, message = "Error code 40401 -- Subject not found"),
+      @ApiResponse(code = 500, message = "Error code 50001 -- Error in the backend datastore")
+  })
+  public void deleteSubjectConfig(
+      final @Suspended AsyncResponse asyncResponse,
+      @Context HttpHeaders headers,
+      @ApiParam(value = "the name of the subject", required = true)
+      @PathParam("subject") String subject) {
+    log.info("Deleting compatibility setting for subject {}", subject);
+    Config deletedConfig;
+    try {
+      CompatibilityLevel currentCompatibility = schemaRegistry.getCompatibilityLevel(subject);
+      if (currentCompatibility == null) {
+        throw Errors.subjectNotFoundException(subject);
+      }
+
+      Map<String, String> headerProperties = requestHeaderBuilder.buildRequestHeaders(
+          headers, schemaRegistry.config().whitelistHeaders());
+      schemaRegistry.deleteSubjectCompatibilityConfigOrForward(subject, headerProperties);
+      deletedConfig = new Config(currentCompatibility.name);
+    } catch (OperationNotPermittedException e) {
+      throw Errors.operationNotPermittedException(e.getMessage());
+    } catch (SchemaRegistryStoreException e) {
+      throw Errors.storeException("Failed to delete compatibility level", e);
+    } catch (UnknownLeaderException e) {
+      throw Errors.unknownLeaderException("Failed to delete compatibility level", e);
+    } catch (SchemaRegistryRequestForwardingException e) {
+      throw Errors.requestForwardingFailedException("Error while forwarding delete config request"
+          + " to the leader", e);
+    }
+    asyncResponse.resume(deletedConfig);
+  }
 }

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

@@ -24,12 +24,15 @@
 
 import javax.validation.constraints.NotNull;
 import javax.ws.rs.Consumes;
+import javax.ws.rs.DELETE;
 import javax.ws.rs.GET;
 import javax.ws.rs.PUT;
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
 import javax.ws.rs.Produces;
 import javax.ws.rs.QueryParam;
+import javax.ws.rs.container.AsyncResponse;
+import javax.ws.rs.container.Suspended;
 import javax.ws.rs.core.Context;
 import javax.ws.rs.core.HttpHeaders;
 import java.util.Locale;
@@ -151,4 +154,41 @@ public ModeUpdateRequest updateTopLevelMode(
   public ModeGetResponse getTopLevelMode() {
     return getMode(null, false);
   }
+
+  @DELETE
+  @Path("/{subject}")
+  @ApiOperation(value = "Deletes the specified subject-level mode and revert to "
+      + "the global default.", response = Mode.class)
+  @ApiResponses(value = {
+      @ApiResponse(code = 404, message = "Error code 40401 -- Subject not found"),
+      @ApiResponse(code = 500, message = "Error code 50001 -- Error in the backend datastore")
+  })
+  public void deleteSubjectMode(
+      final @Suspended AsyncResponse asyncResponse,
+      @Context HttpHeaders headers,
+      @ApiParam(value = "the name of the subject", required = true)
+      @PathParam("subject") String subject) {
+    log.info("Deleting mode for subject {}", subject);
+    Mode deletedMode;
+    try {
+      deletedMode = schemaRegistry.getMode(subject);
+      if (deletedMode == null) {
+        throw Errors.subjectNotFoundException(subject);
+      }
+
+      Map<String, String> headerProperties = requestHeaderBuilder.buildRequestHeaders(
+          headers, schemaRegistry.config().whitelistHeaders());
+      schemaRegistry.deleteSubjectModeOrForward(subject, headerProperties);
+    } catch (OperationNotPermittedException e) {
+      throw Errors.operationNotPermittedException(e.getMessage());
+    } catch (SchemaRegistryStoreException e) {
+      throw Errors.storeException("Failed to delete mode", e);
+    } catch (UnknownLeaderException e) {
+      throw Errors.unknownLeaderException("Failed to delete mode", e);
+    } catch (SchemaRegistryRequestForwardingException e) {
+      throw Errors.requestForwardingFailedException("Error while forwarding delete mode request"
+          + " to the leader", e);
+    }
+    asyncResponse.resume(deletedMode);
+  }
 }