Repository: provectus/kafka-ui
Issue: Support AWS MSK Serverless
Labels: scope/backend
status/accepted
type/feature

**Describe the bug**
The kafka-ui is working but the cluster is shown as "offline" and without topics.

 
**Set up**
I'm trying to use kafka-ui vv0.4.0(521ba0c) as a frontend for AWS MSK Serverless cluster mostly to create and manage topics.
I started a EC2 instance that uses IAM role with full access to the MSK cluster. 
Basically,I just follow the tutorial https://docs.aws.amazon.com/msk/latest/developerguide/serverless-getting-started.html
I tested the setup with a standard kafka-topic.sh tools to make sure the cluster is accessible from the instance and IAM role allows all operations with the cluster.

then I downloaded latest kafka-ui jar from github, and created application.yaml
```
logging.level.root: DEBUG
kafka:
  clusters:
    - name: local
      bootstrapServers: boot-ivd3n3is.c1.kafka-serverless.eu-central-1.amazonaws.com:9098
      properties:
        security.protocol: SASL_SSL
        sasl.mechanism: AWS_MSK_IAM
        sasl.client.callback.handler.class: software.amazon.msk.auth.iam.IAMClientCallbackHandler
        sasl.jaas.config: software.amazon.msk.auth.iam.IAMLoginModule required awsDebugCreds=true
```

and started it
`java -jar kafka-ui-api-v0.4.0.jar`

The web server is starting and working, but with no connection to Kafka.

in console, I see the error
```
2022-06-18 01:47:16,045 DEBUG [parallel-1] o.a.k.c.a.i.AdminMetadataManager: [AdminClient clientId=adminclient-84] Setting bootstrap cluster metadata Cluster(id = null, nodes = [boot-ivd3n3is.c1.kafka-serverless.eu-central-1.amazonaws.com:9098 (id: -1 rack: null)], partitions = [], controller = null).
2022-06-18 01:47:16,045 INFO  [parallel-1] o.a.k.c.m.Metrics: Metrics scheduler closed
2022-06-18 01:47:16,045 INFO  [parallel-1] o.a.k.c.m.Metrics: Closing reporter org.apache.kafka.common.metrics.JmxReporter
2022-06-18 01:47:16,045 INFO  [parallel-1] o.a.k.c.m.Metrics: Metrics reporters closed
2022-06-18 01:47:16,045 ERROR [parallel-1] c.p.k.u.s.MetricsService: Failed to collect cluster local info
java.lang.IllegalStateException: Error while creating AdminClient for Cluster local

```

**Expected behavior**

**Screenshots**
![image](https://user-images.githubusercontent.com/3240529/174418764-33dbf007-ca53-4bfd-97b9-fa3243e13494.png)

**Additional context**

java 18.0.1.1 2022-04-22
Amazon linux 5.10 118-111.515.amzn2.x86_64
[log.txt.gz](https://github.com/provectus/kafka-ui/files/8932807/log.txt.gz)



Issue comments:
Hello there SergKam! 👋

Thank you and congratulations 🎉 for opening your very first issue in this project! 💖

In case you want to claim this issue, please comment down below! We will try to get back to you as soon as we can. 👀
Hey, thanks for reaching out.

There's your problem in your logs:
`Caused by: java.lang.IllegalArgumentException: JAAS config entry not terminated by semi-colon`
Here's an [example](https://github.com/provectus/kafka-ui/blob/master/documentation/compose/kafka-ui-sasl.yaml#L20)
Unfortunately still not working. now the error
```
Caused by: org.apache.kafka.common.errors.InvalidRequestException: Amazon MSK Serverless doesn't support fetching broker or broker-logger configurations.
```

Okay, thanks. That's a totally different issue, it seems like we have to adjust some code to support _serverless_ MSK.
is it supported now?
> is it supported now?

not yet :(
When is the plan to release version 0.5 @Haarolean ?

> When is the plan to release version 0.5 @Haarolean ?

@advissor as soon as roadmap features are done. I'm not sure this will be a part of 0.5, but it has a high priority on our list as well.
Hello @SergKam , can you pls attach full log for latest exception?
> Hello @SergKam , can you pls attach full log for latest exception?

@iliax No, sorry. It was just an experiment, we decided not to use AWS MSK and switch to Confluent instead, so I deleted everything.
But you can try to set it up yourself using AWS MSK guide https://docs.aws.amazon.com/msk/latest/developerguide/serverless-getting-started.html
> > Hello @SergKam , can you pls attach full log for latest exception?
> 
> @iliax No, sorry. It was just an experiment, we decided not to use AWS MSK and switch to Confluent instead, so I deleted everything. But you can try to set it up yourself using AWS MSK guide https://docs.aws.amazon.com/msk/latest/developerguide/serverless-getting-started.html

ok, we will create cluster and check how it goes, thank you
Please note that we need some frontend changes as well (#2739), hopefully will get them live soon as well.
Hai team added msk cluster not getting data,can you please help here what i did mistake , geeting same error as above 

Pull request review comments:

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/TopicMetadataException.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/TopicMetadataException.java
index 7ccceefe614..a659f94f971 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/TopicMetadataException.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/TopicMetadataException.java
@@ -6,6 +6,10 @@ public TopicMetadataException(String message) {
     super(message);
   }
 
+  public TopicMetadataException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
   @Override
   public ErrorCode getErrorCode() {
     return ErrorCode.INVALID_ENTITY_STATE;
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/BrokerService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/BrokerService.java
index cd6ad93ba07..9c258fa8de8 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/BrokerService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/BrokerService.java
@@ -47,10 +47,7 @@ private Mono<Map<Integer, List<ConfigEntry>>> loadBrokersConfig(
   private Mono<List<ConfigEntry>> loadBrokersConfig(
       KafkaCluster cluster, Integer brokerId) {
     return loadBrokersConfig(cluster, Collections.singletonList(brokerId))
-        .map(map -> map.values().stream()
-            .findFirst()
-            .orElseThrow(() -> new NotFoundException(
-                String.format("Config for broker %s not found", brokerId))));
+        .map(map -> map.values().stream().findFirst().orElse(List.of()));
   }
 
   private Flux<InternalBrokerConfig> getBrokersConfig(KafkaCluster cluster, Integer brokerId) {
@@ -81,13 +78,6 @@ public Flux<BrokerDTO> getBrokers(KafkaCluster cluster) {
         .flatMapMany(Flux::fromIterable);
   }
 
-  public Mono<Node> getController(KafkaCluster cluster) {
-    return adminClientService
-        .get(cluster)
-        .flatMap(ReactiveAdminClient::describeCluster)
-        .map(ReactiveAdminClient.ClusterDescription::getController);
-  }
-
   public Mono<Void> updateBrokerLogDir(KafkaCluster cluster,
                                        Integer broker,
                                        BrokerLogdirUpdateDTO brokerLogDir) {
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/FeatureService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/FeatureService.java
index 24a824fb168..9097f2b25e7 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/FeatureService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/FeatureService.java
@@ -60,6 +60,6 @@ private Mono<Boolean> isTopicDeletionEnabled(KafkaCluster cluster, Node controll
                 .filter(e -> e.name().equals(DELETE_TOPIC_ENABLED_SERVER_PROPERTY))
                 .map(e -> Boolean.parseBoolean(e.value()))
                 .findFirst()
-                .orElse(false));
+                .orElse(true));
   }
 }
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ReactiveAdminClient.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ReactiveAdminClient.java
index 0556b3a0461..c08b839410e 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ReactiveAdminClient.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ReactiveAdminClient.java
@@ -18,6 +18,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -54,6 +55,7 @@
 import org.apache.kafka.common.config.ConfigResource;
 import org.apache.kafka.common.errors.GroupIdNotFoundException;
 import org.apache.kafka.common.errors.GroupNotEmptyException;
+import org.apache.kafka.common.errors.InvalidRequestException;
 import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.apache.kafka.common.requests.DescribeLogDirsResponse;
 import reactor.core.publisher.Mono;
@@ -81,7 +83,7 @@ public static class ClusterDescription {
   }
 
   public static Mono<ReactiveAdminClient> create(AdminClient adminClient) {
-    return getClusterVersionImpl(adminClient)
+    return getClusterVersion(adminClient)
         .map(ver ->
             new ReactiveAdminClient(
                 adminClient,
@@ -105,7 +107,13 @@ private static SupportedFeature getSupportedUpdateFeatureForVersion(String versi
   private static <T> Mono<T> toMono(KafkaFuture<T> future) {
     return Mono.<T>create(sink -> future.whenComplete((res, ex) -> {
       if (ex != null) {
-        sink.error(ex);
+        // KafkaFuture doc is unclear about what exception wrapper will be used
+        // (from docs it should be ExecutionException, be we actually see CompletionException, so checking both
+        if (ex instanceof CompletionException || ex instanceof ExecutionException) {
+          sink.error(ex.getCause()); //unwrapping exception
+        } else {
+          sink.error(ex);
+        }
       } else {
         sink.success(res);
       }
@@ -166,17 +174,29 @@ private Mono<Map<String, List<ConfigEntry>>> getTopicsConfigImpl(Collection<Stri
             c -> List.copyOf(c.getValue().entries()))));
   }
 
-  public Mono<Map<Integer, List<ConfigEntry>>> loadBrokersConfig(List<Integer> brokerIds) {
+  private static Mono<Map<Integer, List<ConfigEntry>>> loadBrokersConfig(AdminClient client, List<Integer> brokerIds) {
     List<ConfigResource> resources = brokerIds.stream()
         .map(brokerId -> new ConfigResource(ConfigResource.Type.BROKER, Integer.toString(brokerId)))
         .collect(toList());
     return toMono(client.describeConfigs(resources).all())
+        .doOnError(InvalidRequestException.class,
+            th -> log.trace("Error while getting broker {} configs", brokerIds, th))
+        // some kafka backends (like MSK serverless) do not support broker's configs retrieval,
+        // in that case InvalidRequestException will be thrown
+        .onErrorResume(InvalidRequestException.class, th -> Mono.just(Map.of()))
         .map(config -> config.entrySet().stream()
             .collect(toMap(
                 c -> Integer.valueOf(c.getKey().name()),
                 c -> new ArrayList<>(c.getValue().entries()))));
   }
 
+  /**
+   * Return per-broker configs or empty map if broker's configs retrieval not supported.
+   */
+  public Mono<Map<Integer, List<ConfigEntry>>> loadBrokersConfig(List<Integer> brokerIds) {
+    return loadBrokersConfig(client, brokerIds);
+  }
+
   public Mono<Map<String, TopicDescription>> describeTopics() {
     return listTopics(true).flatMap(this::describeTopics);
   }
@@ -280,20 +300,16 @@ public Mono<ClusterDescription> describeCluster() {
     }));
   }
 
-  private static Mono<String> getClusterVersionImpl(AdminClient client) {
-    return toMono(client.describeCluster().controller()).flatMap(controller ->
-        toMono(client.describeConfigs(
-                List.of(new ConfigResource(
-                    ConfigResource.Type.BROKER, String.valueOf(controller.id()))))
-            .all()
-            .thenApply(configs ->
-                configs.values().stream()
-                    .map(Config::entries)
-                    .flatMap(Collection::stream)
-                    .filter(entry -> entry.name().contains("inter.broker.protocol.version"))
-                    .findFirst().map(ConfigEntry::value)
-                    .orElse("1.0-UNKNOWN")
-            )));
+  private static Mono<String> getClusterVersion(AdminClient client) {
+    return toMono(client.describeCluster().controller())
+        .flatMap(controller -> loadBrokersConfig(client, List.of(controller.id())))
+        .map(configs -> configs.values().stream()
+            .flatMap(Collection::stream)
+            .filter(entry -> entry.name().contains("inter.broker.protocol.version"))
+            .findFirst()
+            .map(ConfigEntry::value)
+            .orElse("1.0-UNKNOWN")
+        );
   }
 
   public Mono<Void> deleteConsumerGroups(Collection<String> groupIds) {
@@ -306,10 +322,14 @@ public Mono<Void> deleteConsumerGroups(Collection<String> groupIds) {
 
   public Mono<Void> createTopic(String name,
                                 int numPartitions,
-                                short replicationFactor,
+                                @Nullable Integer replicationFactor,
                                 Map<String, String> configs) {
-    return toMono(client.createTopics(
-        List.of(new NewTopic(name, numPartitions, replicationFactor).configs(configs))).all());
+    var newTopic = new NewTopic(
+        name,
+        Optional.of(numPartitions),
+        Optional.ofNullable(replicationFactor).map(Integer::shortValue)
+    ).configs(configs);
+    return toMono(client.createTopics(List.of(newTopic)).all());
   }
 
   public Mono<Void> alterPartitionReassignments(
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java
index 1790a3005c2..2dffe0de75e 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java
@@ -170,11 +170,11 @@ private Mono<InternalTopic> createTopic(KafkaCluster c, ReactiveAdminClient admi
             adminClient.createTopic(
                 topicData.getName(),
                 topicData.getPartitions(),
-                topicData.getReplicationFactor().shortValue(),
+                topicData.getReplicationFactor(),
                 topicData.getConfigs()
             ).thenReturn(topicData)
         )
-        .onErrorResume(t -> Mono.error(new TopicMetadataException(t.getMessage())))
+        .onErrorMap(t -> new TopicMetadataException(t.getMessage(), t))
         .flatMap(topicData -> loadTopicAfterCreation(c, topicData.getName()));
   }
 
@@ -194,7 +194,7 @@ public Mono<InternalTopic> recreateTopic(KafkaCluster cluster, String topicName)
                         ac.createTopic(
                                 topic.getName(),
                                 topic.getPartitionCount(),
-                                (short) topic.getReplicationFactor(),
+                                topic.getReplicationFactor(),
                                 topic.getTopicConfigs()
                                     .stream()
                                     .collect(Collectors.toMap(InternalTopicConfig::getName,
@@ -430,7 +430,7 @@ public Mono<InternalTopic> cloneTopic(
                 ac.createTopic(
                     newTopicName,
                     topic.getPartitionCount(),
-                    (short) topic.getReplicationFactor(),
+                    topic.getReplicationFactor(),
                     topic.getTopicConfigs()
                         .stream()
                         .collect(Collectors
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 68b8afa65a3..af6c6c3e6ae 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
@@ -2261,7 +2261,6 @@ components:
       required:
         - name
         - partitions
-        - replicationFactor
 
     TopicUpdate:
       type: object


Test patch: