diff --git a/bom/pom.xml b/bom/pom.xml
index 38f3cb04d3268..6cada6aa54fc1 100644
--- a/bom/pom.xml
+++ b/bom/pom.xml
@@ -363,6 +363,11 @@
                 <artifactId>quarkus-kafka-client</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>io.quarkus</groupId>
+                <artifactId>quarkus-kafka-streams</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>io.quarkus</groupId>
                 <artifactId>quarkus-smallrye-health</artifactId>
@@ -1085,6 +1090,11 @@
                 <artifactId>kafka-clients</artifactId>
                 <version>${kafka-clients.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.apache.kafka</groupId>
+                <artifactId>kafka-streams</artifactId>
+                <version>${kafka-clients.version}</version>
+            </dependency>
             <dependency>
                 <groupId>org.apache.kafka</groupId>
                 <artifactId>kafka_2.12</artifactId>
diff --git a/core/deployment/src/main/java/io/quarkus/deployment/builditem/FeatureBuildItem.java b/core/deployment/src/main/java/io/quarkus/deployment/builditem/FeatureBuildItem.java
index 43e9297bbc72f..597cbfefbb229 100644
--- a/core/deployment/src/main/java/io/quarkus/deployment/builditem/FeatureBuildItem.java
+++ b/core/deployment/src/main/java/io/quarkus/deployment/builditem/FeatureBuildItem.java
@@ -32,6 +32,7 @@ public final class FeatureBuildItem extends MultiBuildItem {
     public static final String JDBC_MARIADB = "jdbc-mariadb";
     public static final String JDBC_POSTGRESQL = "jdbc-postgresql";
     public static final String JDBC_MSSQL = "jdbc-mssql";
+    public static final String KAFKA_STREAMS = "kafka-streams";
     public static final String KEYCLOAK = "keycloak";
     public static final String KOTLIN = "kotlin";
     public static final String MAILER = "mailer";
diff --git a/devtools/common/src/main/filtered/extensions.json b/devtools/common/src/main/filtered/extensions.json
index c39f310335eed..6070f30f0418d 100644
--- a/devtools/common/src/main/filtered/extensions.json
+++ b/devtools/common/src/main/filtered/extensions.json
@@ -497,5 +497,22 @@
     ],
     "groupId": "io.quarkus",
     "artifactId": "quarkus-reactive-pg-client"
+  },
+  {
+    "name": "Apache Kafka Client",
+    "labels": [
+      "kafka"
+    ],
+    "groupId": "io.quarkus",
+    "artifactId": "quarkus-kafka-client"
+  },
+  {
+    "name": "Apache Kafka Streams",
+    "labels": [
+      "kafka",
+      "kafka-streams"
+    ],
+    "groupId": "io.quarkus",
+    "artifactId": "quarkus-kafka-streams"
   }
 ]
diff --git a/extensions/kafka-streams/deployment/pom.xml b/extensions/kafka-streams/deployment/pom.xml
new file mode 100644
index 0000000000000..e13d762d978eb
--- /dev/null
+++ b/extensions/kafka-streams/deployment/pom.xml
@@ -0,0 +1,72 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2019 Red Hat, Inc.
+  ~
+  ~ Licensed under the Apache License, Version 2.0 (the "License");
+  ~ you may not use this file except in compliance with the License.
+  ~ You may obtain a copy of the License at
+  ~
+  ~     http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>quarkus-kafka-streams-parent</artifactId>
+        <groupId>io.quarkus</groupId>
+        <version>999-SNAPSHOT</version>
+        <relativePath>../</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>quarkus-kafka-streams-deployment</artifactId>
+    <name>Quarkus - Kafka Streams - Deployment</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-core-deployment</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-kafka-streams</artifactId>
+        </dependency>
+
+        <!-- test dependencies -->
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-junit5-internal</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.rest-assured</groupId>
+            <artifactId>rest-assured</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <annotationProcessorPaths>
+                        <path>
+                            <groupId>io.quarkus</groupId>
+                            <artifactId>quarkus-extension-processor</artifactId>
+                            <version>${project.version}</version>
+                        </path>
+                    </annotationProcessorPaths>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>
diff --git a/extensions/kafka-streams/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java b/extensions/kafka-streams/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java
new file mode 100644
index 0000000000000..0bde6d29b1630
--- /dev/null
+++ b/extensions/kafka-streams/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java
@@ -0,0 +1,72 @@
+/*
+ * Copyright 2019 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.quarkus.kafka.streams.deployment;
+
+import java.io.IOException;
+
+import org.apache.kafka.common.serialization.Serdes.ByteArraySerde;
+import org.apache.kafka.streams.errors.DefaultProductionExceptionHandler;
+import org.apache.kafka.streams.errors.LogAndFailExceptionHandler;
+import org.apache.kafka.streams.processor.DefaultPartitionGrouper;
+import org.apache.kafka.streams.processor.FailOnInvalidTimestamp;
+import org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor;
+import org.rocksdb.util.Environment;
+
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.ExecutionTime;
+import io.quarkus.deployment.annotations.Record;
+import io.quarkus.deployment.builditem.FeatureBuildItem;
+import io.quarkus.deployment.builditem.substrate.ReflectiveClassBuildItem;
+import io.quarkus.deployment.builditem.substrate.RuntimeReinitializedClassBuildItem;
+import io.quarkus.deployment.builditem.substrate.SubstrateResourceBuildItem;
+import io.quarkus.deployment.recording.RecorderContext;
+import io.quarkus.kafka.streams.runtime.KafkaStreamsTemplate;
+
+class KafkaStreamsProcessor {
+
+    @BuildStep
+    @Record(ExecutionTime.STATIC_INIT)
+    void build(RecorderContext recorder,
+            BuildProducer<FeatureBuildItem> feature,
+            BuildProducer<ReflectiveClassBuildItem> reflectiveClasses,
+            BuildProducer<RuntimeReinitializedClassBuildItem> reinitialized,
+            BuildProducer<SubstrateResourceBuildItem> nativeLibs) throws IOException {
+
+        feature.produce(new FeatureBuildItem(FeatureBuildItem.KAFKA_STREAMS));
+
+        reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, StreamsPartitionAssignor.class));
+        reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, DefaultPartitionGrouper.class));
+        reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, DefaultProductionExceptionHandler.class));
+        reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, LogAndFailExceptionHandler.class));
+        reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, ByteArraySerde.class));
+        reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, FailOnInvalidTimestamp.class));
+
+        nativeLibs.produce(new SubstrateResourceBuildItem(Environment.getJniLibraryFileName("rocksdb")));
+
+        // re-initializing RocksDB to enable load of native libs
+        reinitialized.produce(new RuntimeReinitializedClassBuildItem("org.rocksdb.RocksDB"));
+    }
+
+    @BuildStep
+    @Record(ExecutionTime.RUNTIME_INIT)
+    void build(KafkaStreamsTemplate template) {
+        // Explicitly loading RocksDB native libs, as that's normally done from within
+        // static initializers which already ran during build
+        template.loadRocksDb();
+    }
+}
diff --git a/extensions/kafka-streams/pom.xml b/extensions/kafka-streams/pom.xml
new file mode 100644
index 0000000000000..cb6bbfa5ab669
--- /dev/null
+++ b/extensions/kafka-streams/pom.xml
@@ -0,0 +1,37 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2019 Red Hat, Inc.
+  ~
+  ~ Licensed under the Apache License, Version 2.0 (the "License");
+  ~ you may not use this file except in compliance with the License.
+  ~ You may obtain a copy of the License at
+  ~
+  ~     http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>quarkus-build-parent</artifactId>
+        <groupId>io.quarkus</groupId>
+        <version>999-SNAPSHOT</version>
+        <relativePath>../../build-parent/pom.xml</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>quarkus-kafka-streams-parent</artifactId>
+    <name>Quarkus - Kafka Streams</name>
+
+    <packaging>pom</packaging>
+    <modules>
+        <module>deployment</module>
+        <module>runtime</module>
+    </modules>
+</project>
diff --git a/extensions/kafka-streams/runtime/pom.xml b/extensions/kafka-streams/runtime/pom.xml
new file mode 100644
index 0000000000000..5a576783aec98
--- /dev/null
+++ b/extensions/kafka-streams/runtime/pom.xml
@@ -0,0 +1,71 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2019 Red Hat, Inc.
+  ~
+  ~ Licensed under the Apache License, Version 2.0 (the "License");
+  ~ you may not use this file except in compliance with the License.
+  ~ You may obtain a copy of the License at
+  ~
+  ~     http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>quarkus-kafka-streams-parent</artifactId>
+        <groupId>io.quarkus</groupId>
+        <version>999-SNAPSHOT</version>
+        <relativePath>../</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>quarkus-kafka-streams</artifactId>
+    <name>Quarkus - Kafka Streams - Runtime</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-kafka-client</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-streams</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.oracle.substratevm</groupId>
+            <artifactId>svm</artifactId>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>io.quarkus</groupId>
+                <artifactId>quarkus-bootstrap-maven-plugin</artifactId>
+            </plugin>
+            <plugin>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <annotationProcessorPaths>
+                        <path>
+                            <groupId>io.quarkus</groupId>
+                            <artifactId>quarkus-extension-processor</artifactId>
+                            <version>${project.version}</version>
+                        </path>
+                    </annotationProcessorPaths>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+</project>
diff --git a/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsTemplate.java b/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsTemplate.java
new file mode 100644
index 0000000000000..06332490cb4f4
--- /dev/null
+++ b/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsTemplate.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright 2019 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.quarkus.kafka.streams.runtime;
+
+import org.rocksdb.RocksDB;
+
+import io.quarkus.runtime.annotations.Template;
+
+@Template
+public class KafkaStreamsTemplate {
+
+    public void loadRocksDb() {
+        RocksDB.loadLibrary();
+    }
+}
diff --git a/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/graal/KafkaStreamsSubstitutions.java b/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/graal/KafkaStreamsSubstitutions.java
new file mode 100644
index 0000000000000..007357b8f6cab
--- /dev/null
+++ b/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/graal/KafkaStreamsSubstitutions.java
@@ -0,0 +1,36 @@
+/*
+ * Copyright 2019 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.quarkus.kafka.streams.runtime.graal;
+
+import com.oracle.svm.core.annotate.Alias;
+import com.oracle.svm.core.annotate.RecomputeFieldValue;
+import com.oracle.svm.core.annotate.TargetClass;
+
+/**
+ * Resets the {@code initialized} field, so that the native libs are loaded again at
+ * image runtime, after they have been loaded once at build time via calls from static
+ * initializers.
+ */
+@TargetClass(className = "org.rocksdb.NativeLibraryLoader")
+final class Target_org_rocksdb_NativeLibraryLoader {
+
+    @Alias
+    @RecomputeFieldValue(kind = RecomputeFieldValue.Kind.Reset)
+    private static boolean initialized = false;
+}
+
+public final class KafkaStreamsSubstitutions {
+}
diff --git a/extensions/pom.xml b/extensions/pom.xml
index 28de77bceba89..8b42fa37d2ec7 100644
--- a/extensions/pom.xml
+++ b/extensions/pom.xml
@@ -81,6 +81,7 @@
         <module>hibernate-search-elasticsearch</module>
         <module>elasticsearch-rest-client</module>
         <module>kafka-client</module>
+        <module>kafka-streams</module>
 
         <!-- Spring -->
         <module>spring-di</module>
diff --git a/integration-tests/kafka/pom.xml b/integration-tests/kafka/pom.xml
index 319bc05d480f2..6eb67162f84d5 100644
--- a/integration-tests/kafka/pom.xml
+++ b/integration-tests/kafka/pom.xml
@@ -43,18 +43,30 @@
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-integration-test-shared-library</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+        </dependency>
 
         <!-- JAX-RS -->
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-resteasy</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-resteasy-jsonb</artifactId>
+        </dependency>
 
         <!-- Kafka -->
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-kafka-client</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-kafka-streams</artifactId>
+        </dependency>
 
         <!-- test dependencies -->
         <dependency>
@@ -140,6 +152,7 @@
                                 <configuration>
                                     <cleanupServer>true</cleanupServer>
                                     <enableHttpUrlHandler>true</enableHttpUrlHandler>
+                                    <enableJni>true</enableJni>
                                     <!-- Requires Quarkus Graal fork to work, will fail otherwise
                                       <enableRetainedHeapReporting>true</enableRetainedHeapReporting>
                                       <enableCodeSizeReporting>true</enableCodeSizeReporting>
diff --git a/integration-tests/kafka/src/main/java/io/quarkus/it/kafka/JsonObjectSerde.java b/integration-tests/kafka/src/main/java/io/quarkus/it/kafka/JsonObjectSerde.java
new file mode 100644
index 0000000000000..d5cb3fe43763a
--- /dev/null
+++ b/integration-tests/kafka/src/main/java/io/quarkus/it/kafka/JsonObjectSerde.java
@@ -0,0 +1,84 @@
+package io.quarkus.it.kafka;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Map;
+
+import javax.json.Json;
+import javax.json.JsonObject;
+import javax.json.JsonReader;
+
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serializer;
+
+/**
+ * A {@link Serde} that (de-)serializes JSON.
+ */
+public class JsonObjectSerde implements Serde<JsonObject> {
+
+    public JsonObjectSerde() {
+    }
+
+    @Override
+    public void configure(Map<String, ?> configs, boolean isKey) {
+    }
+
+    @Override
+    public void close() {
+    }
+
+    @Override
+    public Serializer<JsonObject> serializer() {
+        return new JsonSerializer();
+    }
+
+    @Override
+    public Deserializer<JsonObject> deserializer() {
+        return new JsonDeserializer();
+    }
+
+    private final class JsonDeserializer implements Deserializer<JsonObject> {
+
+        @Override
+        public void configure(Map<String, ?> configs, boolean isKey) {
+        }
+
+        @Override
+        public JsonObject deserialize(String topic, byte[] data) {
+            if (data == null) {
+                return null;
+            }
+
+            try (JsonReader reader = Json.createReader(new ByteArrayInputStream(data))) {
+                return reader.readObject();
+            }
+        }
+
+        @Override
+        public void close() {
+        }
+    }
+
+    private final class JsonSerializer implements Serializer<JsonObject> {
+
+        @Override
+        public void configure(Map<String, ?> configs, boolean isKey) {
+        }
+
+        @Override
+        public byte[] serialize(String topic, JsonObject data) {
+            try (ByteArrayOutputStream output = new ByteArrayOutputStream()) {
+                Json.createWriter(output).writeObject(data);
+                return output.toByteArray();
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+        }
+
+        @Override
+        public void close() {
+        }
+    }
+}
diff --git a/integration-tests/kafka/src/main/java/io/quarkus/it/kafka/KafkaStreamsPipeline.java b/integration-tests/kafka/src/main/java/io/quarkus/it/kafka/KafkaStreamsPipeline.java
new file mode 100644
index 0000000000000..d7cebe8a7312f
--- /dev/null
+++ b/integration-tests/kafka/src/main/java/io/quarkus/it/kafka/KafkaStreamsPipeline.java
@@ -0,0 +1,150 @@
+package io.quarkus.it.kafka;
+
+import static org.awaitility.Awaitility.await;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import javax.enterprise.context.ApplicationScoped;
+import javax.enterprise.event.Observes;
+import javax.json.Json;
+import javax.json.JsonObject;
+import javax.json.JsonObjectBuilder;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.ListTopicsResult;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.InvalidStateStoreException;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.streams.kstream.Joined;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KTable;
+import org.apache.kafka.streams.kstream.Materialized;
+import org.apache.kafka.streams.kstream.Produced;
+import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
+import org.apache.kafka.streams.state.QueryableStoreTypes;
+import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
+import org.apache.kafka.streams.state.Stores;
+
+import io.quarkus.runtime.StartupEvent;
+
+@ApplicationScoped
+@Path("/kafkastreams")
+public class KafkaStreamsPipeline {
+
+    private static final String CATEGORIES_TOPIC_NAME = "streams-test-categories";
+    private static final String CUSTOMERS_TOPIC_NAME = "streams-test-customers";
+
+    private KafkaStreams streams;
+
+    private ExecutorService executor;
+
+    void onStart(@Observes StartupEvent ev) {
+        Properties props = new Properties();
+        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-test-pipeline");
+        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:19092");
+        props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 10 * 1024);
+        props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
+        props.put(CommonClientConfigs.METADATA_MAX_AGE_CONFIG, 500);
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        StreamsBuilder builder = new StreamsBuilder();
+
+        JsonObjectSerde jsonNodeSerde = new JsonObjectSerde();
+        KTable<Integer, JsonObject> categories = builder.table(
+                CATEGORIES_TOPIC_NAME,
+                Consumed.with(Serdes.Integer(), jsonNodeSerde));
+
+        KStream<Integer, JsonObject> customers = builder
+                .stream(CUSTOMERS_TOPIC_NAME, Consumed.with(Serdes.Integer(), jsonNodeSerde))
+                .selectKey((k, v) -> v.getJsonNumber("category").intValue())
+                .join(
+                        categories,
+                        (v1, v2) -> {
+                            JsonObjectBuilder target = Json.createObjectBuilder();
+                            v1.forEach(target::add);
+                            target.add("category", v2);
+                            return target.build();
+                        },
+                        Joined.with(Serdes.Integer(), jsonNodeSerde, null));
+
+        KeyValueBytesStoreSupplier storeSupplier = Stores.inMemoryKeyValueStore("countstore");
+        customers.groupByKey()
+                .count(Materialized.<Integer, Long> as(storeSupplier));
+
+        customers.selectKey((k, v) -> v.getJsonNumber("id").intValue())
+                .to("streams-test-customers-processed", Produced.with(Serdes.Integer(), jsonNodeSerde));
+
+        streams = new KafkaStreams(builder.build(), props);
+
+        executor = Executors.newSingleThreadExecutor();
+        executor.execute(() -> {
+            waitForTopicsToBeCreated("localhost:19092");
+            streams.start();
+        });
+    }
+
+    private ReadOnlyKeyValueStore<Integer, Long> getCountstore() {
+        while (true) {
+            try {
+                return streams.store("countstore", QueryableStoreTypes.keyValueStore());
+            } catch (InvalidStateStoreException e) {
+                // ignore, store not ready yet
+            }
+        }
+    }
+
+    @POST
+    @Path("/stop")
+    public void stop() {
+        streams.close();
+    }
+
+    @GET
+    @Path("/category/{id}")
+    @Consumes(MediaType.APPLICATION_JSON)
+    @Produces(MediaType.APPLICATION_JSON)
+    public Long getCategory(@PathParam("id") int id) {
+        return getCountstore().get(id);
+    }
+
+    private void waitForTopicsToBeCreated(String bootstrapServers) {
+        Map<String, Object> config = new HashMap<>();
+        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+
+        try (AdminClient adminClient = AdminClient.create(config)) {
+            await().until(topicsCreated(adminClient, CATEGORIES_TOPIC_NAME, CUSTOMERS_TOPIC_NAME));
+        }
+    }
+
+    private Callable<Boolean> topicsCreated(AdminClient adminClient, String... expectedTopics) {
+        return new Callable<Boolean>() {
+
+            @Override
+            public Boolean call() throws Exception {
+                ListTopicsResult topics = adminClient.listTopics();
+                Set<String> topicNames = topics.names().get(10, TimeUnit.SECONDS);
+
+                return topicNames.containsAll(Arrays.asList(expectedTopics));
+            }
+        };
+    }
+}
diff --git a/integration-tests/kafka/src/test/java/io/quarkus/it/main/KafkaConsumerITCase.java b/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaConsumerITCase.java
similarity index 81%
rename from integration-tests/kafka/src/test/java/io/quarkus/it/main/KafkaConsumerITCase.java
rename to integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaConsumerITCase.java
index 42134366cd911..f8eb48897dbcb 100644
--- a/integration-tests/kafka/src/test/java/io/quarkus/it/main/KafkaConsumerITCase.java
+++ b/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaConsumerITCase.java
@@ -1,4 +1,4 @@
-package io.quarkus.it.main;
+package io.quarkus.it.kafka;
 
 import io.quarkus.test.junit.SubstrateTest;
 
diff --git a/integration-tests/kafka/src/test/java/io/quarkus/it/main/KafkaConsumerTest.java b/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaConsumerTest.java
similarity index 98%
rename from integration-tests/kafka/src/test/java/io/quarkus/it/main/KafkaConsumerTest.java
rename to integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaConsumerTest.java
index 6c77f2bad54c9..f4f4aacaac0f5 100644
--- a/integration-tests/kafka/src/test/java/io/quarkus/it/main/KafkaConsumerTest.java
+++ b/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaConsumerTest.java
@@ -1,4 +1,4 @@
-package io.quarkus.it.main;
+package io.quarkus.it.kafka;
 
 import java.util.Properties;
 
diff --git a/integration-tests/kafka/src/test/java/io/quarkus/it/main/KafkaProducerITCase.java b/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaProducerITCase.java
similarity index 81%
rename from integration-tests/kafka/src/test/java/io/quarkus/it/main/KafkaProducerITCase.java
rename to integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaProducerITCase.java
index 7bf5b3c6d7734..4075439e8cfc1 100644
--- a/integration-tests/kafka/src/test/java/io/quarkus/it/main/KafkaProducerITCase.java
+++ b/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaProducerITCase.java
@@ -1,4 +1,4 @@
-package io.quarkus.it.main;
+package io.quarkus.it.kafka;
 
 import io.quarkus.test.junit.SubstrateTest;
 
diff --git a/integration-tests/kafka/src/test/java/io/quarkus/it/main/KafkaProducerTest.java b/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaProducerTest.java
similarity index 98%
rename from integration-tests/kafka/src/test/java/io/quarkus/it/main/KafkaProducerTest.java
rename to integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaProducerTest.java
index 1b14c552b95b0..af3add9461855 100644
--- a/integration-tests/kafka/src/test/java/io/quarkus/it/main/KafkaProducerTest.java
+++ b/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaProducerTest.java
@@ -1,4 +1,4 @@
-package io.quarkus.it.main;
+package io.quarkus.it.kafka;
 
 import java.time.Duration;
 import java.util.Collections;
diff --git a/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaStreamsITCase.java b/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaStreamsITCase.java
new file mode 100644
index 0000000000000..2410b099dc1bd
--- /dev/null
+++ b/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaStreamsITCase.java
@@ -0,0 +1,8 @@
+package io.quarkus.it.kafka;
+
+import io.quarkus.test.junit.SubstrateTest;
+
+@SubstrateTest
+public class KafkaStreamsITCase extends KafkaStreamsTest {
+
+}
diff --git a/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaStreamsTest.java b/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaStreamsTest.java
new file mode 100644
index 0000000000000..17e0a3afe8230
--- /dev/null
+++ b/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaStreamsTest.java
@@ -0,0 +1,166 @@
+package io.quarkus.it.kafka;
+
+import java.io.StringReader;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+
+import javax.json.Json;
+import javax.json.JsonObject;
+import javax.json.JsonReader;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import io.quarkus.test.common.QuarkusTestResource;
+import io.quarkus.test.junit.QuarkusTest;
+import io.restassured.RestAssured;
+
+@QuarkusTestResource(KafkaTestResource.class)
+@QuarkusTest
+public class KafkaStreamsTest {
+
+    private static Producer<Integer, String> createProducer() {
+        Properties props = new Properties();
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:19092");
+        props.put(ProducerConfig.CLIENT_ID_CONFIG, "streams-test-producer");
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+
+        return new KafkaProducer<Integer, String>(props);
+    }
+
+    private static KafkaConsumer<Integer, String> createConsumer() {
+        Properties props = new Properties();
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:19092");
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, "streams-test-consumer");
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+        KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(props);
+        consumer.subscribe(Collections.singletonList("streams-test-customers-processed"));
+        return consumer;
+    }
+
+    @Test
+    public void testKafkaStreams() throws Exception {
+        produceCustomers();
+
+        Consumer<Integer, String> consumer = createConsumer();
+        List<ConsumerRecord<Integer, String>> records = poll(consumer, 4);
+
+        ConsumerRecord<Integer, String> record = records.get(0);
+        Assertions.assertEquals(101, record.key());
+        JsonObject customer = parse(record.value());
+        Assertions.assertEquals(101, customer.getInt("id"));
+        Assertions.assertEquals("Bob", customer.getString("name"));
+        Assertions.assertEquals("B2B", customer.getJsonObject("category").getString("name"));
+        Assertions.assertEquals("business-to-business", customer.getJsonObject("category").getString("value"));
+
+        record = records.get(1);
+        Assertions.assertEquals(102, record.key());
+        customer = parse(record.value());
+        Assertions.assertEquals(102, customer.getInt("id"));
+        Assertions.assertEquals("Becky", customer.getString("name"));
+        Assertions.assertEquals("B2C", customer.getJsonObject("category").getString("name"));
+        Assertions.assertEquals("business-to-customer", customer.getJsonObject("category").getString("value"));
+
+        record = records.get(2);
+        Assertions.assertEquals(103, record.key());
+        customer = parse(record.value());
+        Assertions.assertEquals(103, customer.getInt("id"));
+        Assertions.assertEquals("Bruce", customer.getString("name"));
+        Assertions.assertEquals("B2B", customer.getJsonObject("category").getString("name"));
+        Assertions.assertEquals("business-to-business", customer.getJsonObject("category").getString("value"));
+
+        record = records.get(3);
+        Assertions.assertEquals(104, record.key());
+        customer = parse(record.value());
+        Assertions.assertEquals(104, customer.getInt("id"));
+        Assertions.assertEquals("Bert", customer.getString("name"));
+        Assertions.assertEquals("B2B", customer.getJsonObject("category").getString("name"));
+        Assertions.assertEquals("business-to-business", customer.getJsonObject("category").getString("value"));
+
+        assertCategoryCount(1, 3);
+        assertCategoryCount(2, 1);
+
+        // explicitly stopping the pipeline *before* the broker is shut down, as it
+        // otherwise will time out
+        RestAssured.post("/kafkastreams/stop");
+    }
+
+    private void produceCustomers() {
+        Producer<Integer, String> producer = createProducer();
+
+        producer.send(new ProducerRecord<>("streams-test-categories", 1,
+                "{ \"name\" : \"B2B\", \"value\" : \"business-to-business\" }"));
+        producer.send(new ProducerRecord<>("streams-test-categories", 2,
+                "{ \"name\" : \"B2C\", \"value\" : \"business-to-customer\" }"));
+
+        producer.send(
+                new ProducerRecord<>("streams-test-customers", 101, "{ \"id\" : 101, \"name\" : \"Bob\", \"category\" : 1 }"));
+        producer.send(new ProducerRecord<>("streams-test-customers", 102,
+                "{ \"id\" : 102, \"name\" : \"Becky\", \"category\" : 2 }"));
+        producer.send(new ProducerRecord<>("streams-test-customers", 103,
+                "{ \"id\" : 103, \"name\" : \"Bruce\", \"category\" : 1 }"));
+        producer.send(new ProducerRecord<>("streams-test-customers", 104,
+                "{ \"id\" : 104, \"name\" : \"Bert\", \"category\" : 1 }"));
+    }
+
+    private void assertCategoryCount(int categoryId, int expectedCount) throws Exception {
+        int i = 0;
+        Integer actual = null;
+
+        // retrying for some time as the aggregation might not have finished yet
+        while (i < 50 && !Integer.valueOf(expectedCount).equals(actual)) {
+            actual = getCategoryCount(categoryId);
+            Thread.sleep(100);
+        }
+
+        Assertions.assertEquals(expectedCount, actual);
+    }
+
+    private Integer getCategoryCount(int categoryId) {
+        String result = RestAssured.when().get("/kafkastreams/category/" + categoryId).asString();
+        if (result != null && !result.trim().isEmpty()) {
+            return Integer.valueOf(result);
+        }
+
+        return null;
+    }
+
+    private List<ConsumerRecord<Integer, String>> poll(Consumer<Integer, String> consumer, int expectedRecordCount) {
+        int fetched = 0;
+        List<ConsumerRecord<Integer, String>> result = new ArrayList<>();
+        while (fetched < expectedRecordCount) {
+            ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofMillis(20000));
+            records.forEach(result::add);
+            fetched = result.size();
+        }
+
+        return result;
+    }
+
+    private JsonObject parse(String json) {
+        try (JsonReader reader = Json.createReader(new StringReader(json))) {
+            return reader.readObject();
+        }
+    }
+}
diff --git a/integration-tests/kafka/src/test/java/io/quarkus/it/main/KafkaTestResource.java b/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaTestResource.java
similarity index 97%
rename from integration-tests/kafka/src/test/java/io/quarkus/it/main/KafkaTestResource.java
rename to integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaTestResource.java
index 5fa03a9b8822a..4f87b8bebbfd3 100644
--- a/integration-tests/kafka/src/test/java/io/quarkus/it/main/KafkaTestResource.java
+++ b/integration-tests/kafka/src/test/java/io/quarkus/it/kafka/KafkaTestResource.java
@@ -1,4 +1,4 @@
-package io.quarkus.it.main;
+package io.quarkus.it.kafka;
 
 import java.io.File;
 import java.util.Collections;