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

ISSUE-309: Topic create & update models split (#312)

Browse source 
* ISSUE-309: Topic create & update models split
This commit is contained in:
iliax 2021-03-25 11:24:51 +03:00 • committed by GitHub
parent 752c19dd9c
commit 8d2f929a52
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 88 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/TopicsController.java
View file

@ -3,8 +3,9 @@ package com.provectus.kafka.ui.controller;
import com.provectus.kafka.ui.api.TopicsApi;
import com.provectus.kafka.ui.model.Topic;
import com.provectus.kafka.ui.model.TopicConfig;
import com.provectus.kafka.ui.model.TopicCreation;
import com.provectus.kafka.ui.model.TopicDetails;
import com.provectus.kafka.ui.model.TopicFormData;
import com.provectus.kafka.ui.model.TopicUpdate;
import com.provectus.kafka.ui.model.TopicsResponse;
import com.provectus.kafka.ui.service.ClusterService;
import java.util.Optional;
@ -26,8 +27,8 @@ public class TopicsController implements TopicsApi {
@Override
public Mono<ResponseEntity<Topic>> createTopic(
String clusterName, @Valid Mono<TopicFormData> topicFormData, ServerWebExchange exchange) {
return clusterService.createTopic(clusterName, topicFormData)
String clusterName, @Valid Mono<TopicCreation> topicCreation, ServerWebExchange exchange) {
return clusterService.createTopic(clusterName, topicCreation)
.map(s -> new ResponseEntity<>(s, HttpStatus.OK))
.switchIfEmpty(Mono.just(ResponseEntity.notFound().build()));
}
@ -70,8 +71,8 @@ public class TopicsController implements TopicsApi {
@Override
public Mono<ResponseEntity<Topic>> updateTopic(
String clusterId, String topicName, @Valid Mono<TopicFormData> topicFormData,
String clusterId, String topicName, @Valid Mono<TopicUpdate> topicUpdate,
ServerWebExchange exchange) {
return clusterService.updateTopic(clusterId, topicName, topicFormData).map(ResponseEntity::ok);
return clusterService.updateTopic(clusterId, topicName, topicUpdate).map(ResponseEntity::ok);
}
}

4
kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/GlobalErrorWebExceptionHandler.java
View file

@ -71,7 +71,7 @@ public class GlobalErrorWebExceptionHandler extends AbstractErrorWebExceptionHan
private Mono<ServerResponse> renderDefault(Throwable throwable, ServerRequest request) {
var response = new ErrorResponse()
.code(ErrorCode.UNEXPECTED.code())
.message(throwable.getMessage())
.message(coalesce(throwable.getMessage(), "Unexpected internal error"))
.requestId(requestId(request))
.timestamp(currentTimestamp());
return ServerResponse
@ -84,7 +84,7 @@ public class GlobalErrorWebExceptionHandler extends AbstractErrorWebExceptionHan
ErrorCode errorCode = baseException.getErrorCode();
var response = new ErrorResponse()
.code(errorCode.code())
.message(baseException.getMessage())
.message(coalesce(baseException.getMessage(), "Internal error"))
.requestId(requestId(request))
.timestamp(currentTimestamp());
return ServerResponse

11
kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ClusterService.java
View file

@ -15,9 +15,10 @@ import com.provectus.kafka.ui.model.InternalTopic;
import com.provectus.kafka.ui.model.KafkaCluster;
import com.provectus.kafka.ui.model.Topic;
import com.provectus.kafka.ui.model.TopicConfig;
import com.provectus.kafka.ui.model.TopicCreation;
import com.provectus.kafka.ui.model.TopicDetails;
import com.provectus.kafka.ui.model.TopicFormData;
import com.provectus.kafka.ui.model.TopicMessage;
import com.provectus.kafka.ui.model.TopicUpdate;
import com.provectus.kafka.ui.model.TopicsResponse;
import com.provectus.kafka.ui.util.ClusterUtil;
import java.util.Collection;
@ -125,9 +126,9 @@ public class ClusterService {
.collect(Collectors.toList()));
}
public Mono<Topic> createTopic(String clusterName, Mono<TopicFormData> topicFormData) {
public Mono<Topic> createTopic(String clusterName, Mono<TopicCreation> topicCreation) {
return clustersStorage.getClusterByName(clusterName).map(cluster ->
kafkaService.createTopic(cluster, topicFormData)
kafkaService.createTopic(cluster, topicCreation)
.doOnNext(t -> updateCluster(t, clusterName, cluster))
.map(clusterMapper::toTopic)
).orElse(Mono.empty());
@ -200,9 +201,9 @@ public class ClusterService {
@SneakyThrows
public Mono<Topic> updateTopic(String clusterName, String topicName,
Mono<TopicFormData> topicFormData) {
Mono<TopicUpdate> topicUpdate) {
return clustersStorage.getClusterByName(clusterName).map(cl ->
topicFormData
topicUpdate
.flatMap(t -> kafkaService.updateTopic(cl, topicName, t))
.doOnNext(t -> updateCluster(t, clusterName, cl))
.map(clusterMapper::toTopic)

25
kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaService.java
View file

@ -12,7 +12,8 @@ import com.provectus.kafka.ui.model.InternalTopicConfig;
import com.provectus.kafka.ui.model.KafkaCluster;
import com.provectus.kafka.ui.model.Metric;
import com.provectus.kafka.ui.model.ServerStatus;
import com.provectus.kafka.ui.model.TopicFormData;
import com.provectus.kafka.ui.model.TopicCreation;
import com.provectus.kafka.ui.model.TopicUpdate;
import com.provectus.kafka.ui.util.ClusterUtil;
import com.provectus.kafka.ui.util.JmxClusterUtil;
import com.provectus.kafka.ui.util.JmxMetricsName;
@ -223,8 +224,8 @@ public class KafkaService {
@SneakyThrows
public Mono<InternalTopic> createTopic(AdminClient adminClient,
Mono<TopicFormData> topicFormData) {
return topicFormData.flatMap(
Mono<TopicCreation> topicCreation) {
return topicCreation.flatMap(
topicData -> {
NewTopic newTopic = new NewTopic(topicData.getName(), topicData.getPartitions(),
topicData.getReplicationFactor().shortValue());
@ -242,9 +243,9 @@ public class KafkaService {
);
}
public Mono<InternalTopic> createTopic(KafkaCluster cluster, Mono<TopicFormData> topicFormData) {
public Mono<InternalTopic> createTopic(KafkaCluster cluster, Mono<TopicCreation> topicCreation) {
return getOrCreateAdminClient(cluster)
.flatMap(ac -> createTopic(ac.getAdminClient(), topicFormData));
.flatMap(ac -> createTopic(ac.getAdminClient(), topicCreation));
}
public Mono<Void> deleteTopic(KafkaCluster cluster, String topicName) {
@ -320,16 +321,16 @@ public class KafkaService {
@SneakyThrows
public Mono<InternalTopic> updateTopic(KafkaCluster cluster, String topicName,
TopicFormData topicFormData) {
TopicUpdate topicUpdate) {
ConfigResource topicCr = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
return getOrCreateAdminClient(cluster)
.flatMap(ac -> {
if (ac.getSupportedFeatures()
.contains(ExtendedAdminClient.SupportedFeature.INCREMENTAL_ALTER_CONFIGS)) {
return incrementalAlterConfig(topicFormData, topicCr, ac)
return incrementalAlterConfig(topicUpdate, topicCr, ac)
.flatMap(c -> getUpdatedTopic(ac, topicName));
} else {
return alterConfig(topicFormData, topicCr, ac)
return alterConfig(topicUpdate, topicCr, ac)
.flatMap(c -> getUpdatedTopic(ac, topicName));
}
});
@ -341,9 +342,9 @@ public class KafkaService {
.filter(t -> t.getName().equals(topicName)).findFirst().orElseThrow());
}
private Mono<String> incrementalAlterConfig(TopicFormData topicFormData, ConfigResource topicCr,
private Mono<String> incrementalAlterConfig(TopicUpdate topicUpdate, ConfigResource topicCr,
ExtendedAdminClient ac) {
List<AlterConfigOp> listOp = topicFormData.getConfigs().entrySet().stream()
List<AlterConfigOp> listOp = topicUpdate.getConfigs().entrySet().stream()
.flatMap(cfg -> Stream.of(new AlterConfigOp(new ConfigEntry(cfg.getKey(), cfg.getValue()),
AlterConfigOp.OpType.SET))).collect(Collectors.toList());
return ClusterUtil.toMono(
@ -352,9 +353,9 @@ public class KafkaService {
}
@SuppressWarnings("deprecation")
private Mono<String> alterConfig(TopicFormData topicFormData, ConfigResource topicCr,
private Mono<String> alterConfig(TopicUpdate topicUpdate, ConfigResource topicCr,
ExtendedAdminClient ac) {
List<ConfigEntry> configEntries = topicFormData.getConfigs().entrySet().stream()
List<ConfigEntry> configEntries = topicUpdate.getConfigs().entrySet().stream()
.flatMap(cfg -> Stream.of(new ConfigEntry(cfg.getKey(), cfg.getValue())))
.collect(Collectors.toList());
Config config = new Config(configEntries);

4
kafka-ui-api/src/test/java/com/provectus/kafka/ui/KafkaConsumerTests.java
View file

@ -1,6 +1,6 @@
package com.provectus.kafka.ui;
import com.provectus.kafka.ui.model.TopicFormData;
import com.provectus.kafka.ui.model.TopicCreation;
import com.provectus.kafka.ui.model.TopicMessage;
import com.provectus.kafka.ui.producer.KafkaTestProducer;
import java.util.Map;
@ -27,7 +27,7 @@ public class KafkaConsumerTests extends AbstractBaseTest {
var topicName = UUID.randomUUID().toString();
webTestClient.post()
.uri("/api/clusters/{clusterName}/topics", LOCAL)
.bodyValue(new TopicFormData()
.bodyValue(new TopicCreation()
.name(topicName)
.partitions(1)
.replicationFactor(1)

19
kafka-ui-api/src/test/java/com/provectus/kafka/ui/ReadOnlyModeTests.java
View file

@ -1,6 +1,7 @@
package com.provectus.kafka.ui;
import com.provectus.kafka.ui.model.TopicFormData;
import com.provectus.kafka.ui.model.TopicCreation;
import com.provectus.kafka.ui.model.TopicUpdate;
import java.util.Map;
import java.util.UUID;
import lombok.extern.log4j.Log4j2;
@ -24,7 +25,7 @@ public class ReadOnlyModeTests extends AbstractBaseTest {
var topicName = UUID.randomUUID().toString();
webTestClient.post()
.uri("/api/clusters/{clusterName}/topics", LOCAL)
.bodyValue(new TopicFormData()
.bodyValue(new TopicCreation()
.name(topicName)
.partitions(1)
.replicationFactor(1)
@ -40,7 +41,7 @@ public class ReadOnlyModeTests extends AbstractBaseTest {
var topicName = UUID.randomUUID().toString();
webTestClient.post()
.uri("/api/clusters/{clusterName}/topics", SECOND_LOCAL)
.bodyValue(new TopicFormData()
.bodyValue(new TopicCreation()
.name(topicName)
.partitions(1)
.replicationFactor(1)
@ -56,7 +57,7 @@ public class ReadOnlyModeTests extends AbstractBaseTest {
var topicName = UUID.randomUUID().toString();
webTestClient.post()
.uri("/api/clusters/{clusterName}/topics", LOCAL)
.bodyValue(new TopicFormData()
.bodyValue(new TopicCreation()
.name(topicName)
.partitions(1)
.replicationFactor(1)
@ -67,10 +68,7 @@ public class ReadOnlyModeTests extends AbstractBaseTest {
.isOk();
webTestClient.patch()
.uri("/api/clusters/{clusterName}/topics/{topicName}", LOCAL, topicName)
.bodyValue(new TopicFormData()
.name(topicName)
.partitions(2)
.replicationFactor(1)
.bodyValue(new TopicUpdate()
.configs(Map.of())
)
.exchange()
@ -83,10 +81,7 @@ public class ReadOnlyModeTests extends AbstractBaseTest {
var topicName = UUID.randomUUID().toString();
webTestClient.patch()
.uri("/api/clusters/{clusterName}/topics/{topicName}", SECOND_LOCAL, topicName)
.bodyValue(new TopicFormData()
.name(topicName)
.partitions(1)
.replicationFactor(1)
.bodyValue(new TopicUpdate()
.configs(Map.of())
)
.exchange()

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

@ -162,7 +162,7 @@ paths:
content:
application/json:
schema:
$ref: '#/components/schemas/TopicFormData'
$ref: '#/components/schemas/TopicCreation'
responses:
201:
description: Created
@ -215,7 +215,7 @@ paths:
content:
application/json:
schema:
$ref: '#/components/schemas/TopicFormData'
$ref: '#/components/schemas/TopicUpdate'
responses:
200:
description: Updated
@ -1281,7 +1281,7 @@ components:
required:
- name
TopicFormData:
TopicCreation:
type: object
properties:
name:
@ -1296,6 +1296,18 @@ components:
type: string
required:
- name
- partitions
- replicationFactor
TopicUpdate:
type: object
properties:
configs:
type: object
additionalProperties:
type: string
required:
- configs
Broker:
type: object

39
kafka-ui-react-app/src/redux/actions/thunks/topics.ts
View file

@ -4,7 +4,8 @@ import {
MessagesApi,
Configuration,
Topic,
TopicFormData,
TopicCreation,
TopicUpdate,
TopicConfig,
} from 'generated-sources';
import {
@ -136,7 +137,7 @@ export const fetchTopicConfig = (
}
};
const formatTopicFormData = (form: TopicFormDataRaw): TopicFormData => {
const formatTopicCreation = (form: TopicFormDataRaw): TopicCreation => {
const {
name,
partitions,
@ -172,6 +173,36 @@ const formatTopicFormData = (form: TopicFormDataRaw): TopicFormData => {
};
};
const formatTopicUpdate = (form: TopicFormDataRaw): TopicUpdate => {
const {
cleanupPolicy,
retentionBytes,
retentionMs,
maxMessageBytes,
minInSyncReplicas,
customParams,
} = form;
return {
configs: {
'cleanup.policy': cleanupPolicy,
'retention.ms': retentionMs,
'retention.bytes': retentionBytes,
'max.message.bytes': maxMessageBytes,
'min.insync.replicas': minInSyncReplicas,
...Object.values(customParams || {}).reduce(
(result: TopicFormFormattedParams, customParam: TopicConfig) => {
return {
...result,
[customParam.name]: customParam.value,
};
},
{}
),
},
};
};
export const createTopic = (
clusterName: ClusterName,
form: TopicFormDataRaw
@ -180,7 +211,7 @@ export const createTopic = (
try {
const topic: Topic = await topicsApiClient.createTopic({
clusterName,
topicFormData: formatTopicFormData(form),
topicCreation: formatTopicCreation(form),
});
const state = getState().topics;
@ -210,7 +241,7 @@ export const updateTopic = (
const topic: Topic = await topicsApiClient.updateTopic({
clusterName,
topicName: form.name,
topicFormData: formatTopicFormData(form),
topicUpdate: formatTopicUpdate(form),
});
const state = getState().topics;

4
kafka-ui-react-app/src/redux/interfaces/topic.ts
View file

@ -3,7 +3,7 @@ import {
TopicDetails,
TopicMessage,
TopicConfig,
TopicFormData,
TopicCreation,
GetTopicMessagesRequest,
} from 'generated-sources';
@ -50,7 +50,7 @@ export interface TopicsState {
messages: TopicMessage[];
}
export type TopicFormFormattedParams = TopicFormData['configs'];
export type TopicFormFormattedParams = TopicCreation['configs'];
export interface TopicFormDataRaw {
name: string;