class KafkaBroker
implements MessageBrokerAdapter
Since 0.1.0

Kafka message broker implementation.

Constructors

KafkaBroker(
runtime: IRuntimeServices,
serializer: ISerializer,
options?: KafkaOptions
)

Creates a new Kafka broker.

Parameters
runtime: IRuntimeServices

Runtime services for uuid, timestamps, and timers

serializer: ISerializer

Serializer for message payloads

optional
options: KafkaOptions

Kafka connection and configuration options

Methods

Since 0.1.0
connect(): Promise<void>

Connects to Kafka and creates producer.

Since 0.1.0
disconnect(): Promise<void>

Disconnects from Kafka.

Since 0.1.0
isHealthy(): Promise<boolean>

Boolean port member (M70c): false only when positively unreachable.

Since 0.1.0
isReady(): boolean

Checks if the broker is connected.

Since 0.1.0
publish<T>(
topic: string,
message: T
): Promise<void>

Publishes a message to a topic.

publishWithHeaders<T>(
topic: string,
message: T,
headers: Readonly<Record<string, string>>
): Promise<void>

Publishes a message with framework-owned transport headers. @internal

Since 0.1.0
reachability(): Promise<boolean | undefined>

Tri-state backend reachability (M70c).

kafkajs retries internally, so the broker runs the supervisor in observe mode: the producer's producer.disconnect/producer.connect events (the VALUES of kafkajs's producer.events.DISCONNECT / producer.events.CONNECT — the uppercase KEYS are not accepted listener names and kafkajs throws for them, X28-1) mark the fault window (consumer.crash with restart: false is terminal and surfaces as a producer.disconnect the client does not recover from). false while the window is active, true otherwise, undefined when the producer exposes no event surface (a minimal fake) — the indicator then reports reachable: 'unknown'.

Since 0.1.0
request<TReq, TRes>(
topic: string,
message: TReq,
options?: RequestOptions
): Promise<TRes>

Sends a request and awaits its single correlated reply.

Replies arrive on the shared reply topic (KafkaOptions.replyTopic, default 'messaging.replies'), which must exist — this broker creates no topics, because IKafkaFactory exposes no admin surface. Either pre-create it or enable auto.create.topics.enable; otherwise the underlying producer error surfaces from this call rather than hanging until the timeout.

requestWithHeaders<TReq, TRes>(
topic: string,
message: TReq,
headers: Readonly<Record<string, string>>,
options?: RequestOptions
): Promise<TRes>

Sends request-reply traffic with framework-owned headers. @internal

Since 0.1.0
respond<TReq, TRes>(
topic: string,
handler: RequestHandler<TReq, TRes>,
): Promise<ISubscription>

Registers a responder for a request topic. The handler's resolved value is sent back to the caller, correlated to the originating request.

Since 0.1.0
subscribe<T>(
topic: string,
handler: MessageHandler<T>,
): Promise<ISubscription>

Subscribes to a topic using a consumer group.

subscribeWithHeaders<T>(
topic: string,
handler: MessageHandler<T>,
): Promise<ISubscription>

Subscribes through the header-aware internal path. @internal

Usage

import { KafkaBroker } from "messaging-plugin/src/index.ts";