|
@@ -2,6 +2,7 @@ package com.provectus.kafka.ui.service;
|
|
|
|
|
|
import com.provectus.kafka.ui.exception.TopicMetadataException;
|
|
|
import com.provectus.kafka.ui.exception.ValidationException;
|
|
|
+import com.provectus.kafka.ui.model.CleanupPolicy;
|
|
|
import com.provectus.kafka.ui.model.CreateTopicMessage;
|
|
|
import com.provectus.kafka.ui.model.ExtendedAdminClient;
|
|
|
import com.provectus.kafka.ui.model.InternalBrokerDiskUsage;
|
|
@@ -207,12 +208,18 @@ public class KafkaService {
|
|
|
|
|
|
private Map<String, InternalTopic> mergeWithConfigs(
|
|
|
List<InternalTopic> topics, Map<String, List<InternalTopicConfig>> configs) {
|
|
|
- return topics.stream().map(
|
|
|
- t -> t.toBuilder().topicConfigs(configs.get(t.getName())).build()
|
|
|
- ).collect(Collectors.toMap(
|
|
|
- InternalTopic::getName,
|
|
|
- e -> e
|
|
|
- ));
|
|
|
+ return topics.stream()
|
|
|
+ .map(t -> t.toBuilder().topicConfigs(configs.get(t.getName())).build())
|
|
|
+ .map(t -> t.toBuilder().cleanUpPolicy(
|
|
|
+ CleanupPolicy.fromString(t.getTopicConfigs().stream()
|
|
|
+ .filter(config -> config.getName().equals("cleanup.policy"))
|
|
|
+ .findFirst()
|
|
|
+ .orElseGet(() -> InternalTopicConfig.builder().value("unknown").build())
|
|
|
+ .getValue())).build())
|
|
|
+ .collect(Collectors.toMap(
|
|
|
+ InternalTopic::getName,
|
|
|
+ e -> e
|
|
|
+ ));
|
|
|
}
|
|
|
|
|
|
@SneakyThrows
|
|
@@ -225,11 +232,12 @@ public class KafkaService {
|
|
|
final Mono<Map<String, List<InternalTopicConfig>>> configsMono =
|
|
|
loadTopicsConfig(adminClient, topics);
|
|
|
|
|
|
- return ClusterUtil.toMono(adminClient.describeTopics(topics).all()).map(
|
|
|
- m -> m.values().stream().map(ClusterUtil::mapToInternalTopic).collect(Collectors.toList())
|
|
|
- ).flatMap(internalTopics -> configsMono.map(configs ->
|
|
|
- mergeWithConfigs(internalTopics, configs).values()
|
|
|
- )).flatMapMany(Flux::fromIterable);
|
|
|
+ return ClusterUtil.toMono(adminClient.describeTopics(topics).all())
|
|
|
+ .map(m -> m.values().stream()
|
|
|
+ .map(ClusterUtil::mapToInternalTopic).collect(Collectors.toList()))
|
|
|
+ .flatMap(internalTopics -> configsMono
|
|
|
+ .map(configs -> mergeWithConfigs(internalTopics, configs).values()))
|
|
|
+ .flatMapMany(Flux::fromIterable);
|
|
|
}
|
|
|
|
|
|
|