0ct0pu5/kafka-ui
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor schema api contract

Browse source
This commit is contained in:
Ildar Almakaev 2021-01-29 10:35:54 +03:00
parent 08d855f975
commit d853862790
4 changed files with 67 additions and 66 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/exception/NotFoundException.java
View file

@ -5,6 +5,11 @@ import org.springframework.web.bind.annotation.ResponseStatus;
@ResponseStatus(HttpStatus.NOT_FOUND)
public class NotFoundException extends RuntimeException {
public NotFoundException() {
super();
}
public NotFoundException(String message) {
super(message);
}

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

@ -3,7 +3,7 @@ package com.provectus.kafka.ui.cluster.service;
import com.provectus.kafka.ui.cluster.exception.NotFoundException;
import com.provectus.kafka.ui.cluster.model.ClustersStorage;
import com.provectus.kafka.ui.model.NewSchemaSubject;
import com.provectus.kafka.ui.model.SubjectSchema;
import com.provectus.kafka.ui.model.SchemaSubject;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import org.springframework.http.HttpStatus;
@ -19,10 +19,10 @@ import reactor.core.publisher.Mono;
@Log4j2
@RequiredArgsConstructor
public class SchemaRegistryService {
public static final String URL_SUBJECTS = "/subjects";
public static final String URL_SUBJECT = "/subjects/{subjectName}";
public static final String URL_SUBJECT_VERSIONS = "/subjects/{subjectName}/versions";
public static final String URL_SUBJECT_BY_VERSION = "/subjects/{subjectName}/versions/{version}";
private static final String URL_SUBJECTS = "/subjects";
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 final ClustersStorage clustersStorage;
private final WebClient webClient;
@ -33,7 +33,7 @@ public class SchemaRegistryService {
.uri(cluster.getSchemaRegistry() + URL_SUBJECTS)
.retrieve()
.bodyToFlux(String.class))
.orElse(Flux.empty());
.orElse(Flux.error(new NotFoundException("No such cluster")));
}
public Flux<Integer> getSchemaSubjectVersions(String clusterName, String subjectName) {
@ -43,47 +43,47 @@ public class SchemaRegistryService {
.retrieve()
.onStatus(HttpStatus.NOT_FOUND::equals, resp -> Mono.error(new NotFoundException("No such subject")))
.bodyToFlux(Integer.class))
.orElse(Flux.empty());
.orElse(Flux.error(new NotFoundException("No such cluster")));
}
public Flux<SubjectSchema> getSchemaSubjectByVersion(String clusterName, String subjectName, Integer version) {
public Flux<SchemaSubject> getSchemaSubjectByVersion(String clusterName, String subjectName, Integer version) {
return clustersStorage.getClusterByName(clusterName)
.map(cluster -> webClient.get()
.uri(cluster.getSchemaRegistry() + URL_SUBJECT_BY_VERSION, subjectName, version)
.retrieve()
.onStatus(HttpStatus.NOT_FOUND::equals, resp -> Mono.error(new NotFoundException("No such subject or version")))
.bodyToFlux(SubjectSchema.class))
.orElse(Flux.empty());
.bodyToFlux(SchemaSubject.class))
.orElse(Flux.error(new NotFoundException()));
}
public Mono<Object> deleteSchemaSubjectByVersion(String clusterName, String subjectName, Integer version) {
public Mono<ResponseEntity<Void>> deleteSchemaSubjectByVersion(String clusterName, String subjectName, Integer version) {
return clustersStorage.getClusterByName(clusterName)
.map(cluster -> webClient.delete()
.uri(cluster.getSchemaRegistry() + URL_SUBJECT_BY_VERSION, subjectName, version)
.retrieve()
.onStatus(HttpStatus.NOT_FOUND::equals, resp -> Mono.error(new NotFoundException("No such subject or version")))
.bodyToMono(Object.class))
.orElse(Mono.empty());
.toBodilessEntity())
.orElse(Mono.error(new NotFoundException("No such cluster")));
}
public Mono<Object> deleteSchemaSubject(String clusterName, String subjectName) {
public Mono<ResponseEntity<Void>> deleteSchemaSubject(String clusterName, String subjectName) {
return clustersStorage.getClusterByName(clusterName)
.map(cluster -> webClient.delete()
.uri(cluster.getSchemaRegistry() + URL_SUBJECT, subjectName)
.retrieve()
.onStatus(HttpStatus.NOT_FOUND::equals, resp -> Mono.error(new NotFoundException("No such subject or version")))
.bodyToMono(Object.class))
.orElse(Mono.empty());
.toBodilessEntity())
.orElse(Mono.error(new NotFoundException("No such cluster")));
}
public Mono<ResponseEntity<SubjectSchema>> createNewSubject(String clusterName, String subjectSchema, Mono<NewSchemaSubject> newSchemaSubject) {
public Mono<ResponseEntity<SchemaSubject>> createNewSubject(String clusterName, String subjectSchema, Mono<NewSchemaSubject> newSchemaSubject) {
return clustersStorage.getClusterByName(clusterName)
.map(cluster -> webClient.post()
.uri(cluster.getSchemaRegistry() + URL_SUBJECT_VERSIONS, subjectSchema)
.contentType(MediaType.APPLICATION_JSON)
.body(BodyInserters.fromPublisher(newSchemaSubject, NewSchemaSubject.class))
.retrieve()
.toEntity(SubjectSchema.class))
.orElse(Mono.empty());
.toEntity(SchemaSubject.class))
.orElse(Mono.error(new NotFoundException("No such cluster")));
}
}

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

@ -103,35 +103,41 @@ public class MetricsRestController implements ApiClustersApi {
}
@Override
public Mono<ResponseEntity<Flux<SubjectSchema>>> getSchemaSubjectByVersion(String clusterName, String subjectName, Integer version, ServerWebExchange exchange) {
Flux<SubjectSchema> flux = schemaRegistryService.getSchemaSubjectByVersion(clusterName, subjectName, version);
return Mono.just(ResponseEntity.ok(flux));
public Mono<ResponseEntity<Flux<SchemaSubject>>> getSchemaByVersion(String clusterName, String schemaName, Integer version, ServerWebExchange exchange) {
Flux<SchemaSubject> flux = schemaRegistryService.getSchemaSubjectByVersion(clusterName, schemaName, version);
return Mono.just(ResponseEntity.ok(flux)).onErrorReturn(ResponseEntity.notFound().build());
}
@Override
public Mono<ResponseEntity<Flux<String>>> getSchemaSubjects(String clusterName, ServerWebExchange exchange) {
public Mono<ResponseEntity<Flux<String>>> getSchemas(String clusterName, ServerWebExchange exchange) {
Flux<String> subjects = schemaRegistryService.getAllSchemaSubjects(clusterName);
return Mono.just(ResponseEntity.ok(subjects));
}
@Override
public Mono<ResponseEntity<Flux<Integer>>> getSchemaSubjectVersions(String clusterName, String subjectName, ServerWebExchange exchange) {
return Mono.just(ResponseEntity.ok(schemaRegistryService.getSchemaSubjectVersions(clusterName, subjectName)));
public Mono<ResponseEntity<Flux<Integer>>> getSchemaVersions(String clusterName, String subjectName, ServerWebExchange exchange) {
return Mono.just(ResponseEntity.ok(schemaRegistryService.getSchemaSubjectVersions(clusterName, subjectName)))
.onErrorReturn(ResponseEntity.notFound().build());
}
@Override
public Mono<ResponseEntity<Object>> deleteSchemaByVersion(String clusterName, String subjectName, Integer version, ServerWebExchange exchange) {
return Mono.just(ResponseEntity.ok(schemaRegistryService.deleteSchemaSubjectByVersion(clusterName, subjectName, version)));
public Mono<ResponseEntity<Void>> deleteSchemaByVersion(String clusterName, String subjectName, Integer version, ServerWebExchange exchange) {
return schemaRegistryService.deleteSchemaSubjectByVersion(clusterName, subjectName, version)
.onErrorReturn(ResponseEntity.notFound().build());
}
@Override
public Mono<ResponseEntity<Object>> deleteSchemaSubject(String clusterName, String subjectName, ServerWebExchange exchange) {
return Mono.just(ResponseEntity.ok(schemaRegistryService.deleteSchemaSubject(clusterName, subjectName)));
public Mono<ResponseEntity<Void>> deleteSchema(String clusterName, String subjectName, ServerWebExchange exchange) {
return schemaRegistryService.deleteSchemaSubject(clusterName, subjectName)
.onErrorReturn(ResponseEntity.notFound().build());
}
@Override
public Mono<ResponseEntity<SubjectSchema>> createNewSubjectSchema(String clusterName, String subjectName, @Valid Mono<NewSchemaSubject> newSchemaSubject, ServerWebExchange exchange) {
return schemaRegistryService.createNewSubject(clusterName, subjectName, newSchemaSubject);
public Mono<ResponseEntity<SchemaSubject>> createNewSchema(String clusterName, String schemaName,
@Valid Mono<NewSchemaSubject> newSchemaSubject,
ServerWebExchange exchange) {
return schemaRegistryService.createNewSubject(clusterName, schemaName, newSchemaSubject)
.onErrorReturn(ResponseEntity.notFound().build());
}
@Override

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

@ -335,12 +335,12 @@ paths:
items:
$ref: '#/components/schemas/ConsumerGroup'
/api/clusters/{clusterName}/schema/subjects:
/api/clusters/{clusterName}/schemas:
get:
tags:
- /api/clusters
summary: get all subjects from schema registry
operationId: getSchemaSubjects
summary: get all schemas from Schema Registry service
operationId: getSchemas
parameters:
- name: clusterName
in: path
@ -357,19 +357,19 @@ paths:
items:
type: string
/api/clusters/{clusterName}/schema/subjects/{subjectName}:
/api/clusters/{clusterName}/schemas/{schemaName}:
delete:
tags:
- /api/clusters
summary: delete subject from schema registry
operationId: deleteSchemaSubject
summary: delete schema from Schema Registry service
operationId: deleteSchema
parameters:
- name: clusterName
in: path
required: true
schema:
type: string
- name: subjectName
- name: schemaName
in: path
required: true
schema:
@ -377,24 +377,22 @@ paths:
responses:
200:
description: OK
content:
application/json:
schema:
type: object
404:
description: Not found
/api/clusters/{clusterName}/schema/subjects/{subjectName}/versions:
/api/clusters/{clusterName}/schemas/{schemaName}/versions:
get:
tags:
- /api/clusters
summary: get subject versions from schema registry
operationId: getSchemaSubjectVersions
summary: get all version of schema from Schema Registry service
operationId: getSchemaVersions
parameters:
- name: clusterName
in: path
required: true
schema:
type: string
- name: subjectName
- name: schemaName
in: path
required: true
schema:
@ -412,14 +410,14 @@ paths:
tags:
- /api/clusters
summary: create a new subject schema
operationId: createNewSubjectSchema
operationId: createNewSchema
parameters:
- name: clusterName
in: path
required: true
schema:
type: string
- name: subjectName
- name: schemaName
in: path
required: true
schema:
@ -435,21 +433,21 @@ paths:
content:
application/json:
schema:
$ref: '#/components/schemas/SubjectSchema'
$ref: '#/components/schemas/SchemaSubject'
/api/clusters/{clusterName}/schema/subjects/{subjectName}/versions/{version}:
/api/clusters/{clusterName}/schemas/{schemaName}/versions/{version}:
get:
tags:
- /api/clusters
summary: get schema of subject by version from schema registry
operationId: getSchemaSubjectByVersion
summary: get schema by version from Schema Registry service
operationId: getSchemaByVersion
parameters:
- name: clusterName
in: path
required: true
schema:
type: string
- name: subjectName
- name: schemaName
in: path
required: true
schema:
@ -467,7 +465,7 @@ paths:
schema:
type: array
items:
$ref: '#/components/schemas/SubjectSchema'
$ref: '#/components/schemas/SchemaSubject'
delete:
tags:
- /api/clusters
@ -479,7 +477,7 @@ paths:
required: true
schema:
type: string
- name: subjectName
- name: schemaName
in: path
required: true
schema:
@ -490,18 +488,10 @@ paths:
schema:
type: integer
responses:
204:
description: Deleted
content:
application/json:
schema:
type: object
200:
description: OK
404:
description: Not found
content:
application/json:
schema:
type: object
components:
schemas:
@ -827,7 +817,7 @@ components:
additionalProperties:
type: number
SubjectSchema:
SchemaSubject:
type: object
properties:
subject: