Repository: provectus/kafka-ui
Issue: Ability to run KSQL queries directly from UI
Labels: type/enhancement
scope/backend
scope/frontend


Issue comments:
Hi guys, on backend side this feature can be implemented by adding routing from kafka-ui backend to ksqldb server or by creating endpoints in the kafka-ui backend itself. @IldarAlmakaev @workshur @germanosin any thoughts?
I've merged to master backend side of this ticket
cc @workshur

Pull request review comments:

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/strategy/ksql/statement/ShowStrategy.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/strategy/ksql/statement/ShowStrategy.java
index 1b80c6648bf..1a2250e850d 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/strategy/ksql/statement/ShowStrategy.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/strategy/ksql/statement/ShowStrategy.java
@@ -1,6 +1,7 @@
 package com.provectus.kafka.ui.strategy.ksql.statement;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.provectus.kafka.ui.model.KsqlCommand;
 import com.provectus.kafka.ui.model.KsqlCommandResponse;
 import java.util.List;
 import java.util.Optional;
@@ -36,6 +37,15 @@ protected String getTestRegExp() {
     return "";
   }
 
+  @Override
+  public BaseStrategy ksqlCommand(KsqlCommand ksqlCommand) {
+    // return new instance to avoid conflicts for parallel requests
+    ShowStrategy clone = new ShowStrategy();
+    clone.setResponseValueKey(responseValueKey);
+    clone.ksqlCommand = ksqlCommand;
+    return clone;
+  }
+
   protected String getShowRegExp(String key) {
     return "show " + key + ";";
   }


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 aef12d8ec83..dd8c3ab8295 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
@@ -14,6 +14,7 @@
 import com.provectus.kafka.ui.model.KsqlCommand;
 import com.provectus.kafka.ui.model.KsqlCommandResponse;
 import com.provectus.kafka.ui.strategy.ksql.statement.BaseStrategy;
+import com.provectus.kafka.ui.strategy.ksql.statement.DescribeStrategy;
 import com.provectus.kafka.ui.strategy.ksql.statement.ShowStrategy;
 import java.util.List;
 import java.util.Optional;
@@ -30,6 +31,7 @@
 class KsqlServiceTest {
   private KsqlService ksqlService;
   private BaseStrategy baseStrategy;
+  private BaseStrategy alternativeStrategy;
 
   @Mock
   private ClustersStorage clustersStorage;
@@ -40,10 +42,11 @@ class KsqlServiceTest {
   @BeforeEach
   public void setUp() {
     this.baseStrategy = new ShowStrategy();
+    this.alternativeStrategy = new DescribeStrategy();
     this.ksqlService = new KsqlService(
         this.ksqlClient,
         this.clustersStorage,
-        List.of(baseStrategy)
+        List.of(baseStrategy, alternativeStrategy)
     );
   }
 
@@ -91,7 +94,7 @@ void shouldThrowUnprocessableEntityExceptionOnExecuteKsqlCommand() {
   void shouldSetHostToStrategy() {
     String clusterName = "test";
     String host = "localhost:8088";
-    KsqlCommand command = (new KsqlCommand()).ksql("show streams;");
+    KsqlCommand command = (new KsqlCommand()).ksql("describe streams;");
     KafkaCluster kafkaCluster = Mockito.mock(KafkaCluster.class);
 
     when(clustersStorage.getClusterByName(clusterName))
@@ -100,13 +103,13 @@ void shouldSetHostToStrategy() {
     when(ksqlClient.execute(any())).thenReturn(Mono.just(new KsqlCommandResponse()));
 
     ksqlService.executeKsqlCommand(clusterName, Mono.just(command)).block();
-    assertThat(baseStrategy.getUri()).isEqualTo(host + "/ksql");
+    assertThat(alternativeStrategy.getUri()).isEqualTo(host + "/ksql");
   }
 
   @Test
   void shouldCallClientAndReturnResponse() {
     String clusterName = "test";
-    KsqlCommand command = (new KsqlCommand()).ksql("show streams;");
+    KsqlCommand command = (new KsqlCommand()).ksql("describe streams;");
     KafkaCluster kafkaCluster = Mockito.mock(KafkaCluster.class);
     KsqlCommandResponse response = new KsqlCommandResponse().message("success");
 
@@ -117,7 +120,7 @@ void shouldCallClientAndReturnResponse() {
 
     KsqlCommandResponse receivedResponse =
         ksqlService.executeKsqlCommand(clusterName, Mono.just(command)).block();
-    verify(ksqlClient, times(1)).execute(baseStrategy);
+    verify(ksqlClient, times(1)).execute(alternativeStrategy);
     assertThat(receivedResponse).isEqualTo(response);
 
   }