KafkaStore (Apache Kafka)

KafkaStore (Apache Kafka) Reference for the KafkaStore NoSQL store, which wraps a kafkajs client and exposes Apache Kafka's core operations. Source: kafka · Tests: kafka Overview Apache Kafka is a distributed, partitioned, append-only commit log used as a message broker / event-streaming platform. It has no query language and does not fit the SQL-shaped Dialect interface, so KafkaStore implements the minimal NoSqlStore marker interface (store) instead. Rather than forcing everything into a query(sql) shape, it exposes Kafka's real operations: producing to topics, creating/listing topics via the admin client, and subscribing consumers to topics. Identifying properties: The kafkajs driver is not a hard dependency. It is lazy-loaded via require('kafkajs') inside connect(), so importing this module does not require the driver to be installed unless a real connection is made. A pre-built kafkajs Kafka instance can also be injected via the client option, in which case connect() uses it directly and never calls require('kafkajs'). KafkaStore satisfies the NoSqlStore contract: Connection Options These options mirror the kafkajs Kafka constructor. brokers is the list of host:port seed brokers (defaults to [] if omitted). Real connection connect() builds a Kafka client from the options (passing clientId, brokers, ssl, and sasl straight through to kafkajs), then creates a producer and connects it. After a successful call, isConnected() returns true. connect() is idempotent: if already connected with a live client, it returns immediately without reconnecting the producer. If connection fails, the client, producer, and connected flag are all reset and a ConnectionError is thrown. An ECONNREFUSED error is reported as "Connection refused"; any other error becomes "Unable to connect to Kafka: <message>". Injected client Supply a pre-built Kafka instance to skip the require('kafkajs') call entirely. This is how the test suite injects a mock: Methods Lifecycle connect(): Promise<void> Builds (or adopts the injected) Kafka client, then creates and connects the producer. Idempotent when already connected. Throws ConnectionError on failure (see Real connection). disconnect(): Promise<void> Best-effort teardown. Stops all consumers created via subscribe() first (so they stop pulling before the producer/client go), then disconnects the producer, then clears the client and sets connected to false. Individual disconnect failures are swallowed so teardown of the remaining resources still proceeds. isConnected(): boolean Returns true only when the store is marked connected and the client is non-null. getClient(): any Returns the underlying kafkajs Kafka client for anything not wrapped here. Throws ConnectionError('Not connected to Kafka') if there is no client. Producing produce(topic: string, messages: KafkaMessage[]): Promise<any> Produces one or more messages to topic by calling producer.send({ topic, messages }). Returns kafkajs's record metadata (per-partition base offsets). Requires a connected producer, otherwise throws ConnectionError('Not connected to Kafka'). A failed send is wrapped in a DatabaseError with the message Kafka produce to "<topic>" failed: <cause>. Topic administration createTopic(topic: string, opts?: CreateTopicOptions): Promise<boolean> Creates topic via a short-lived admin connection: it calls client.admin(), connects, calls admin.createTopics(...) with a single topic entry, and disconnects the admin in a finally block (best-effort). Returns kafkajs's boolean indicating whether the topic was newly created (false if it already existed). Requires a client (throws ConnectionError otherwise). Failures are wrapped in DatabaseError with Kafka createTopic "<topic>" failed: <cause>. listTopics(): Promise<string[]> Lists all topic names known to the cluster, also via a short-lived admin connection (connect → admin.listTopics() → disconnect in finally). Requires a client. Failures are wrapped in DatabaseError with Kafka listTopics failed: <cause>. Consuming subscribe(topics, handler, options?): Promise<any> Creates a new consumer and subscribes it to topics: 1. Normalizes topics to an array (a single string is accepted). 2. Resolves the group id from options.groupId, or generates one as ${clientId ?? 'kafka'}-group-${Date.now()} when omitted. 3. Creates the consumer with client.consumer({ groupId }), connects it, and calls consumer.subscribe({ topics, fromBeginning }) (fromBeginning defaults to false). 4. Registers the handler via consumer.run({ eachMessage: handler }). The handler receives kafkajs's eachMessage payload ({ topic, partition, message }). 5. Tracks the consumer internally so disconnect() stops it, and returns the created consumer. Requires a client (throws ConnectionError before connect()). If any step fails, the half-set-up consumer is disconnected to avoid leaking its connection, and the error is wrapped in DatabaseError with Kafka subscribe to "<topics>" failed: <cause>. Example Verification status Unit / mock-verified only. The kafkajs driver is not installed in this repository. Every test in kafka injects a mock Kafka instance via the client constructor option, so connect() never requires the real driver and no network I/O occurs. The mock's producer/consumer/admin methods are Jest spies, which lets the tests assert the exact calls KafkaStore makes against the kafkajs API shape. Behaviors covered by the mock-based tests: name / library reporting, and connecting the producer via the injected client. getClient() returning the underlying client, and throwing ConnectionError when not connected. connect() idempotency and clean disconnect() (with isConnected() flipping to false). Wrapping connection failures (e.g. ECONNREFUSED) in ConnectionError. produce() sending messages and wrapping send failures in DatabaseError. createTopic() / listTopics() opening and closing a short-lived admin connection. subscribe() wiring groupId, topics, fromBeginning, and the eachMessage handler; single-string topic normalization; generated group ids; tracked-consumer teardown on disconnect(); and cleanup + DatabaseError on subscribe failure. produce() / createTopic() / subscribe() throwing ConnectionError when called before connect(). No integration testing against a live Kafka broker has been performed. Related reading All data stores — the full catalogue, grouped by purpose Database types — where this sits among the 22 categories