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

Return 'Bad request' if there is 'Internal Server Error' from Schema Registry

Browse source
This commit is contained in:
Ildar Almakaev 2021-01-29 11:21:18 +03:00
parent d853862790
commit 2525772162
3 changed files with 14 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -11,6 +11,7 @@ import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity; import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.BodyInserters; import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
@ -78,12 +79,16 @@ public class SchemaRegistryService {
public Mono<ResponseEntity<SchemaSubject>> createNewSubject(String clusterName, String subjectSchema, Mono<NewSchemaSubject> newSchemaSubject) { public Mono<ResponseEntity<SchemaSubject>> createNewSubject(String clusterName, String subjectSchema, Mono<NewSchemaSubject> newSchemaSubject) {
return clustersStorage.getClusterByName(clusterName) return clustersStorage.getClusterByName(clusterName)
.map(cluster -> webClient.post() .map(cluster -> {
.uri(cluster.getSchemaRegistry() + URL_SUBJECT_VERSIONS, subjectSchema) log.info("Submitting a new subject: {} to Schema Registry: {}", subjectSchema, cluster.getSchemaRegistry());
.contentType(MediaType.APPLICATION_JSON) return webClient.post()
.body(BodyInserters.fromPublisher(newSchemaSubject, NewSchemaSubject.class)) .uri(cluster.getSchemaRegistry() + URL_SUBJECT_VERSIONS, subjectSchema)
.retrieve() .contentType(MediaType.APPLICATION_JSON)
.toEntity(SchemaSubject.class)) .body(BodyInserters.fromPublisher(newSchemaSubject, NewSchemaSubject.class))
.retrieve()
.onStatus(HttpStatus.INTERNAL_SERVER_ERROR::equals, ClientResponse::createException)
.toEntity(SchemaSubject.class);
})
.orElse(Mono.error(new NotFoundException("No such cluster"))); .orElse(Mono.error(new NotFoundException("No such cluster")));
} }
} }

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

@ -137,7 +137,7 @@ public class MetricsRestController implements ApiClustersApi {
@Valid Mono<NewSchemaSubject> newSchemaSubject, @Valid Mono<NewSchemaSubject> newSchemaSubject,
ServerWebExchange exchange) { ServerWebExchange exchange) {
return schemaRegistryService.createNewSubject(clusterName, schemaName, newSchemaSubject) return schemaRegistryService.createNewSubject(clusterName, schemaName, newSchemaSubject)
.onErrorReturn(ResponseEntity.notFound().build()); .onErrorReturn(ResponseEntity.badRequest().build());
} }
@Override @Override

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

@ -434,6 +434,8 @@ paths:
application/json: application/json:
schema: schema:
$ref: '#/components/schemas/SchemaSubject' $ref: '#/components/schemas/SchemaSubject'
400:
description: Bad request
/api/clusters/{clusterName}/schemas/{schemaName}/versions/{version}: /api/clusters/{clusterName}/schemas/{schemaName}/versions/{version}:
get: get: