implement basic kafka dev ui

bom/application/pom.xml

@@ -1265,6 +1265,11 @@
                 <artifactId>quarkus-kafka-client-deployment</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>io.quarkus</groupId>
+                <artifactId>quarkus-kafka-client-ui</artifactId>
+                <version>${project.version}</version>
+            </dependency>            
             <dependency>
                 <groupId>io.quarkus</groupId>
                 <artifactId>quarkus-kafka-streams</artifactId>

extensions/kafka-client/deployment/pom.xml

@@ -44,6 +44,14 @@
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-caffeine-deployment</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-vertx-http-deployment</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-kafka-client-ui</artifactId>
+        </dependency>                
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>testcontainers</artifactId>

extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaBuildTimeConfig.java

@@ -1,5 +1,6 @@
 package io.quarkus.kafka.client.deployment;
 
+import io.quarkus.runtime.annotations.ConfigDocSection;
 import io.quarkus.runtime.annotations.ConfigItem;
 import io.quarkus.runtime.annotations.ConfigPhase;
 import io.quarkus.runtime.annotations.ConfigRoot;
@@ -28,4 +29,11 @@ public class KafkaBuildTimeConfig {
      */
     @ConfigItem
     public KafkaDevServicesBuildTimeConfig devservices;
+
+    /**
+     * Kafka UI configuration
+     */
+    @ConfigItem
+    @ConfigDocSection
+    public KafkaBuildTimeUiConfig ui;
 }

extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaBuildTimeUiConfig.java

@@ -0,0 +1,28 @@
+package io.quarkus.kafka.client.deployment;
+
+import io.quarkus.runtime.annotations.ConfigGroup;
+import io.quarkus.runtime.annotations.ConfigItem;
+
+@ConfigGroup
+public class KafkaBuildTimeUiConfig {
+
+    /**
+     * The path where Kafka UI is available.
+     * The value `/` is not allowed as it blocks the application from serving anything else.
+     * By default, this URL will be resolved as a path relative to `${quarkus.http.non-application-root-path}`.
+     */
+    @ConfigItem(defaultValue = "kafka-ui")
+    public String rootPath;
+    /**
+     * Whether or not to enable Kafka Dev UI in non-development native mode.
+     */
+    @ConfigItem(name = "handlerpath", defaultValue = "kafka-admin")
+    public String handlerRootPath;
+    /**
+     * Always include the UI. By default, this will only be included in dev and test.
+     * Setting this to true will also include the UI in Prod
+     */
+    @ConfigItem(defaultValue = "false")
+    public boolean alwaysInclude;
+
+}

extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java

@@ -1,14 +1,21 @@
 package io.quarkus.kafka.client.deployment;
 
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Scanner;
 import java.util.Set;
 import java.util.function.Consumer;
 import java.util.logging.Level;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import javax.security.auth.spi.LoginModule;
 
@@ -70,10 +77,13 @@
 import io.quarkus.deployment.builditem.CombinedIndexBuildItem;
 import io.quarkus.deployment.builditem.ExtensionSslNativeSupportBuildItem;
 import io.quarkus.deployment.builditem.FeatureBuildItem;
+import io.quarkus.deployment.builditem.HotDeploymentWatchedFileBuildItem;
 import io.quarkus.deployment.builditem.IndexDependencyBuildItem;
+import io.quarkus.deployment.builditem.LaunchModeBuildItem;
 import io.quarkus.deployment.builditem.LogCategoryBuildItem;
 import io.quarkus.deployment.builditem.RunTimeConfigurationDefaultBuildItem;
 import io.quarkus.deployment.builditem.RuntimeConfigSetupCompleteBuildItem;
+import io.quarkus.deployment.builditem.ShutdownContextBuildItem;
 import io.quarkus.deployment.builditem.nativeimage.NativeImageProxyDefinitionBuildItem;
 import io.quarkus.deployment.builditem.nativeimage.NativeImageResourceBuildItem;
 import io.quarkus.deployment.builditem.nativeimage.NativeImageSecurityProviderBuildItem;
@@ -82,9 +92,11 @@
 import io.quarkus.deployment.builditem.nativeimage.ServiceProviderBuildItem;
 import io.quarkus.deployment.logging.LogCleanupFilterBuildItem;
 import io.quarkus.deployment.pkg.NativeConfig;
-import io.quarkus.kafka.client.runtime.KafkaBindingConverter;
-import io.quarkus.kafka.client.runtime.KafkaRecorder;
+import io.quarkus.kafka.client.runtime.*;
 import io.quarkus.kafka.client.runtime.KafkaRuntimeConfigProducer;
+import io.quarkus.kafka.client.runtime.ui.KafkaTopicClient;
+import io.quarkus.kafka.client.runtime.ui.KafkaUiRecorder;
+import io.quarkus.kafka.client.runtime.ui.KafkaUiUtils;
 import io.quarkus.kafka.client.serialization.BufferDeserializer;
 import io.quarkus.kafka.client.serialization.BufferSerializer;
 import io.quarkus.kafka.client.serialization.JsonArrayDeserializer;
@@ -95,7 +107,19 @@
 import io.quarkus.kafka.client.serialization.JsonbSerializer;
 import io.quarkus.kafka.client.serialization.ObjectMapperDeserializer;
 import io.quarkus.kafka.client.serialization.ObjectMapperSerializer;
+import io.quarkus.maven.dependency.GACT;
+import io.quarkus.runtime.LaunchMode;
+import io.quarkus.runtime.configuration.ConfigurationException;
 import io.quarkus.smallrye.health.deployment.spi.HealthBuildItem;
+import io.quarkus.vertx.http.deployment.BodyHandlerBuildItem;
+import io.quarkus.vertx.http.deployment.HttpRootPathBuildItem;
+import io.quarkus.vertx.http.deployment.NonApplicationRootPathBuildItem;
+import io.quarkus.vertx.http.deployment.RouteBuildItem;
+import io.quarkus.vertx.http.deployment.webjar.WebJarBuildItem;
+import io.quarkus.vertx.http.deployment.webjar.WebJarResourcesFilter;
+import io.quarkus.vertx.http.deployment.webjar.WebJarResultsBuildItem;
+import io.vertx.core.Handler;
+import io.vertx.ext.web.RoutingContext;
 
 public class KafkaProcessor {
 
@@ -144,6 +168,25 @@ public class KafkaProcessor {
 
     static final DotName PARTITION_ASSIGNER = DotName
             .createSimple("org.apache.kafka.clients.consumer.internals.PartitionAssignor");
+    // For the UI
+    private static final GACT KAFKA_UI_WEBJAR_ARTIFACT_KEY = new GACT("io.quarkus", "quarkus-kafka-client-ui", null, "jar");
+    private static final String KAFKA_UI_WEBJAR_STATIC_RESOURCES_PATH = "META-INF/resources/kafka-ui/";
+    private static final String FILE_TO_UPDATE = "config.js";
+    private static final String LINE_TO_UPDATE = "export const api = '";
+    private static final String LINE_FORMAT = LINE_TO_UPDATE + "%s';";
+    private static final String UI_LINE_TO_UPDATE = "export const ui = '";
+    private static final String UI_LINE_FORMAT = UI_LINE_TO_UPDATE + "%s';";
+    private static final String LOGO_LINE_TO_UPDATE = "export const logo = '";
+    private static final String LOGO_LINE_FORMAT = LOGO_LINE_TO_UPDATE + "%s';";
+    private static final String UI_LOGO_PATH = "logo.png";
+    // UI brandibg
+    private static final String BRANDING_DIR = "META-INF/branding/";
+    private static final String BRANDING_LOGO_GENERAL = BRANDING_DIR + "logo.png";
+    private static final String BRANDING_LOGO_MODULE = BRANDING_DIR + "quarkus-kafka-client-ui.png";
+    private static final String BRANDING_STYLE_GENERAL = BRANDING_DIR + "style.css";
+    private static final String BRANDING_STYLE_MODULE = BRANDING_DIR + "quarkus-kafka-client-ui.css";
+    private static final String BRANDING_FAVICON_GENERAL = BRANDING_DIR + "favicon.ico";
+    private static final String BRANDING_FAVICON_MODULE = BRANDING_DIR + "quarkus-kafka-client-ui.ico";
 
     @BuildStep
     FeatureBuildItem feature() {
@@ -165,7 +208,8 @@ void silenceUnwantedConfigLogs(BuildProducer<LogCleanupFilterBuildItem> logClean
 
         List<String> ignoredMessages = new ArrayList<>();
         for (String ignoredConfigProperty : ignoredConfigProperties) {
-            ignoredMessages.add("The configuration '" + ignoredConfigProperty + "' was supplied but isn't a known config.");
+            ignoredMessages
+                    .add("The configuration '" + ignoredConfigProperty + "' was supplied but isn't a known config.");
         }
 
         logCleanupFilters.produce(new LogCleanupFilterBuildItem("org.apache.kafka.clients.consumer.ConsumerConfig",
@@ -478,4 +522,166 @@ void registerServiceBinding(Capabilities capabilities,
                             KafkaBindingConverter.class.getName()));
         }
     }
+
+    // Kafka UI related stuff
+
+    @BuildStep
+    public AdditionalBeanBuildItem kafkaClientBeans() {
+        return AdditionalBeanBuildItem.builder()
+                .addBeanClass(KafkaAdminClient.class)
+                .addBeanClass(KafkaTopicClient.class)
+                .addBeanClass(KafkaUiUtils.class)
+                .setUnremovable()
+                .build();
+    }
+
+    @BuildStep
+    @Record(ExecutionTime.RUNTIME_INIT)
+    public void registerKafkaUiExecHandler(
+            BuildProducer<RouteBuildItem> routeProducer,
+            KafkaUiRecorder recorder,
+            LaunchModeBuildItem launchMode,
+            HttpRootPathBuildItem httpRootPathBuildItem,
+            NonApplicationRootPathBuildItem nonApplicationRootPathBuildItem,
+            KafkaBuildTimeConfig buildConfig,
+            BodyHandlerBuildItem bodyHandlerBuildItem,
+            ShutdownContextBuildItem shutdownContext) {
+
+        if (shouldIncludeUi(launchMode, buildConfig)) {
+            String handlerPath = nonApplicationRootPathBuildItem.resolvePath(buildConfig.ui.handlerRootPath);
+            Handler<RoutingContext> executionHandler = recorder.kafkaControlHandler();
+            HttpRootPathBuildItem.Builder requestBuilder = httpRootPathBuildItem.routeBuilder()
+                    .routeFunction(handlerPath, recorder.routeFunction(bodyHandlerBuildItem.getHandler()))
+                    .handler(executionHandler)
+                    .routeConfigKey("quarkus.kafka-client-ui.root-path")
+                    .displayOnNotFoundPage("Kafka UI Endpoint");
+
+            routeProducer.produce(requestBuilder.build());
+        }
+    }
+
+    @BuildStep
+    List<HotDeploymentWatchedFileBuildItem> uiBrandingFiles() {
+        return Stream.of(BRANDING_LOGO_GENERAL,
+                BRANDING_STYLE_GENERAL,
+                BRANDING_FAVICON_GENERAL,
+                BRANDING_LOGO_MODULE,
+                BRANDING_STYLE_MODULE,
+                BRANDING_FAVICON_MODULE).map(HotDeploymentWatchedFileBuildItem::new)
+                .collect(Collectors.toList());
+    }
+
+    @BuildStep
+    void getKafkaUiFinalDestination(
+            HttpRootPathBuildItem httpRootPath,
+            NonApplicationRootPathBuildItem nonApplicationRootPathBuildItem,
+            LaunchModeBuildItem launchMode,
+            KafkaBuildTimeConfig buildConfig,
+            BuildProducer<WebJarBuildItem> webJarBuildProducer) {
+
+        if (shouldIncludeUi(launchMode, buildConfig)) {
+
+            if ("/".equals(buildConfig.ui.rootPath)) {
+                throw new ConfigurationException(
+                        "quarkus.kafka-client-ui.root-path was set to \"/\", this is not allowed as it blocks the application from serving anything else.",
+                        Collections.singleton("quarkus.kafka-client-ui.root-path"));
+            }
+
+            String devUiPath = nonApplicationRootPathBuildItem.resolvePath("dev");
+            String kafkaUiPath = nonApplicationRootPathBuildItem.resolvePath(buildConfig.ui.rootPath);
+            String kafkaHandlerPath = nonApplicationRootPathBuildItem.resolvePath(buildConfig.ui.handlerRootPath);
+            webJarBuildProducer.produce(
+                    WebJarBuildItem.builder().artifactKey(KAFKA_UI_WEBJAR_ARTIFACT_KEY)
+                            .root(KAFKA_UI_WEBJAR_STATIC_RESOURCES_PATH)
+                            .filter(new WebJarResourcesFilter() {
+                                @Override
+                                public WebJarResourcesFilter.FilterResult apply(String fileName, InputStream file)
+                                        throws IOException {
+                                    if (fileName.endsWith(FILE_TO_UPDATE)) {
+                                        String content = new String(file.readAllBytes(), StandardCharsets.UTF_8);
+                                        content = updateUrl(content, kafkaHandlerPath,
+                                                LINE_TO_UPDATE,
+                                                LINE_FORMAT);
+                                        content = updateUrl(content, kafkaUiPath,
+                                                UI_LINE_TO_UPDATE,
+                                                UI_LINE_FORMAT);
+                                        content = updateUrl(content,
+                                                getLogoUrl(launchMode, kafkaUiPath + "/" + UI_LOGO_PATH,
+                                                        kafkaUiPath + "/" + UI_LOGO_PATH),
+                                                LOGO_LINE_TO_UPDATE,
+                                                LOGO_LINE_FORMAT);
+
+                                        return new WebJarResourcesFilter.FilterResult(
+                                                new ByteArrayInputStream(content.getBytes(StandardCharsets.UTF_8)), true);
+                                    }
+
+                                    return new WebJarResourcesFilter.FilterResult(file, false);
+                                }
+                            })
+                            .build());
+        }
+    }
+
+    @BuildStep
+    @Record(ExecutionTime.RUNTIME_INIT)
+    void registerKafkaUiHandler(
+            BuildProducer<RouteBuildItem> routeProducer,
+            KafkaUiRecorder recorder,
+            LaunchModeBuildItem launchMode,
+            NonApplicationRootPathBuildItem nonApplicationRootPathBuildItem,
+            KafkaBuildTimeConfig buildConfig,
+            WebJarResultsBuildItem webJarResultsBuildItem,
+            ShutdownContextBuildItem shutdownContext) {
+
+        WebJarResultsBuildItem.WebJarResult result = webJarResultsBuildItem.byArtifactKey(KAFKA_UI_WEBJAR_ARTIFACT_KEY);
+        if (result == null) {
+            return;
+        }
+
+        if (shouldIncludeUi(launchMode, buildConfig)) {
+            String kafkaUiPath = nonApplicationRootPathBuildItem.resolvePath(buildConfig.ui.rootPath);
+            String finalDestination = result.getFinalDestination();
+
+            Handler<RoutingContext> handler = recorder.uiHandler(finalDestination,
+                    kafkaUiPath, result.getWebRootConfigurations(), shutdownContext);
+            routeProducer.produce(nonApplicationRootPathBuildItem.routeBuilder()
+                    .route(buildConfig.ui.rootPath)
+                    .displayOnNotFoundPage("Kafka UI")
+                    .routeConfigKey("quarkus.kafka-client.ui.root-path")
+                    .handler(handler)
+                    .build());
+
+            routeProducer.produce(nonApplicationRootPathBuildItem.routeBuilder()
+                    .route(buildConfig.ui.rootPath + "*")
+                    .handler(handler)
+                    .build());
+
+        }
+    }
+
+    // In dev mode, when you click on the logo, you should go to Dev UI
+    private String getLogoUrl(LaunchModeBuildItem launchMode, String devUIValue, String defaultValue) {
+        if (launchMode.getLaunchMode().equals(LaunchMode.DEVELOPMENT)) {
+            return devUIValue;
+        }
+        return defaultValue;
+    }
+
+    private String updateUrl(String original, String path, String lineStartsWith, String format) {
+        try (Scanner scanner = new Scanner(original)) {
+            while (scanner.hasNextLine()) {
+                String line = scanner.nextLine();
+                if (line.trim().startsWith(lineStartsWith)) {
+                    String newLine = String.format(format, path);
+                    return original.replace(line.trim(), newLine);
+                }
+            }
+        }
+
+        return original;
+    }
+
+    private static boolean shouldIncludeUi(LaunchModeBuildItem launchMode, KafkaBuildTimeConfig config) {
+        return launchMode.getLaunchMode().isDevOrTest() || config.ui.alwaysInclude;
+    }
 }

extensions/kafka-client/deployment/src/main/resources/dev-templates/embedded.html

@@ -0,0 +1,3 @@
+  <a href="{config:http-path('quarkus.kafka-client.ui.root-path')}/" class="badge badge-light">
+  <i class="fa fa-project-diagram fa-fw"></i>
+  Kafka UI</a>

extensions/kafka-client/pom.xml

@@ -15,6 +15,7 @@
    <packaging>pom</packaging>
 
    <modules>
+      <module>ui</module>
       <module>deployment</module>
       <module>runtime</module>
    </modules>

extensions/kafka-client/runtime/pom.xml

@@ -59,6 +59,11 @@
             <scope>provided</scope>
         </dependency>
 
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-vertx-http</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-junit5-internal</artifactId>