diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 96ce8b7..0024b7a 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -8,12 +8,14 @@ on:
     paths:
       - "rabbitmq/**"
       - "servicebus/**"
+      - "kafka/**"
       - ".github/workflows/release.yml"
   pull_request:
     branches: [ "main" ]
     paths:
       - "rabbitmq/**"
       - "servicebus/**"
+      - "kafka/**"
       - ".github/workflows/release.yml"
   workflow_dispatch:
     inputs:
@@ -34,7 +36,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        plugin: [rabbitmq, servicebus]
+        plugin: [rabbitmq, servicebus, kafka]
         platform: [windows-latest, windows-11-arm, ubuntu-22.04, ubuntu-22.04-arm, macos-26-intel, macos-14]
 
     steps:
@@ -76,29 +78,44 @@ jobs:
               ;;
           esac
 
-          if [ "${{ matrix.plugin }}" = "servicebus" ]; then
-            echo "DIRECTORY=servicebus" >> "$GITHUB_ENV"
-            echo "ARTIFACT_PATH=servicebus/build/ServiceBusPlugin" >> "$GITHUB_ENV"
-            echo "CONFIGURE_ARGS=-DCMAKE_TOOLCHAIN_FILE=${VCPKG_INSTALLATION_ROOT}/scripts/buildsystems/vcpkg.cmake" >> "$GITHUB_ENV"
-            if [ "${{ runner.os }}" = "Linux" ]; then
-              echo "INSTALL_PACKAGES=build-essential cmake git ninja-build pkg-config" >> "$GITHUB_ENV"
-            elif [ "${{ runner.os }}" = "macOS" ]; then
-              echo "INSTALL_PACKAGES=cmake ninja pkg-config" >> "$GITHUB_ENV"
-            else
-              echo "INSTALL_PACKAGES=cmake" >> "$GITHUB_ENV"
-            fi
-          else
-            echo "DIRECTORY=rabbitmq" >> "$GITHUB_ENV"
-            echo "ARTIFACT_PATH=rabbitmq/build/RabbitmqPlugin" >> "$GITHUB_ENV"
-            echo "CONFIGURE_ARGS=" >> "$GITHUB_ENV"
-            if [ "${{ runner.os }}" = "Linux" ]; then
-              echo "INSTALL_PACKAGES=build-essential cmake git" >> "$GITHUB_ENV"
-            elif [ "${{ runner.os }}" = "macOS" ]; then
-              echo "INSTALL_PACKAGES=cmake" >> "$GITHUB_ENV"
-            else
-              echo "INSTALL_PACKAGES=cmake" >> "$GITHUB_ENV"
-            fi
-          fi
+          case "${{ matrix.plugin }}" in
+            servicebus)
+              echo "DIRECTORY=servicebus" >> "$GITHUB_ENV"
+              echo "ARTIFACT_PATH=servicebus/build/ServiceBusPlugin" >> "$GITHUB_ENV"
+              echo "CONFIGURE_ARGS=-DCMAKE_TOOLCHAIN_FILE=${VCPKG_INSTALLATION_ROOT}/scripts/buildsystems/vcpkg.cmake" >> "$GITHUB_ENV"
+              if [ "${{ runner.os }}" = "Linux" ]; then
+                echo "INSTALL_PACKAGES=build-essential cmake git ninja-build pkg-config" >> "$GITHUB_ENV"
+              elif [ "${{ runner.os }}" = "macOS" ]; then
+                echo "INSTALL_PACKAGES=cmake ninja pkg-config" >> "$GITHUB_ENV"
+              else
+                echo "INSTALL_PACKAGES=cmake" >> "$GITHUB_ENV"
+              fi
+              ;;
+            kafka)
+              echo "DIRECTORY=kafka" >> "$GITHUB_ENV"
+              echo "ARTIFACT_PATH=kafka/build/KafkaPlugin" >> "$GITHUB_ENV"
+              echo "CONFIGURE_ARGS=" >> "$GITHUB_ENV"
+              if [ "${{ runner.os }}" = "Linux" ]; then
+                echo "INSTALL_PACKAGES=build-essential cmake git libssl-dev libsasl2-dev zlib1g-dev" >> "$GITHUB_ENV"
+              elif [ "${{ runner.os }}" = "macOS" ]; then
+                echo "INSTALL_PACKAGES=cmake openssl" >> "$GITHUB_ENV"
+              else
+                echo "INSTALL_PACKAGES=cmake" >> "$GITHUB_ENV"
+              fi
+              ;;
+            *)
+              echo "DIRECTORY=rabbitmq" >> "$GITHUB_ENV"
+              echo "ARTIFACT_PATH=rabbitmq/build/RabbitmqPlugin" >> "$GITHUB_ENV"
+              echo "CONFIGURE_ARGS=" >> "$GITHUB_ENV"
+              if [ "${{ runner.os }}" = "Linux" ]; then
+                echo "INSTALL_PACKAGES=build-essential cmake git" >> "$GITHUB_ENV"
+              elif [ "${{ runner.os }}" = "macOS" ]; then
+                echo "INSTALL_PACKAGES=cmake" >> "$GITHUB_ENV"
+              else
+                echo "INSTALL_PACKAGES=cmake" >> "$GITHUB_ENV"
+              fi
+              ;;
+          esac
 
           echo "PLUGIN_NAME=${{ matrix.plugin }}" >> "$GITHUB_ENV"
 
diff --git a/.gitignore b/.gitignore
index d6e96c3..a8ffd37 100644
--- a/.gitignore
+++ b/.gitignore
@@ -3,4 +3,7 @@
 .cursor
 
 out/
-build/
\ No newline at end of file
+build/
+bin/
+obj/
+Binaries/
diff --git a/README.md b/README.md
index 80605e2..b903cea 100644
--- a/README.md
+++ b/README.md
@@ -8,6 +8,7 @@ Official open-source Graftcode Gateway plugins for carrying Graft calls over ext
 |--------|---------|
 | [rabbitmq](rabbitmq/) | RabbitMQ (AMQP 0-9-1), request/reply |
 | [servicebus](servicebus/) | Azure Service Bus (AMQP 1.0), request/reply and one-way |
+| [kafka](kafka/) | Apache Kafka, request/reply (correlation-id) |
 | [observability/opentelemetry](observability/opentelemetry/) | OpenTelemetry / Azure Application Insights connector |
 
 Each plugin has its own README with build and configuration steps. For how the Gateway loads a plugin, see the "Plugin server config" section of the [Graftcode Gateway](https://github.com/grft-dev/graftcode-gateway) README.
diff --git a/kafka/.gitignore b/kafka/.gitignore
new file mode 100644
index 0000000..5c9d681
--- /dev/null
+++ b/kafka/.gitignore
@@ -0,0 +1,9 @@
+build/
+out/
+.vs/
+CMakeUserPresets.json
+cmake-build-*/
+vcpkg/
+vcpkg_installed/
+.cache/
+*.user
diff --git a/kafka/BUILD_NOTES.md b/kafka/BUILD_NOTES.md
new file mode 100644
index 0000000..7524eb5
--- /dev/null
+++ b/kafka/BUILD_NOTES.md
@@ -0,0 +1,5 @@
+# Kafka plugin build notes
+
+- Dependencies: CMake `FetchContent` only (nlohmann/json + librdkafka). No vcpkg.
+- Smoke test links factory exports; full RPC needs a live broker (see docker-compose).
+- First configure needs network/git for FetchContent downloads.
diff --git a/kafka/CMakeLists.txt b/kafka/CMakeLists.txt
new file mode 100644
index 0000000..a7f50a4
--- /dev/null
+++ b/kafka/CMakeLists.txt
@@ -0,0 +1,22 @@
+set(CMAKE_MIN 3.22)
+cmake_minimum_required(VERSION ${CMAKE_MIN})
+set(CMAKE_POLICY_VERSION_MINIMUM ${CMAKE_MIN})
+cmake_policy(VERSION ${CMAKE_MIN})
+
+project("KafkaPlugin" VERSION 1.0.0)
+
+set(CMAKE_CXX_STANDARD 20)
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
+set(CMAKE_POSITION_INDEPENDENT_CODE ON)
+
+add_definitions(-DUNICODE)
+enable_testing()
+
+set(CMAKE_POLICY_DEFAULT_CMP0135 NEW)
+
+if(MSVC)
+    add_compile_options(/utf-8)
+endif()
+
+add_subdirectory("KafkaPlugin")
+add_subdirectory("KafkaPluginTest")
diff --git a/kafka/GraftcodePluginsInterfaces/IServer.h b/kafka/GraftcodePluginsInterfaces/IServer.h
new file mode 100644
index 0000000..4574ab8
--- /dev/null
+++ b/kafka/GraftcodePluginsInterfaces/IServer.h
@@ -0,0 +1,16 @@
+#pragma once
+#include <cstddef>
+
+namespace GraftcodeGateway {
+class IServer {
+public:
+    using byte = unsigned char;
+    using WriteResponseFn = void(*)(void* context, const byte* data, std::size_t size);
+    using ProcessMessageFn = bool(*)(const byte* requestData, std::size_t requestSize, WriteResponseFn writeResponse, void* writeContext);
+
+    virtual ~IServer() = default;
+    virtual void configure(const char* jsonConfig, ProcessMessageFn processMessage) = 0;
+    virtual void start() = 0;
+    virtual void stop() = 0;
+};
+}
diff --git a/kafka/GraftcodePluginsInterfaces/ITransport.h b/kafka/GraftcodePluginsInterfaces/ITransport.h
new file mode 100644
index 0000000..3ab68b9
--- /dev/null
+++ b/kafka/GraftcodePluginsInterfaces/ITransport.h
@@ -0,0 +1,12 @@
+#pragma once
+#include "_common.h"
+
+namespace Hypertube::Native::Interfaces {
+class ITransport {
+public:
+    virtual ~ITransport() = default;
+    virtual int Initialize(byte callingRuntimeNumber, byte calledRuntimeNumber, byte calledRuntimeVersion) = 0;
+    virtual int SendCommand(byte* messageByteArray, int32_t messageByteArrayLen) = 0;
+    virtual int ReadResponse(byte* responseByteArray, int32_t responseByteArrayLen) = 0;
+};
+}
diff --git a/kafka/GraftcodePluginsInterfaces/_common.h b/kafka/GraftcodePluginsInterfaces/_common.h
new file mode 100644
index 0000000..f878601
--- /dev/null
+++ b/kafka/GraftcodePluginsInterfaces/_common.h
@@ -0,0 +1,9 @@
+#pragma once
+#ifndef byte
+typedef unsigned char byte;
+#endif
+#ifndef int32_t
+typedef int int32_t;
+#endif
+#include <string>
+#include <iostream>
diff --git a/kafka/KafkaPlugin/CMakeLists.txt b/kafka/KafkaPlugin/CMakeLists.txt
new file mode 100644
index 0000000..58671ea
--- /dev/null
+++ b/kafka/KafkaPlugin/CMakeLists.txt
@@ -0,0 +1,56 @@
+# Graftcode Kafka plugin (request/reply via librdkafka C++ API).
+set(target_name KafkaPlugin)
+
+add_library(${target_name} SHARED
+    TransportKafka.cpp
+    KafkaClient.cpp
+    KafkaServer.cpp
+)
+
+target_include_directories(${target_name} PUBLIC
+    "${CMAKE_SOURCE_DIR}/GraftcodePluginsInterfaces"
+)
+
+include(FetchContent)
+
+FetchContent_Declare(
+    json
+    GIT_REPOSITORY https://github.com/nlohmann/json.git
+    GIT_TAG v3.12.0
+    GIT_SHALLOW TRUE
+)
+FetchContent_MakeAvailable(json)
+
+set(RDKAFKA_BUILD_STATIC ON CACHE BOOL "Build librdkafka as static library" FORCE)
+set(RDKAFKA_BUILD_EXAMPLES OFF CACHE BOOL "Skip librdkafka examples" FORCE)
+set(RDKAFKA_BUILD_TESTS OFF CACHE BOOL "Skip librdkafka tests" FORCE)
+
+FetchContent_Declare(
+    librdkafka
+    GIT_REPOSITORY https://github.com/confluentinc/librdkafka.git
+    GIT_TAG v2.8.0
+    GIT_SHALLOW TRUE
+)
+FetchContent_MakeAvailable(librdkafka)
+
+if(NOT TARGET rdkafka++)
+    message(FATAL_ERROR "librdkafka FetchContent succeeded but rdkafka++ target was not found")
+endif()
+
+target_link_libraries(${target_name} PUBLIC
+    nlohmann_json::nlohmann_json
+    rdkafka++
+)
+
+if(TARGET rdkafka)
+    target_link_libraries(${target_name} PUBLIC rdkafka)
+endif()
+
+if(WIN32 OR CMAKE_SYSTEM_NAME STREQUAL "Windows")
+    target_link_libraries(${target_name} PUBLIC ws2_32)
+endif()
+
+if(UNIX AND NOT APPLE)
+    find_package(Threads REQUIRED)
+    target_link_libraries(${target_name} PUBLIC Threads::Threads)
+endif()
diff --git a/kafka/KafkaPlugin/KafkaClient.cpp b/kafka/KafkaPlugin/KafkaClient.cpp
new file mode 100644
index 0000000..fd7718b
--- /dev/null
+++ b/kafka/KafkaPlugin/KafkaClient.cpp
@@ -0,0 +1,217 @@
+#include "KafkaClient.h"
+#include "KafkaUtil.h"
+
+#include <algorithm>
+#include <chrono>
+#include <stdexcept>
+#include <utility>
+
+#if __has_include(<librdkafka/rdkafkacpp.h>)
+#include <librdkafka/rdkafkacpp.h>
+#elif __has_include(<rdkafkacpp.h>)
+#include <rdkafkacpp.h>
+#else
+#error "librdkafka C++ header not found (expected librdkafka/rdkafkacpp.h or rdkafkacpp.h)"
+#endif
+
+namespace {
+
+void drainProducer(RdKafka::Producer* producer, int timeoutMs) {
+    if (!producer) {
+        return;
+    }
+    producer->flush(timeoutMs);
+    while (producer->outq_len() > 0) {
+        producer->poll(50);
+    }
+}
+
+}  // namespace
+
+KafkaClient::KafkaClient(Config cfg) : cfg_(std::move(cfg)) {
+    instanceId_ = KafkaPluginUtil::makeUuidV4();
+    // Unique group so concurrent clients sharing one reply topic each get a full
+    // copy of replies, then filter by correlation-id (broadcast-per-group model).
+    consumerGroupId_ = cfg_.groupId.empty()
+                           ? ("graft-client-" + instanceId_)
+                           : (cfg_.groupId + "-" + instanceId_);
+}
+
+KafkaClient::~KafkaClient() {
+    std::lock_guard<std::mutex> lock(callMutex_);
+    if (consumer_) {
+        consumer_->close();
+        consumer_.reset();
+    }
+    if (producer_) {
+        drainProducer(producer_.get(), 5000);
+        producer_.reset();
+    }
+}
+
+void KafkaClient::applySecurity(RdKafka::Conf* conf) const {
+    using KafkaPluginUtil::setConfOrThrow;
+    if (!cfg_.securityProtocol.empty()) {
+        setConfOrThrow(conf, "security.protocol", cfg_.securityProtocol);
+    }
+    if (!cfg_.saslMechanism.empty()) {
+        setConfOrThrow(conf, "sasl.mechanism", cfg_.saslMechanism);
+    }
+    if (!cfg_.saslUsername.empty()) {
+        setConfOrThrow(conf, "sasl.username", cfg_.saslUsername);
+    }
+    if (!cfg_.saslPassword.empty()) {
+        setConfOrThrow(conf, "sasl.password", cfg_.saslPassword);
+    }
+    if (!cfg_.sslCaLocation.empty()) {
+        setConfOrThrow(conf, "ssl.ca.location", cfg_.sslCaLocation);
+    }
+}
+
+void KafkaClient::ensureStarted() {
+    if (started_) {
+        return;
+    }
+
+    using KafkaPluginUtil::setConfOrThrow;
+    std::string errstr;
+
+    {
+        std::unique_ptr<RdKafka::Conf> pconf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
+        setConfOrThrow(pconf.get(), "bootstrap.servers", cfg_.brokers);
+        setConfOrThrow(pconf.get(), "client.id", "graft-kafka-client-producer-" + instanceId_);
+        setConfOrThrow(pconf.get(), "message.timeout.ms", std::to_string(cfg_.rpcTimeoutMs));
+        applySecurity(pconf.get());
+
+        RdKafka::Producer* raw = RdKafka::Producer::create(pconf.release(), errstr);
+        if (!raw) {
+            throw std::runtime_error("Failed to create Kafka producer: " + errstr);
+        }
+        producer_.reset(raw);
+    }
+
+    {
+        std::unique_ptr<RdKafka::Conf> cconf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
+        setConfOrThrow(cconf.get(), "bootstrap.servers", cfg_.brokers);
+        setConfOrThrow(cconf.get(), "group.id", consumerGroupId_);
+        setConfOrThrow(cconf.get(), "client.id", "graft-kafka-client-consumer-" + instanceId_);
+        setConfOrThrow(cconf.get(), "enable.auto.commit", "true");
+        setConfOrThrow(cconf.get(), "auto.offset.reset", "latest");
+        setConfOrThrow(cconf.get(), "allow.auto.create.topics", "true");
+        applySecurity(cconf.get());
+
+        RdKafka::KafkaConsumer* raw = RdKafka::KafkaConsumer::create(cconf.release(), errstr);
+        if (!raw) {
+            throw std::runtime_error("Failed to create Kafka consumer: " + errstr);
+        }
+        consumer_.reset(raw);
+
+        const RdKafka::ErrorCode subErr = consumer_->subscribe({cfg_.replyTopic});
+        if (subErr != RdKafka::ERR_NO_ERROR) {
+            throw std::runtime_error("Failed to subscribe to reply topic '" + cfg_.replyTopic +
+                                     "': " + RdKafka::err2str(subErr));
+        }
+
+        // Brief poll so the consumer joins the group / gets assignment before produce.
+        // Without this, a fast reply can be published before the client is assigned.
+        const auto warmDeadline =
+            std::chrono::steady_clock::now() + std::chrono::milliseconds(1500);
+        while (std::chrono::steady_clock::now() < warmDeadline) {
+            std::unique_ptr<RdKafka::Message> msg(consumer_->consume(100));
+            if (!msg) {
+                continue;
+            }
+            if (msg->err() == RdKafka::ERR_NO_ERROR) {
+                // Unexpected early message — leave for the next call() filter.
+                break;
+            }
+            if (msg->err() != RdKafka::ERR__TIMED_OUT &&
+                msg->err() != RdKafka::ERR__PARTITION_EOF) {
+                // Keep warming on transient errors.
+            }
+        }
+    }
+
+    started_ = true;
+}
+
+std::vector<unsigned char> KafkaClient::call(const unsigned char* data, std::size_t len) {
+    if (data == nullptr && len > 0) {
+        throw std::invalid_argument("KafkaClient::call: null data with non-zero length");
+    }
+
+    std::lock_guard<std::mutex> lock(callMutex_);
+    ensureStarted();
+
+    const std::string correlationId = KafkaPluginUtil::makeUuidV4();
+    RdKafka::Headers* headers =
+        KafkaPluginUtil::makeRpcHeaders(correlationId, cfg_.replyTopic);
+
+    const RdKafka::ErrorCode produceErr = producer_->produce(
+        cfg_.requestTopic,
+        RdKafka::Topic::PARTITION_UA,
+        RdKafka::Producer::RK_MSG_COPY,
+        /*payload=*/const_cast<unsigned char*>(data ? data : reinterpret_cast<const unsigned char*>("")),
+        /*len=*/len,
+        /*key=*/nullptr,
+        /*key_len=*/0,
+        /*timestamp=*/0,
+        headers,
+        /*msg_opaque=*/nullptr);
+
+    if (produceErr != RdKafka::ERR_NO_ERROR) {
+        // produce() takes ownership of headers only on success.
+        delete headers;
+        throw std::runtime_error("Kafka produce to '" + cfg_.requestTopic +
+                                 "' failed: " + RdKafka::err2str(produceErr));
+    }
+
+    producer_->poll(0);
+    // Wait until the request is handed off to the broker (best-effort).
+    if (producer_->flush(std::min(cfg_.rpcTimeoutMs, 10000)) != RdKafka::ERR_NO_ERROR) {
+        throw std::runtime_error("Kafka produce flush timed out for correlation-id=" +
+                                 correlationId);
+    }
+
+    const auto deadline =
+        std::chrono::steady_clock::now() + std::chrono::milliseconds(cfg_.rpcTimeoutMs);
+
+    while (std::chrono::steady_clock::now() < deadline) {
+        const auto remaining = std::chrono::duration_cast<std::chrono::milliseconds>(
+                                   deadline - std::chrono::steady_clock::now())
+                                   .count();
+        const int pollMs = static_cast<int>(std::min<std::int64_t>(remaining, 200));
+
+        std::unique_ptr<RdKafka::Message> msg(consumer_->consume(pollMs));
+        if (!msg) {
+            continue;
+        }
+
+        if (msg->err() == RdKafka::ERR__TIMED_OUT ||
+            msg->err() == RdKafka::ERR__PARTITION_EOF) {
+            continue;
+        }
+        if (msg->err() != RdKafka::ERR_NO_ERROR) {
+            throw std::runtime_error("Kafka consume error while waiting for reply: " +
+                                     msg->errstr());
+        }
+
+        const std::string gotId =
+            KafkaPluginUtil::headerValue(msg->headers(), "correlation-id");
+        if (gotId != correlationId) {
+            // Shared reply topic / other clients — skip.
+            continue;
+        }
+
+        const void* payload = msg->payload();
+        const std::size_t payloadLen = static_cast<std::size_t>(msg->len());
+        if (!payload || payloadLen == 0) {
+            return {};
+        }
+        const auto* bytes = static_cast<const unsigned char*>(payload);
+        return std::vector<unsigned char>(bytes, bytes + payloadLen);
+    }
+
+    throw std::runtime_error("Kafka RPC timed out after " + std::to_string(cfg_.rpcTimeoutMs) +
+                             " ms waiting for correlation-id=" + correlationId);
+}
diff --git a/kafka/KafkaPlugin/KafkaClient.h b/kafka/KafkaPlugin/KafkaClient.h
new file mode 100644
index 0000000..8c3cc7f
--- /dev/null
+++ b/kafka/KafkaPlugin/KafkaClient.h
@@ -0,0 +1,52 @@
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <vector>
+
+namespace RdKafka {
+class Conf;
+class Producer;
+class KafkaConsumer;
+}  // namespace RdKafka
+
+// Client-side RPC over Kafka: produce to request topic, consume correlated reply.
+class KafkaClient {
+public:
+    struct Config {
+        std::string brokers = "localhost:9092";
+        std::string requestTopic = "graft.requests";
+        std::string replyTopic = "graft.replies";
+        std::string groupId = "graft-client";
+        int rpcTimeoutMs = 30000;
+        // Optional SASL/SSL pass-through (empty = leave librdkafka defaults).
+        std::string securityProtocol;   // e.g. "SASL_SSL", "SSL", "PLAINTEXT"
+        std::string saslMechanism;      // e.g. "PLAIN", "SCRAM-SHA-512"
+        std::string saslUsername;
+        std::string saslPassword;
+        std::string sslCaLocation;
+    };
+
+    explicit KafkaClient(Config cfg);
+    ~KafkaClient();
+
+    KafkaClient(const KafkaClient&) = delete;
+    KafkaClient& operator=(const KafkaClient&) = delete;
+
+    // Returns response bytes; throws on timeout or produce/consume failure.
+    std::vector<unsigned char> call(const unsigned char* data, std::size_t len);
+
+private:
+    void applySecurity(RdKafka::Conf* conf) const;
+    void ensureStarted();
+
+    Config cfg_;
+    std::string instanceId_;
+    std::string consumerGroupId_;
+    std::unique_ptr<RdKafka::Producer> producer_;
+    std::unique_ptr<RdKafka::KafkaConsumer> consumer_;
+    std::mutex callMutex_;
+    bool started_ = false;
+};
diff --git a/kafka/KafkaPlugin/KafkaServer.cpp b/kafka/KafkaPlugin/KafkaServer.cpp
new file mode 100644
index 0000000..dd22d45
--- /dev/null
+++ b/kafka/KafkaPlugin/KafkaServer.cpp
@@ -0,0 +1,308 @@
+#include "KafkaServer.h"
+#include "KafkaUtil.h"
+
+#include <nlohmann/json.hpp>
+
+#include <chrono>
+#include <cstring>
+#include <iostream>
+#include <stdexcept>
+#include <vector>
+
+#if __has_include(<librdkafka/rdkafkacpp.h>)
+#include <librdkafka/rdkafkacpp.h>
+#elif __has_include(<rdkafkacpp.h>)
+#include <rdkafkacpp.h>
+#else
+#error "librdkafka C++ header not found (expected librdkafka/rdkafkacpp.h or rdkafkacpp.h)"
+#endif
+
+namespace {
+
+void logInfo(const std::string& msg) {
+    std::cout << "[KafkaServer][INFO] " << msg << std::endl;
+}
+
+void logWarn(const std::string& msg) {
+    std::cout << "[KafkaServer][WARN] " << msg << std::endl;
+}
+
+}  // namespace
+
+KafkaServer::~KafkaServer() { stop(); }
+
+void KafkaServer::configure(const char* jsonConfig, ProcessMessageFn processMessage) {
+    {
+        std::lock_guard<std::mutex> lock(processMutex_);
+        process_ = processMessage;
+    }
+    if (!jsonConfig) {
+        return;
+    }
+    auto j = nlohmann::json::parse(jsonConfig, nullptr, false);
+    if (j.is_discarded() || !j.is_object()) {
+        return;
+    }
+    if (j.contains("brokers") && j["brokers"].is_string()) {
+        brokers_ = j["brokers"].get<std::string>();
+    } else if (j.contains("host") && j["host"].is_string()) {
+        brokers_ = j["host"].get<std::string>();  // compatibility with other plugins
+    }
+    if (j.contains("requestTopic") && j["requestTopic"].is_string()) {
+        requestTopic_ = j["requestTopic"].get<std::string>();
+    } else if (j.contains("queue") && j["queue"].is_string()) {
+        requestTopic_ = j["queue"].get<std::string>();
+    }
+    if (j.contains("replyTopic") && j["replyTopic"].is_string()) {
+        replyTopic_ = j["replyTopic"].get<std::string>();
+    } else if (j.contains("replyQueue") && j["replyQueue"].is_string()) {
+        replyTopic_ = j["replyQueue"].get<std::string>();
+    }
+    if (j.contains("groupId") && j["groupId"].is_string()) {
+        groupId_ = j["groupId"].get<std::string>();
+    }
+    if (j.contains("securityProtocol") && j["securityProtocol"].is_string()) {
+        securityProtocol_ = j["securityProtocol"].get<std::string>();
+    }
+    if (j.contains("saslMechanism") && j["saslMechanism"].is_string()) {
+        saslMechanism_ = j["saslMechanism"].get<std::string>();
+    }
+    if (j.contains("saslUsername") && j["saslUsername"].is_string()) {
+        saslUsername_ = j["saslUsername"].get<std::string>();
+    }
+    if (j.contains("saslPassword") && j["saslPassword"].is_string()) {
+        saslPassword_ = j["saslPassword"].get<std::string>();
+    }
+    if (j.contains("sslCaLocation") && j["sslCaLocation"].is_string()) {
+        sslCaLocation_ = j["sslCaLocation"].get<std::string>();
+    }
+}
+
+void KafkaServer::applySecurity(RdKafka::Conf* conf) const {
+    using KafkaPluginUtil::setConfOrThrow;
+    if (!securityProtocol_.empty()) {
+        setConfOrThrow(conf, "security.protocol", securityProtocol_);
+    }
+    if (!saslMechanism_.empty()) {
+        setConfOrThrow(conf, "sasl.mechanism", saslMechanism_);
+    }
+    if (!saslUsername_.empty()) {
+        setConfOrThrow(conf, "sasl.username", saslUsername_);
+    }
+    if (!saslPassword_.empty()) {
+        setConfOrThrow(conf, "sasl.password", saslPassword_);
+    }
+    if (!sslCaLocation_.empty()) {
+        setConfOrThrow(conf, "ssl.ca.location", sslCaLocation_);
+    }
+}
+
+void KafkaServer::start() {
+    if (running_.exchange(true)) {
+        return;
+    }
+    worker_ = std::thread([this] { loop(); });
+}
+
+void KafkaServer::stop() {
+    if (!running_.exchange(false)) {
+        if (worker_.joinable()) {
+            worker_.join();
+        }
+        return;
+    }
+    if (worker_.joinable()) {
+        worker_.join();
+    }
+}
+
+void KafkaServer::publishReply(RdKafka::Producer* producer,
+                               const std::string& replyTopic,
+                               const std::string& correlationId,
+                               const std::vector<unsigned char>& response) {
+    RdKafka::Headers* headers = RdKafka::Headers::create();
+    if (!correlationId.empty()) {
+        headers->add("correlation-id", correlationId);
+    }
+
+    void* payload = nullptr;
+    std::size_t len = 0;
+    if (!response.empty()) {
+        payload = const_cast<unsigned char*>(response.data());
+        len = response.size();
+    }
+
+    const RdKafka::ErrorCode err = producer->produce(
+        replyTopic,
+        RdKafka::Topic::PARTITION_UA,
+        RdKafka::Producer::RK_MSG_COPY,
+        payload,
+        len,
+        nullptr,
+        0,
+        0,
+        headers,
+        nullptr);
+
+    if (err != RdKafka::ERR_NO_ERROR) {
+        delete headers;
+        throw std::runtime_error("Failed to publish reply to '" + replyTopic +
+                                 "': " + RdKafka::err2str(err));
+    }
+    producer->poll(0);
+}
+
+void KafkaServer::loop() {
+    logInfo("starting consumer on topic='" + requestTopic_ + "' group='" + groupId_ +
+            "' brokers='" + brokers_ + "'");
+
+    while (running_) {
+        std::unique_ptr<RdKafka::Producer> producer;
+        std::unique_ptr<RdKafka::KafkaConsumer> consumer;
+
+        try {
+            using KafkaPluginUtil::setConfOrThrow;
+            std::string errstr;
+
+            {
+                std::unique_ptr<RdKafka::Conf> pconf(
+                    RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
+                setConfOrThrow(pconf.get(), "bootstrap.servers", brokers_);
+                setConfOrThrow(pconf.get(), "client.id", "graft-kafka-server-producer");
+                applySecurity(pconf.get());
+                RdKafka::Producer* raw = RdKafka::Producer::create(pconf.release(), errstr);
+                if (!raw) {
+                    throw std::runtime_error("Failed to create reply producer: " + errstr);
+                }
+                producer.reset(raw);
+            }
+
+            {
+                std::unique_ptr<RdKafka::Conf> cconf(
+                    RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
+                setConfOrThrow(cconf.get(), "bootstrap.servers", brokers_);
+                setConfOrThrow(cconf.get(), "group.id", groupId_);
+                setConfOrThrow(cconf.get(), "client.id", "graft-kafka-server-consumer");
+                setConfOrThrow(cconf.get(), "enable.auto.commit", "false");
+                setConfOrThrow(cconf.get(), "auto.offset.reset", "earliest");
+                setConfOrThrow(cconf.get(), "allow.auto.create.topics", "true");
+                applySecurity(cconf.get());
+                RdKafka::KafkaConsumer* raw =
+                    RdKafka::KafkaConsumer::create(cconf.release(), errstr);
+                if (!raw) {
+                    throw std::runtime_error("Failed to create request consumer: " + errstr);
+                }
+                consumer.reset(raw);
+
+                const RdKafka::ErrorCode subErr = consumer->subscribe({requestTopic_});
+                if (subErr != RdKafka::ERR_NO_ERROR) {
+                    throw std::runtime_error("Failed to subscribe to '" + requestTopic_ +
+                                             "': " + RdKafka::err2str(subErr));
+                }
+            }
+
+            logInfo("Kafka server loop connected");
+
+            while (running_) {
+                std::unique_ptr<RdKafka::Message> msg(consumer->consume(200));
+                if (!msg) {
+                    continue;
+                }
+                if (msg->err() == RdKafka::ERR__TIMED_OUT ||
+                    msg->err() == RdKafka::ERR__PARTITION_EOF) {
+                    producer->poll(0);
+                    continue;
+                }
+                if (msg->err() != RdKafka::ERR_NO_ERROR) {
+                    logWarn("consume error: " + msg->errstr());
+                    continue;
+                }
+
+                ProcessMessageFn process = nullptr;
+                {
+                    std::lock_guard<std::mutex> lock(processMutex_);
+                    process = process_;
+                }
+                if (!process) {
+                    logWarn("processMessage callback is not configured; skipping message");
+                    consumer->commitSync(msg.get());
+                    continue;
+                }
+
+                const std::string correlationId =
+                    KafkaPluginUtil::headerValue(msg->headers(), "correlation-id");
+                std::string replyTo =
+                    KafkaPluginUtil::headerValue(msg->headers(), "reply-to");
+                if (replyTo.empty()) {
+                    replyTo = replyTopic_;
+                }
+
+                std::vector<unsigned char> request;
+                if (msg->payload() && msg->len() > 0) {
+                    const auto* bytes = static_cast<const unsigned char*>(msg->payload());
+                    request.assign(bytes, bytes + msg->len());
+                }
+
+                std::vector<unsigned char> response;
+                auto writeResponse = [](void* context, const byte* data, std::size_t size) {
+                    auto* out = static_cast<std::vector<unsigned char>*>(context);
+                    if (!out) {
+                        return;
+                    }
+                    if (!data || size == 0) {
+                        out->clear();
+                        return;
+                    }
+                    out->assign(data, data + size);
+                };
+
+                const bool ok =
+                    process(request.data(), request.size(), writeResponse, &response);
+                if (!ok) {
+                    logWarn("processMessage returned false; not publishing reply");
+                    // Still commit to avoid poison-pill loops; gateway owns retry policy.
+                    consumer->commitSync(msg.get());
+                    continue;
+                }
+
+                if (!replyTo.empty()) {
+                    publishReply(producer.get(), replyTo, correlationId, response);
+                    const RdKafka::ErrorCode flushErr = producer->flush(10000);
+                    if (flushErr != RdKafka::ERR_NO_ERROR) {
+                        logWarn("reply flush incomplete: " + RdKafka::err2str(flushErr));
+                    }
+                } else {
+                    logWarn("no reply-to / replyTopic configured; dropping response");
+                }
+
+                consumer->commitSync(msg.get());
+            }
+        } catch (const std::exception& ex) {
+            logWarn(std::string("loop error: ") + ex.what());
+        } catch (...) {
+            logWarn("loop error: unknown");
+        }
+
+        if (consumer) {
+            try {
+                consumer->close();
+            } catch (...) {
+            }
+            consumer.reset();
+        }
+        if (producer) {
+            try {
+                producer->flush(2000);
+            } catch (...) {
+            }
+            producer.reset();
+        }
+
+        if (running_) {
+            logWarn("reconnecting in 2s...");
+            std::this_thread::sleep_for(std::chrono::seconds(2));
+        }
+    }
+
+    logInfo("stopped");
+}
diff --git a/kafka/KafkaPlugin/KafkaServer.h b/kafka/KafkaPlugin/KafkaServer.h
new file mode 100644
index 0000000..487f44f
--- /dev/null
+++ b/kafka/KafkaPlugin/KafkaServer.h
@@ -0,0 +1,50 @@
+#pragma once
+
+#include "IServer.h"
+
+#include <atomic>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <thread>
+#include <vector>
+
+namespace RdKafka {
+class Producer;
+class KafkaConsumer;
+class Conf;
+}  // namespace RdKafka
+
+class KafkaServer : public GraftcodeGateway::IServer {
+public:
+    KafkaServer() = default;
+    ~KafkaServer() override;
+
+    void configure(const char* jsonConfig, ProcessMessageFn processMessage) override;
+    void start() override;
+    void stop() override;
+
+private:
+    void loop();
+    void applySecurity(RdKafka::Conf* conf) const;
+    void publishReply(RdKafka::Producer* producer,
+                      const std::string& replyTopic,
+                      const std::string& correlationId,
+                      const std::vector<unsigned char>& response);
+
+    ProcessMessageFn process_{nullptr};
+    std::mutex processMutex_;
+
+    std::string brokers_ = "localhost:9092";
+    std::string requestTopic_ = "graft.requests";
+    std::string replyTopic_ = "graft.replies";
+    std::string groupId_ = "graft-gateway";
+    std::string securityProtocol_;
+    std::string saslMechanism_;
+    std::string saslUsername_;
+    std::string saslPassword_;
+    std::string sslCaLocation_;
+
+    std::atomic<bool> running_{false};
+    std::thread worker_;
+};
diff --git a/kafka/KafkaPlugin/KafkaUtil.h b/kafka/KafkaPlugin/KafkaUtil.h
new file mode 100644
index 0000000..e118e4f
--- /dev/null
+++ b/kafka/KafkaPlugin/KafkaUtil.h
@@ -0,0 +1,74 @@
+#pragma once
+
+#include <cstdint>
+#include <random>
+#include <sstream>
+#include <iomanip>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+#if __has_include(<librdkafka/rdkafkacpp.h>)
+#include <librdkafka/rdkafkacpp.h>
+#elif __has_include(<rdkafkacpp.h>)
+#include <rdkafkacpp.h>
+#else
+#error "librdkafka C++ header not found (expected librdkafka/rdkafkacpp.h or rdkafkacpp.h)"
+#endif
+
+namespace KafkaPluginUtil {
+
+inline std::string makeUuidV4() {
+    std::random_device rd;
+    std::mt19937_64 gen(rd());
+    std::uniform_int_distribution<std::uint64_t> dist;
+
+    const std::uint64_t a = dist(gen);
+    const std::uint64_t b = dist(gen);
+
+    // RFC 4122 variant 1, version 4
+    const std::uint64_t time_low = a & 0xFFFFFFFFULL;
+    const std::uint64_t time_mid = (a >> 32) & 0xFFFFULL;
+    const std::uint64_t time_hi = ((a >> 48) & 0x0FFFULL) | 0x4000ULL;
+    const std::uint64_t clock_seq = ((b >> 48) & 0x3FFFULL) | 0x8000ULL;
+    const std::uint64_t node = b & 0xFFFFFFFFFFFFULL;
+
+    std::ostringstream oss;
+    oss << std::hex << std::setfill('0')
+        << std::setw(8) << time_low << '-'
+        << std::setw(4) << time_mid << '-'
+        << std::setw(4) << time_hi << '-'
+        << std::setw(4) << clock_seq << '-'
+        << std::setw(12) << node;
+    return oss.str();
+}
+
+inline std::string headerValue(const RdKafka::Headers* headers, const std::string& key) {
+    if (!headers) {
+        return {};
+    }
+    const auto result = headers->get(key);
+    if (result.empty() || !result[0].value() || result[0].value_size() == 0) {
+        return {};
+    }
+    return std::string(static_cast<const char*>(result[0].value()), result[0].value_size());
+}
+
+inline RdKafka::Headers* makeRpcHeaders(const std::string& correlationId,
+                                        const std::string& replyTo) {
+    RdKafka::Headers* headers = RdKafka::Headers::create();
+    headers->add("correlation-id", correlationId);
+    if (!replyTo.empty()) {
+        headers->add("reply-to", replyTo);
+    }
+    return headers;
+}
+
+inline void setConfOrThrow(RdKafka::Conf* conf, const std::string& key, const std::string& value) {
+    std::string errstr;
+    if (conf->set(key, value, errstr) != RdKafka::Conf::CONF_OK) {
+        throw std::runtime_error("librdkafka conf set '" + key + "' failed: " + errstr);
+    }
+}
+
+}  // namespace KafkaPluginUtil
diff --git a/kafka/KafkaPlugin/TransportKafka.cpp b/kafka/KafkaPlugin/TransportKafka.cpp
new file mode 100644
index 0000000..a25129c
--- /dev/null
+++ b/kafka/KafkaPlugin/TransportKafka.cpp
@@ -0,0 +1,97 @@
+#include "TransportKafka.h"
+#include "KafkaServer.h"
+
+#include <nlohmann/json.hpp>
+
+#include <cstring>
+
+#if defined(_WIN32)
+#define KAFKA_PLUGIN_EXPORT extern "C" __declspec(dllexport)
+#else
+#define KAFKA_PLUGIN_EXPORT extern "C"
+#endif
+
+static KafkaClient::Config parseClientConfig(const char* configJson) {
+    KafkaClient::Config c;
+    if (!configJson) {
+        return c;
+    }
+    auto j = nlohmann::json::parse(configJson, nullptr, false);
+    if (j.is_discarded() || !j.is_object()) {
+        return c;
+    }
+    if (j.contains("brokers") && j["brokers"].is_string()) {
+        c.brokers = j["brokers"].get<std::string>();
+    } else if (j.contains("host") && j["host"].is_string()) {
+        c.brokers = j["host"].get<std::string>();
+    }
+    if (j.contains("requestTopic") && j["requestTopic"].is_string()) {
+        c.requestTopic = j["requestTopic"].get<std::string>();
+    } else if (j.contains("queue") && j["queue"].is_string()) {
+        c.requestTopic = j["queue"].get<std::string>();
+    }
+    if (j.contains("replyTopic") && j["replyTopic"].is_string()) {
+        c.replyTopic = j["replyTopic"].get<std::string>();
+    } else if (j.contains("replyQueue") && j["replyQueue"].is_string()) {
+        c.replyTopic = j["replyQueue"].get<std::string>();
+    }
+    if (j.contains("groupId") && j["groupId"].is_string()) {
+        c.groupId = j["groupId"].get<std::string>();
+    }
+    if (j.contains("rpcTimeoutMs") && j["rpcTimeoutMs"].is_number_integer()) {
+        c.rpcTimeoutMs = j["rpcTimeoutMs"].get<int>();
+    }
+    if (j.contains("securityProtocol") && j["securityProtocol"].is_string()) {
+        c.securityProtocol = j["securityProtocol"].get<std::string>();
+    }
+    if (j.contains("saslMechanism") && j["saslMechanism"].is_string()) {
+        c.saslMechanism = j["saslMechanism"].get<std::string>();
+    }
+    if (j.contains("saslUsername") && j["saslUsername"].is_string()) {
+        c.saslUsername = j["saslUsername"].get<std::string>();
+    }
+    if (j.contains("saslPassword") && j["saslPassword"].is_string()) {
+        c.saslPassword = j["saslPassword"].get<std::string>();
+    }
+    if (j.contains("sslCaLocation") && j["sslCaLocation"].is_string()) {
+        c.sslCaLocation = j["sslCaLocation"].get<std::string>();
+    }
+    return c;
+}
+
+TransportKafka::TransportKafka(const char*, unsigned short, const char* configJson)
+    : client_(std::make_unique<KafkaClient>(parseClientConfig(configJson))) {}
+
+int TransportKafka::Initialize(byte, byte, byte) { return 0; }
+
+int TransportKafka::SendCommand(byte* messageByteArray, int32_t messageByteArrayLen) {
+    lastResponse_ = client_->call(messageByteArray, static_cast<std::size_t>(messageByteArrayLen));
+    return static_cast<int>(lastResponse_.size());
+}
+
+int TransportKafka::ReadResponse(byte* responseByteArray, int32_t responseByteArrayLen) {
+    if (responseByteArrayLen < static_cast<int32_t>(lastResponse_.size())) {
+        return -1;
+    }
+    if (!lastResponse_.empty()) {
+        std::memcpy(responseByteArray, lastResponse_.data(), lastResponse_.size());
+    }
+    return 0;
+}
+
+KAFKA_PLUGIN_EXPORT Hypertube::Native::Interfaces::ITransport*
+CreateTransportChannel(const char* ipAddress, const unsigned short port, const char* configSource) {
+    return new TransportKafka(ipAddress, port, configSource);
+}
+
+KAFKA_PLUGIN_EXPORT void DestroyTransportChannel(Hypertube::Native::Interfaces::ITransport* transport) {
+    delete transport;
+}
+
+KAFKA_PLUGIN_EXPORT GraftcodeGateway::IServer* CreateServer() {
+    return new KafkaServer();
+}
+
+KAFKA_PLUGIN_EXPORT void DestroyServer(GraftcodeGateway::IServer* server) {
+    delete server;
+}
diff --git a/kafka/KafkaPlugin/TransportKafka.h b/kafka/KafkaPlugin/TransportKafka.h
new file mode 100644
index 0000000..e8dcbdc
--- /dev/null
+++ b/kafka/KafkaPlugin/TransportKafka.h
@@ -0,0 +1,17 @@
+#pragma once
+#include "ITransport.h"
+#include "KafkaClient.h"
+#include <memory>
+#include <vector>
+
+class TransportKafka : public Hypertube::Native::Interfaces::ITransport {
+public:
+    TransportKafka(const char* /*ip*/, unsigned short /*port*/, const char* configJson);
+    int Initialize(byte callingRuntimeNumber, byte calledRuntimeNumber, byte calledRuntimeVersion) override;
+    int SendCommand(byte* messageByteArray, int32_t messageByteArrayLen) override;
+    int ReadResponse(byte* responseByteArray, int32_t responseByteArrayLen) override;
+
+private:
+    std::unique_ptr<KafkaClient> client_;
+    std::vector<unsigned char> lastResponse_;
+};
diff --git a/kafka/KafkaPluginTest/CMakeLists.txt b/kafka/KafkaPluginTest/CMakeLists.txt
new file mode 100644
index 0000000..d65dd1b
--- /dev/null
+++ b/kafka/KafkaPluginTest/CMakeLists.txt
@@ -0,0 +1,6 @@
+add_executable(KafkaPluginTest smoke.cpp)
+target_link_libraries(KafkaPluginTest PRIVATE KafkaPlugin)
+target_include_directories(KafkaPluginTest PRIVATE
+    "${CMAKE_SOURCE_DIR}/GraftcodePluginsInterfaces"
+)
+add_test(NAME KafkaPluginSmoke COMMAND KafkaPluginTest)
diff --git a/kafka/KafkaPluginTest/smoke.cpp b/kafka/KafkaPluginTest/smoke.cpp
new file mode 100644
index 0000000..b68b717
--- /dev/null
+++ b/kafka/KafkaPluginTest/smoke.cpp
@@ -0,0 +1,62 @@
+#include "IServer.h"
+#include "ITransport.h"
+
+#include <cstdio>
+#include <cstring>
+
+#if defined(_WIN32)
+#define KAFKA_PLUGIN_IMPORT extern "C" __declspec(dllimport)
+#else
+#define KAFKA_PLUGIN_IMPORT extern "C"
+#endif
+
+KAFKA_PLUGIN_IMPORT GraftcodeGateway::IServer* CreateServer();
+KAFKA_PLUGIN_IMPORT void DestroyServer(GraftcodeGateway::IServer* server);
+KAFKA_PLUGIN_IMPORT Hypertube::Native::Interfaces::ITransport* CreateTransportChannel(
+    const char* ipAddress, unsigned short port, const char* configSource);
+KAFKA_PLUGIN_IMPORT void DestroyTransportChannel(Hypertube::Native::Interfaces::ITransport* transport);
+
+static bool g_processCalled = false;
+
+static bool smokeProcess(const GraftcodeGateway::IServer::byte* /*requestData*/,
+                         std::size_t /*requestSize*/,
+                         GraftcodeGateway::IServer::WriteResponseFn writeResponse,
+                         void* writeContext) {
+    g_processCalled = true;
+    static const unsigned char kPayload[] = {'o', 'k'};
+    if (writeResponse) {
+        writeResponse(writeContext, kPayload, sizeof(kPayload));
+    }
+    return true;
+}
+
+int main() {
+    // Link / symbol smoke — no broker required.
+    GraftcodeGateway::IServer* server = CreateServer();
+    if (!server) {
+        std::puts("FAIL: CreateServer returned null");
+        return 1;
+    }
+
+    const char* cfg =
+        R"({"brokers":"127.0.0.1:1","requestTopic":"graft.requests","replyTopic":"graft.replies","groupId":"graft-smoke"})";
+    server->configure(cfg, smokeProcess);
+    // Do not start() — that would spin a reconnect loop against a missing broker.
+    DestroyServer(server);
+
+    Hypertube::Native::Interfaces::ITransport* transport =
+        CreateTransportChannel("127.0.0.1", 0, cfg);
+    if (!transport) {
+        std::puts("FAIL: CreateTransportChannel returned null");
+        return 1;
+    }
+    if (transport->Initialize(1, 2, 1) != 0) {
+        std::puts("FAIL: Initialize returned non-zero");
+        DestroyTransportChannel(transport);
+        return 1;
+    }
+    DestroyTransportChannel(transport);
+
+    std::puts("KafkaPlugin smoke OK — factories linked; run against local Kafka to exercise RPC");
+    return 0;
+}
diff --git a/kafka/Readme.md b/kafka/Readme.md
new file mode 100644
index 0000000..a8714c7
--- /dev/null
+++ b/kafka/Readme.md
@@ -0,0 +1,106 @@
+# Kafka Plugin Build (CMake)
+
+This plugin is the Apache Kafka counterpart of the RabbitMQ / Azure Service Bus
+plugins. It implements the same Graftcode plugin interfaces
+(`Hypertube::Native::Interfaces::ITransport` for the calling runtime and
+`GraftcodeGateway::IServer` for the gateway) and exposes the same exported
+factory symbols (`CreateTransportChannel` / `DestroyTransportChannel` and
+`CreateServer` / `DestroyServer`).
+
+It is written in C++ and talks to Kafka using **librdkafka** (`rdkafka++`),
+acquired through CMake `FetchContent` (same pattern as the RabbitMQ plugin).
+
+## RPC model
+
+Kafka has no native request/reply. This plugin mirrors the Service Bus / AMQP
+pattern with message headers:
+
+1. **Client** produces to `requestTopic` with headers:
+   - `correlation-id` — UUID per call
+   - `reply-to` — reply topic the server should use
+2. **Server** (GG plugin) consumes `requestTopic`, runs `processMessage`, then
+   produces to the `reply-to` topic (falling back to configured `replyTopic`)
+   echoing the same `correlation-id`.
+3. **Client** consumes `replyTopic` until a message with a matching
+   `correlation-id` arrives, or `rpcTimeoutMs` elapses.
+
+## 1) Clone repository
+
+```bash
+git clone https://github.com/grft-dev/graftcode-extensions.git
+cd graftcode-extensions/kafka
+```
+
+## 2) Configure with CMake
+
+```bash
+cmake -S . -B build -DCMAKE_BUILD_TYPE=Release
+```
+
+## 3) Build
+
+```bash
+cmake --build build --config Release
+```
+
+CMake downloads **nlohmann/json** and **librdkafka** (`v2.8.0`) via FetchContent
+and links a static `rdkafka++` into the shared plugin.
+
+As a result, you will receive:
+- `kafka/build/KafkaPlugin/KafkaPlugin.dll` (Windows)
+- `kafka/build/KafkaPlugin/libKafkaPlugin.so` / `.dylib` (Linux / macOS)
+
+If the generated library is `libKafkaPlugin.*`, use plugin name
+`libKafkaPlugin` in config.
+
+## 4) Download GG
+
+Download `gg` from:
+https://github.com/grft-dev/graftcode-gateway/releases/
+
+## 5) Run a local broker
+
+### Apache Kafka (KRaft)
+
+```bash
+docker compose -f docker-compose.yml up -d
+./scripts/create-topics.sh
+```
+
+### Redpanda (lighter alternative)
+
+```bash
+docker compose -f docker-compose.redpanda.yml up -d
+# brokers: localhost:19092
+```
+
+## 6) Run GG with sample library
+
+Create `pluginConfig.json` (see also the example in this folder):
+
+```json
+{
+  "name": "KafkaPlugin",
+  "brokers": "localhost:9092",
+  "requestTopic": "graft.requests",
+  "replyTopic": "graft.replies",
+  "groupId": "graft-gateway",
+  "rpcTimeoutMs": 30000
+}
+```
+
+Then run:
+
+```powershell
+./gg .\PhysicsCalculator.dll --config .\pluginConfig.json
+```
+
+## Configuration reference
+
+| Field | Required | Description |
+|-------|----------|-------------|
+| `brokers` / `host` | yes | Kafka bootstrap servers |
+| `requestTopic` / `queue` | yes | Topic for requests |
+| `replyTopic` / `replyQueue` | yes (RPC client) | Topic for replies |
+| `groupId` | no | Consumer group base name |
+| `rpcTimeoutMs` | no | Request/response timeout (default 30000) |
diff --git a/samples/kafka/README.md b/samples/kafka/README.md
new file mode 100644
index 0000000..e7a0106
--- /dev/null
+++ b/samples/kafka/README.md
@@ -0,0 +1,29 @@
+# Kafka sample configs
+
+Local broker + UI and Graftcode plugin connection files.
+
+## Start Kafka
+
+```bash
+cd samples/kafka
+docker compose up -d
+```
+
+- Broker: `localhost:9092`
+- UI: http://localhost:8080
+
+Create topics (if needed):
+
+```bash
+docker exec graftcode-kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --if-not-exists --topic graft.requests --partitions 1 --replication-factor 1
+docker exec graftcode-kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --if-not-exists --topic graft.replies --partitions 1 --replication-factor 1
+```
+
+## Config files
+
+| File | Use |
+|------|-----|
+| `pluginConfig.gateway.json` | `gg YourModule.dll --config pluginConfig.gateway.json` |
+| `pluginConfig.client.json` | Client-side plugin block |
+| `graftConfig.kafka.example.json` | Full GraftConfig example for .NET |
+| `pluginConfig.json` | Same as gateway (default name) |
diff --git a/samples/kafka/docker-compose.redpanda.yml b/samples/kafka/docker-compose.redpanda.yml
new file mode 100644
index 0000000..eff7477
--- /dev/null
+++ b/samples/kafka/docker-compose.redpanda.yml
@@ -0,0 +1,10 @@
+services:
+  redpanda:
+    image: redpandadata/redpanda:v24.2.4
+    command:
+      - redpanda start
+      - --kafka-addr internal://0.0.0.0:9092,external://0.0.0.0:19092
+      - --advertise-kafka-addr internal://redpanda:9092,external://localhost:19092
+      - --mode dev-container
+    ports:
+      - "19092:19092"
diff --git a/samples/kafka/docker-compose.yml b/samples/kafka/docker-compose.yml
new file mode 100644
index 0000000..4e4f838
--- /dev/null
+++ b/samples/kafka/docker-compose.yml
@@ -0,0 +1,32 @@
+# Lokalny Kafka (KRaft) + UI — testy Graftcode KafkaPlugin.
+services:
+  kafka:
+    image: apache/kafka:3.8.1
+    container_name: graftcode-kafka
+    ports:
+      - "9092:9092"
+    environment:
+      KAFKA_NODE_ID: 1
+      KAFKA_PROCESS_ROLES: broker,controller
+      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093,PLAINTEXT_INTERNAL://0.0.0.0:9094
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://kafka:9094
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT_INTERNAL
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
+
+  kafka-ui:
+    image: provectuslabs/kafka-ui:v0.7.2
+    container_name: graftcode-kafka-ui
+    depends_on:
+      - kafka
+    ports:
+      - "8080:8080"
+    environment:
+      KAFKA_CLUSTERS_0_NAME: local
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9094
diff --git a/samples/kafka/graftConfig.kafka.example.json b/samples/kafka/graftConfig.kafka.example.json
new file mode 100644
index 0000000..120c636
--- /dev/null
+++ b/samples/kafka/graftConfig.kafka.example.json
@@ -0,0 +1,16 @@
+{
+  "configurations": {
+    "graft.nuget.PhysicsCalculator": {
+      "runtime": "netcore",
+      "stateless": true,
+      "plugin": {
+        "name": "KafkaPlugin",
+        "brokers": "localhost:9092",
+        "requestTopic": "graft.requests",
+        "replyTopic": "graft.replies",
+        "groupId": "graft-client",
+        "rpcTimeoutMs": 30000
+      }
+    }
+  }
+}
diff --git a/samples/kafka/pluginConfig.client.json b/samples/kafka/pluginConfig.client.json
new file mode 100644
index 0000000..8cac248
--- /dev/null
+++ b/samples/kafka/pluginConfig.client.json
@@ -0,0 +1,8 @@
+{
+  "name": "KafkaPlugin",
+  "brokers": "localhost:9092",
+  "requestTopic": "graft.requests",
+  "replyTopic": "graft.replies",
+  "groupId": "graft-client",
+  "rpcTimeoutMs": 30000
+}
diff --git a/samples/kafka/pluginConfig.gateway.json b/samples/kafka/pluginConfig.gateway.json
new file mode 100644
index 0000000..e607326
--- /dev/null
+++ b/samples/kafka/pluginConfig.gateway.json
@@ -0,0 +1,8 @@
+{
+  "name": "KafkaPlugin",
+  "brokers": "localhost:9092",
+  "requestTopic": "graft.requests",
+  "replyTopic": "graft.replies",
+  "groupId": "graft-gateway",
+  "rpcTimeoutMs": 30000
+}
diff --git a/samples/kafka/pluginConfig.json b/samples/kafka/pluginConfig.json
new file mode 100644
index 0000000..e607326
--- /dev/null
+++ b/samples/kafka/pluginConfig.json
@@ -0,0 +1,8 @@
+{
+  "name": "KafkaPlugin",
+  "brokers": "localhost:9092",
+  "requestTopic": "graft.requests",
+  "replyTopic": "graft.replies",
+  "groupId": "graft-gateway",
+  "rpcTimeoutMs": 30000
+}
diff --git a/samples/kafka/scripts/create-topics.sh b/samples/kafka/scripts/create-topics.sh
new file mode 100644
index 0000000..efbb103
--- /dev/null
+++ b/samples/kafka/scripts/create-topics.sh
@@ -0,0 +1,53 @@
+#!/usr/bin/env bash
+# Create the default Graftcode Kafka request/reply topics.
+set -euo pipefail
+
+BROKERS="${BROKERS:-localhost:9092}"
+REQUEST_TOPIC="${REQUEST_TOPIC:-graft.requests}"
+REPLY_TOPIC="${REPLY_TOPIC:-graft.replies}"
+CONTAINER="${KAFKA_CONTAINER:-}"
+
+create_with_kafka_topics() {
+  local bin="$1"
+  "$bin" --bootstrap-server "$BROKERS" --create --if-not-exists --topic "$REQUEST_TOPIC" --partitions 1 --replication-factor 1
+  "$bin" --bootstrap-server "$BROKERS" --create --if-not-exists --topic "$REPLY_TOPIC" --partitions 1 --replication-factor 1
+}
+
+if [[ -n "$CONTAINER" ]]; then
+  echo "Creating topics via docker exec on container '$CONTAINER' (bootstrap $BROKERS)..."
+  # apache/kafka image path
+  if docker exec "$CONTAINER" test -x /opt/kafka/bin/kafka-topics.sh; then
+    docker exec "$CONTAINER" /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \
+      --create --if-not-exists --topic "$REQUEST_TOPIC" --partitions 1 --replication-factor 1
+    docker exec "$CONTAINER" /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \
+      --create --if-not-exists --topic "$REPLY_TOPIC" --partitions 1 --replication-factor 1
+  # bitnami / other layouts
+  elif docker exec "$CONTAINER" sh -c 'command -v kafka-topics.sh' >/dev/null 2>&1; then
+    docker exec "$CONTAINER" kafka-topics.sh --bootstrap-server localhost:9092 \
+      --create --if-not-exists --topic "$REQUEST_TOPIC" --partitions 1 --replication-factor 1
+    docker exec "$CONTAINER" kafka-topics.sh --bootstrap-server localhost:9092 \
+      --create --if-not-exists --topic "$REPLY_TOPIC" --partitions 1 --replication-factor 1
+  else
+    echo "Could not find kafka-topics.sh inside $CONTAINER" >&2
+    exit 1
+  fi
+elif command -v kafka-topics.sh >/dev/null 2>&1; then
+  create_with_kafka_topics kafka-topics.sh
+elif command -v rpk >/dev/null 2>&1; then
+  echo "Creating topics via rpk (brokers $BROKERS)..."
+  rpk topic create "$REQUEST_TOPIC" -X brokers="$BROKERS" || true
+  rpk topic create "$REPLY_TOPIC" -X brokers="$BROKERS" || true
+else
+  # Auto-detect a running compose kafka container
+  CONTAINER="$(docker ps --format '{{.Names}}' | grep -E 'kafka|redpanda' | head -n1 || true)"
+  if [[ -z "$CONTAINER" ]]; then
+    echo "No kafka-topics.sh/rpk on PATH and no kafka/redpanda container running." >&2
+    echo "Start the broker first: docker compose up -d" >&2
+    echo "Or set KAFKA_CONTAINER=<name>." >&2
+    exit 1
+  fi
+  KAFKA_CONTAINER="$CONTAINER" "$0"
+  exit $?
+fi
+
+echo "Topics ready: $REQUEST_TOPIC, $REPLY_TOPIC"