Add GET/DELETE actions for the latest schema version

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/service/SchemaRegistryService.java

@@ -28,6 +28,7 @@ public class SchemaRegistryService {
     private static final String URL_SUBJECT = "/subjects/{subjectName}";
     private static final String URL_SUBJECT_VERSIONS = "/subjects/{subjectName}/versions";
     private static final String URL_SUBJECT_BY_VERSION = "/subjects/{subjectName}/versions/{version}";
+    private static final String LATEST = "latest";
 
     private final ClustersStorage clustersStorage;
     private final WebClient webClient;
@@ -37,7 +38,9 @@ public class SchemaRegistryService {
                 .map(cluster -> webClient.get()
                         .uri(cluster.getSchemaRegistry() + URL_SUBJECTS)
                         .retrieve()
-                        .bodyToFlux(String.class))
+                        .onStatus(HttpStatus::is5xxServerError, ClientResponse::createException)
+                        .bodyToFlux(String.class)
+                        .doOnError(log::error))
                 .orElse(Flux.error(new NotFoundException("No such cluster")));
     }
 
@@ -52,6 +55,14 @@ public class SchemaRegistryService {
     }
 
     public Flux<SchemaSubject> getSchemaSubjectByVersion(String clusterName, String subjectName, Integer version) {
+        return this.getSchemaSubject(clusterName, subjectName, String.valueOf(version));
+    }
+
+    public Flux<SchemaSubject> getLatestSchemaSubject(String clusterName, String subjectName) {
+        return this.getSchemaSubject(clusterName, subjectName, LATEST);
+    }
+
+    private Flux<SchemaSubject> getSchemaSubject(String clusterName, String subjectName, String version) {
         return clustersStorage.getClusterByName(clusterName)
                 .map(cluster -> webClient.get()
                         .uri(cluster.getSchemaRegistry() + URL_SUBJECT_BY_VERSION, subjectName, version)
@@ -62,6 +73,14 @@ public class SchemaRegistryService {
     }
 
     public Mono<ResponseEntity<Void>> deleteSchemaSubjectByVersion(String clusterName, String subjectName, Integer version) {
+        return this.deleteSchemaSubject(clusterName, subjectName, String.valueOf(version));
+    }
+
+    public Mono<ResponseEntity<Void>> deleteLatestSchemaSubject(String clusterName, String subjectName) {
+        return this.deleteSchemaSubject(clusterName, subjectName, LATEST);
+    }
+
+    private Mono<ResponseEntity<Void>> deleteSchemaSubject(String clusterName, String subjectName, String version) {
         return clustersStorage.getClusterByName(clusterName)
                 .map(cluster -> webClient.delete()
                         .uri(cluster.getSchemaRegistry() + URL_SUBJECT_BY_VERSION, subjectName, version)
@@ -83,16 +102,14 @@ public class SchemaRegistryService {
 
     public Mono<ResponseEntity<SchemaSubject>> createNewSubject(String clusterName, String subjectSchema, Mono<NewSchemaSubject> newSchemaSubject) {
         return clustersStorage.getClusterByName(clusterName)
-                .map(cluster -> {
+                .map(cluster -> webClient.post()
-                    log.info("Submitting a new subject: {} to Schema Registry: {}", subjectSchema, cluster.getSchemaRegistry());
+                        .uri(cluster.getSchemaRegistry() + URL_SUBJECT_VERSIONS, subjectSchema)
-                    return webClient.post()
+                        .contentType(MediaType.APPLICATION_JSON)
-                            .uri(cluster.getSchemaRegistry() + URL_SUBJECT_VERSIONS, subjectSchema)
+                        .body(BodyInserters.fromPublisher(newSchemaSubject, NewSchemaSubject.class))
-                            .contentType(MediaType.APPLICATION_JSON)
+                        .retrieve()
-                            .body(BodyInserters.fromPublisher(newSchemaSubject, NewSchemaSubject.class))
+                        .onStatus(HttpStatus::isError, ClientResponse::createException)
-                            .retrieve()
+                        .toEntity(SchemaSubject.class)
-                            .onStatus(HttpStatus.INTERNAL_SERVER_ERROR::equals, ClientResponse::createException)
+                        .doOnError(log::error))
-                            .toEntity(SchemaSubject.class);
-                })
                 .orElse(Mono.error(new NotFoundException("No such cluster")));
     }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/rest/MetricsRestController.java

@@ -104,6 +104,12 @@ public class MetricsRestController implements ApiClustersApi {
                 .switchIfEmpty(Mono.just(ResponseEntity.notFound().build())); // TODO: check behaviour on cluster not found and empty groups list
     }
 
+    @Override
+    public Mono<ResponseEntity<Flux<SchemaSubject>>> getLatestSchemaByVersion(String clusterName, String schemaName, ServerWebExchange exchange) {
+        Flux<SchemaSubject> flux = schemaRegistryService.getLatestSchemaSubject(clusterName, schemaName);
+        return Mono.just(ResponseEntity.ok(flux)).onErrorReturn(ResponseEntity.notFound().build());
+    }
+
     @Override
     public Mono<ResponseEntity<Flux<SchemaSubject>>> getSchemaByVersion(String clusterName, String schemaName, Integer version, ServerWebExchange exchange) {
         Flux<SchemaSubject> flux = schemaRegistryService.getSchemaSubjectByVersion(clusterName, schemaName, version);
@@ -122,6 +128,12 @@ public class MetricsRestController implements ApiClustersApi {
                 .onErrorReturn(ResponseEntity.notFound().build());
     }
 
+    @Override
+    public Mono<ResponseEntity<Void>> deleteLatestSchema(String clusterName, String schemaName, ServerWebExchange exchange) {
+        return schemaRegistryService.deleteLatestSchemaSubject(clusterName, schemaName)
+                .onErrorReturn(ResponseEntity.notFound().build());
+    }
+
     @Override
     public Mono<ResponseEntity<Void>> deleteSchemaByVersion(String clusterName, String subjectName, Integer version, ServerWebExchange exchange) {
         return schemaRegistryService.deleteSchemaSubjectByVersion(clusterName, subjectName, version)

kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml

@@ -437,6 +437,55 @@ paths:
         400:
           description: Bad request
 
+  /api/clusters/{clusterName}/schemas/{schemaName}/latest:
+    get:
+      tags:
+        - /api/clusters
+      summary: get the latest schema from Schema Registry service
+      operationId: getLatestSchemaByVersion
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: schemaName
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        200:
+          description: OK
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: '#/components/schemas/SchemaSubject'
+    delete:
+      tags:
+        - /api/clusters
+      summary: delete the latest schema from schema registry
+      operationId: deleteLatestSchema
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: schemaName
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        200:
+          description: OK
+        404:
+          description: Not found
+
+
   /api/clusters/{clusterName}/schemas/{schemaName}/versions/{version}:
     get:
       tags: