Merge pull request #30135 from ozangunalp/kafka_ui_fixes

Kafka Dev UI Fixes
quarkusio · Jan 3, 2023 · 9d6601f · 9d6601f
2 parents 533f315 + 7093aa6
commit 9d6601f
Show file tree

Hide file tree

Showing 6 changed files with 65 additions and 9 deletions.
diff --git a/extensions/kafka-client/deployment/src/main/resources/dev-static/js/config.js b/extensions/kafka-client/deployment/src/main/resources/dev-static/js/config.js
@@ -1,2 +1,5 @@
-export const api = '/q/dev/io.quarkus.quarkus-kafka-client/kafka-admin';
+export const api = () => {
+    let path = window.location.pathname
+    return path.replace('/kafka-dev-ui', '/kafka-admin')
+}
 export const ui = 'kafka-ui';
diff --git a/extensions/kafka-client/deployment/src/main/resources/dev-static/js/web/web.js b/extensions/kafka-client/deployment/src/main/resources/dev-static/js/web/web.js
@@ -2,7 +2,7 @@ import {api} from "../config.js"
 
 export function doPost(data, successCallback, errorCallback) {
     $.ajax({
-        url: api,
+        url: api(),
         type: 'POST',
         data: JSON.stringify(data),
         contentType: "application/json; charset=utf-8",

diff --git a/...kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/ui/KafkaUiHandler.java b/...kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/ui/KafkaUiHandler.java
@@ -24,7 +24,8 @@ public void handlePost(RoutingContext event) {
             endResponse(event, BAD_REQUEST, "Request body is null");
             return;
         }
-        var body = event.body().asJsonObject();
+        var webUtils = kafkaWebUiUtils();
+        var body = webUtils.fromJson(event.body().buffer());
         if (body == null) {
             endResponse(event, BAD_REQUEST, "Request JSON body is null");
             return;
@@ -34,7 +35,6 @@ public void handlePost(RoutingContext event) {
         var message = "OK";
         var error = "";
 
-        var webUtils = kafkaWebUiUtils();
         var adminClient = kafkaAdminClient();
 
         boolean res = false;
@@ -50,7 +50,7 @@ public void handlePost(RoutingContext event) {
                         res = true;
                         break;
                     case "createTopic":
-                        var topicCreateRq = event.body().asPojo(KafkaCreateTopicRequest.class);
+                        var topicCreateRq = webUtils.fromJson(event.body().buffer(), KafkaCreateTopicRequest.class);
                         res = adminClient.createTopic(topicCreateRq);
                         message = webUtils.toJson(webUtils.getTopics());
                         break;
@@ -64,17 +64,17 @@ public void handlePost(RoutingContext event) {
                         res = true;
                         break;
                     case "topicMessages":
-                        var msgRequest = event.body().asPojo(KafkaMessagesRequest.class);
+                        var msgRequest = webUtils.fromJson(event.body().buffer(), KafkaMessagesRequest.class);
                         message = webUtils.toJson(webUtils.getMessages(msgRequest));
                         res = true;
                         break;
                     case "getOffset":
-                        var request = event.body().asPojo(KafkaOffsetRequest.class);
+                        var request = webUtils.fromJson(event.body().buffer(), KafkaOffsetRequest.class);
                         message = webUtils.toJson(webUtils.getOffset(request));
                         res = true;
                         break;
                     case "createMessage":
-                        var rq = event.body().asPojo(KafkaMessageCreateRequest.class);
+                        var rq = webUtils.fromJson(event.body().buffer(), KafkaMessageCreateRequest.class);
                         webUtils.createMessage(rq);
                         message = "{}";
                         res = true;

diff --git a/...s/kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/ui/KafkaUiUtils.java b/...s/kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/ui/KafkaUiUtils.java
@@ -2,6 +2,8 @@
 
 import static io.quarkus.kafka.client.runtime.ui.util.ConsumerFactory.createConsumer;
 
+import java.io.IOException;
+import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -21,19 +23,23 @@
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
 
+import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
 import com.fasterxml.jackson.databind.json.JsonMapper;
 
+import io.netty.buffer.ByteBufInputStream;
 import io.quarkus.kafka.client.runtime.KafkaAdminClient;
 import io.quarkus.kafka.client.runtime.ui.model.Order;
 import io.quarkus.kafka.client.runtime.ui.model.request.KafkaMessageCreateRequest;
 import io.quarkus.kafka.client.runtime.ui.model.request.KafkaMessagesRequest;
 import io.quarkus.kafka.client.runtime.ui.model.request.KafkaOffsetRequest;
 import io.quarkus.kafka.client.runtime.ui.model.response.*;
 import io.smallrye.common.annotation.Identifier;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.json.JsonObject;
 
 @Singleton
 public class KafkaUiUtils {
@@ -229,9 +235,22 @@ public String toJson(Object o) {
         try {
             res = objectMapper.writeValueAsString(o);
         } catch (JsonProcessingException ex) {
-            //FIXME:
             res = "";
         }
         return res;
     }
+
+    public JsonObject fromJson(Buffer buffer) {
+        return new JsonObject(fromJson(buffer, Map.class));
+    }
+
+    public <T> T fromJson(Buffer buffer, Class<T> type) {
+        try {
+            JsonParser parser = objectMapper.createParser((InputStream) new ByteBufInputStream(buffer.getByteBuf()));
+            return objectMapper.readValue(parser, type);
+        } catch (IOException e) {
+            return null;
+        }
+    }
+
 }
diff --git a/integration-tests/devmode/pom.xml b/integration-tests/devmode/pom.xml
@@ -34,6 +34,11 @@
             <artifactId>quarkus-grpc-deployment</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-kafka-client-deployment</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-resteasy-reactive-qute-deployment</artifactId>

diff --git a/...tion-tests/devmode/src/test/java/io/quarkus/test/devconsole/DevConsoleKafkaSmokeTest.java b/...tion-tests/devmode/src/test/java/io/quarkus/test/devconsole/DevConsoleKafkaSmokeTest.java
@@ -0,0 +1,29 @@
+package io.quarkus.test.devconsole;
+
+import org.hamcrest.Matchers;
+import org.jboss.shrinkwrap.api.asset.StringAsset;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusDevModeTest;
+import io.restassured.RestAssured;
+
+/**
+ * Note that this test cannot be placed under the relevant {@code -deployment} module because then the DEV UI processor would
+ * not be able to locate the template resources correctly.
+ */
+public class DevConsoleKafkaSmokeTest {
+
+    @RegisterExtension
+    static final QuarkusDevModeTest config = new QuarkusDevModeTest()
+            .withApplicationRoot(
+                    (jar) -> jar.addAsResource(new StringAsset("quarkus.http.root-path=testing"), "application.properties"));
+
+    @Test
+    public void testServices() {
+        RestAssured.get("testing/q/dev/io.quarkus.quarkus-kafka-client/kafka-dev-ui")
+                .then()
+                .statusCode(200).body(Matchers.containsString("Kafka Dev UI"));
+    }
+
+}