From 217b36307615effe6dad8c272e4472d82b74425e Mon Sep 17 00:00:00 2001
From: Ramazan Yapparov <ramazan.yapparov@gmail.com>
Date: Thu, 18 Mar 2021 21:16:59 +0300
Subject: [PATCH] added features field to kafka cluster (#256)

---
 .../kafka/ui/mapper/ClusterMapper.java        |  1 +
 .../com/provectus/kafka/ui/model/Feature.java | 27 +++++++++++++++++++
 .../kafka/ui/model/KafkaCluster.java          |  1 +
 .../kafka/ui/service/ClustersStorage.java     |  9 ++++++-
 .../main/resources/swagger/kafka-ui-api.yaml  |  7 +++++
 5 files changed, 44 insertions(+), 1 deletion(-)
 create mode 100644 kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/Feature.java

diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ClusterMapper.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ClusterMapper.java
index 7a47620f2f..1b6e446e51 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ClusterMapper.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ClusterMapper.java
@@ -37,6 +37,7 @@ public interface ClusterMapper {
     TopicConfig toTopicConfig(InternalTopicConfig topic);
     Replica toReplica(InternalReplica replica);
     Connect toKafkaConnect(KafkaConnectCluster connect);
+    List<Cluster.FeaturesEnum> toFeaturesEnum(List<Feature> features);
 
     @Mapping(target = "isCompatible", source = "compatible")
     CompatibilityCheckResponse toCompatibilityCheckResponse(InternalCompatibilityCheck dto);
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/Feature.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/Feature.java
new file mode 100644
index 0000000000..ef23ed344e
--- /dev/null
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/Feature.java
@@ -0,0 +1,27 @@
+package com.provectus.kafka.ui.model;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+public enum Feature {
+    KAFKA_CONNECT(cluster -> Optional.ofNullable(cluster.getKafkaConnect())
+            .filter(Predicate.not(List::isEmpty))
+            .isPresent()
+    ),
+    SCHEMA_REGISTRY(cluster -> cluster.getSchemaRegistry() != null);
+
+    private final Predicate<KafkaCluster> isEnabled;
+
+    Feature(Predicate<KafkaCluster> isEnabled) {
+        this.isEnabled = isEnabled;
+    }
+
+    public static List<Feature> getEnabledFeatures(KafkaCluster cluster) {
+        return Arrays.stream(values())
+                .filter(feature -> feature.isEnabled.test(cluster))
+                .collect(Collectors.toList());
+    }
+}
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java
index 361b214194..790f994cf5 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java
@@ -28,4 +28,5 @@ public class KafkaCluster {
     private final String protobufMessageName;
     private final Properties properties;
     private final Boolean readOnly;
+    private final List<Feature> features;
 }
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ClustersStorage.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ClustersStorage.java
index 50da1cdd52..1bc7b7d5c3 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ClustersStorage.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ClustersStorage.java
@@ -2,6 +2,7 @@ package com.provectus.kafka.ui.service;
 
 import com.provectus.kafka.ui.config.ClustersProperties;
 import com.provectus.kafka.ui.mapper.ClusterMapper;
+import com.provectus.kafka.ui.model.Feature;
 import com.provectus.kafka.ui.model.KafkaCluster;
 import lombok.RequiredArgsConstructor;
 import org.mapstruct.factory.Mappers;
@@ -29,7 +30,13 @@ public class ClustersStorage {
             if (kafkaClusters.get(clusterProperties.getName()) != null) {
                 throw new IllegalStateException("Application config isn't correct. Two clusters can't have the same name");
             }
-            kafkaClusters.put(clusterProperties.getName(), clusterMapper.toKafkaCluster(clusterProperties));
+            KafkaCluster cluster = clusterMapper.toKafkaCluster(clusterProperties);
+            kafkaClusters.put(
+                    clusterProperties.getName(),
+                    cluster.toBuilder()
+                            .features(Feature.getEnabledFeatures(cluster))
+                            .build()
+            );
         }
     }
 
diff --git a/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml b/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
index 80c34ce4c4..493290c0f2 100644
--- a/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
+++ b/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
@@ -1079,6 +1079,13 @@ components:
           type: number
         readOnly:
           type: boolean
+        features:
+          type: array
+          items:
+            type: string
+            enum:
+              - SCHEMA_REGISTRY
+              - KAFKA_CONNECT
       required:
         - id
         - name