v0.5.0: Kafka Bus integration (Optiq KFK model)

Add KafkaBus multi-topic publisher mirroring Optiq KFK between ME and
downstream consumers. MECoreActor publishes orders, trades, and BBO
snapshots to Kafka when EUNEX_KAFKA_BROKERS is set. No-op stub compiles
when EUNEX_USE_KAFKA is OFF. Switch Docker from bitnami/kafka to
apache/kafka:3.9.0 (KRaft mode). Add .dockerignore, update docs.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

.dockerignore

@@ -0,0 +1,9 @@
+build/
+.git/
+*.exe
+*.obj
+*.pdb
+__pycache__/
+*.pyc
+.vs/
+.vscode/

CMakeLists.txt

@@ -1,5 +1,5 @@
 cmake_minimum_required(VERSION 3.16)
-project(EuNEx VERSION 0.4.0 LANGUAGES CXX)
+project(EuNEx VERSION 0.5.0 LANGUAGES CXX)
 
 set(CMAKE_CXX_STANDARD 20)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)

README.md

@@ -27,7 +27,7 @@ ch_ai_trader.py          →   ClearingHouseActor        →   Clearing House (P
   AI strategies          →   AITraderActor             →   Trading obligations
 ```
 
-## Actor Topology (v0.4)
+## Actor Topology (v0.5)
 
 ```
  Core 0: OEGActor + FIXAcceptorActor     ← Order entry & FIX protocol
@@ -109,17 +109,25 @@ cmake --build build --config Release
 # Run matching engine
 ./build/Release/eunex_me
 
-# Run all tests (6 suites)
+# Run all tests (7 suites)
 cd build && ctest -C Release
 ```
 
 ## With Kafka Persistence
 
 ```bash
+# Compile with Kafka support (requires librdkafka-dev)
 cmake -B build -DEUNEX_USE_KAFKA=ON
 cmake --build build --config Release
+
+# Run with Kafka (set broker address)
+EUNEX_KAFKA_BROKERS=localhost:9092 ./build/Release/eunex_me
+
+# Topics: eunex.orders, eunex.trades, eunex.market-data, eunex.recovery.fragments
 ```
 
+Without `EUNEX_USE_KAFKA`, the engine compiles with a no-op stub and runs standalone.
+
 ## FIX Gateway
 
 The C++ engine includes a built-in FIX 4.4 acceptor on TCP port 9001.
@@ -163,6 +171,7 @@ EuNEx/
 │   ├── net/SocketCompat.hpp           # Cross-platform socket abstraction
 │   ├── persistence/
 │   │   ├── PersistenceStore.hpp        # Abstract store + InMemoryStore
+│   │   ├── KafkaBus.hpp                # Multi-topic Kafka publisher (Optiq KFK)
 │   │   └── KafkaStore.hpp              # Kafka persistence (optional)
 │   ├── recovery/RecoveryProxy.hpp/cpp  # Recovery Cause/Effect
 │   └── iaca/

docker/Dockerfile

@@ -32,6 +32,10 @@ COPY --from=builder /app/build/eunex_me /app/
 COPY --from=builder /app/build/test_orderbook /app/
 COPY --from=builder /app/build/test_matching_engine /app/
 COPY --from=builder /app/build/test_threaded_engine /app/
+COPY --from=builder /app/build/test_clearing_house /app/
+COPY --from=builder /app/build/test_fix_gateway /app/
+COPY --from=builder /app/build/test_ai_trader /app/
+COPY --from=builder /app/build/test_stop_orders /app/
 COPY --from=builder /app/build/simple_match /app/
 COPY --from=builder /app/dashboard/ /app/dashboard/
 COPY --from=builder /app/shared/ /app/shared/

docker/docker-compose.yml

@@ -1,44 +1,45 @@
 services:
   kafka:
-    image: bitnami/kafka:3.9
+    image: apache/kafka:3.9.0
     ports:
       - "9092:9092"
     environment:
-      KAFKA_CFG_NODE_ID: 1
-      KAFKA_CFG_PROCESS_ROLES: controller,broker
-      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
-      KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
-      KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
-      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
-      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
-      KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "true"
-      KAFKA_CFG_NUM_PARTITIONS: 4
-      KAFKA_CFG_LOG_RETENTION_HOURS: 24
+      KAFKA_NODE_ID: 1
+      KAFKA_PROCESS_ROLES: broker,controller
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
+      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
+      KAFKA_NUM_PARTITIONS: 4
+      KAFKA_LOG_RETENTION_HOURS: 24
+      CLUSTER_ID: "EuNEx-kafka-cluster-001"
     volumes:
-      - kafka_data:/bitnami/kafka
+      - kafka_data:/tmp/kraft-combined-logs
     healthcheck:
-      test: ["CMD-SHELL", "kafka-topics.sh --bootstrap-server localhost:9092 --list || exit 1"]
+      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list || exit 1"]
       interval: 10s
       timeout: 5s
       retries: 5
 
   kafka-init:
-    image: bitnami/kafka:3.9
+    image: apache/kafka:3.9.0
     depends_on:
       kafka:
         condition: service_healthy
     entrypoint: ["/bin/bash", "-c"]
     command:
       - |
-        kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists \
+        /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists \
           --topic eunex.recovery.fragments --partitions 4 --replication-factor 1
-        kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists \
+        /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists \
           --topic eunex.trades --partitions 4 --replication-factor 1
-        kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists \
+        /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists \
           --topic eunex.market-data --partitions 4 --replication-factor 1
-        kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists \
+        /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists \
           --topic eunex.orders --partitions 4 --replication-factor 1
-        kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists \
+        /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists \
           --topic eunex.control --partitions 1 --replication-factor 1
         echo "Topics created."
 

docs/developers-guide.md

@@ -1,6 +1,6 @@
 # EuNEx Developers Guide
 
-**Version 0.4.0** | Euronext Optiq-Modeled Exchange Simulator
+**Version 0.5.0** | Euronext Optiq-Modeled Exchange Simulator
 
 ---
 
@@ -15,13 +15,14 @@
 7. [Event System](#7-event-system)
 8. [Order Book & Matching](#8-order-book--matching)
 9. [Recovery & IACA](#9-recovery--iaca)
-10. [FIX Protocol Gateway](#10-fix-protocol-gateway)
-11. [Clearing House](#11-clearing-house)
-12. [AI Trading Members](#12-ai-trading-members)
-13. [Project Structure](#13-project-structure)
-14. [Build & Test](#14-build--test)
-15. [Configuration](#15-configuration)
-16. [Extending EuNEx](#16-extending-eunex)
+10. [Kafka Bus](#10-kafka-bus)
+11. [FIX Protocol Gateway](#11-fix-protocol-gateway)
+12. [Clearing House](#12-clearing-house)
+13. [AI Trading Members](#13-ai-trading-members)
+14. [Project Structure](#14-project-structure)
+15. [Build & Test](#15-build--test)
+16. [Configuration](#16-configuration)
+17. [Extending EuNEx](#17-extending-eunex)
 
 ---
 
@@ -523,7 +524,72 @@ The `Book` class implements price-time priority matching:
 
 ---
 
-## 10. FIX Protocol Gateway
+## 10. Kafka Bus
+
+### Overview
+
+The Kafka Bus (`src/persistence/KafkaBus.hpp`) mirrors the Optiq KFK (Kafka Bus) that connects the Matching Engine to downstream consumers: MDG, PTB, Clearing, IDS, and SATURN.
+
+When `EUNEX_USE_KAFKA` is enabled at compile time and `EUNEX_KAFKA_BROKERS` is set at runtime, the bus publishes events to Kafka topics in real time. When disabled, a no-op stub compiles in its place so the engine runs standalone.
+
+### Topics
+
+```
+  Topic                        Content                  Key
+  ─────────────────────────────────────────────────────────────
+  eunex.orders                 Raw Order structs        symbolIdx
+  eunex.trades                 Trade structs            symbolIdx
+  eunex.market-data            BBO snapshots            symbolIdx
+  eunex.recovery.fragments     Recovery fragments       originId:originKey
+  eunex.control                Control messages         (reserved)
+```
+
+### Architecture
+
+```
+  MECoreActor (per symbol)
+       │
+       ├─ onEvent(NewOrderEvent)
+       │     ├─ kafka_->publishOrder(order)         → eunex.orders
+       │     ├─ onTrade callback:
+       │     │     ├─ kafka_->publishTrade(trade)    → eunex.trades
+       │     │     └─ mdPipe_ / chPipe_ (actors)
+       │     └─ publishBookUpdate()
+       │           └─ kafka_->publishMarketData(...) → eunex.market-data
+       │
+       └─ KafkaBus* kafka_  (nullptr when disabled)
+```
+
+### Compile-Time Toggle
+
+```cmake
+cmake .. -DEUNEX_USE_KAFKA=ON    # requires librdkafka-dev
+cmake .. -DEUNEX_USE_KAFKA=OFF   # no-op stub (default)
+```
+
+### Runtime Configuration
+
+Set `EUNEX_KAFKA_BROKERS` environment variable:
+
+```bash
+export EUNEX_KAFKA_BROKERS=kafka:9092
+./eunex_me
+```
+
+The engine prints Kafka connection status at startup and publishes cumulative stats (orders, trades, market-data messages) every 10 rounds.
+
+### Docker Deployment
+
+`docker/docker-compose.yml` runs Kafka in KRaft mode (no ZooKeeper) using `apache/kafka:3.9.0`, creates all topics via `kafka-init`, then starts the engine with `EUNEX_KAFKA_BROKERS=kafka:9092`.
+
+```bash
+cd docker
+docker compose up --build
+```
+
+---
+
+## 11. FIX Protocol Gateway
 
 ### FIXAcceptorActor (`src/actors/FIXAcceptorActor.hpp`)
 
@@ -583,7 +649,7 @@ The `Book` class implements price-time priority matching:
 
 ---
 
-## 11. Clearing House
+## 12. Clearing House
 
 ### ClearingHouseActor (`src/actors/ClearingHouseActor.hpp`)
 
@@ -630,7 +696,7 @@ The `Book` class implements price-time priority matching:
 
 ---
 
-## 12. AI Trading Members
+## 13. AI Trading Members
 
 ### AITraderActor (`src/actors/AITraderActor.hpp`)
 
@@ -682,7 +748,7 @@ The `Book` class implements price-time priority matching:
 
 ---
 
-## 13. Project Structure
+## 14. Project Structure
 
 ```
  EuNEx/
@@ -744,7 +810,7 @@ The `Book` class implements price-time priority matching:
 
 ---
 
-## 14. Build & Test
+## 15. Build & Test
 
 ### Prerequisites
 
@@ -803,7 +869,7 @@ cd build && ctest -C Release --output-on-failure
 
 ---
 
-## 15. Configuration
+## 16. Configuration
 
 ### Runtime Configuration (main.cpp)
 
@@ -829,7 +895,7 @@ The engine pre-populates order books with spread-defining orders:
 
 ---
 
-## 16. Extending EuNEx
+## 17. Extending EuNEx
 
 ### Adding a New Symbol
 

src/actors/MECoreActor.cpp

@@ -6,10 +6,12 @@ namespace eunex {
 MECoreActor::MECoreActor(SymbolIndex_t symbolIdx,
                                const tredzone::ActorId& oeGatewayId,
                                const tredzone::ActorId& marketDataId,
-                               const tredzone::ActorId& clearingHouseId)
+                               const tredzone::ActorId& clearingHouseId,
+                               KafkaBus* kafkaBus)
     : book_(symbolIdx)
     , oePipe_(*this, oeGatewayId)
     , mdPipe_(*this, marketDataId)
+    , kafka_(kafkaBus)
 {
     if (clearingHouseId.id != 0) {
         chPipe_.emplace(*this, clearingHouseId);
@@ -19,10 +21,6 @@ MECoreActor::MECoreActor(SymbolIndex_t symbolIdx,
     registerEventHandler<ModifyOrderEvent>(*this);
 }
 
-// ── NewOrder ───────────────────────────────────────────────────────
-// StockEx: match_order(order, producer)
-// Optiq:   RecoveryCause.onInput → CauseOperator → forwardToBook()
-
 void MECoreActor::onEvent(const NewOrderEvent& event) {
     Order order{};
     order.clOrdId    = event.clOrdId;
@@ -35,9 +33,12 @@ void MECoreActor::onEvent(const NewOrderEvent& event) {
     order.sessionId  = event.sessionId;
     order.stopPrice  = event.stopPrice;
 
+    if (kafka_) kafka_->publishOrder(order);
+
     auto onTrade = [this](const Trade& trade) {
         mdPipe_.push<TradeEvent>(trade);
         if (chPipe_) chPipe_->push<TradeEvent>(trade);
+        if (kafka_) kafka_->publishTrade(trade);
     };
     auto onExec = [this, &event](const ExecutionReport& rpt) {
         oePipe_.push<ExecReportEvent>(rpt, event.sessionId);
@@ -45,7 +46,6 @@ void MECoreActor::onEvent(const NewOrderEvent& event) {
 
     book_.newOrder(order, onTrade, onExec);
 
-    // Trigger any stop orders whose stopPrice has been crossed
     if (book_.stopOrderCount() > 0 && book_.lastTradePrice() != 0) {
         book_.triggerStopOrders(book_.lastTradePrice(), onTrade, onExec);
     }
@@ -53,9 +53,6 @@ void MECoreActor::onEvent(const NewOrderEvent& event) {
     publishBookUpdate();
 }
 
-// ── Cancel ─────────────────────────────────────────────────────────
-// StockEx: handle_cancel(msg, producer)
-
 void MECoreActor::onEvent(const CancelOrderEvent& event) {
     ExecutionReport rpt{};
     if (book_.cancelOrder(event.orderId, rpt)) {
@@ -68,9 +65,6 @@ void MECoreActor::onEvent(const CancelOrderEvent& event) {
     }
 }
 
-// ── Modify (Cancel-Replace) ────────────────────────────────────────
-// StockEx: handle_amend(msg, producer)
-
 void MECoreActor::onEvent(const ModifyOrderEvent& event) {
     ExecutionReport rpt{};
     if (book_.modifyOrder(event.orderId, event.newPrice, event.newQuantity, rpt)) {
@@ -83,10 +77,6 @@ void MECoreActor::onEvent(const ModifyOrderEvent& event) {
     }
 }
 
-// ── Publish book snapshot to MarketData actor ──────────────────────
-// StockEx: the Dashboard reads orderbook via REST GET /orderbook/<symbol>
-// Optiq:   publishLimitEffect → push to MDLimitLogicalCoreHandler
-
 void MECoreActor::publishBookUpdate() {
     BookUpdateEvent update;
     update.symbolIdx = book_.symbolIndex();
@@ -104,6 +94,14 @@ void MECoreActor::publishBookUpdate() {
     }
 
     mdPipe_.push<BookUpdateEvent>(update);
+
+    if (kafka_) {
+        Price_t bb = bidLevels.empty() ? 0 : bidLevels[0].price;
+        Price_t ba = askLevels.empty() ? 0 : askLevels[0].price;
+        Quantity_t bq = bidLevels.empty() ? 0 : bidLevels[0].totalQty;
+        Quantity_t aq = askLevels.empty() ? 0 : askLevels[0].totalQty;
+        kafka_->publishMarketData(book_.symbolIndex(), bb, ba, bq, aq);
+    }
 }
 
 } // namespace eunex

src/actors/MECoreActor.hpp

@@ -1,24 +1,8 @@
 #pragma once
-// ════════════════════════════════════════════════════════════════════
-// MECoreActor — The matching engine core
-//
-// StockEx equivalent: matcher.py (match_order, handle_cancel, handle_amend)
-// Optiq equivalent:   LogicalCoreActor + RecoveryHelperCore + Book
-//
-// One actor per symbol (or group of symbols). Owns the OrderBook,
-// processes incoming orders, and emits trades + execution reports.
-//
-// In Optiq, this would be a LogicalCoreActor with:
-//   - RecoveryProxy::Cause for persisting each incoming event
-//   - IACA Cause/Effect chain for producing IA messages
-//   - Effects gated by Master/Mirror role
-//
-// Here we implement the core matching with simplified recovery hooks.
-// ════════════════════════════════════════════════════════════════════
-
 #include "engine/SimplxShim.hpp"
 #include "common/Book.hpp"
 #include "actors/Events.hpp"
+#include "persistence/KafkaBus.hpp"
 #include <optional>
 
 namespace eunex {
@@ -30,7 +14,8 @@ public:
     MECoreActor(SymbolIndex_t symbolIdx,
                    const tredzone::ActorId& oeGatewayId,
                    const tredzone::ActorId& marketDataId,
-                   const tredzone::ActorId& clearingHouseId = tredzone::ActorId{});
+                   const tredzone::ActorId& clearingHouseId = tredzone::ActorId{},
+                   KafkaBus* kafkaBus = nullptr);
 
     void onEvent(const NewOrderEvent& event);
     void onEvent(const CancelOrderEvent& event);
@@ -41,6 +26,7 @@ private:
     tredzone::Actor::Event::Pipe oePipe_;
     tredzone::Actor::Event::Pipe mdPipe_;
     std::optional<tredzone::Actor::Event::Pipe> chPipe_;
+    KafkaBus* kafka_ = nullptr;
 
     void publishBookUpdate();
 };

src/main.cpp

@@ -21,11 +21,13 @@
 #include "actors/ClearingHouseActor.hpp"
 #include "actors/FIXAcceptorActor.hpp"
 #include "actors/AITraderActor.hpp"
+#include "persistence/KafkaBus.hpp"
 #include <iostream>
 #include <thread>
 #include <chrono>
 #include <csignal>
 #include <atomic>
+#include <cstdlib>
 
 using namespace tredzone;
 using namespace eunex;
@@ -43,6 +45,15 @@ int main() {
     std::signal(SIGINT, signalHandler);
     std::signal(SIGTERM, signalHandler);
 
+    // --Kafka Bus (optional) --------------------------------------
+    std::unique_ptr<KafkaBus> kafkaBus;
+    const char* kafkaBrokers = std::getenv("EUNEX_KAFKA_BROKERS");
+    if (kafkaBrokers && kafkaBrokers[0] != '\0') {
+        KafkaBusConfig cfg;
+        cfg.brokers = kafkaBrokers;
+        kafkaBus = std::make_unique<KafkaBus>(cfg);
+    }
+
     // --Symbol definitions ----------------------------------------
     constexpr SymbolIndex_t SYM_AAPL  = 1;
     constexpr SymbolIndex_t SYM_MSFT  = 2;
@@ -72,14 +83,15 @@ int main() {
     }
 
     // --Core 1: Order Books (per symbol) --------------------------
+    KafkaBus* kb = kafkaBus.get();
     auto bookAAPL = std::make_unique<MECoreActor>(
-        SYM_AAPL, oeGateway->getActorId(), mdActor->getActorId(), chActor->getActorId());
+        SYM_AAPL, oeGateway->getActorId(), mdActor->getActorId(), chActor->getActorId(), kb);
     auto bookMSFT = std::make_unique<MECoreActor>(
-        SYM_MSFT, oeGateway->getActorId(), mdActor->getActorId(), chActor->getActorId());
+        SYM_MSFT, oeGateway->getActorId(), mdActor->getActorId(), chActor->getActorId(), kb);
     auto bookGOOGL = std::make_unique<MECoreActor>(
-        SYM_GOOGL, oeGateway->getActorId(), mdActor->getActorId(), chActor->getActorId());
+        SYM_GOOGL, oeGateway->getActorId(), mdActor->getActorId(), chActor->getActorId(), kb);
     auto bookEURO50 = std::make_unique<MECoreActor>(
-        SYM_EURO50, oeGateway->getActorId(), mdActor->getActorId(), chActor->getActorId());
+        SYM_EURO50, oeGateway->getActorId(), mdActor->getActorId(), chActor->getActorId(), kb);
 
     oeGateway->mapSymbol(SYM_AAPL, bookAAPL->getActorId());
     oeGateway->mapSymbol(SYM_MSFT, bookMSFT->getActorId());
@@ -111,7 +123,13 @@ int main() {
     std::cout << "Services:\n";
     std::cout << "  FIX Gateway:  TCP port 9001\n";
     std::cout << "  AI Traders:   10 members (MBR01-MBR10)\n";
-    std::cout << "  Symbols:      AAPL, MSFT, GOOGL, EURO50\n\n";
+    std::cout << "  Symbols:      AAPL, MSFT, GOOGL, EURO50\n";
+    if (kafkaBus && kafkaBus->isConnected()) {
+        std::cout << "  Kafka Bus:    " << kafkaBrokers << " (connected)\n";
+    } else {
+        std::cout << "  Kafka Bus:    disabled (set EUNEX_KAFKA_BROKERS to enable)\n";
+    }
+    std::cout << "\n";
 
     // --Seed initial orders for AI to have market data ------------
     std::cout << "Seeding initial order book...\n";
@@ -179,6 +197,11 @@ int main() {
             if (fixGateway->isRunning()) {
                 std::cout << "  FIX clients: " << fixGateway->clientCount() << "\n";
             }
+            if (kafkaBus && kafkaBus->isConnected()) {
+                std::cout << "  Kafka: orders=" << kafkaBus->orderCount()
+                          << " trades=" << kafkaBus->tradeCount()
+                          << " md=" << kafkaBus->mdCount() << "\n";
+            }
             std::cout << "\n";
         }
 
@@ -210,6 +233,14 @@ int main() {
     }
 
     std::cout << "\nTrades processed: " << mdActor->getRecentTrades().size() << "\n";
+
+    if (kafkaBus) {
+        kafkaBus->flush();
+        std::cout << "Kafka totals: orders=" << kafkaBus->orderCount()
+                  << " trades=" << kafkaBus->tradeCount()
+                  << " md=" << kafkaBus->mdCount() << "\n";
+    }
+
     std::cout << "===========================================\n";
     std::cout << "  Engine stopped.\n";
     return 0;

src/persistence/KafkaBus.hpp

@@ -0,0 +1,156 @@
+#pragma once
+// ====================================================================
+// KafkaBus -- Multi-topic Kafka publisher for the EuNEx event bus
+//
+// Optiq equivalent: Kafka Bus (KFK) between ME and downstream
+// consumers (MDG, PTB, Clearing, IDS, SATURN).
+//
+// Topics:
+//   eunex.orders           -- incoming orders (NewOrder, Cancel, Modify)
+//   eunex.trades           -- executed trades
+//   eunex.market-data      -- BBO / depth snapshots
+//   eunex.recovery.fragments -- recovery fragments (Master/Mirror)
+//
+// When EUNEX_USE_KAFKA is OFF, KafkaBus is replaced by a no-op stub
+// so actors compile and run without librdkafka.
+// ====================================================================
+
+#include "common/Types.hpp"
+#include <string>
+#include <cstring>
+#include <vector>
+#include <iostream>
+#include <atomic>
+
+#ifdef EUNEX_USE_KAFKA
+#include <librdkafka/rdkafkacpp.h>
+#endif
+
+namespace eunex {
+
+struct KafkaBusConfig {
+    std::string brokers         = "localhost:9092";
+    std::string ordersTopic     = "eunex.orders";
+    std::string tradesTopic     = "eunex.trades";
+    std::string marketDataTopic = "eunex.market-data";
+    std::string recoveryTopic   = "eunex.recovery.fragments";
+    int         flushTimeoutMs  = 5000;
+};
+
+#ifdef EUNEX_USE_KAFKA
+
+class KafkaBus {
+public:
+    explicit KafkaBus(const KafkaBusConfig& cfg) : config_(cfg) {
+        std::string errstr;
+        auto conf = std::unique_ptr<RdKafka::Conf>(
+            RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
+        conf->set("bootstrap.servers", config_.brokers, errstr);
+        conf->set("queue.buffering.max.messages", "100000", errstr);
+        conf->set("linger.ms", "5", errstr);
+
+        producer_.reset(RdKafka::Producer::create(conf.get(), errstr));
+        if (!producer_) {
+            std::cerr << "[KafkaBus] producer creation failed: " << errstr << "\n";
+            return;
+        }
+        connected_ = true;
+        std::cout << "  Kafka Bus:    " << config_.brokers << "\n";
+    }
+
+    ~KafkaBus() {
+        if (producer_) producer_->flush(config_.flushTimeoutMs);
+    }
+
+    void publishTrade(const Trade& trade) {
+        std::string key = std::to_string(trade.symbolIdx);
+        publish(config_.tradesTopic, key,
+                reinterpret_cast<const char*>(&trade), sizeof(Trade));
+        tradeCount_.fetch_add(1);
+    }
+
+    void publishOrder(const Order& order) {
+        std::string key = std::to_string(order.symbolIdx);
+        publish(config_.ordersTopic, key,
+                reinterpret_cast<const char*>(&order), sizeof(Order));
+        orderCount_.fetch_add(1);
+    }
+
+    void publishMarketData(SymbolIndex_t sym,
+                            Price_t bestBid, Price_t bestAsk,
+                            Quantity_t bidQty, Quantity_t askQty) {
+        struct MDSnapshot {
+            SymbolIndex_t sym;
+            Price_t bestBid;
+            Price_t bestAsk;
+            Quantity_t bidQty;
+            Quantity_t askQty;
+            Timestamp_ns ts;
+        };
+        MDSnapshot snap{sym, bestBid, bestAsk, bidQty, askQty, nowNs()};
+        std::string key = std::to_string(sym);
+        publish(config_.marketDataTopic, key,
+                reinterpret_cast<const char*>(&snap), sizeof(snap));
+        mdCount_.fetch_add(1);
+    }
+
+    void publishRecoveryFragment(const void* data, size_t len,
+                                  uint16_t originId, uint32_t originKey) {
+        std::string key = std::to_string(originId) + ":" + std::to_string(originKey);
+        publish(config_.recoveryTopic, key,
+                reinterpret_cast<const char*>(data), len);
+    }
+
+    bool isConnected() const { return connected_; }
+    size_t tradeCount() const { return tradeCount_.load(); }
+    size_t orderCount() const { return orderCount_.load(); }
+    size_t mdCount() const { return mdCount_.load(); }
+
+    void flush() {
+        if (producer_) producer_->flush(config_.flushTimeoutMs);
+    }
+
+private:
+    KafkaBusConfig config_;
+    std::unique_ptr<RdKafka::Producer> producer_;
+    bool connected_ = false;
+    std::atomic<size_t> tradeCount_{0};
+    std::atomic<size_t> orderCount_{0};
+    std::atomic<size_t> mdCount_{0};
+
+    void publish(const std::string& topic, const std::string& key,
+                 const char* data, size_t len) {
+        if (!producer_) return;
+        RdKafka::ErrorCode err = producer_->produce(
+            topic, RdKafka::Topic::PARTITION_UA,
+            RdKafka::Producer::RK_MSG_COPY,
+            const_cast<char*>(data), len,
+            key.data(), key.size(),
+            0, nullptr);
+        if (err != RdKafka::ERR_NO_ERROR) {
+            std::cerr << "[KafkaBus] produce to " << topic
+                      << " failed: " << RdKafka::err2str(err) << "\n";
+        }
+        producer_->poll(0);
+    }
+};
+
+#else // No Kafka -- stub
+
+class KafkaBus {
+public:
+    explicit KafkaBus(const KafkaBusConfig&) {}
+    void publishTrade(const Trade&) {}
+    void publishOrder(const Order&) {}
+    void publishMarketData(SymbolIndex_t, Price_t, Price_t, Quantity_t, Quantity_t) {}
+    void publishRecoveryFragment(const void*, size_t, uint16_t, uint32_t) {}
+    bool isConnected() const { return false; }
+    size_t tradeCount() const { return 0; }
+    size_t orderCount() const { return 0; }
+    size_t mdCount() const { return 0; }
+    void flush() {}
+};
+
+#endif // EUNEX_USE_KAFKA
+
+} // namespace eunex