Repository: provectus/kafka-ui
Issue: KSQLDB: Support Basic Auth
Labels: type/enhancement
good first issue
scope/backend
status/accepted

Hi Team,

I'm trying to connect KSQL DB which is on Confluent Cloud but I'm getting error "Failed to fetch tables and streams".

I've added below property in Docker-Compose file,
KAFKA_CLUSTERS_0_KSQLDBSERVER: https://test-db-xxx.confluent.cloud

To connect that DB, I guess we need to provide a Key/Secret and I don't see properties to provide the same.

Could you pls advise, what am I missing here?


Thank you.

Issue comments:
Hello there niyatidoshi! 👋

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.

If your ksqldb server uses basic auth, unfortunately we currently do not support auth. We're planning on implementing this in future releases.
Hey Haarolean,

Thanks for the update. Our KSql DB is in Confluent Cloud. Here is the doc that was followed https://docs.confluent.io/cloud/current/get-started/ksql.html and for the 1st step (Access Control) we chose Granular access.  For the connectivity, an additional Key/Secret was created. Hope this helps.

Could you pls include this feature in your future release?


Thank you.
@niyatidoshi we'll try to for sure :) thanks for the instructions
When will this be released :/
> When will this be released :/

@Diyaa1 pull `master`-labeled image. Caution: it's a dev branch.

Pull request review comments:
let's extract both username and password as variables here
let's add `@ToString` and exclude password field
the same with tostring

Patch that resolved the issue:
diff --git a/README.md b/README.md
index 5a0f51cf07e..494909e6730 100644
--- a/README.md
+++ b/README.md
@@ -163,6 +163,8 @@ For example, if you want to use an environment variable to set the `name` parame
 |`KAFKA_CLUSTERS_0_NAME` | Cluster name
 |`KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS` 	|Address where to connect 
 |`KAFKA_CLUSTERS_0_KSQLDBSERVER` 	| KSQL DB server address 
+|`KAFKA_CLUSTERS_0_KSQLDBSERVERAUTH_USERNAME` 	| KSQL DB server's basic authentication username 
+|`KAFKA_CLUSTERS_0_KSQLDBSERVERAUTH_PASSWORD` 	| KSQL DB server's basic authentication password 
 |`KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL` 	|Security protocol to connect to the brokers. For SSL connection use "SSL", for plaintext connection don't set this environment variable
 |`KAFKA_CLUSTERS_0_SCHEMAREGISTRY`   	|SchemaRegistry's address
 |`KAFKA_CLUSTERS_0_SCHEMAREGISTRYAUTH_USERNAME`   	|SchemaRegistry's basic authentication username
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/KsqlClient.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/KsqlClient.java
index 2e8026648d7..8d051234abb 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/KsqlClient.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/KsqlClient.java
@@ -3,7 +3,9 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.provectus.kafka.ui.exception.UnprocessableEntityException;
+import com.provectus.kafka.ui.model.KafkaCluster;
 import com.provectus.kafka.ui.model.KsqlCommandResponseDTO;
+import com.provectus.kafka.ui.service.ksql.KsqlApiClient;
 import com.provectus.kafka.ui.strategy.ksql.statement.BaseStrategy;
 import lombok.RequiredArgsConstructor;
 import lombok.SneakyThrows;
@@ -23,9 +25,10 @@ public class KsqlClient {
   private final WebClient webClient;
   private final ObjectMapper mapper;
 
-  public Mono<KsqlCommandResponseDTO> execute(BaseStrategy ksqlStatement) {
+  public Mono<KsqlCommandResponseDTO> execute(BaseStrategy ksqlStatement, KafkaCluster cluster) {
     return webClient.post()
         .uri(ksqlStatement.getUri())
+        .headers(httpHeaders -> KsqlApiClient.setBasicAuthIfEnabled(httpHeaders, cluster))
         .accept(new MediaType("application", "vnd.ksql.v1+json"))
         .body(BodyInserters.fromValue(ksqlStatement.getKsqlCommand()))
         .retrieve()
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java
index 0d83e143c5d..de4c8bc3477 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java
@@ -8,6 +8,7 @@
 import java.util.Set;
 import javax.annotation.PostConstruct;
 import lombok.Data;
+import lombok.ToString;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.util.StringUtils;
@@ -26,6 +27,7 @@ public static class Cluster {
     String schemaRegistry;
     SchemaRegistryAuth schemaRegistryAuth;
     String ksqldbServer;
+    KsqldbServerAuth ksqldbServerAuth;
     String schemaNameTemplate = "%s-value";
     String keySchemaNameTemplate = "%s-key";
     String protobufFile;
@@ -57,6 +59,13 @@ public static class SchemaRegistryAuth {
     String password;
   }
 
+  @Data
+  @ToString(exclude = "password")
+  public static class KsqldbServerAuth {
+    String username;
+    String password;
+  }
+
   @PostConstruct
   public void validateAndSetDefaults() {
     validateClusterNames();
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 1eb6199f96c..5709709e9b4 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
@@ -17,6 +17,7 @@
 import com.provectus.kafka.ui.model.InternalBrokerConfig;
 import com.provectus.kafka.ui.model.InternalBrokerDiskUsage;
 import com.provectus.kafka.ui.model.InternalClusterState;
+import com.provectus.kafka.ui.model.InternalKsqlServer;
 import com.provectus.kafka.ui.model.InternalPartition;
 import com.provectus.kafka.ui.model.InternalReplica;
 import com.provectus.kafka.ui.model.InternalSchemaRegistry;
@@ -53,6 +54,7 @@ public interface ClusterMapper {
   @Mapping(target = "protobufFile", source = "protobufFile", qualifiedByName = "resolvePath")
   @Mapping(target = "properties", source = "properties", qualifiedByName = "setProperties")
   @Mapping(target = "schemaRegistry", source = ".", qualifiedByName = "setSchemaRegistry")
+  @Mapping(target = "ksqldbServer", source = ".", qualifiedByName = "setKsqldbServer")
   KafkaCluster toKafkaCluster(ClustersProperties.Cluster clusterProperties);
 
   ClusterStatsDTO toClusterStats(InternalClusterState clusterState);
@@ -110,6 +112,24 @@ default InternalSchemaRegistry setSchemaRegistry(ClustersProperties.Cluster clus
     return internalSchemaRegistry.build();
   }
 
+  @Named("setKsqldbServer")
+  default InternalKsqlServer setKsqldbServer(ClustersProperties.Cluster clusterProperties) {
+    if (clusterProperties == null
+            || clusterProperties.getKsqldbServer() == null) {
+      return null;
+    }
+
+    InternalKsqlServer.InternalKsqlServerBuilder internalKsqlServerBuilder =
+            InternalKsqlServer.builder().url(clusterProperties.getKsqldbServer());
+
+    if (clusterProperties.getKsqldbServerAuth() != null) {
+      internalKsqlServerBuilder.username(clusterProperties.getKsqldbServerAuth().getUsername());
+      internalKsqlServerBuilder.password(clusterProperties.getKsqldbServerAuth().getPassword());
+    }
+
+    return internalKsqlServerBuilder.build();
+  }
+
   TopicDetailsDTO toTopicDetails(InternalTopic topic);
 
   @Mapping(target = "isReadOnly", source = "readOnly")
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalKsqlServer.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalKsqlServer.java
new file mode 100644
index 00000000000..a1c715bb586
--- /dev/null
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalKsqlServer.java
@@ -0,0 +1,14 @@
+package com.provectus.kafka.ui.model;
+
+import lombok.Builder;
+import lombok.Data;
+import lombok.ToString;
+
+@Data
+@ToString(exclude = "password")
+@Builder(toBuilder = true)
+public class InternalKsqlServer {
+  private final String url;
+  private final String username;
+  private final String password;
+}
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 b9f1ea96768..eab02e789f0 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
@@ -21,7 +21,7 @@ public class KafkaCluster {
   private final String jmxPassword;
   private final String bootstrapServers;
   private final InternalSchemaRegistry schemaRegistry;
-  private final String ksqldbServer;
+  private final InternalKsqlServer ksqldbServer;
   private final List<KafkaConnectCluster> kafkaConnect;
   private final String schemaNameTemplate;
   private final String keySchemaNameTemplate;
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KsqlService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KsqlService.java
index 4e970dc0c8c..6f74ede75ec 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KsqlService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KsqlService.java
@@ -28,10 +28,10 @@ public Mono<KsqlCommandResponseDTO> executeKsqlCommand(KafkaCluster cluster,
               e instanceof ClusterNotFoundException ? e : new KsqlDbNotFoundException();
           return Mono.error(throwable);
         })
-        .flatMap(host -> getStatementStrategyForKsqlCommand(ksqlCommand)
-            .map(statement -> statement.host(host))
+        .flatMap(ksqlServer -> getStatementStrategyForKsqlCommand(ksqlCommand)
+            .map(statement -> statement.host(ksqlServer.getUrl()))
         )
-        .flatMap(ksqlClient::execute);
+        .flatMap(baseStrategy -> ksqlClient.execute(baseStrategy, cluster));
   }
 
   private Mono<BaseStrategy> getStatementStrategyForKsqlCommand(
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlApiClient.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlApiClient.java
index 161c284c972..b83da666aaf 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlApiClient.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlApiClient.java
@@ -8,6 +8,7 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.TextNode;
+import com.provectus.kafka.ui.exception.ValidationException;
 import com.provectus.kafka.ui.model.KafkaCluster;
 import com.provectus.kafka.ui.service.ksql.response.ResponseParser;
 import java.util.List;
@@ -18,6 +19,7 @@
 import lombok.Value;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.core.codec.DecodingException;
+import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.json.Jackson2JsonDecoder;
 import org.springframework.util.MimeTypeUtils;
@@ -79,12 +81,25 @@ private WebClient webClient() {
         .build();
     return WebClient.builder()
         .codecs(c -> c.defaultCodecs().maxInMemorySize((int) maxBuffSize.toBytes()))
+        .defaultHeaders(httpHeaders -> setBasicAuthIfEnabled(httpHeaders, cluster))
         .exchangeStrategies(exchangeStrategies)
         .build();
   }
 
+  public static void setBasicAuthIfEnabled(HttpHeaders headers, KafkaCluster cluster) {
+    String username = cluster.getKsqldbServer().getUsername();
+    String password = cluster.getKsqldbServer().getPassword();
+    if (username != null && password != null) {
+      headers.setBasicAuth(username, password);
+    } else if (username != null) {
+      throw new ValidationException("You specified username but did not specify password");
+    } else if (password != null) {
+      throw new ValidationException("You specified password but did not specify username");
+    }
+  }
+
   private String baseKsqlDbUri() {
-    return cluster.getKsqldbServer();
+    return cluster.getKsqldbServer().getUrl();
   }
 
   private KsqlRequest ksqlRequest(String ksql, Map<String, String> streamProperties) {


Test patch:
diff --git a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/KsqlServiceTest.java b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/KsqlServiceTest.java
index f41b595e791..ed434efba44 100644
--- a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/KsqlServiceTest.java
+++ b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/KsqlServiceTest.java
@@ -2,6 +2,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -9,6 +10,7 @@
 import com.provectus.kafka.ui.client.KsqlClient;
 import com.provectus.kafka.ui.exception.KsqlDbNotFoundException;
 import com.provectus.kafka.ui.exception.UnprocessableEntityException;
+import com.provectus.kafka.ui.model.InternalKsqlServer;
 import com.provectus.kafka.ui.model.KafkaCluster;
 import com.provectus.kafka.ui.model.KsqlCommandDTO;
 import com.provectus.kafka.ui.model.KsqlCommandResponseDTO;
@@ -62,7 +64,7 @@ void shouldThrowUnprocessableEntityExceptionOnExecuteKsqlCommand() {
     KsqlCommandDTO command =
         (new KsqlCommandDTO()).ksql("CREATE STREAM users WITH (KAFKA_TOPIC='users');");
     KafkaCluster kafkaCluster = Mockito.mock(KafkaCluster.class);
-    when(kafkaCluster.getKsqldbServer()).thenReturn("localhost:8088");
+    when(kafkaCluster.getKsqldbServer()).thenReturn(InternalKsqlServer.builder().url("localhost:8088").build());
 
     StepVerifier.create(ksqlService.executeKsqlCommand(kafkaCluster, Mono.just(command)))
         .verifyError(UnprocessableEntityException.class);
@@ -77,8 +79,8 @@ void shouldSetHostToStrategy() {
     KsqlCommandDTO command = (new KsqlCommandDTO()).ksql("describe streams;");
     KafkaCluster kafkaCluster = Mockito.mock(KafkaCluster.class);
 
-    when(kafkaCluster.getKsqldbServer()).thenReturn(host);
-    when(ksqlClient.execute(any())).thenReturn(Mono.just(new KsqlCommandResponseDTO()));
+    when(kafkaCluster.getKsqldbServer()).thenReturn(InternalKsqlServer.builder().url(host).build());
+    when(ksqlClient.execute(any(), any())).thenReturn(Mono.just(new KsqlCommandResponseDTO()));
 
     ksqlService.executeKsqlCommand(kafkaCluster, Mono.just(command)).block();
     assertThat(alternativeStrategy.getUri()).isEqualTo(host + "/ksql");
@@ -90,12 +92,12 @@ void shouldCallClientAndReturnResponse() {
     KafkaCluster kafkaCluster = Mockito.mock(KafkaCluster.class);
     KsqlCommandResponseDTO response = new KsqlCommandResponseDTO().message("success");
 
-    when(kafkaCluster.getKsqldbServer()).thenReturn("host");
-    when(ksqlClient.execute(any())).thenReturn(Mono.just(response));
+    when(kafkaCluster.getKsqldbServer()).thenReturn(InternalKsqlServer.builder().url("host").build());
+    when(ksqlClient.execute(any(), any())).thenReturn(Mono.just(response));
 
     KsqlCommandResponseDTO receivedResponse =
         ksqlService.executeKsqlCommand(kafkaCluster, Mono.just(command)).block();
-    verify(ksqlClient, times(1)).execute(alternativeStrategy);
+    verify(ksqlClient, times(1)).execute(eq(alternativeStrategy), any());
     assertThat(receivedResponse).isEqualTo(response);
 
   }
diff --git a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlApiClientTest.java b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlApiClientTest.java
index 5956d730829..a797175091f 100644
--- a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlApiClientTest.java
+++ b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlApiClientTest.java
@@ -9,6 +9,7 @@
 import com.fasterxml.jackson.databind.node.TextNode;
 import com.provectus.kafka.ui.AbstractIntegrationTest;
 import com.provectus.kafka.ui.container.KsqlDbContainer;
+import com.provectus.kafka.ui.model.InternalKsqlServer;
 import com.provectus.kafka.ui.model.KafkaCluster;
 import java.time.Duration;
 import java.util.List;
@@ -42,7 +43,8 @@ static void stopContainer() {
   // Tutorial is here: https://ksqldb.io/quickstart.html
   @Test
   void ksqTutorialQueriesWork() {
-    var client = new KsqlApiClient(KafkaCluster.builder().ksqldbServer(KSQL_DB.url()).build(), maxBuffSize);
+    var client = new KsqlApiClient(KafkaCluster.builder().ksqldbServer(
+            InternalKsqlServer.builder().url(KSQL_DB.url()).build()).build(), maxBuffSize);
     execCommandSync(client,
         "CREATE STREAM riderLocations (profileId VARCHAR, latitude DOUBLE, longitude DOUBLE) "
             + "WITH (kafka_topic='locations', value_format='json', partitions=1);",
diff --git a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlServiceV2Test.java b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlServiceV2Test.java
index d670680ea21..89823d643ea 100644
--- a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlServiceV2Test.java
+++ b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlServiceV2Test.java
@@ -4,6 +4,7 @@
 
 import com.provectus.kafka.ui.AbstractIntegrationTest;
 import com.provectus.kafka.ui.container.KsqlDbContainer;
+import com.provectus.kafka.ui.model.InternalKsqlServer;
 import com.provectus.kafka.ui.model.KafkaCluster;
 import com.provectus.kafka.ui.model.KsqlStreamDescriptionDTO;
 import com.provectus.kafka.ui.model.KsqlTableDescriptionDTO;
@@ -34,7 +35,8 @@ static void init() {
 
   @AfterAll
   static void cleanup() {
-    var client = new KsqlApiClient(KafkaCluster.builder().ksqldbServer(KSQL_DB.url()).build(), maxBuffSize);
+    var client = new KsqlApiClient(KafkaCluster.builder().ksqldbServer(
+        InternalKsqlServer.builder().url(KSQL_DB.url()).build()).build(), maxBuffSize);
 
     TABLES_TO_DELETE.forEach(t ->
         client.execute(String.format("DROP TABLE IF EXISTS %s DELETE TOPIC;", t), Map.of())
@@ -51,7 +53,7 @@ static void cleanup() {
 
   @Test
   void listStreamsReturnsAllKsqlStreams() {
-    var cluster = KafkaCluster.builder().ksqldbServer(KSQL_DB.url()).build();
+    var cluster = KafkaCluster.builder().ksqldbServer(InternalKsqlServer.builder().url(KSQL_DB.url()).build()).build();
     var streamName = "stream_" + System.currentTimeMillis();
     STREAMS_TO_DELETE.add(streamName);
 
@@ -80,7 +82,7 @@ void listStreamsReturnsAllKsqlStreams() {
 
   @Test
   void listTablesReturnsAllKsqlTables() {
-    var cluster = KafkaCluster.builder().ksqldbServer(KSQL_DB.url()).build();
+    var cluster = KafkaCluster.builder().ksqldbServer(InternalKsqlServer.builder().url(KSQL_DB.url()).build()).build();
     var tableName = "table_" + System.currentTimeMillis();
     TABLES_TO_DELETE.add(tableName);