Repository: provectus/kafka-ui
Issue: Support running kafka-ui on subset of topics to get around auth errors
Labels: type/enhancement
scope/backend

### Is your proposal related to a problem?
In our company we have kafka cluster common for all teams but it does not have convenient UI for our needs. Our team tried to deploy kafka-ui with our account that has only access to our topics but it failed with access error. It would be useful to have possibility to run kafka-ui on subset of topics to which account have the needed rights.

Addition question on described problem: Which rights kafka-ui needs to run correctly? Can't find it on project page.

### Describe the solution you'd like
The easiest solution (from my perspective) is to provide the config with a list of topics to which specified account has access.

As alternative kafka-ui can just skip authorization error on topics.


Issue comments:
Hey, thanks for reaching out.

This is for sure to be done via ignoring permission issues. We'll get this done in future releases :)


#928 #753 #482
@MaksimMyshkin Currently kafka-ui needs WRITE and DESCRIBE_CONFIGS permissions on a topic to display it correctly. If the account used to configure kafka-ui  has some access to any topic where DESCRIBE_CONFIGS is not allowed no topics is displayed at all. You might want to add DESCRIBE_CONFIGS  to all your topics as a workaround. 
Thank you for information. Unfortunately our team can't get access to all topics in the cluster (only to our topics). 

Also I dug a little and found out that errors is appearing on creation of AdminClient. It is even before getting information about topics. Is this code a reason why rights to all topics required? Looks like AdminClient requires wider permissions than it is needed for simpler scenarios (testing by producing and reading topics messages for example).
@MaksimMyshkin Sorry, looks like I didn't make myself clear enough. It was not about all topics in the cluster, only about "your" topics. If you have any access to a topic then WRITE and DESCRIBE_CONFIGS should be allowed for that topic.
@vdanylenko hey! Any updates here? What's the status on this issue? 
I afraid permissions that @vdanylenko described is not enough to work with topics from kafka-ui. At the moment I think that problem due to using AdminClient and I don't known how to work around it right. I just trying to find a reason to the problem that I (or someone else) can help to fix it sooner(
@MaksimMyshkin hey, wanna hop on our discord? We can try to investigate this. The link is available in README.
This issue has been automatically marked as stale because no requested feedback has been provided. It will be closed if no further activity occurs. Thank you for your contributions.
Using Kafka-UI in a multi-tenant scenario is not working currently. Main reason is, that there are no clusterwide permissions assigned to users. So referring to this table the following permissions are granted:
https://docs.confluent.io/platform/current/kafka/authorization.html

Topics:
Describe, DescribeConfigs, Read and if needed Write (single topic oder wildcard) ; note: read implies describe
Group:
Describe
Transactional ID:
Describe

The following data is available then (without cluster permissions) in a different tool:
![topics](https://user-images.githubusercontent.com/3028772/173685181-2454d5ee-9fab-4f02-b665-a0f938894583.png)
![brokers](https://user-images.githubusercontent.com/3028772/173693478-f97cff0c-4a56-4686-aa3a-b531ac9c4b11.png)
![overview](https://user-images.githubusercontent.com/3028772/173693494-6ad50a3d-a5e0-4e4a-81b2-595dfbb99184.jpg)
@moinmoin thanks for the feedback!
Faced with similar issue as @MaksimMyshkin where same proposed solution would be good enough for our use (relatively small tradeoff in situations where the list of topics is anyways known and does not grow with such speed that would cause the manual maintenance on the list to be a problem).

Code requiring alternative or limited-permissions approach (rather than it just crashing): https://github.com/provectus/kafka-ui/blob/4660bb8759e33a2a229bcdc6343d3be1bdfd2f28/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaService.java#L323-L337

Reached this realization by running the application with `LOGGING_LEVEL_ROOT: debug` where it became clear that from the list of commands executed on startup:

```txt
API_VERSIONS
METADATA
DESCRIBE_CLUSTER
DESCRIBE_CONFIGS <=== this one fails
```

Usage context: getTopicsData > loadTopicsConfig > describeConfigs
> The following data is available then (without cluster permissions) in a different tool:

@moinmoin - Don't hold out on us - what tool is this?
@TaylorSMarks This is the https://www.conduktor.io/download fat client.
I have opened a small PR which fixes the root issue (you lack permissions to run describe cluster), although I accomplished it by just ignoring the exception you get instead of going into a crash loop. I don't think adding the feature to be able to run on a subset of topics would actually fix the root issue, since the Admin Client would still fail to create with an auth exception.

Everything runs pretty ok with the change.

Here's the PR:
https://github.com/provectus/kafka-ui/pull/3325

The directions for opening a PR said I should request reviewers and add myself as the owner for the PR, but I lack permissions to do either of those things. Here's hoping the right people see, review, approve, and merge the PR...

Edit: My PR failed the E2E testing but it seems unrelated to my change. Every build from the last 6 days has been failing this same check.
Can confirm that this solves our issue :) Thank you @Haarolean @TaylorSMarks @iliax!
@allanpaiste great to hear that! 

Pull request review comments:

Patch that resolved the issue:
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 ea56edcf96e..7cdf2ef16d7 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
@@ -62,9 +62,11 @@
 import org.apache.kafka.common.TopicPartitionReplica;
 import org.apache.kafka.common.acl.AclOperation;
 import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.ClusterAuthorizationException;
 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.TopicAuthorizationException;
 import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.requests.DescribeLogDirsResponse;
@@ -176,6 +178,7 @@ public Mono<Map<String, List<ConfigEntry>>> getTopicsConfig() {
   }
 
   //NOTE: skips not-found topics (for which UnknownTopicOrPartitionException was thrown by AdminClient)
+  //and topics for which DESCRIBE_CONFIGS permission is not set (TopicAuthorizationException was thrown)
   public Mono<Map<String, List<ConfigEntry>>> getTopicsConfig(Collection<String> topicNames, boolean includeDoc) {
     var includeDocFixed = features.contains(SupportedFeature.CONFIG_DOCUMENTATION_RETRIEVAL) && includeDoc;
     // we need to partition calls, because it can lead to AdminClient timeouts in case of large topics count
@@ -196,7 +199,8 @@ private Mono<Map<String, List<ConfigEntry>>> getTopicsConfigImpl(Collection<Stri
         client.describeConfigs(
             resources,
             new DescribeConfigsOptions().includeSynonyms(true).includeDocumentation(includeDoc)).values(),
-        UnknownTopicOrPartitionException.class
+        UnknownTopicOrPartitionException.class,
+        TopicAuthorizationException.class
     ).map(config -> config.entrySet().stream()
         .collect(toMap(
             c -> c.getKey().name(),
@@ -208,11 +212,17 @@ private static Mono<Map<Integer, List<ConfigEntry>>> loadBrokersConfig(AdminClie
         .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()))
+        .onErrorResume(InvalidRequestException.class, th -> {
+          log.trace("Error while getting broker {} configs", brokerIds, th);
+          return Mono.just(Map.of());
+        })
+        // there are situations when kafka-ui user has no DESCRIBE_CONFIGS permission on cluster
+        .onErrorResume(ClusterAuthorizationException.class, th -> {
+          log.trace("AuthorizationException while getting configs for brokers {}", brokerIds, th);
+          return Mono.just(Map.of());
+        })
         .map(config -> config.entrySet().stream()
             .collect(toMap(
                 c -> Integer.valueOf(c.getKey().name()),
@@ -242,13 +252,16 @@ public Mono<Map<String, TopicDescription>> describeTopics(Collection<String> top
 
   private Mono<Map<String, TopicDescription>> describeTopicsImpl(Collection<String> topics) {
     return toMonoWithExceptionFilter(
-        client.describeTopics(topics).values(),
-        UnknownTopicOrPartitionException.class
+        client.describeTopics(topics).topicNameValues(),
+        UnknownTopicOrPartitionException.class,
+        // we only describe topics that we see from listTopics() API, so we should have permission to do it,
+        // but also adding this exception here for rare case when access restricted after we called listTopics()
+        TopicAuthorizationException.class
     );
   }
 
   /**
-   * Returns TopicDescription mono, or Empty Mono if topic not found.
+   * Returns TopicDescription mono, or Empty Mono if topic not visible.
    */
   public Mono<TopicDescription> describeTopic(String topic) {
     return describeTopics(List.of(topic)).flatMap(m -> Mono.justOrEmpty(m.get(topic)));
@@ -262,10 +275,11 @@ public Mono<TopicDescription> describeTopic(String topic) {
    * such topics in resulting map.
    * <p/>
    * This method converts input map into Mono[Map] ignoring keys for which KafkaFutures
-   * finished with <code>clazz</code> exception and empty Monos.
+   * finished with <code>classes</code> exceptions and empty Monos.
    */
+  @SafeVarargs
   static <K, V> Mono<Map<K, V>> toMonoWithExceptionFilter(Map<K, KafkaFuture<V>> values,
-                                                          Class<? extends KafkaException> clazz) {
+                                                          Class<? extends KafkaException>... classes) {
     if (values.isEmpty()) {
       return Mono.just(Map.of());
     }
@@ -277,7 +291,7 @@ static <K, V> Mono<Map<K, V>> toMonoWithExceptionFilter(Map<K, KafkaFuture<V>> v
                 .defaultIfEmpty(Tuples.of(e.getKey(), Optional.empty())) //tracking empty Monos
                 .onErrorResume(
                     // tracking Monos with suppressible error
-                    th -> th.getClass().isAssignableFrom(clazz),
+                    th -> Stream.of(classes).anyMatch(clazz -> th.getClass().isAssignableFrom(clazz)),
                     th -> Mono.just(Tuples.of(e.getKey(), Optional.empty()))))
         .toList();
 
@@ -300,6 +314,7 @@ public Mono<Map<Integer, Map<String, DescribeLogDirsResponse.LogDirInfo>>> descr
       Collection<Integer> brokerIds) {
     return toMono(client.describeLogDirs(brokerIds).all())
         .onErrorResume(UnsupportedVersionException.class, th -> Mono.just(Map.of()))
+        .onErrorResume(ClusterAuthorizationException.class, th -> Mono.just(Map.of()))
         .onErrorResume(th -> true, th -> {
           log.warn("Error while calling describeLogDirs", th);
           return Mono.just(Map.of());
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 ab22cecaafc..b9038677267 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
@@ -162,9 +162,14 @@ public Mono<InternalTopic> getTopicDetails(KafkaCluster cluster, String topicNam
   }
 
   public Mono<List<ConfigEntry>> getTopicConfigs(KafkaCluster cluster, String topicName) {
+    // there 2 case that we cover here:
+    // 1. topic not found/visible - describeTopic() will be empty and we will throw TopicNotFoundException
+    // 2. topic is visible, but we don't have DESCRIBE_CONFIG permission - we should return empty list
     return adminClientService.get(cluster)
-        .flatMap(ac -> ac.getTopicsConfig(List.of(topicName), true))
-        .map(m -> m.values().stream().findFirst().orElseThrow(TopicNotFoundException::new));
+        .flatMap(ac -> ac.describeTopic(topicName)
+            .switchIfEmpty(Mono.error(new TopicNotFoundException()))
+            .then(ac.getTopicsConfig(List.of(topicName), true))
+            .map(m -> m.values().stream().findFirst().orElse(List.of())));
   }
 
   private Mono<InternalTopic> createTopic(KafkaCluster c, ReactiveAdminClient adminClient,


Test patch: