From 91beb6dc894e5697d3269d9a461d624c758abbff Mon Sep 17 00:00:00 2001 From: Dmitry Werner Date: Mon, 3 Feb 2025 17:57:38 +0500 Subject: [PATCH 1/2] BE: Chore: Cleanup api module --- .../ui/client/RetryingKafkaConnectClient.java | 4 +- .../config/auth/BasicAuthSecurityConfig.java | 4 - .../logout/CognitoLogoutSuccessHandler.java | 2 +- .../io/kafbat/ui/emitter/OffsetsInfo.java | 2 +- .../kafbat/ui/emitter/ResultSizeLimiter.java | 23 ----- .../io/kafbat/ui/exception/CelException.java | 2 +- .../exception/ConnectNotFoundException.java | 13 --- .../exception/DuplicateEntityException.java | 13 --- .../io/kafbat/ui/exception/ErrorCode.java | 6 -- .../GlobalErrorWebExceptionHandler.java | 7 +- ...afkaConnectConflictResponseException.java} | 5 +- .../ui/exception/KsqlDbNotFoundException.java | 13 --- .../SchemaFailedToDeleteException.java | 13 --- .../UnprocessableEntityException.java | 14 --- .../kafbat/ui/mapper/ConsumerGroupMapper.java | 26 ++--- .../ui/mapper/DescribeLogDirsMapper.java | 5 +- .../io/kafbat/ui/serdes/SerdeInstance.java | 2 +- .../kafbat/ui/serdes/builtin/Base64Serde.java | 2 +- .../io/kafbat/ui/serdes/builtin/HexSerde.java | 2 +- .../ui/serdes/builtin/ProtobufFileSerde.java | 2 +- .../ui/service/KafkaConnectService.java | 2 - .../ui/service/ReactiveAdminClient.java | 6 -- .../ui/service/SchemaRegistryService.java | 2 +- .../io/kafbat/ui/service/TopicsService.java | 12 +-- .../io/kafbat/ui/service/acl/AclsService.java | 4 +- .../kafbat/ui/service/ksql/KsqlApiClient.java | 2 +- .../kafbat/ui/service/ksql/KsqlGrammar.java | 11 +-- .../service/ksql/response/ResponseParser.java | 97 ++++++++----------- .../ui/service/metrics/MetricsCollector.java | 9 +- .../kafbat/ui/util/EmptyRedirectStrategy.java | 22 +---- .../ui/util/KafkaServicesValidation.java | 5 +- .../io/kafbat/ui/util/ReactiveFailover.java | 4 +- .../jsonschema/AvroJsonSchemaConverter.java | 1 - .../util/jsonschema/JsonAvroConversion.java | 42 +++----- .../kafbat/ui/util/jsonschema/JsonSchema.java | 9 -- 35 files changed, 97 insertions(+), 291 deletions(-) delete mode 100644 api/src/main/java/io/kafbat/ui/emitter/ResultSizeLimiter.java delete mode 100644 api/src/main/java/io/kafbat/ui/exception/ConnectNotFoundException.java delete mode 100644 api/src/main/java/io/kafbat/ui/exception/DuplicateEntityException.java rename api/src/main/java/io/kafbat/ui/exception/{KafkaConnectConflictReponseException.java => KafkaConnectConflictResponseException.java} (67%) delete mode 100644 api/src/main/java/io/kafbat/ui/exception/KsqlDbNotFoundException.java delete mode 100644 api/src/main/java/io/kafbat/ui/exception/SchemaFailedToDeleteException.java delete mode 100644 api/src/main/java/io/kafbat/ui/exception/UnprocessableEntityException.java diff --git a/api/src/main/java/io/kafbat/ui/client/RetryingKafkaConnectClient.java b/api/src/main/java/io/kafbat/ui/client/RetryingKafkaConnectClient.java index df2da3e55..cdf5bce14 100644 --- a/api/src/main/java/io/kafbat/ui/client/RetryingKafkaConnectClient.java +++ b/api/src/main/java/io/kafbat/ui/client/RetryingKafkaConnectClient.java @@ -12,7 +12,7 @@ import io.kafbat.ui.connect.model.ConnectorTopics; import io.kafbat.ui.connect.model.NewConnector; import io.kafbat.ui.connect.model.TaskStatus; -import io.kafbat.ui.exception.KafkaConnectConflictReponseException; +import io.kafbat.ui.exception.KafkaConnectConflictResponseException; import io.kafbat.ui.exception.ValidationException; import io.kafbat.ui.util.WebClientConfigurator; import jakarta.validation.constraints.NotNull; @@ -48,7 +48,7 @@ private static Retry conflictCodeRetry() { .fixedDelay(MAX_RETRIES, RETRIES_DELAY) .filter(e -> e instanceof WebClientResponseException.Conflict) .onRetryExhaustedThrow((spec, signal) -> - new KafkaConnectConflictReponseException( + new KafkaConnectConflictResponseException( (WebClientResponseException.Conflict) signal.failure())); } diff --git a/api/src/main/java/io/kafbat/ui/config/auth/BasicAuthSecurityConfig.java b/api/src/main/java/io/kafbat/ui/config/auth/BasicAuthSecurityConfig.java index db8ef8153..788c33bdd 100644 --- a/api/src/main/java/io/kafbat/ui/config/auth/BasicAuthSecurityConfig.java +++ b/api/src/main/java/io/kafbat/ui/config/auth/BasicAuthSecurityConfig.java @@ -1,8 +1,6 @@ package io.kafbat.ui.config.auth; -import io.kafbat.ui.util.EmptyRedirectStrategy; import io.kafbat.ui.util.StaticFileWebFilter; -import java.net.URI; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.context.annotation.Bean; @@ -12,8 +10,6 @@ import org.springframework.security.config.web.server.SecurityWebFiltersOrder; import org.springframework.security.config.web.server.ServerHttpSecurity; import org.springframework.security.web.server.SecurityWebFilterChain; -import org.springframework.security.web.server.authentication.RedirectServerAuthenticationSuccessHandler; -import org.springframework.security.web.server.authentication.logout.RedirectServerLogoutSuccessHandler; import org.springframework.security.web.server.util.matcher.ServerWebExchangeMatchers; @Configuration diff --git a/api/src/main/java/io/kafbat/ui/config/auth/logout/CognitoLogoutSuccessHandler.java b/api/src/main/java/io/kafbat/ui/config/auth/logout/CognitoLogoutSuccessHandler.java index d98ea22af..e58f51ab3 100644 --- a/api/src/main/java/io/kafbat/ui/config/auth/logout/CognitoLogoutSuccessHandler.java +++ b/api/src/main/java/io/kafbat/ui/config/auth/logout/CognitoLogoutSuccessHandler.java @@ -40,7 +40,7 @@ public Mono handle(WebFilterExchange exchange, Authentication authenticati requestUri.getPath(), requestUri.getQuery()); final UriComponents baseUrl = UriComponentsBuilder - .fromHttpUrl(fullUrl) + .fromUriString(fullUrl) .replacePath("/") .replaceQuery(null) .fragment(null) diff --git a/api/src/main/java/io/kafbat/ui/emitter/OffsetsInfo.java b/api/src/main/java/io/kafbat/ui/emitter/OffsetsInfo.java index a361834d0..7f34e1708 100644 --- a/api/src/main/java/io/kafbat/ui/emitter/OffsetsInfo.java +++ b/api/src/main/java/io/kafbat/ui/emitter/OffsetsInfo.java @@ -53,7 +53,7 @@ private Map firstOffsetsForPolling(Consumer consumer Collection partitions) { try { // we try to use offsetsForTimes() to find earliest offsets, since for - // some topics (like compacted) beginningOffsets() ruturning 0 offsets + // some topics (like compacted) beginningOffsets() returning 0 offsets // even when effectively first offset can be very high var offsets = consumer.offsetsForTimes( partitions.stream().collect(Collectors.toMap(p -> p, p -> 0L)) diff --git a/api/src/main/java/io/kafbat/ui/emitter/ResultSizeLimiter.java b/api/src/main/java/io/kafbat/ui/emitter/ResultSizeLimiter.java deleted file mode 100644 index 3e0ec2a43..000000000 --- a/api/src/main/java/io/kafbat/ui/emitter/ResultSizeLimiter.java +++ /dev/null @@ -1,23 +0,0 @@ -package io.kafbat.ui.emitter; - -import io.kafbat.ui.model.TopicMessageEventDTO; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.Predicate; - -public class ResultSizeLimiter implements Predicate { - private final AtomicInteger processed = new AtomicInteger(); - private final int limit; - - public ResultSizeLimiter(int limit) { - this.limit = limit; - } - - @Override - public boolean test(TopicMessageEventDTO event) { - if (event.getType().equals(TopicMessageEventDTO.TypeEnum.MESSAGE)) { - final int i = processed.incrementAndGet(); - return i <= limit; - } - return true; - } -} diff --git a/api/src/main/java/io/kafbat/ui/exception/CelException.java b/api/src/main/java/io/kafbat/ui/exception/CelException.java index e904368f6..191984afb 100644 --- a/api/src/main/java/io/kafbat/ui/exception/CelException.java +++ b/api/src/main/java/io/kafbat/ui/exception/CelException.java @@ -1,7 +1,7 @@ package io.kafbat.ui.exception; public class CelException extends CustomBaseException { - private String celOriginalExpression; + private final String celOriginalExpression; public CelException(String celOriginalExpression, String errorMessage) { super("CEL error. Original expression: %s. Error message: %s".formatted(celOriginalExpression, errorMessage)); diff --git a/api/src/main/java/io/kafbat/ui/exception/ConnectNotFoundException.java b/api/src/main/java/io/kafbat/ui/exception/ConnectNotFoundException.java deleted file mode 100644 index 5978c2e93..000000000 --- a/api/src/main/java/io/kafbat/ui/exception/ConnectNotFoundException.java +++ /dev/null @@ -1,13 +0,0 @@ -package io.kafbat.ui.exception; - -public class ConnectNotFoundException extends CustomBaseException { - - public ConnectNotFoundException() { - super("Connect not found"); - } - - @Override - public ErrorCode getErrorCode() { - return ErrorCode.CONNECT_NOT_FOUND; - } -} diff --git a/api/src/main/java/io/kafbat/ui/exception/DuplicateEntityException.java b/api/src/main/java/io/kafbat/ui/exception/DuplicateEntityException.java deleted file mode 100644 index 23ba0c5af..000000000 --- a/api/src/main/java/io/kafbat/ui/exception/DuplicateEntityException.java +++ /dev/null @@ -1,13 +0,0 @@ -package io.kafbat.ui.exception; - -public class DuplicateEntityException extends CustomBaseException { - - public DuplicateEntityException(String message) { - super(message); - } - - @Override - public ErrorCode getErrorCode() { - return ErrorCode.DUPLICATED_ENTITY; - } -} diff --git a/api/src/main/java/io/kafbat/ui/exception/ErrorCode.java b/api/src/main/java/io/kafbat/ui/exception/ErrorCode.java index 6d4a732e3..32cf5c5c8 100644 --- a/api/src/main/java/io/kafbat/ui/exception/ErrorCode.java +++ b/api/src/main/java/io/kafbat/ui/exception/ErrorCode.java @@ -4,11 +4,8 @@ import org.slf4j.LoggerFactory; import org.springframework.http.HttpStatus; - public enum ErrorCode { - FORBIDDEN(403, HttpStatus.FORBIDDEN), - UNEXPECTED(5000, HttpStatus.INTERNAL_SERVER_ERROR), KSQL_API_ERROR(5001, HttpStatus.INTERNAL_SERVER_ERROR), BINDING_FAIL(4001, HttpStatus.BAD_REQUEST), @@ -16,13 +13,10 @@ public enum ErrorCode { VALIDATION_FAIL(4002, HttpStatus.BAD_REQUEST), READ_ONLY_MODE_ENABLE(4003, HttpStatus.METHOD_NOT_ALLOWED), CONNECT_CONFLICT_RESPONSE(4004, HttpStatus.CONFLICT), - DUPLICATED_ENTITY(4005, HttpStatus.CONFLICT), UNPROCESSABLE_ENTITY(4006, HttpStatus.UNPROCESSABLE_ENTITY), CLUSTER_NOT_FOUND(4007, HttpStatus.NOT_FOUND), TOPIC_NOT_FOUND(4008, HttpStatus.NOT_FOUND), SCHEMA_NOT_FOUND(4009, HttpStatus.NOT_FOUND), - CONNECT_NOT_FOUND(4010, HttpStatus.NOT_FOUND), - KSQLDB_NOT_FOUND(4011, HttpStatus.NOT_FOUND), DIR_NOT_FOUND(4012, HttpStatus.BAD_REQUEST), TOPIC_OR_PARTITION_NOT_FOUND(4013, HttpStatus.BAD_REQUEST), INVALID_REQUEST(4014, HttpStatus.BAD_REQUEST), diff --git a/api/src/main/java/io/kafbat/ui/exception/GlobalErrorWebExceptionHandler.java b/api/src/main/java/io/kafbat/ui/exception/GlobalErrorWebExceptionHandler.java index 61236f801..b9bf58c37 100644 --- a/api/src/main/java/io/kafbat/ui/exception/GlobalErrorWebExceptionHandler.java +++ b/api/src/main/java/io/kafbat/ui/exception/GlobalErrorWebExceptionHandler.java @@ -102,8 +102,7 @@ private Mono render(CustomBaseException baseException, ServerReq private Mono render(WebExchangeBindException exception, ServerRequest request) { Map> fieldErrorsMap = exception.getFieldErrors().stream() - .collect(Collectors - .toMap(FieldError::getField, f -> Set.of(extractFieldErrorMsg(f)), Sets::union)); + .collect(Collectors.toMap(FieldError::getField, f -> Set.of(extractFieldErrorMsg(f)), Sets::union)); var fieldsErrors = fieldErrorsMap.entrySet().stream() .map(e -> { @@ -151,9 +150,7 @@ private String requestId(ServerRequest request) { } private Consumer headers(ServerRequest request) { - return (HttpHeaders headers) -> { - CorsGlobalConfiguration.fillCorsHeader(headers, request.exchange().getRequest()); - }; + return (HttpHeaders headers) -> CorsGlobalConfiguration.fillCorsHeader(headers, request.exchange().getRequest()); } private BigDecimal currentTimestamp() { diff --git a/api/src/main/java/io/kafbat/ui/exception/KafkaConnectConflictReponseException.java b/api/src/main/java/io/kafbat/ui/exception/KafkaConnectConflictResponseException.java similarity index 67% rename from api/src/main/java/io/kafbat/ui/exception/KafkaConnectConflictReponseException.java rename to api/src/main/java/io/kafbat/ui/exception/KafkaConnectConflictResponseException.java index 48376d1ac..ad356b7e5 100644 --- a/api/src/main/java/io/kafbat/ui/exception/KafkaConnectConflictReponseException.java +++ b/api/src/main/java/io/kafbat/ui/exception/KafkaConnectConflictResponseException.java @@ -1,11 +1,10 @@ package io.kafbat.ui.exception; - import org.springframework.web.reactive.function.client.WebClientResponseException; -public class KafkaConnectConflictReponseException extends CustomBaseException { +public class KafkaConnectConflictResponseException extends CustomBaseException { - public KafkaConnectConflictReponseException(WebClientResponseException.Conflict e) { + public KafkaConnectConflictResponseException(WebClientResponseException.Conflict e) { super("Kafka Connect responded with 409 (Conflict) code. Response body: " + e.getResponseBodyAsString()); } diff --git a/api/src/main/java/io/kafbat/ui/exception/KsqlDbNotFoundException.java b/api/src/main/java/io/kafbat/ui/exception/KsqlDbNotFoundException.java deleted file mode 100644 index 255ccec80..000000000 --- a/api/src/main/java/io/kafbat/ui/exception/KsqlDbNotFoundException.java +++ /dev/null @@ -1,13 +0,0 @@ -package io.kafbat.ui.exception; - -public class KsqlDbNotFoundException extends CustomBaseException { - - public KsqlDbNotFoundException() { - super("KSQL DB not found"); - } - - @Override - public ErrorCode getErrorCode() { - return ErrorCode.KSQLDB_NOT_FOUND; - } -} diff --git a/api/src/main/java/io/kafbat/ui/exception/SchemaFailedToDeleteException.java b/api/src/main/java/io/kafbat/ui/exception/SchemaFailedToDeleteException.java deleted file mode 100644 index 05ba55c70..000000000 --- a/api/src/main/java/io/kafbat/ui/exception/SchemaFailedToDeleteException.java +++ /dev/null @@ -1,13 +0,0 @@ -package io.kafbat.ui.exception; - -public class SchemaFailedToDeleteException extends CustomBaseException { - - public SchemaFailedToDeleteException(String schemaName) { - super(String.format("Unable to delete schema with name %s", schemaName)); - } - - @Override - public ErrorCode getErrorCode() { - return ErrorCode.SCHEMA_NOT_DELETED; - } -} diff --git a/api/src/main/java/io/kafbat/ui/exception/UnprocessableEntityException.java b/api/src/main/java/io/kafbat/ui/exception/UnprocessableEntityException.java deleted file mode 100644 index fcd9e41fd..000000000 --- a/api/src/main/java/io/kafbat/ui/exception/UnprocessableEntityException.java +++ /dev/null @@ -1,14 +0,0 @@ -package io.kafbat.ui.exception; - - -public class UnprocessableEntityException extends CustomBaseException { - - public UnprocessableEntityException(String message) { - super(message); - } - - @Override - public ErrorCode getErrorCode() { - return ErrorCode.UNPROCESSABLE_ENTITY; - } -} diff --git a/api/src/main/java/io/kafbat/ui/mapper/ConsumerGroupMapper.java b/api/src/main/java/io/kafbat/ui/mapper/ConsumerGroupMapper.java index 72b3d65b4..800eab757 100644 --- a/api/src/main/java/io/kafbat/ui/mapper/ConsumerGroupMapper.java +++ b/api/src/main/java/io/kafbat/ui/mapper/ConsumerGroupMapper.java @@ -97,23 +97,15 @@ private static BrokerDTO mapCoordinator(Node node) { return new BrokerDTO().host(node.host()).id(node.id()).port(node.port()); } - private static ConsumerGroupStateDTO mapConsumerGroupState( - org.apache.kafka.common.ConsumerGroupState state) { - switch (state) { - case DEAD: - return ConsumerGroupStateDTO.DEAD; - case EMPTY: - return ConsumerGroupStateDTO.EMPTY; - case STABLE: - return ConsumerGroupStateDTO.STABLE; - case PREPARING_REBALANCE: - return ConsumerGroupStateDTO.PREPARING_REBALANCE; - case COMPLETING_REBALANCE: - return ConsumerGroupStateDTO.COMPLETING_REBALANCE; - default: - return ConsumerGroupStateDTO.UNKNOWN; - } + private static ConsumerGroupStateDTO mapConsumerGroupState(org.apache.kafka.common.ConsumerGroupState state) { + return switch (state) { + case DEAD -> ConsumerGroupStateDTO.DEAD; + case EMPTY -> ConsumerGroupStateDTO.EMPTY; + case STABLE -> ConsumerGroupStateDTO.STABLE; + case PREPARING_REBALANCE -> ConsumerGroupStateDTO.PREPARING_REBALANCE; + case COMPLETING_REBALANCE -> ConsumerGroupStateDTO.COMPLETING_REBALANCE; + default -> ConsumerGroupStateDTO.UNKNOWN; + }; } - } diff --git a/api/src/main/java/io/kafbat/ui/mapper/DescribeLogDirsMapper.java b/api/src/main/java/io/kafbat/ui/mapper/DescribeLogDirsMapper.java index 8a4e44e5c..bccd3a66b 100644 --- a/api/src/main/java/io/kafbat/ui/mapper/DescribeLogDirsMapper.java +++ b/api/src/main/java/io/kafbat/ui/mapper/DescribeLogDirsMapper.java @@ -42,7 +42,7 @@ private BrokersLogdirsDTO toBrokerLogDirs(Integer broker, String dirName, private BrokerTopicLogdirsDTO toTopicLogDirs(Integer broker, String name, List> partitions) { + DescribeLogDirsResponse.ReplicaInfo>> partitions) { BrokerTopicLogdirsDTO topic = new BrokerTopicLogdirsDTO(); topic.setName(name); topic.setPartitions( @@ -54,8 +54,7 @@ private BrokerTopicLogdirsDTO toTopicLogDirs(Integer broker, String name, } private BrokerTopicPartitionLogdirDTO topicPartitionLogDir(Integer broker, Integer partition, - DescribeLogDirsResponse.ReplicaInfo - replicaInfo) { + DescribeLogDirsResponse.ReplicaInfo replicaInfo) { BrokerTopicPartitionLogdirDTO logDir = new BrokerTopicPartitionLogdirDTO(); logDir.setBroker(broker); logDir.setPartition(partition); diff --git a/api/src/main/java/io/kafbat/ui/serdes/SerdeInstance.java b/api/src/main/java/io/kafbat/ui/serdes/SerdeInstance.java index 7c1826257..b0fdc9834 100644 --- a/api/src/main/java/io/kafbat/ui/serdes/SerdeInstance.java +++ b/api/src/main/java/io/kafbat/ui/serdes/SerdeInstance.java @@ -97,7 +97,7 @@ public void close() { try { serde.close(); } catch (Exception e) { - log.error("Error closing serde " + name, e); + log.error("Error closing serde {}", name, e); } return null; }); diff --git a/api/src/main/java/io/kafbat/ui/serdes/builtin/Base64Serde.java b/api/src/main/java/io/kafbat/ui/serdes/builtin/Base64Serde.java index 02e56ff22..515354695 100644 --- a/api/src/main/java/io/kafbat/ui/serdes/builtin/Base64Serde.java +++ b/api/src/main/java/io/kafbat/ui/serdes/builtin/Base64Serde.java @@ -40,7 +40,7 @@ public Serde.Serializer serializer(String topic, Serde.Target type) { return inputString -> { inputString = inputString.trim(); // it is actually a hack to provide ability to sent empty array as a key/value - if (inputString.length() == 0) { + if (inputString.isEmpty()) { return new byte[] {}; } return decoder.decode(inputString); diff --git a/api/src/main/java/io/kafbat/ui/serdes/builtin/HexSerde.java b/api/src/main/java/io/kafbat/ui/serdes/builtin/HexSerde.java index a3c958a06..ab7f66ebb 100644 --- a/api/src/main/java/io/kafbat/ui/serdes/builtin/HexSerde.java +++ b/api/src/main/java/io/kafbat/ui/serdes/builtin/HexSerde.java @@ -62,7 +62,7 @@ public Serializer serializer(String topic, Target type) { return input -> { input = input.trim(); // it is a hack to provide ability to sent empty array as a key/value - if (input.length() == 0) { + if (input.isEmpty()) { return new byte[] {}; } return HexFormat.of().parseHex(prepareInputForParse(input)); diff --git a/api/src/main/java/io/kafbat/ui/serdes/builtin/ProtobufFileSerde.java b/api/src/main/java/io/kafbat/ui/serdes/builtin/ProtobufFileSerde.java index 2c0939c03..19fe2f437 100644 --- a/api/src/main/java/io/kafbat/ui/serdes/builtin/ProtobufFileSerde.java +++ b/api/src/main/java/io/kafbat/ui/serdes/builtin/ProtobufFileSerde.java @@ -380,7 +380,7 @@ private Map knownProtoFiles() { } private ProtoFile loadKnownProtoFile(String path, Descriptors.FileDescriptor fileDescriptor) { - String protoFileString = null; + String protoFileString; // know type file contains either message or enum if (!fileDescriptor.getMessageTypes().isEmpty()) { protoFileString = new ProtobufSchema(fileDescriptor.getMessageTypes().get(0)).canonicalString(); diff --git a/api/src/main/java/io/kafbat/ui/service/KafkaConnectService.java b/api/src/main/java/io/kafbat/ui/service/KafkaConnectService.java index 31e4268a0..92bfc260b 100644 --- a/api/src/main/java/io/kafbat/ui/service/KafkaConnectService.java +++ b/api/src/main/java/io/kafbat/ui/service/KafkaConnectService.java @@ -1,6 +1,5 @@ package io.kafbat.ui.service; -import com.fasterxml.jackson.databind.ObjectMapper; import io.kafbat.ui.connect.api.KafkaConnectClientApi; import io.kafbat.ui.connect.model.ConnectorStatus; import io.kafbat.ui.connect.model.ConnectorStatusConnector; @@ -44,7 +43,6 @@ public class KafkaConnectService { private final ClusterMapper clusterMapper; private final KafkaConnectMapper kafkaConnectMapper; - private final ObjectMapper objectMapper; private final KafkaConfigSanitizer kafkaConfigSanitizer; public Flux getConnects(KafkaCluster cluster) { diff --git a/api/src/main/java/io/kafbat/ui/service/ReactiveAdminClient.java b/api/src/main/java/io/kafbat/ui/service/ReactiveAdminClient.java index 651f6d531..6aea290c3 100644 --- a/api/src/main/java/io/kafbat/ui/service/ReactiveAdminClient.java +++ b/api/src/main/java/io/kafbat/ui/service/ReactiveAdminClient.java @@ -389,12 +389,6 @@ static Mono> toMonoWithExceptionFilter(Map> v ); } - public Mono>> describeLogDirs() { - return describeCluster() - .map(d -> d.getNodes().stream().map(Node::id).collect(toList())) - .flatMap(this::describeLogDirs); - } - public Mono>> describeLogDirs( Collection brokerIds) { return toMono(client.describeLogDirs(brokerIds).all()) diff --git a/api/src/main/java/io/kafbat/ui/service/SchemaRegistryService.java b/api/src/main/java/io/kafbat/ui/service/SchemaRegistryService.java index 1bac22235..c725a787e 100644 --- a/api/src/main/java/io/kafbat/ui/service/SchemaRegistryService.java +++ b/api/src/main/java/io/kafbat/ui/service/SchemaRegistryService.java @@ -63,7 +63,7 @@ public Mono> getAllSubjectNames(KafkaCluster cluster) { @SneakyThrows private List parseSubjectListString(String subjectNamesStr) { //workaround for https://github.com/spring-projects/spring-framework/issues/24734 - return new JsonMapper().readValue(subjectNamesStr, new TypeReference>() { + return new JsonMapper().readValue(subjectNamesStr, new TypeReference<>() { }); } diff --git a/api/src/main/java/io/kafbat/ui/service/TopicsService.java b/api/src/main/java/io/kafbat/ui/service/TopicsService.java index 015a86838..95ad7bc5a 100644 --- a/api/src/main/java/io/kafbat/ui/service/TopicsService.java +++ b/api/src/main/java/io/kafbat/ui/service/TopicsService.java @@ -97,7 +97,7 @@ private Mono loadTopic(KafkaCluster c, String topicName) { /** * After creation topic can be invisible via API for some time. - * To workaround this, we retyring topic loading until it becomes visible. + * To workaround this, we're retrying topic loading until it becomes visible. */ private Mono loadTopicAfterCreation(KafkaCluster c, String topicName) { return loadTopic(c, topicName) @@ -137,8 +137,7 @@ private List createList(List orderedNames, .collect(toList()); } - private Mono getPartitionOffsets(Map - descriptionsMap, + private Mono getPartitionOffsets(Map descriptionsMap, ReactiveAdminClient ac) { var descriptions = descriptionsMap.values(); return ac.listOffsets(descriptions, OffsetSpec.earliest()) @@ -225,8 +224,7 @@ private Mono updateTopic(KafkaCluster cluster, .then(loadTopic(cluster, topicName))); } - public Mono updateTopic(KafkaCluster cl, String topicName, - Mono topicUpdate) { + public Mono updateTopic(KafkaCluster cl, String topicName, Mono topicUpdate) { return topicUpdate .flatMap(t -> updateTopic(cl, topicName, t)); } @@ -298,7 +296,7 @@ private Map> getPartitionsRea var brokers = brokersUsage.entrySet().stream() .sorted(Map.Entry.comparingByValue()) .map(Map.Entry::getKey) - .collect(toList()); + .toList(); // Iterate brokers and try to add them in assignment // while partition replicas count != requested replication factor @@ -326,7 +324,7 @@ private Map> getPartitionsRea var brokersUsageList = brokersUsage.entrySet().stream() .sorted(Map.Entry.comparingByValue(Comparator.reverseOrder())) .map(Map.Entry::getKey) - .collect(toList()); + .toList(); // Iterate brokers and try to remove them from assignment // while partition replicas count != requested replication factor diff --git a/api/src/main/java/io/kafbat/ui/service/acl/AclsService.java b/api/src/main/java/io/kafbat/ui/service/acl/AclsService.java index b3877a336..30078d435 100644 --- a/api/src/main/java/io/kafbat/ui/service/acl/AclsService.java +++ b/api/src/main/java/io/kafbat/ui/service/acl/AclsService.java @@ -112,13 +112,13 @@ private void logAclSyncPlan(KafkaCluster cluster, Set toBeAdded, Set if (!toBeAdded.isEmpty()) { log.info("ACLs to be added ({}): ", toBeAdded.size()); for (AclBinding aclBinding : toBeAdded) { - log.info(" " + AclCsv.createAclString(aclBinding)); + log.info(" {}", AclCsv.createAclString(aclBinding)); } } if (!toBeDeleted.isEmpty()) { log.info("ACLs to be deleted ({}): ", toBeDeleted.size()); for (AclBinding aclBinding : toBeDeleted) { - log.info(" " + AclCsv.createAclString(aclBinding)); + log.info(" {}", AclCsv.createAclString(aclBinding)); } } } diff --git a/api/src/main/java/io/kafbat/ui/service/ksql/KsqlApiClient.java b/api/src/main/java/io/kafbat/ui/service/ksql/KsqlApiClient.java index 90192eb2d..13daee2bc 100644 --- a/api/src/main/java/io/kafbat/ui/service/ksql/KsqlApiClient.java +++ b/api/src/main/java/io/kafbat/ui/service/ksql/KsqlApiClient.java @@ -176,7 +176,7 @@ public Flux execute(String ksql, Map streamPr if (statements.size() > 1) { return errorTableFlux("Only single statement supported now"); } - if (statements.size() == 0) { + if (statements.isEmpty()) { return errorTableFlux("No valid ksql statement found"); } if (isUnsupportedStatementType(statements.get(0))) { diff --git a/api/src/main/java/io/kafbat/ui/service/ksql/KsqlGrammar.java b/api/src/main/java/io/kafbat/ui/service/ksql/KsqlGrammar.java index 3243841eb..1068ac193 100644 --- a/api/src/main/java/io/kafbat/ui/service/ksql/KsqlGrammar.java +++ b/api/src/main/java/io/kafbat/ui/service/ksql/KsqlGrammar.java @@ -74,13 +74,10 @@ public static CaseInsensitiveStream from(CharStream stream) { @Override public int LA(final int i) { final int result = stream.LA(i); - switch (result) { - case 0: - case IntStream.EOF: - return result; - default: - return Character.toUpperCase(result); - } + return switch (result) { + case 0, IntStream.EOF -> result; + default -> Character.toUpperCase(result); + }; } }; } diff --git a/api/src/main/java/io/kafbat/ui/service/ksql/response/ResponseParser.java b/api/src/main/java/io/kafbat/ui/service/ksql/response/ResponseParser.java index f353ea578..a5d54f369 100644 --- a/api/src/main/java/io/kafbat/ui/service/ksql/response/ResponseParser.java +++ b/api/src/main/java/io/kafbat/ui/service/ksql/response/ResponseParser.java @@ -99,64 +99,45 @@ public static List parseStatementResponse(JsonN .orElse("unknown"); // messages structure can be inferred from https://github.com/confluentinc/ksql/blob/master/ksqldb-rest-model/src/main/java/io/confluent/ksql/rest/entity/KsqlEntity.java - switch (type) { - case "currentStatus": - return parseObject( - "Status", - List.of("status", "message"), - jsonNode.get("commandStatus") - ); - case "properties": - return parseProperties(jsonNode); - case "queries": - return parseArray("Queries", "queries", jsonNode); - case "sourceDescription": - return parseObjectDynamically("Source Description", jsonNode.get("sourceDescription")); - case "queryDescription": - return parseObjectDynamically("Queries Description", jsonNode.get("queryDescription")); - case "topicDescription": - return parseObject( - "Topic Description", - List.of("name", "kafkaTopic", "format", "schemaString"), - jsonNode - ); - case "streams": - return parseArray("Streams", "streams", jsonNode); - case "tables": - return parseArray("Tables", "tables", jsonNode); - case "kafka_topics": - return parseArray("Topics", "topics", jsonNode); - case "kafka_topics_extended": - return parseArray("Topics extended", "topics", jsonNode); - case "executionPlan": - return parseObject("Execution plan", List.of("executionPlanText"), jsonNode); - case "source_descriptions": - return parseArray("Source descriptions", "sourceDescriptions", jsonNode); - case "query_descriptions": - return parseArray("Queries", "queryDescriptions", jsonNode); - case "describe_function": - return parseObject("Function description", - List.of("name", "author", "version", "description", "functions", "path", "type"), - jsonNode - ); - case "function_names": - return parseArray("Function Names", "functions", jsonNode); - case "connector_info": - return parseObjectDynamically("Connector Info", jsonNode.get("info")); - case "drop_connector": - return parseObject("Dropped connector", List.of("connectorName"), jsonNode); - case "connector_list": - return parseArray("Connectors", "connectors", jsonNode); - case "connector_plugins_list": - return parseArray("Connector Plugins", "connectorPlugins", jsonNode); - case "connector_description": - return parseObject("Connector Description", - List.of("connectorClass", "status", "sources", "topics"), - jsonNode - ); - default: - return parseUnknownResponse(jsonNode); - } + return switch (type) { + case "currentStatus" -> parseObject( + "Status", + List.of("status", "message"), + jsonNode.get("commandStatus") + ); + case "properties" -> parseProperties(jsonNode); + case "queries" -> parseArray("Queries", "queries", jsonNode); + case "sourceDescription" -> parseObjectDynamically("Source Description", jsonNode.get("sourceDescription")); + case "queryDescription" -> parseObjectDynamically("Queries Description", jsonNode.get("queryDescription")); + case "topicDescription" -> parseObject( + "Topic Description", + List.of("name", "kafkaTopic", "format", "schemaString"), + jsonNode + ); + case "streams" -> parseArray("Streams", "streams", jsonNode); + case "tables" -> parseArray("Tables", "tables", jsonNode); + case "kafka_topics" -> parseArray("Topics", "topics", jsonNode); + case "kafka_topics_extended" -> parseArray("Topics extended", "topics", jsonNode); + case "executionPlan" -> parseObject("Execution plan", List.of("executionPlanText"), jsonNode); + case "source_descriptions" -> parseArray("Source descriptions", "sourceDescriptions", jsonNode); + case "query_descriptions" -> parseArray("Queries", "queryDescriptions", jsonNode); + case "describe_function" -> parseObject( + "Function description", + List.of("name", "author", "version", "description", "functions", "path", "type"), + jsonNode + ); + case "function_names" -> parseArray("Function Names", "functions", jsonNode); + case "connector_info" -> parseObjectDynamically("Connector Info", jsonNode.get("info")); + case "drop_connector" -> parseObject("Dropped connector", List.of("connectorName"), jsonNode); + case "connector_list" -> parseArray("Connectors", "connectors", jsonNode); + case "connector_plugins_list" -> parseArray("Connector Plugins", "connectorPlugins", jsonNode); + case "connector_description" -> parseObject( + "Connector Description", + List.of("connectorClass", "status", "sources", "topics"), + jsonNode + ); + default -> parseUnknownResponse(jsonNode); + }; } private static List parseObjectDynamically( diff --git a/api/src/main/java/io/kafbat/ui/service/metrics/MetricsCollector.java b/api/src/main/java/io/kafbat/ui/service/metrics/MetricsCollector.java index b124feb8d..e9a08e8cb 100644 --- a/api/src/main/java/io/kafbat/ui/service/metrics/MetricsCollector.java +++ b/api/src/main/java/io/kafbat/ui/service/metrics/MetricsCollector.java @@ -28,7 +28,7 @@ public Mono getBrokerMetrics(KafkaCluster cluster, Collection nod return Flux.fromIterable(nodes) .flatMap(n -> getMetrics(cluster, n).map(lst -> Tuples.of(n, lst))) .collectMap(Tuple2::getT1, Tuple2::getT2) - .map(nodeMetrics -> collectMetrics(cluster, nodeMetrics)) + .map(this::collectMetrics) .defaultIfEmpty(Metrics.empty()); } @@ -45,20 +45,19 @@ private Mono> getMetrics(KafkaCluster kafkaCluster, Node node) { return metricFlux.collectList(); } - public Metrics collectMetrics(KafkaCluster cluster, Map> perBrokerMetrics) { + public Metrics collectMetrics(Map> perBrokerMetrics) { Metrics.MetricsBuilder builder = Metrics.builder() .perBrokerMetrics( perBrokerMetrics.entrySet() .stream() .collect(Collectors.toMap(e -> e.getKey().id(), Map.Entry::getValue))); - populateWellknowMetrics(cluster, perBrokerMetrics) - .apply(builder); + populateWellknowMetrics(perBrokerMetrics).apply(builder); return builder.build(); } - private WellKnownMetrics populateWellknowMetrics(KafkaCluster cluster, Map> perBrokerMetrics) { + private WellKnownMetrics populateWellknowMetrics(Map> perBrokerMetrics) { WellKnownMetrics wellKnownMetrics = new WellKnownMetrics(); perBrokerMetrics.forEach((node, metrics) -> metrics.forEach(metric -> diff --git a/api/src/main/java/io/kafbat/ui/util/EmptyRedirectStrategy.java b/api/src/main/java/io/kafbat/ui/util/EmptyRedirectStrategy.java index c6f80a113..9ebbba7f8 100644 --- a/api/src/main/java/io/kafbat/ui/util/EmptyRedirectStrategy.java +++ b/api/src/main/java/io/kafbat/ui/util/EmptyRedirectStrategy.java @@ -10,27 +10,18 @@ public class EmptyRedirectStrategy implements ServerRedirectStrategy { - private HttpStatus httpStatus = HttpStatus.FOUND; - - private boolean contextRelative = true; - public Mono sendRedirect(ServerWebExchange exchange, URI location) { Assert.notNull(exchange, "exchange cannot be null"); Assert.notNull(location, "location cannot be null"); return Mono.fromRunnable(() -> { ServerHttpResponse response = exchange.getResponse(); - response.setStatusCode(this.httpStatus); + response.setStatusCode(HttpStatus.FOUND); response.getHeaders().setLocation(createLocation(exchange, location)); }); } private URI createLocation(ServerWebExchange exchange, URI location) { - if (!this.contextRelative) { - return location; - } - - String url = location.getPath().isEmpty() ? "/" - : location.toASCIIString(); + String url = location.getPath().isEmpty() ? "/" : location.toASCIIString(); if (url.startsWith("/")) { String context = exchange.getRequest().getPath().contextPath().value(); @@ -38,13 +29,4 @@ private URI createLocation(ServerWebExchange exchange, URI location) { } return location; } - - public void setHttpStatus(HttpStatus httpStatus) { - Assert.notNull(httpStatus, "httpStatus cannot be null"); - this.httpStatus = httpStatus; - } - - public void setContextRelative(boolean contextRelative) { - this.contextRelative = contextRelative; - } } diff --git a/api/src/main/java/io/kafbat/ui/util/KafkaServicesValidation.java b/api/src/main/java/io/kafbat/ui/util/KafkaServicesValidation.java index 397fa3839..019a33543 100644 --- a/api/src/main/java/io/kafbat/ui/util/KafkaServicesValidation.java +++ b/api/src/main/java/io/kafbat/ui/util/KafkaServicesValidation.java @@ -62,8 +62,7 @@ public static Optional validateTruststore(TruststoreConfig truststoreCon public static Mono validateClusterConnection(String bootstrapServers, Properties clusterProps, - @Nullable - TruststoreConfig ssl) { + @Nullable TruststoreConfig ssl) { Properties properties = new Properties(); KafkaClientSslPropertiesUtil.addKafkaSslProperties(ssl, properties); properties.putAll(clusterProps); @@ -73,7 +72,7 @@ public static Mono validateClusterConnection(S properties.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 5_000); properties.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 5_000); properties.put(AdminClientConfig.CLIENT_ID_CONFIG, "kui-admin-client-validation-" + System.currentTimeMillis()); - AdminClient adminClient = null; + AdminClient adminClient; try { adminClient = AdminClient.create(properties); } catch (Exception e) { diff --git a/api/src/main/java/io/kafbat/ui/util/ReactiveFailover.java b/api/src/main/java/io/kafbat/ui/util/ReactiveFailover.java index b46384d2e..872e9ddf9 100644 --- a/api/src/main/java/io/kafbat/ui/util/ReactiveFailover.java +++ b/api/src/main/java/io/kafbat/ui/util/ReactiveFailover.java @@ -59,8 +59,8 @@ public static ReactiveFailover create(List args, } private ReactiveFailover(List> publishers, - Predicate failoverExceptionsPredicate, - String noAvailablePublishersMsg) { + Predicate failoverExceptionsPredicate, + String noAvailablePublishersMsg) { Preconditions.checkArgument(!publishers.isEmpty()); this.publishers = publishers; this.failoverExceptionsPredicate = failoverExceptionsPredicate; diff --git a/api/src/main/java/io/kafbat/ui/util/jsonschema/AvroJsonSchemaConverter.java b/api/src/main/java/io/kafbat/ui/util/jsonschema/AvroJsonSchemaConverter.java index b5640eb06..475dd3091 100644 --- a/api/src/main/java/io/kafbat/ui/util/jsonschema/AvroJsonSchemaConverter.java +++ b/api/src/main/java/io/kafbat/ui/util/jsonschema/AvroJsonSchemaConverter.java @@ -153,7 +153,6 @@ private JsonType convertType(Schema schema) { case INT, LONG -> new SimpleJsonType(JsonType.Type.INTEGER); case MAP, RECORD -> new SimpleJsonType(JsonType.Type.OBJECT); case ENUM -> new EnumJsonType(schema.getEnumSymbols()); - case BYTES, STRING -> new SimpleJsonType(JsonType.Type.STRING); case NULL -> new SimpleJsonType(JsonType.Type.NULL); case ARRAY -> new SimpleJsonType(JsonType.Type.ARRAY); case FIXED, FLOAT, DOUBLE -> new SimpleJsonType(JsonType.Type.NUMBER); diff --git a/api/src/main/java/io/kafbat/ui/util/jsonschema/JsonAvroConversion.java b/api/src/main/java/io/kafbat/ui/util/jsonschema/JsonAvroConversion.java index 52b6913f6..de23d40cd 100644 --- a/api/src/main/java/io/kafbat/ui/util/jsonschema/JsonAvroConversion.java +++ b/api/src/main/java/io/kafbat/ui/util/jsonschema/JsonAvroConversion.java @@ -49,7 +49,7 @@ public class JsonAvroConversion { // converts json into Object that is expected input for KafkaAvroSerializer // (with AVRO_USE_LOGICAL_TYPE_CONVERTERS flat enabled!) public static Object convertJsonToAvro(String jsonString, Schema avroSchema) { - JsonNode rootNode = null; + JsonNode rootNode; try { rootNode = MAPPER.readTree(jsonString); } catch (JsonProcessingException e) { @@ -221,9 +221,7 @@ public static JsonNode convertAvroToJson(Object obj, Schema avroSchema) { list.forEach(e -> node.add(convertAvroToJson(e, avroSchema.getElementType()))); yield node; } - case ENUM -> { - yield new TextNode(obj.toString()); - } + case ENUM -> new TextNode(obj.toString()); case UNION -> { ObjectNode node = MAPPER.createObjectNode(); int unionIdx = AvroData.getGenericData().resolveUnion(avroSchema, obj); @@ -343,9 +341,7 @@ enum LogicalTypeConversion { assertJsonType(node, JsonNodeType.STRING); return java.util.UUID.fromString(node.asText()); }, - (obj, schema) -> { - return new TextNode(obj.toString()); - }, + (obj, schema) -> new TextNode(obj.toString()), new SimpleFieldSchema( new SimpleJsonType( JsonType.Type.STRING, @@ -363,9 +359,7 @@ enum LogicalTypeConversion { "node '%s' can't be converted to decimal logical type" .formatted(node)); }, - (obj, schema) -> { - return new DecimalNode((BigDecimal) obj); - }, + (obj, schema) -> new DecimalNode((BigDecimal) obj), new SimpleFieldSchema(new SimpleJsonType(JsonType.Type.NUMBER)) ), @@ -381,9 +375,7 @@ enum LogicalTypeConversion { .formatted(node)); } }, - (obj, schema) -> { - return new TextNode(obj.toString()); - }, + (obj, schema) -> new TextNode(obj.toString()), new SimpleFieldSchema( new SimpleJsonType( JsonType.Type.STRING, @@ -402,9 +394,7 @@ enum LogicalTypeConversion { .formatted(node)); } }, - (obj, schema) -> { - return new TextNode(obj.toString()); - }, + (obj, schema) -> new TextNode(obj.toString()), new SimpleFieldSchema( new SimpleJsonType( JsonType.Type.STRING, @@ -423,9 +413,7 @@ enum LogicalTypeConversion { .formatted(node)); } }, - (obj, schema) -> { - return new TextNode(obj.toString()); - }, + (obj, schema) -> new TextNode(obj.toString()), new SimpleFieldSchema( new SimpleJsonType( JsonType.Type.STRING, @@ -444,9 +432,7 @@ enum LogicalTypeConversion { .formatted(node)); } }, - (obj, schema) -> { - return new TextNode(obj.toString()); - }, + (obj, schema) -> new TextNode(obj.toString()), new SimpleFieldSchema( new SimpleJsonType( JsonType.Type.STRING, @@ -469,9 +455,7 @@ enum LogicalTypeConversion { .formatted(node)); } }, - (obj, schema) -> { - return new TextNode(obj.toString()); - }, + (obj, schema) -> new TextNode(obj.toString()), new SimpleFieldSchema( new SimpleJsonType( JsonType.Type.STRING, @@ -487,9 +471,7 @@ enum LogicalTypeConversion { Instant instant = (Instant) TIMESTAMP_MILLIS.jsonToAvroConversion.apply(node, schema); return LocalDateTime.ofInstant(instant, ZoneOffset.UTC); }, - (obj, schema) -> { - return new TextNode(obj.toString()); - }, + (obj, schema) -> new TextNode(obj.toString()), new SimpleFieldSchema( new SimpleJsonType( JsonType.Type.STRING, @@ -504,9 +486,7 @@ enum LogicalTypeConversion { Instant instant = (Instant) TIMESTAMP_MICROS.jsonToAvroConversion.apply(node, schema); return LocalDateTime.ofInstant(instant, ZoneOffset.UTC); }, - (obj, schema) -> { - return new TextNode(obj.toString()); - }, + (obj, schema) -> new TextNode(obj.toString()), new SimpleFieldSchema( new SimpleJsonType( JsonType.Type.STRING, diff --git a/api/src/main/java/io/kafbat/ui/util/jsonschema/JsonSchema.java b/api/src/main/java/io/kafbat/ui/util/jsonschema/JsonSchema.java index 090010dac..491acebc7 100644 --- a/api/src/main/java/io/kafbat/ui/util/jsonschema/JsonSchema.java +++ b/api/src/main/java/io/kafbat/ui/util/jsonschema/JsonSchema.java @@ -9,7 +9,6 @@ import java.util.stream.Collectors; import lombok.Builder; import lombok.Data; -import lombok.SneakyThrows; import reactor.util.function.Tuple2; import reactor.util.function.Tuples; @@ -59,12 +58,4 @@ public String toJson() { } return objectNode.toString(); } - - @SneakyThrows - public static JsonSchema stringSchema() { - return JsonSchema.builder() - .id(new URI("http://unknown.unknown")) - .type(new SimpleJsonType(JsonType.Type.STRING)) - .build(); - } } From ecd5f245ab11ee47e2b5e7a6259d6efab9b6abc6 Mon Sep 17 00:00:00 2001 From: Dmitry Werner Date: Tue, 4 Feb 2025 16:13:22 +0500 Subject: [PATCH 2/2] BE: Chore: Cleanup model package reverted --- .../io/kafbat/ui/util/jsonschema/AvroJsonSchemaConverter.java | 1 + 1 file changed, 1 insertion(+) diff --git a/api/src/main/java/io/kafbat/ui/util/jsonschema/AvroJsonSchemaConverter.java b/api/src/main/java/io/kafbat/ui/util/jsonschema/AvroJsonSchemaConverter.java index 475dd3091..b5640eb06 100644 --- a/api/src/main/java/io/kafbat/ui/util/jsonschema/AvroJsonSchemaConverter.java +++ b/api/src/main/java/io/kafbat/ui/util/jsonschema/AvroJsonSchemaConverter.java @@ -153,6 +153,7 @@ private JsonType convertType(Schema schema) { case INT, LONG -> new SimpleJsonType(JsonType.Type.INTEGER); case MAP, RECORD -> new SimpleJsonType(JsonType.Type.OBJECT); case ENUM -> new EnumJsonType(schema.getEnumSymbols()); + case BYTES, STRING -> new SimpleJsonType(JsonType.Type.STRING); case NULL -> new SimpleJsonType(JsonType.Type.NULL); case ARRAY -> new SimpleJsonType(JsonType.Type.ARRAY); case FIXED, FLOAT, DOUBLE -> new SimpleJsonType(JsonType.Type.NUMBER);