need to refactor

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/service/ClusterService.java

@@ -62,6 +62,12 @@ public class ClusterService {
         return kafkaService.createTopic(cluster, topicFormData);
     }
 
+    public Mono<ResponseEntity<Topic>> updateTopic(String clusterName, String topicName, Mono<TopicFormData> topicFormData) {
+        KafkaCluster cluster = clustersStorage.getClusterByName(clusterName);
+        if (cluster == null) return null;
+        return topicFormData.flatMap(t -> kafkaService.updateTopic(cluster, topicName, t)).map(ResponseEntity::ok);
+    }
+
     @SneakyThrows
     public Mono<ResponseEntity<Flux<ConsumerGroup>>> getConsumerGroup (String clusterName) {
             var cluster = clustersStorage.getClusterByName(clusterName);

kafka-ui-api/src/main/java/com/provectus/kafka/ui/kafka/KafkaService.java

@@ -1,6 +1,7 @@
 package com.provectus.kafka.ui.kafka;
 
 import com.provectus.kafka.ui.cluster.model.KafkaCluster;
+import com.provectus.kafka.ui.cluster.util.ClusterUtil;
 import com.provectus.kafka.ui.model.*;
 import lombok.RequiredArgsConstructor;
 import lombok.SneakyThrows;
@@ -15,6 +16,8 @@ import org.springframework.stereotype.Service;
 import reactor.core.publisher.Mono;
 
 import java.util.*;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static com.provectus.kafka.ui.kafka.KafkaConstants.*;
 import static org.apache.kafka.common.config.TopicConfig.MESSAGE_FORMAT_VERSION_CONFIG;
@@ -72,6 +75,30 @@ public class KafkaService {
         );
     }
 
+    public Mono<Topic> updateTopic(KafkaCluster cluster, String topicName, TopicFormData topicFormData) {
+        return ClusterUtil.toMono(cluster.getAdminClient().describeCluster().controller()).map(Node::id)
+                .flatMap(n -> {
+                    ConfigResource cr = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
+
+                    return ClusterUtil.toMono(cluster.getAdminClient().describeConfigs(Collections.singletonList(new ConfigResource(ConfigResource.Type.BROKER, n))).all())
+                            .flatMap(c -> {
+                                String e = c.entrySet().stream().map(en -> en.getValue().entries().stream().filter(en1 -> en1.name().contains("inter.broker.protocol.version")).findFirst().orElseThrow()).findFirst().orElseThrow().value();
+                                if (Float.parseFloat(e.split("-")[0]) <= 2.3f) {
+                                    List<ConfigEntry> ce = topicFormData.getConfigs().entrySet().stream()
+                                            .flatMap(cfg -> Stream.of(new ConfigEntry(cfg.getKey(), cfg.getValue()))).collect(Collectors.toList());
+                                    Config config = new Config(ce);
+                                    Map<ConfigResource, Config> map = Collections.singletonMap(cr, config);
+                                    cluster.getAdminClient().alterConfigs(map);
+                                } else {
+                                    List<AlterConfigOp> listOp = topicFormData.getConfigs().entrySet().stream()
+                                            .flatMap(cfg -> Stream.of(new AlterConfigOp(new ConfigEntry(cfg.getKey(), cfg.getValue()), AlterConfigOp.OpType.SET))).collect(Collectors.toList());
+                                    cluster.getAdminClient().incrementalAlterConfigs(Collections.singletonMap(cr, listOp));
+                                }
+                                return ClusterUtil.toMono(cluster.getAdminClient().describeTopics(Collections.singletonList(topicName)).all()).map(t -> collectTopicData(cluster, t.get(topicName)));
+                            });
+                            });
+    }
+
     @SneakyThrows
     private String getClusterId(KafkaCluster kafkaCluster) {
         return kafkaCluster.getAdminClient().describeCluster().clusterId().get();

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

@@ -4,7 +4,6 @@ import com.provectus.kafka.ui.api.ApiClustersApi;
 import com.provectus.kafka.ui.cluster.service.ClusterService;
 import com.provectus.kafka.ui.model.*;
 import lombok.RequiredArgsConstructor;
-import org.apache.kafka.clients.admin.ListConsumerGroupsResult;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.server.ServerWebExchange;
@@ -40,6 +39,11 @@ public class MetricsRestController implements ApiClustersApi {
         return clusterService.getTopicDetails(clusterId, topicName);
     }
 
+    @Override
+    public Mono<ResponseEntity<Topic>> updateTopic(String clusterId, String topicName, @Valid Mono<TopicFormData> topicFormData, ServerWebExchange exchange) {
+        return clusterService.updateTopic(clusterId, topicName, topicFormData);
+    }
+
     @Override
     public Mono<ResponseEntity<Flux<TopicConfig>>> getTopicConfigs(String clusterId, String topicName, ServerWebExchange exchange) {
         return clusterService.getTopicConfigs(clusterId, topicName);

kafka-ui-api/src/main/java/com/provectus/kafka/ui/zookeeper/ZooKeeperConstants.java

@@ -7,4 +7,7 @@ public final class ZooKeeperConstants {
     public static int ONLINE = 1;
     public static int OFFLINE = 0;
 
+    public static int CONNECTION_TIMEOUT_MS = 1000;
+    public static int SESSION_TIMEOUT_MS = 3000;
+
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/zookeeper/ZookeeperService.java

@@ -34,7 +34,7 @@ public class ZookeeperService {
 
     private boolean createZookeeperConnection(KafkaCluster kafkaCluster) {
         try {
-            kafkaCluster.setZkClient(new ZkClient(kafkaCluster.getZookeeper(), 1000));
+            kafkaCluster.setZkClient(new ZkClient(kafkaCluster.getZookeeper(), ZooKeeperConstants.CONNECTION_TIMEOUT_MS));
 
             return true;
         } catch (Exception e) {

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

@@ -141,6 +141,34 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/TopicDetails'
+    patch:
+      tags:
+        - /api/clusters
+      summary: updateTopic
+      operationId: updateTopic
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: topicName
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/TopicFormData'
+      responses:
+        201:
+          description: Created
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Topic'
 
   /api/clusters/{clusterName}/topics/{topicName}/config:
     get:

pom.xml

@@ -19,7 +19,7 @@
 		<org.projectlombok.version>1.18.10</org.projectlombok.version>
 		<git.revision>latest</git.revision>
 		<zkclient.version>0.11</zkclient.version>
-		<kafka-clients.version>2.4.0</kafka-clients.version>
+		<kafka-clients.version>2.5.0</kafka-clients.version>
 		<node.version>v12.13.1</node.version>
 		<dockerfile-maven-plugin.version>1.4.10</dockerfile-maven-plugin.version>
 		<frontend-maven-plugin.version>1.8.0</frontend-maven-plugin.version>