RedpandaStore — Redpanda streaming (Kafka-compatible)

RedpandaStore — Redpanda streaming (Kafka-compatible) Overview Redpanda is a streaming data platform that is wire-compatible with the Apache Kafka protocol — it speaks the exact same protocol, so this store reuses the kafkajs driver unchanged. Like Kafka, it has no query language and does not fit the SQL-shaped Dialect interface, so RedpandaStore implements the minimal NoSqlStore marker interface (src/nosql/store.ts) — connection lifecycle plus a getClient() escape hatch — and exposes Redpanda's real operations (produce / topic admin / subscribe) directly rather than a query(sql) shape. Because Redpanda speaks the Kafka protocol, this store is a mirror of KafkaStore: same kafkajs client, same producer/consumer/admin surface. Identity: Property | Value | --------- | ------------ | name | 'redpanda' | library | 'kafkajs' | On connect(), the store builds a kafkajs Kafka instance, creates a single shared producer, and connects it. Lazy loading — not a hard dependency kafkajs is not a hard dependency of this package. The driver is loaded lazily via require('kafkajs') inside connect(), rather than a top-level import. Importing this module therefore does not require the driver to be installed — it is only needed when a Redpanda store is actually connected against a real cluster. Injected client RedpandaStoreOptions accepts a pre-built client (a kafkajs Kafka instance, or a compatible mock). When provided, connect() uses it directly and does not require('kafkajs'). This is how the test suite injects a mock client (no driver, no network), and how callers can supply a custom-configured Kafka. Connection Build a store from connection options and call connect(): RedpandaStoreOptions (mirrors kafkajs's Kafka constructor options): Option | Type | Purpose | ---------- | ----------------------------------- | -------------------------------------------------------------------------------- | clientId | string | Client identifier reported to the broker. | brokers | string[] | Seed broker list, e.g. ['localhost:9092']. Defaults to [] when omitted. | ssl | boolean \| Record<string, unknown> | Enable TLS, or pass a kafkajs TLS options object. | sasl | RedpandaSaslOptions | SASL authentication configuration ({ mechanism, username, password, ... }). | client | any | A pre-built kafkajs Kafka instance. When set, the driver is not required. | connect() is idempotent: if already connected with a live client, it returns immediately. Connection failures are wrapped in a ConnectionError (an ECONNREFUSED message is normalized to 'Connection refused'), and connection state is reset to disconnected. Injected-client form Supply your own client (or a mock) to bypass driver-based client construction: Methods Using any operation before connect() (or after disconnect()) throws a ConnectionError via the internal requireClient() / requireProducer() guards. Operation failures are wrapped in a DatabaseError (preserving the original error). Lifecycle Method | Signature | Behavior | ------------- | ----------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------- | connect | connect(): Promise<void> | Uses an injected client if provided, otherwise lazy-requires kafkajs and builds a Kafka({ clientId, brokers, ssl, sasl }). Then creates and connects a shared producer. Idempotent. Wraps failures in ConnectionError. | disconnect | disconnect(): Promise<void> | Best-effort disconnects every tracked consumer first, then the producer, then clears the client and connection state. | isConnected | isConnected(): boolean | true only when connected and a client is present. | getClient | getClient(): any | Returns the underlying kafkajs Kafka client for operations not wrapped here. Throws ConnectionError if not connected. | Producing Method | Signature | Behavior | --------- | ----------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------ | produce | produce(topic: string, messages: RedpandaMessage[]): Promise<any> | Sends one or more messages to topic via the shared producer. Returns kafkajs's record metadata (per-partition base offsets). | Topic administration Both admin methods open a short-lived admin connection, do their work, and disconnect the admin in a finally (best-effort). Method | Signature | Behavior | ------------- | -------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------- | createTopic | createTopic(topic: string, opts?: CreateTopicOptions): Promise<boolean> | Creates topic via a temporary admin. Returns kafkajs's boolean: true if newly created, false if it already existed. | listTopics | listTopics(): Promise<string[]> | Returns all topic names known to the cluster via a temporary admin. | Consuming Method | Signature | Behavior | ----------- | --------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | subscribe | subscribe(topics: string \| string[], handler: (payload: any) => Promise<void> \| void, options?: SubscribeOptions): Promise<any> | Creates a new consumer in group options.groupId (a generated ${clientId ?? 'redpanda'}-group-${Date.now()} id is used if omitted), connects it, subscribes to topics (a single string is wrapped in an array; fromBeginning defaults to false), and runs it with { eachMessage: handler }. On failure, the half-set-up consumer is disconnected before the error is re-thrown. The consumer is tracked so disconnect() stops it. Returns the created kafkajs consumer. | The handler receives kafkajs's eachMessage payload ({ topic, partition, message }). Example Verification status Unit / mock-verified only. The tests in tests/nosql/redpanda.test.ts are fully mock-driven: a fake Kafka instance whose producer()/consumer()/admin() return spy-backed sub-clients is injected via RedpandaStoreOptions.client. The real kafkajs package is not installed, and there is no live cluster and no network in the test run — connect() never reaches the require. What this proves: connect() creates and connects a single producer via the injected client, and is idempotent (producer connected exactly once). produce() calls producer.send({ topic, messages }) and returns the record metadata. createTopic() / listTopics() connect an admin, perform the operation with the expected arguments (topic + partitions/replication), and disconnect the admin. subscribe() wires the consumer with the right groupId, topic list (single string wrapped to an array), fromBeginning, and { eachMessage: handler }; generates a groupId containing the clientId when none is given; and disconnects the consumer on subscribe failure. Lifecycle: connecting via an injected client without loading the driver, disconnect() disconnecting tracked consumers and the producer, and ConnectionError from getClient() / operations before connect(). Error handling: ECONNREFUSED at connect wrapped in ConnectionError; produce/subscribe failures wrapped so the message matches /produce.failed/i / /subscribe.failed/i. What this does not prove: live execution against a real Redpanda (or Kafka) cluster. The call shapes are verified against the kafkajs API contract, but end-to-end streaming over the wire has not been exercised here. Related reading All data stores — the full catalogue, grouped by purpose Database types — where this sits among the 22 categories