Hello there ubergarm! 👋

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. 👀 Hello @ubergarm, thank you for opening as issue. 
We found that if contoller node is unknown(which is Msk Serverless case) we disabling topics deletion, which is not right. We created PR to fix it (https://github.com/provectus/kafka-ui/pull/3548), we will merge it soon. I confirmed this works now in `0.6.1` thanks so much for the quick point release!!
Bug: Topic deletion is unavailable on MSK Serverless for 0.6.0
## Issue
I can successfully configure and run `kafka-ui-api-v0.6.0.jar` to connect to our MSK Serverless cluster, list topics, and create a topic. However when I attempt to delete the topic it always pops up a red toast saying `400 Bad Request Topic deletion restricted`. I've tried many config options after grep'ing through the code but nothing seems to change it.

## Expected Behavior
When I click on `Topics->Delete Selected Topics->Confirm` I expect it to delete the topic and remove it from the list. (see screenshot at bottom)

## AWS EC2 IAM Role Setup
I have an EC2 instance with the correct security group, IAM policy with wildcard permissions on the cluster, and IAM role. To be sure I've confirmed that I can run the kafka cli tools to create, list, and delete topics successfully on the same EC2 instance running ubuntu LTS and all this works fine:
```bash
export BS="boot-blahblahblah.c3.kafka-serverless.us-east-2.amazonaws.com:9098" 
# CREATE TOPIC
bin/kafka-topics.sh \
    --bootstrap-server $BS \
    --command-config client.properties \
    --create \
    --topic myTestTopic \
    --partitions 2
# LIST TOPICS
bin/kafka-topics.sh \
    --bootstrap-server $BS \
    --command-config client.properties \
    --list
# DELETE TOPIC
bin/kafka-topics.sh \
    --bootstrap-server $BS \
    --command-config client.properties \
    --delete \
    --topic myTestTopic
```

## kafka-ui invokation
Now on the same EC2 instance, I'm running kafka-ui as such:
```bash
# using sdkman
$ sdk current java
Using java version 17.0.5-amzn
# java details
$ java --version
openjdk 17.0.5 2022-10-18 LTS
OpenJDK Runtime Environment Corretto-17.0.5.8.1 (build 17.0.5+8-LTS)
OpenJDK 64-Bit Server VM Corretto-17.0.5.8.1 (build 17.0.5+8-LTS, mixed mode, sharing)
# start up kafka-ui
$ java \
    --add-opens java.rmi/javax.rmi.ssl=ALL-UNNAMED \
    -Dspring.profiles.active=default \
    -jar kafka-ui-api-v0.6.0.jar
```

Here is an example log from when i run like this:
```
 _   _ ___    __             _                _          _  __      __ _
| | | |_ _|  / _|___ _ _    /_\  _ __ __ _ __| |_  ___  | |/ /__ _ / _| |_____
| |_| || |  |  _/ _ | '_|  / _ \| '_ / _` / _| ' \/ -_) | ' </ _` |  _| / / _`|
 \___/|___| |_| \___|_|   /_/ \_| .__\__,_\__|_||_\___| |_|\_\__,_|_| |_\_\__,|
                                 |_|

2023-03-22 17:53:11,991 INFO  [background-preinit] o.h.v.i.u.Version: HV000001: Hibernate Validator 6.2.5.Final
2023-03-22 17:53:12,101 INFO  [main] c.p.k.u.KafkaUiApplication: Starting KafkaUiApplication using Java 17.0.5 on ec2.staging with PID 41483 (/home/ubuntu/kafka-ui/kafka-ui-api-v0.6.0.jar started by ubuntu in /home/ubuntu/kafka-ui)
2023-03-22 17:53:12,102 DEBUG [main] c.p.k.u.KafkaUiApplication: Running with Spring Boot v2.7.5, Spring v5.3.23
2023-03-22 17:53:12,103 INFO  [main] c.p.k.u.KafkaUiApplication: The following 1 profile is active: "default"
2023-03-22 17:53:16,382 INFO  [main] o.s.c.s.PostProcessorRegistrationDelegate$BeanPostProcessorChecker: Bean 'org.springframework.security.config.annotation.method.configuration.ReactiveMethodSecurityConfiguration' of type [org.springframework.security.config.annotation.method.configuration.ReactiveMethodSecurityConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2023-03-22 17:53:18,049 DEBUG [main] c.p.k.u.s.SerdesInitializer: Configuring serdes for cluster staging
2023-03-22 17:53:19,485 INFO  [main] o.s.b.a.e.w.EndpointLinksResolver: Exposing 2 endpoint(s) beneath base path '/actuator'
2023-03-22 17:53:19,765 INFO  [main] o.s.b.a.s.r.ReactiveUserDetailsServiceAutoConfiguration:

Using generated security password: BLAHBLAHB-BLAH-BLAH-BLAHBLAH-BLAH

2023-03-22 17:53:19,947 INFO  [main] c.p.k.u.c.a.OAuthSecurityConfig: Configuring OAUTH2 authentication.
2023-03-22 17:53:20,967 INFO  [main] o.s.b.w.e.n.NettyWebServer: Netty started on port 8080
2023-03-22 17:53:21,056 INFO  [main] c.p.k.u.KafkaUiApplication: Started KafkaUiApplication in 10.707 seconds (JVM running for 12.153)
2023-03-22 17:53:21,099 DEBUG [parallel-1] c.p.k.u.s.ClustersStatisticsScheduler: Start getting metrics for kafkaCluster: staging
2023-03-22 17:53:21,130 INFO  [parallel-1] o.a.k.c.a.AdminClientConfig: AdminClientConfig values:
	bootstrap.servers = [boot-blahblah.c3.kafka-serverless.us-east-2.amazonaws.com:9098]
	client.dns.lookup = use_all_dns_ips
	client.id = kafka-ui-admin-1679507601-1
	connections.max.idle.ms = 300000
	default.api.timeout.ms = 60000
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	receive.buffer.bytes = 65536
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retries = 2147483647
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = class software.amazon.msk.auth.iam.IAMClientCallbackHandler
	sasl.jaas.config = [hidden]
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.connect.timeout.ms = null
	sasl.login.read.timeout.ms = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.login.retry.backoff.max.ms = 10000
	sasl.login.retry.backoff.ms = 100
	sasl.mechanism = AWS_MSK_IAM
	sasl.oauthbearer.clock.skew.seconds = 30
	sasl.oauthbearer.expected.audience = null
	sasl.oauthbearer.expected.issuer = null
	sasl.oauthbearer.jwks.endpoint.refresh.ms = 3600000
	sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms = 10000
	sasl.oauthbearer.jwks.endpoint.retry.backoff.ms = 100
	sasl.oauthbearer.jwks.endpoint.url = null
	sasl.oauthbearer.scope.claim.name = scope
	sasl.oauthbearer.sub.claim.name = sub
	sasl.oauthbearer.token.endpoint.url = null
	security.protocol = SASL_SSL
	security.providers = null
	send.buffer.bytes = 131072
	socket.connection.setup.timeout.max.ms = 30000
	socket.connection.setup.timeout.ms = 10000
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
	ssl.endpoint.identification.algorithm = https
	ssl.engine.factory.class = null
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.certificate.chain = null
	ssl.keystore.key = null
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLSv1.3
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.certificates = null
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS

2023-03-22 17:53:21,554 INFO  [parallel-1] o.a.k.c.s.a.AbstractLogin: Successfully logged in.
2023-03-22 17:53:21,609 INFO  [parallel-1] o.a.k.c.u.AppInfoParser: Kafka version: 3.3.1
2023-03-22 17:53:21,609 INFO  [parallel-1] o.a.k.c.u.AppInfoParser: Kafka commitId: e23c59d00e687ff5
2023-03-22 17:53:21,609 INFO  [parallel-1] o.a.k.c.u.AppInfoParser: Kafka startTimeMs: 1679507601606
2023-03-22 17:53:24,183 DEBUG [parallel-2] c.p.k.u.s.ClustersStatisticsScheduler: Metrics updated for cluster: staging
```

## kafka-ui config
Here is the contents of my config file `application-default.yml`:

```yaml
# first specify the kafka cluster connection and config options
# https://github.com/provectus/kafka-ui/blob/master/documentation/guides/AWS_IAM.md
dynamic.config.enabled: false
delete.topic.enable: true
delete.topic.enabled: true
topic.deletion: true
kafka:
  clusters:
    - name: staging
      bootstrapServers: "boot-blahblah.c3.kafka-serverless.us-east-2.amazonaws.com:9098"
      readOnly: false
      delete.topic.enable: true
      delete.topic.enabled: true
      delete:
        topic:
          enable: true
          enabled: true
      topic.deletion: true
      topic:
        deletion: true
      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;

## how to authenticate with the kafka-ui app
auth:
  type: OAUTH2
  oauth2:
    client:
      github:
        provider: github
        clientId: xxx
        clientSecret: yyy
        scope:
          - read:org
        user-name-attribute: login
        custom-params:
          type: github

# next specify the role based access control stuff
# https://github.com/provectus/kafka-ui/wiki/RBAC-(role-based-access-control)
rbac:
  roles:
    - name: "admins"
      clusters:
        - staging
      subjects:
        - provider: oauth_github
          type: user
          value: ubergarm
      permissions:
        - resource: clusterconfig
          actions: all
        - resource: topic
          value: ".*"
          actions: all
        - resource: topic
          value: "foobar"
          actions:
            - VIEW # can be upper or lower case
            - CREATE
            - EDIT
            - DELETE
            - MESSAGES_READ
            - MESSAGES_PRODUCE
            - MESSAGES_DELETE
        - resource: consumer
          value: ".*"
          actions: all
        - resource: schema
          value: ".*"
          actions: all
        - resource: connect
          value: ".*"
          actions: all
        - resource: ksql
          actions: all
```

I initially tried without OAUTH or RBAC, but that didn't work, so I tried with it and even explicitly set permissions to delete the test topic. I've also tried many combinations of variable names and even tried some ENV VAR versions to allow deleting of topics but nothing has yet worked.

Finally I tried to use the previous release `kafka-ui-api-v0.5.0.jar` but it can't even list topics and gives this error at runtime:
```
2023-03-22 17:28:28,138 INFO  [parallel-1] o.a.k.c.s.a.AbstractLogin: Successfully logged in.
2023-03-22 17:28:28,391 INFO  [parallel-1] o.a.k.c.u.AppInfoParser: Kafka version: 3.3.1
2023-03-22 17:28:28,392 INFO  [parallel-1] o.a.k.c.u.AppInfoParser: Kafka commitId: e23c59d00e687ff5
2023-03-22 17:28:28,392 INFO  [parallel-1] o.a.k.c.u.AppInfoParser: Kafka startTimeMs: 1679506108389
2023-03-22 17:28:30,977 ERROR [parallel-2] c.p.k.u.s.StatisticsService: Failed to collect cluster staging info
org.apache.kafka.common.errors.UnsupportedVersionException: The broker does not support DESCRIBE_LOG_DIRS
2023-03-22 17:28:30,977 DEBUG [parallel-2] c.p.k.u.s.ClustersStatisticsScheduler: Metrics updated for cluster: staging
2023-03-22 17:28:57,644 DEBUG [parallel-2] c.p.k.u.s.ClustersStatisticsScheduler: Start getting metrics for kafkaCluster: staging
2023-03-22 17:28:57,708 ERROR [parallel-2] c.p.k.u.s.StatisticsService: Failed to collect cluster staging info
org.apache.kafka.common.errors.UnsupportedVersionException: The broker does not support DESCRIBE_LOG_DIRS
2023-03-22 17:28:57,709 DEBUG [parallel-2] c.p.k.u.s.ClustersStatisticsScheduler: Metrics updated for cluster: staging
```

## Screenshots

Here is a screenshot of the webapp where I just created a new topic fine, but am unable to delete it.
![image](https://user-images.githubusercontent.com/4724321/226995328-a16f1437-57b2-4a7c-9a96-f8d5f3775444.png)

## Thanks
Thanks much! Hopefully it is just some config confusion and we can get this going. Cheers!
kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/FeatureService.java kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/StatisticsService.java
kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/FeatureService.java kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/StatisticsService.java
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 f5fbf262649..ec749abd14e 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
@@ -25,7 +25,8 @@ public class FeatureService {
 
   private final AdminClientService adminClientService;
 
-  public Mono<List<ClusterFeature>> getAvailableFeatures(KafkaCluster cluster, @Nullable Node controller) {
+  public Mono<List<ClusterFeature>> getAvailableFeatures(KafkaCluster cluster,
+                                                         ReactiveAdminClient.ClusterDescription clusterDescription) {
     List<Mono<ClusterFeature>> features = new ArrayList<>();
 
     if (Optional.ofNullable(cluster.getConnectsClients())
@@ -42,17 +43,15 @@ public Mono<List<ClusterFeature>> getAvailableFeatures(KafkaCluster cluster, @Nu
       features.add(Mono.just(ClusterFeature.SCHEMA_REGISTRY));
     }
 
-    if (controller != null) {
-      features.add(
-          isTopicDeletionEnabled(cluster, controller)
-              .flatMap(r -> Boolean.TRUE.equals(r) ? Mono.just(ClusterFeature.TOPIC_DELETION) : Mono.empty())
-      );
-    }
+    features.add(topicDeletionEnabled(cluster, clusterDescription.getController()));
 
     return Flux.fromIterable(features).flatMap(m -> m).collectList();
   }
 
-  private Mono<Boolean> isTopicDeletionEnabled(KafkaCluster cluster, Node controller) {
+  private Mono<ClusterFeature> topicDeletionEnabled(KafkaCluster cluster, @Nullable Node controller) {
+    if (controller == null) {
+      return Mono.just(ClusterFeature.TOPIC_DELETION); // assuming it is enabled by default
+    }
     return adminClientService.get(cluster)
         .flatMap(ac -> ac.loadBrokersConfig(List.of(controller.id())))
         .map(config ->
@@ -61,6 +60,9 @@ 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(true));
+                .orElse(true))
+        .flatMap(enabled -> enabled
+            ? Mono.just(ClusterFeature.TOPIC_DELETION)
+            : Mono.empty());
   }
 }
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/StatisticsService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/StatisticsService.java
index a36a64ff6dc..994c30714ae 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/StatisticsService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/StatisticsService.java
@@ -41,7 +41,7 @@ private Mono<Statistics> getStatistics(KafkaCluster cluster) {
                     List.of(
                         metricsCollector.getBrokerMetrics(cluster, description.getNodes()),
                         getLogDirInfo(description, ac),
-                        featureService.getAvailableFeatures(cluster, description.getController()),
+                        featureService.getAvailableFeatures(cluster, description),
                         loadTopicConfigs(cluster),
                         describeTopics(cluster)),
                     results ->
