Prosody: JavaScript Bindings for Kafka

Prosody offers JavaScript bindings to the Prosody Kafka client, providing features for message production and consumption, including configurable retry mechanisms, failure handling strategies, and integrated OpenTelemetry support for distributed tracing.

Features

• Kafka Consumer: Per-key ordering with cross-key concurrency, offset management, consumer groups
• Kafka Producer: Idempotent delivery with configurable retries
• Timer System: Persistent scheduled execution backed by Cassandra or in-memory store
• Quality of Service: Fair scheduling limits concurrency and prevents failures from starving fresh traffic. Pipeline mode adds deferred retry and monopolization detection
• Distributed Tracing: OpenTelemetry integration for tracing message flow across services
• Backpressure: Pauses partitions when handlers fall behind
• Mocking: In-memory Kafka broker for tests (mock: true)
• Failure Handling: Pipeline (retry forever), Low-Latency (dead letter), Best-Effort (log and skip)

Installation

npm install @prosody-events/prosody

The package ships TypeScript declarations for the public API. EventHandler<P, R> carries application payload and response types through each handler method. Keyed-state definitions carry their item types through context.state(). Unparameterized API types default to JsonValue. See the strict TypeScript examples for IDE-ready projects compiled by the repository typecheck.

Quick Start

Run each example in an asynchronous function unless the example defines one.

const { ProsodyClient } = require("@prosody-events/prosody");

async function main() {
  // Initialize the client with Kafka bootstrap servers, consumer group, and topics
  const client = await ProsodyClient.create({
    // Bootstrap servers should normally be set using the PROSODY_BOOTSTRAP_SERVERS environment variable
    bootstrapServers: "localhost:9092",

    // To allow loopbacks, sourceSystem must be different from groupId.
    // Normally, sourceSystem is omitted and defaults to groupId.
    sourceSystem: "my-application-source",

    // groupId should be set to the name of your application
    groupId: "my-consumer-group",

    // Topics the client should subscribe to
    subscribedTopics: "my-topic",
  });

  // Define a message handler
  const messageHandler = {
    onExcise: async (context, message, signal) => {
      console.log(`Excise key: ${message.key}`);
      await context.clearScheduled();
      return null;
    },

    onMessage: async (context, message, signal) => {
      // Process the received message
      console.log(`Received message: ${JSON.stringify(message)}`);

      // Schedule a timer for delayed processing
      if (message.payload.scheduleFollowup) {
        const followupTime = new Date(Date.now() + 30000); // 30 seconds from now
        await context.schedule(followupTime);
      }
      return null;
    },

    onTimer: async (context, timer, signal) => {
      // Handle timer firing
      console.log(`Timer fired for key: ${timer.key} at ${timer.time}`);
    },
  };

  // Subscribe to messages using the message handler
  await client.subscribe(messageHandler);

  // Send a message to a topic
  await client.send("my-topic", "message-key", { content: "Hello, Kafka!" });
  await client.excise("my-topic", "obsolete-key");

  // Shut down all client services when done
  await client.shutdown();
}

main().catch(console.error);

Excise records

Applications can copy event data into keyed state and external stores. A regulatory or contractual deletion must remove every copy for one key.

An excise record carries this deletion command. Kafka encodes the command as a key with no payload. During topic compaction, Kafka deletes earlier values for the key. Call excise(topic, key) to send the record. Prosody routes the record to onExcise. The handler must delete all consumer-owned data for the key.

Each handler must implement onMessage, onExcise, and onTimer. Subscription fails before consumption if a method is missing.

If an excise record is a request, return a response from onExcise. Prosody uses this response as the subsystem result.

Architecture

Prosody enables efficient, parallel processing of Kafka messages while maintaining order for messages with the same key:

• Partition-Level Parallelism: Separate management of each Kafka partition
• Key-Based Queuing: Ordered processing for each key within a partition
• Concurrent Processing: Simultaneous processing of different keys
• Backpressure Management: Pause consumption from backed-up partitions

Quality of Service

All modes use fair scheduling to limit concurrency and distribute execution time. Pipeline mode adds deferred retry and monopolization detection.

Fair Scheduling (All Modes)

The scheduler controls which message runs next and how many run concurrently.

Virtual Time (VT): Each key accumulates VT equal to its handler execution time. The scheduler picks the key with the lowest VT. A key that runs for 500ms accumulates 500ms of VT; a key that hasn't run recently has zero VT and gets priority.

Two-Class Split: Normal messages and failure retries have separate VT pools. The scheduler allocates execution time between them (default: 70% normal, 30% failure). During a failure spike, retries get at most 30% of execution time—fresh messages continue processing.

Starvation Prevention: Tasks receive a quadratic priority boost based on wait time. A task waiting 2 minutes (configurable) gets maximum boost, overriding VT disadvantage.

Deferred Retry (Pipeline Mode)

Moves failing keys to timer-based retry so the partition can continue processing other keys.

On transient failure: store the message offset in Cassandra, schedule a timer, return success. The partition advances. When the timer fires, reload the message from Kafka and retry.

// Configure defer behavior
const client = await ProsodyClient.create({
  groupId: "my-consumer-group",
  subscribedTopics: "my-topic",
  deferEnabled: true, // Enable deferral (default: true)
  deferBaseMs: 1000, // Wait 1s before first retry
  deferMaxDelayMs: 86400000, // Cap at 24 hours
  deferFailureThreshold: 0.9, // Disable when >90% failing
});

Failure Rate Gating: When >90% of recent messages fail, deferral disables. The retry middleware blocks the partition, applying backpressure upstream.

Monopolization Detection (Pipeline Mode)

Rejects keys that consume too much execution time.

The middleware tracks per-key execution time in 5-minute rolling windows. Keys exceeding 90% of window time are rejected with a transient error, routing them through defer.

// Configure monopolization detection
const client = await ProsodyClient.create({
  groupId: "my-consumer-group",
  subscribedTopics: "my-topic",
  monopolizationEnabled: true, // Enable detection (default: true)
  monopolizationThreshold: 0.9, // Reject keys using >90% of window
  monopolizationWindowMs: 300000, // 5-minute window
});

Handler Timeout

Handlers are automatically cancelled if they exceed a deadline:

const client = await ProsodyClient.create({
  groupId: "my-consumer-group",
  subscribedTopics: "my-topic",
  timeoutMs: 30000, // Cancel after 30 seconds
  stallThresholdMs: 60000, // Report unhealthy after 60 seconds
});

When a handler times out, context.shouldCancel becomes true and context.onCancel() resolves. The handler should exit promptly. If not specified, timeout defaults to 80% of stallThresholdMs.

Configuration

For the complete configuration reference, see CONFIGURATION.md.

Constructor options take precedence. Unset options use environment variables, then library defaults.

Client construction is asynchronous. Replace new ProsodyClient(config) with await ProsodyClient.create(config).

Liveness and Readiness Probes

Prosody includes a built-in probe server for consumer-based applications that provides health check endpoints. The probe server is tied to the consumer's lifecycle and offers two main endpoints:

1. /readyz: A readiness probe that checks if any partitions are assigned to the consumer. Returns a success status only when the consumer has at least one partition assigned, indicating it's ready to process messages.

2. /livez: A liveness probe that checks if any partitions have stalled (haven't processed a message within a configured time threshold).

Configure the probe server using either the client constructor:

const client = await ProsodyClient.create({
  groupId: "my-consumer-group",
  subscribedTopics: "my-topic",
  probePort: 8000, // Set to null to disable
  stallThresholdMs: 15000, // 15 seconds before considering a partition stalled
});

Or via environment variables:

PROSODY_PROBE_PORT=8000  # Set to 'none' to disable
PROSODY_STALL_THRESHOLD=15s  # Default stall detection threshold

Important Notes

1. The probe server starts automatically when the consumer is subscribed and stops when unsubscribed.
2. A partition is considered "stalled" if it hasn't processed a message within the stallThreshold duration.
3. The stall threshold should be set based on your application's message processing latency and expected message frequency.
4. Setting the threshold too low might cause false positives, while setting it too high could delay detection of actual issues.
5. The probe server is only active when consuming messages (not for producer-only usage).

You can monitor the stall state programmatically using the client's properties:

// Get the number of partitions currently assigned to this consumer
const partitionCount = client.assignedPartitionCount;

// You can use these in your own health checks or monitoring
if (client.isStalled) {
  console.warn("Consumer has stalled partitions");
}

Subsystems

A consumer group ID identifies a set of processes that share records and the keyed state that the group owns. A subsystem can include one or more services and consumer groups. If callers use these IDs, a refactor can require changes to each caller.

A subsystem gives requests and published state one stable public name. Callers use this name instead of consumer group IDs. You can change its services and consumer groups without changing callers. Prosody uses the first response to a subsystem request. For each published-state read, it uses one consumer group that publishes the collection.

Requests

Kafka decouples producers from consumers, so a send does not return consumer results. This asynchronous model lets each service process records independently. Some operations must wait for consumer results before they continue. A request recovers synchrony for the caller while consumers continue asynchronous processing.

Send a request from a handler or other application code. The Prosody client does not need an active subscription. The result map uses canonical subsystem names as keys. Each value is a Success or Failure outcome. Use requestExcise to send an excise record and collect the same outcome type.

Do not rely on map order. The map contains one entry for each selected subsystem. A missing response becomes a timeout Failure; Prosody does not omit the subsystem. The request rejects for request-level failures, such as invalid input, a Kafka send failure, or shutdown. Do not await a request if the current consumer group must process it for the same key. That group cannot process it until the handler returns.

Message and excise handler return values become successful outcomes. Each return value must have a JSON representation.

Set subsystem to inventory on the client that subscribes this handler.

await client.subscribe({
  onMessage: async (_context, message) => ({ accepted: message.key }),
  onExcise: async (_context, message) => ({ accepted: message.key }),
  onTimer: async () => {},
});

Send the request:

const subsystems = ["inventory", "billing"];
const results = await client.request(
  "orders",
  "order-1",
  { type: "order.created" },
  { subsystems, timeoutMs: 2_000 },
);

for (const [subsystem, outcome] of results) {
  if (outcome.ok) console.log(`${subsystem}:`, outcome.value);
  else console.error(`${subsystem}: ${outcome.error.message}`);
}

The example can print these results:

inventory: { accepted: 'order-1' }
billing: no response arrived before the deadline

Each failure contains one typed response error.

Each response error has one message.

Advanced Usage

Pipeline Mode

All messages must be processed. Retries indefinitely. Uses defer and monopolization detection.

Middleware stack:

Kafka → Deduplication → Retry → Defer → Monopolization → Shutdown → Scheduler → Timeout → Telemetry → Handler

| Layer | Purpose | | -------------- | ------------------------------------------------- | | Deduplication | Skips messages whose ID was already processed | | Retry | Retries transient errors indefinitely | | Defer | Stores failing messages for timer-based retry | | Monopolization | Rejects keys exceeding execution time threshold | | Shutdown | Drains in-flight work on partition revocation | | Scheduler | Enforces concurrency limits and VT-based priority | | Timeout | Cancels handlers exceeding deadline | | Telemetry | Emits handler lifecycle events |

const client = await ProsodyClient.create({
  mode: Mode.Pipeline, // Default mode
  groupId: "my-consumer-group",
  subscribedTopics: "my-topic",
});

Low-Latency Mode

Tries a few times, then routes failures to a dead letter topic.

• Retries up to maxRetries times, then writes to failure topic
• Fair scheduling limits how much time retries consume
• Use when you need to keep moving and can reprocess failures later

const client = await ProsodyClient.create({
  mode: Mode.LowLatency,
  groupId: "my-consumer-group",
  subscribedTopics: "my-topic",
  failureTopic: "failed-messages", // Required for low-latency mode
  maxRetries: 3, // Retry up to 3 times after the initial attempt
});

Best-Effort Mode

Logs failures and moves on.

• No retries; failed messages are logged and committed
• Fair scheduling still enforces concurrency limits
• Use for development or when message loss is acceptable

const client = await ProsodyClient.create({
  mode: Mode.BestEffort,
  groupId: "my-consumer-group",
  subscribedTopics: "my-topic",
});

Event Type Filtering

Prosody supports filtering messages based on event type prefixes, allowing your consumer to process only specific types of events:

// Process only events with types starting with "user." or "account."
const client = await ProsodyClient.create({
  groupId: "my-consumer-group",
  subscribedTopics: "my-topic",
  allowedEvents: ["user.", "account."],
});

Or via environment variables:

PROSODY_ALLOWED_EVENTS=user.,account.

Matching Behavior

Prefixes must match exactly from the start of the event type:

✓ Matches:

• {"type": "user.created"} matches prefix user.
• {"type": "account.deleted"} matches prefix account.

✗ No Match:

• {"type": "admin.user.created"} doesn't match user.
• {"type": "my.account.deleted"} doesn't match account.
• {"type": "notification"} doesn't match any prefix

If no prefixes are configured, all messages are processed. Messages without a type field are always processed.

Source System Deduplication

Prosody prevents processing loops in distributed systems by tracking the source of each message:

// Consumer and producer in one application
const client = await ProsodyClient.create({
  groupId: "my-service",
  sourceSystem: "my-service-producer", // Must differ from groupId to allow loopbacks; defaults to groupId
  subscribedTopics: "my-topic",
});

Or via environment variable:

PROSODY_SOURCE_SYSTEM=my-service-producer

How It Works

1. Producers add a source-system header to all outgoing messages.
2. Consumers check this header on incoming messages.
3. If a message's source system matches the consumer's group ID, the message is skipped.

This prevents endless loops where a service consumes its own produced messages.

Message Deduplication

Prosody automatically deduplicates messages using the id field in their JSON payload. Consecutive messages with the same ID and key are processed only once.

Deduplication uses a two-tier approach:

• Global in-memory cache: A single cache shared across all partitions within the same consumer instance. Survives partition reassignments within the same process. Controlled by idempotenceCacheSize (default 8192).
• Cassandra-backed persistent store: Survives restarts and rebalances across instances. TTL controlled by idempotenceTtlS (default 7 days, i.e. 604800s).

Deduplication is always active. idempotenceCacheSize must be greater than 0; a value of 0 (via either the option or PROSODY_IDEMPOTENCE_CACHE_SIZE=0) is rejected when the client is constructed.

// Messages with IDs are deduplicated per key
await client.send("my-topic", "key1", {
  id: "msg-123", // Message will be processed
  content: "Hello!",
});

await client.send("my-topic", "key1", {
  id: "msg-123", // Message will be skipped (duplicate)
  content: "Hello again!",
});

await client.send("my-topic", "key2", {
  id: "msg-123", // Message will be processed (different key)
  content: "Hello!",
});

To invalidate all previously recorded dedup entries (forcing reprocessing of messages), change the version:

const client = await ProsodyClient.create({
  groupId: "my-consumer-group",
  subscribedTopics: "my-topic",
  idempotenceVersion: "2", // Changing this invalidates all previously recorded entries
});

Or via environment variable:

PROSODY_IDEMPOTENCE_VERSION=2

Timer Functionality

Prosody supports timer-based delayed execution within message handlers. When a timer fires, your handler's onTimer method will be called:

const messageHandler = {
  onMessage: async (context, message, signal) => {
    // Schedule a timer to fire in 30 seconds
    const futureTime = new Date(Date.now() + 30000);
    await context.schedule(futureTime);

    // Schedule multiple timers
    const oneMinute = new Date(Date.now() + 60000);
    const twoMinutes = new Date(Date.now() + 120000);
    await context.schedule(oneMinute);
    await context.schedule(twoMinutes);

    // Check what's scheduled
    const scheduled = await context.scheduled();
    console.log(`Scheduled timers: ${scheduled.length}`);
    return null;
  },

  onTimer: async (context, timer, signal) => {
    console.log("Timer fired!");
    console.log(`Key: ${timer.key}`);
    console.log(`Scheduled time: ${timer.time}`);
  },

  onExcise: async (context) => {
    await context.clearScheduled();
    return null;
  },
};

Timer Methods

The context provides timer scheduling methods that allow you to delay execution or implement timeout behavior:

• schedule(time): Schedules a timer to fire at the specified time
• clearAndSchedule(time): Clears all timers and schedules a new one
• unschedule(time): Removes a timer scheduled for the specified time
• clearScheduled(): Removes all scheduled timers
• scheduled(): Returns an array of all scheduled timer times

Timer Object

When a timer fires, the onTimer method receives a timer object with these properties:

• key (string): The entity key identifying what this timer belongs to
• time (Date): The time when this timer was scheduled to fire

Note: Timer precision is limited to seconds due to the underlying storage format. Sub-second precision in scheduled times will be rounded to the nearest second.

Timer Configuration

Timer functionality requires Cassandra for persistence unless running in mock mode. Configure Cassandra connection via environment variable:

PROSODY_CASSANDRA_NODES=localhost:9042  # Required for timer persistence

Or programmatically when creating the client:

const client = await ProsodyClient.create({
  bootstrapServers: "localhost:9092",
  groupId: "my-application",
  subscribedTopics: "my-topic",
  cassandraNodes: "localhost:9042", // Required unless mock: true
});

For testing, you can use mock mode to avoid Cassandra dependency:

// Mock mode for testing (timers work but aren't persisted)
const client = await ProsodyClient.create({
  bootstrapServers: "localhost:9092",
  groupId: "my-application",
  subscribedTopics: "my-topic",
  mock: true, // No Cassandra required in mock mode
});

Error Handling

Prosody classifies errors as transient (temporary, can be retried) or permanent (won't be resolved by retrying). By default, all errors are considered transient.

The error classes and decorators apply to onMessage, onExcise, and onTimer.

Using Decorators

If you're using TypeScript or a JavaScript environment that supports decorators, you can use the @permanent decorator to classify exceptions that should not be retried:

import { permanent, ProsodyClient } from "@prosody-events/prosody";

class MyHandler {
  @permanent(TypeError, AttributeError)
  async onMessage(context, message, signal) {
    // Your message handling logic here
    // TypeError and AttributeError will be treated as permanent
    // All other exceptions will be treated as transient (default behavior)
    return null;
  }

  async onExcise() {
    return null;
  }

  async onTimer() {}
}

const client = await ProsodyClient.create(config);
client.subscribe(new MyHandler());

Without Decorators

If you're not using decorators, you can still classify errors as permanent by throwing a PermanentError:

import { PermanentError, ProsodyClient } from "@prosody-events/prosody";

const messageHandler = {
  onMessage: async (context, message, signal) => {
    try {
      // Your message handling logic here
    } catch (error) {
      if (error instanceof TypeError || error instanceof AttributeError) {
        throw new PermanentError(error.message);
      }
      // All other exceptions will be treated as transient (default behavior)
      throw error;
    }
    return null;
  },
  onExcise: async () => null,
  onTimer: async () => {},
};

const client = await ProsodyClient.create(config);
client.subscribe(messageHandler);

Best Practices for Error Handling

• Use permanent errors for issues like malformed data or business logic violations.
• Use transient errors for temporary issues like network problems.
• Be cautious with permanent errors as they prevent retries and can result in data loss.
• Consider system reliability and data consistency when classifying errors.

Keyed State

Many stream transformations must reason across multiple events or timer firings. Windows, state machines, aggregates, and complex event processing all require state.

A Kafka key identifies an entity, such as a customer or order. Keyed state gives each key independent working state for these transformations. With Cassandra, the state survives restarts and partition reassignment.

Prosody selects the current message or timer key. It processes one event at a time for that key but can process other keys concurrently. By default, Prosody commits pending keyed-state changes only when the handler succeeds. If the handler returns an error, Prosody discards those changes.

Give most collections a time to live (TTL). Set the TTL beyond the longest timer or workflow that uses the collection. Omit it when state must remain for inactive keys.

A counter for each key

Declare each collection once. Register it on the client. In a handler, get the current key's state from the event context:

const COUNT = value<number>("count", { ttlSeconds: 30 * 24 * 60 * 60 });

const client = await ProsodyClient.create({
  ...config,
  stateCollections: [COUNT],
});

client.subscribe({
  async onMessage(context) {
    const count = context.state(COUNT);
    await count.set(((await count.get()) ?? 0) + 1);
    return null;
  },
  async onExcise(context) {
    await context.state(COUNT).clear();
    return null;
  },
  async onTimer() {},
});

Each Kafka key now has an independent counter. A counter expires when that key has no update for 30 days.

Window activity into one notification

This example sends the first event for a user immediately. It collects later events for five minutes and then sends one summary.

The user ID is the Kafka key. Each user has an independent window.

const WINDOW = value<boolean>("window", { ttlSeconds: 24 * 60 * 60 });
const PENDING = messageDeque<Activity>("pending", {
  capacity: 100,
  ttlSeconds: 24 * 60 * 60,
});

const handler = {
  async onMessage(context, message) {
    const window = context.state(WINDOW);
    const pending = context.state(PENDING);

    if (await window.get()) {
      await pending.push(message);
      return null;
    }

    await notify(message.key, [message]);
    await window.set(true);
    await context.clearAndSchedule(new Date(Date.now() + 5 * 60_000));
    return null;
  },

  async onTimer(context, timer) {
    const pending = context.state(PENDING);
    const batch: Message<Activity>[] = [];
    for await (const message of pending.values()) batch.push(message);

    if (batch.length > 0) await notify(timer.key, batch);
    await pending.clear();
    await context.state(WINDOW).clear();
  },
  async onExcise(context) {
    await context.state(PENDING).clear();
    await context.state(WINDOW).clear();
    await context.clearScheduled();
    return null;
  },
} satisfies EventHandler<Activity>;

See the complete, type-checked example for imports, types, client setup, and notify: examples/windowing.ts.

Why this works:

• Register both definitions in stateCollections before you subscribe. Keyed state uses Cassandra unless mock: true.
• Use clearAndSchedule, not schedule, so a retried event does not add another timer for the same key.
• capacity: 100 and the one-day TTL bound the saved backlog. Overflow drops the oldest message because this example only pushes.
• A messageDeque requires the original Kafka messages during the window. Use deque when topic retention or compaction cannot provide them.
• Prosody runs one handler at a time for each key, so a user's message and timer handlers cannot overlap.
• A notification is outside the state transaction. A retry can send it again. Use a stable operation ID to reject duplicate notifications.

Collections and handles

A definition sets a collection's durable name, kind, and options. Register it once. Pass it to context.state in a handler.

Do not reuse a durable name for a different collection kind or payload type. Create handles inside the handler. Do not retain handles or iterators.

| Collection | JSON payload | Kafka message | Main operations | | ------------------ | ------------ | ----------------- | -------------------------------------------------------------------- | | Value | value<T> | messageValue<P> | get, set, clear | | Ordered string map | map<V> | messageMap<P> | get, getMany, has, set, delete, entries, keys, clear | | Deque | deque<T> | messageDeque<P> | push, unshift, pop, shift, at, length, values, clear |

All operations are asynchronous. Map and deque scans are asynchronous iterables. A for await loop can stop early safely.

Map keys are strings. null and undefined mean absence. Do not store these values. Use clear() or delete().

When keyed-state changes become visible

By default, retries do not see pending state from a failed attempt. Reads in a handler see its earlier keyed-state writes. Prosody commits pending changes when the event succeeds and discards them when the handler throws.

This transaction applies only to keyed state. Some workflows need state changes before the handler ends, so each collection also provides explicit controls:

• readUncommitted: true persists keyed-state changes before Prosody records the event as complete. If the process stops between these steps, Prosody can process the same event again. The retry sees state changes from the earlier attempt. You must make these keyed-state changes idempotent. Each retry must produce the same state.
• commit() commits the collection's pending changes before the handler ends. A later handler failure does not remove them.
• rollback() discards pending changes since the last commit(). It cannot undo committed changes.

Published state

Some callers need only the current value for a key. They can accept a stale value or a race with a concurrent update.

Use topics and event sourcing when a consumer must process each state change in order. Use published state for direct, read-only lookup of persisted keyed state. The caller does not need to consume the owner's topics or maintain a separate lookup store.

Configure the subsystem name on each publisher. Enable publication on the collection definition. Register the definition on the Prosody client:

const CURRENT_ORDER = value("current-order", { published: true });
const owner = await ProsodyClient.create({
  ...config,
  subsystem: "checkout",
  stateCollections: [CURRENT_ORDER],
});

// The handler uses the key from its current event.
const currentOrder = context.state(CURRENT_ORDER);
await currentOrder.set({ sku: "book" });

Read published state from a handler or other application code. The Prosody client does not need an active subscription.

Use the subsystem and the same definition to open a reader:

const orderReader = await client.state("checkout", CURRENT_ORDER);
const currentOrder = await orderReader.get("customer-123");

The reader cannot see pending changes that exist only in a handler. It cannot change the collection. Each read takes an explicit key because no handler supplies one.

Map and deque readers fetch data in chunks. They do not load the complete collection before iteration starts.

The default cache window is five seconds. Set readCache: { ttlMs } to select a different window. Set readCache: false to bypass the cache.

To stop publication, deploy the definition with published: false. Keep the definition registered during that deployment. Keep the subsystem configured during that deployment.

OpenTelemetry Tracing

Prosody supports OpenTelemetry tracing, allowing you to monitor and analyze the performance of your Kafka-based applications. The library will emit traces using the OTLP protocol if the OTEL_EXPORTER_OTLP_ENDPOINT environment variable is defined.

Note: Prosody emits its own traces separately because it uses its own tracing runtime, as it would be expensive to send all traces to JavaScript.

Required Packages

To use OpenTelemetry tracing with Prosody, you need to install the following packages:

npm install @opentelemetry/api @opentelemetry/sdk-node @opentelemetry/exporter-trace-otlp-http

Initializing Tracing

To initialize tracing in your application:

const opentelemetry = require("@opentelemetry/api");
const { NodeSDK } = require("@opentelemetry/sdk-node");
const {
  OTLPTraceExporter,
} = require("@opentelemetry/exporter-trace-otlp-http");

const sdk = new NodeSDK({
  traceExporter: new OTLPTraceExporter(),
  serviceName: "my-service-name",
});

sdk.start();

// Creates a tracer from the global tracer provider
const tracer = opentelemetry.trace.getTracer("my-service-name");

Setting OpenTelemetry Environment Variables

Set the following standard OpenTelemetry environment variables:

OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317
OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf
OTEL_SERVICE_NAME=my-service-name

For more information on these and other OpenTelemetry environment variables, refer to the OpenTelemetry specification.

Call flushTelemetry() to export pending telemetry. Call shutdownTelemetry() before the process exits.

Using Tracing in Your Application

After initializing tracing, you can define spans in your application, and they will be properly propagated through Kafka:

const { ProsodyClient } = require("@prosody-events/prosody");
const opentelemetry = require("@opentelemetry/api");

const tracer = opentelemetry.trace.getTracer("my-service-name");

const client = await ProsodyClient.create({
  groupId: "my-consumer-group",
  subscribedTopics: "my-topic",
});

const messageHandler = {
  onMessage: async (context, message, signal) => {
    const span = tracer.startSpan("process-message");
    try {
      // Process the received message
      span.addEvent("message.received", {
        "message.payload": JSON.stringify(message),
      });
    } finally {
      span.end();
    }
    return null;
  },
  onExcise: async () => null,
  onTimer: async () => {},
};

client.subscribe(messageHandler);

Span Linking

By default, message execution spans use child (child-of relationship — the execution span is part of the same trace as the producer). Timer execution spans use follows_from (the execution span starts a new trace with a span link back to the scheduling span, since timer execution is causally related but not part of the same operation).

Both strategies are configurable via the messageSpans / PROSODY_MESSAGE_SPANS and timerSpans / PROSODY_TIMER_SPANS options. Accepted values: 'child', 'follows_from'.

Best Practices

Ensuring Idempotent Message Handlers

Idempotent message handlers are crucial for maintaining data consistency, fault tolerance, and scalability when working with distributed, event-based systems. They ensure that processing a message multiple times has the same effect as processing it once, which is essential for recovering from failures.

Strategies for achieving idempotence:

1. Natural Idempotence: Use inherently idempotent operations (e.g., setting a value in a key-value store).

2. Deduplication with Unique Identifiers:

• Kafka messages can be uniquely identified by their partition and offset.
• Before processing, check if the message has been handled before.
• Store processed message identifiers with an appropriate TTL.

3. Database Upserts: Use upsert operations for database writes.

4. Partition Offset Tracking:

• Store the latest processed offset for each partition.
• Only process messages with higher offsets than the last processed one.
• Critically, store these offsets transactionally with other state updates to ensure consistency.

5. Idempotency Keys for External APIs: Utilize idempotency keys when supported by external APIs.

6. Check-then-Act Pattern:

• For non-idempotent external systems, verify if an operation was previously completed before execution.
• Maintain a record of completed operations, keyed by a unique message identifier.

7. Saga Pattern:

• Implement a state machine in your database for multi-step operations.
• Each message advances the state machine, allowing for idempotent processing and easy failure recovery.
• Particularly useful for complex, distributed transactions across multiple services.

Application shutdown

A Prosody client runs a subscription, timers, and other services in the background. Before an application terminates, it must stop all client services. unsubscribe() stops only the active subscription.

Call shutdown() when the application terminates. It stops all client services and rejects new operations. Call unsubscribe() only when the application will use the client again. You do not need to call unsubscribe() before shutdown().

await client.shutdown();

Handle application shutdown:

const { ProsodyClient } = require("@prosody-events/prosody");

async function main() {
  const client = await ProsodyClient.create({
    groupId: "my-consumer-group",
    subscribedTopics: "my-topic",
  });

  const messageHandler = {
    onMessage: async (context, message, signal) => {
      // Process the message.
      return null;
    },
    onExcise: async () => null,
    onTimer: async () => {},
  };

  client.subscribe(messageHandler);

  // Resolve this promise after a shutdown signal.
  const shutdownPromise = new Promise((resolve) => {
    const shutdown = async (signal) => {
      console.log(`Received ${signal}. Client shutdown starts.`);
      await client.shutdown();
      resolve();
    };

    process.on("SIGTERM", () => shutdown("SIGTERM"));
    process.on("SIGINT", () => shutdown("SIGINT"));
    process.on("SIGHUP", () => shutdown("SIGHUP"));
  });

  // Wait for a shutdown signal.
  await shutdownPromise;
}

main().catch(console.error);

Handling Task Cancellation

Prosody cancels tasks during partition rebalancing or shutdown. How you handle cancellation is critical:

• Prosody interprets task success based on exception propagation.
• A task that exits without an exception is considered successful.
• Any exception signals task failure.

The library uses AbortSignals in both the send method and onMessage handler. It's crucial to pass this abort signal to any I/O operations, fetch calls, or database queries to ensure prompt task cancellation.

Best practices:

1. Exit promptly when cancelled to avoid rebalancing delays.
2. Use try/catch blocks to handle cancellation gracefully.
3. Use try/finally or equivalent constructs for clean resource handling.
4. Pass the AbortSignal to all async operations that support it.

Example of using AbortSignal in message processing:

const messageHandler = {
  onMessage: async (context, message, signal) => {
    // Pass the signal to fetch calls
    const response = await fetch("https://api.example.com", { signal });
    const data = await response.json();

    // Pass the signal to database operations
    await db.query(
      "INSERT INTO messages (payload) VALUES ($1)",
      [message.payload],
      { signal },
    );

    // Process the data...

    // Send a message, passing the abort signal
    await client.send("topic", "key", { data: "value" }, signal);
    return null;
  },
  onExcise: async () => null,
  onTimer: async () => {},
};

For the send method, note that while an abort signal will cause the method to return early, it may not cancel the message being sent, depending on when the abort is signaled. If the abort occurs after the message has been handed off to the Kafka client, the message may still be sent.

Failing to follow these practices can lead to:

• Slower message processing due to delayed rebalancing.
• Data loss from missed messages when cancellation errors are suppressed.
• Resource leaks if long-running operations aren't properly cancelled.

Release Process

Prosody uses an automated release process managed by GitHub Actions. Here's an overview of how releases are handled:

1. Trigger: The release process is triggered automatically on pushes to the main branch.

2. Release Please: The process starts with the "Release Please" action, which:

   • Analyzes commit messages since the last release.
   • Creates or updates a release pull request with changelog updates and version bumps.
   • When the PR is merged, it creates a GitHub release and a git tag.

3. Build Process: If a new release is created, the following build jobs are triggered:

   • Linux builds for x86_64 and aarch64 (glibc).
   • Windows build for x64.
   • macOS build for aarch64 (Apple Silicon).

4. Testing: The built binaries are tested on Linux (x86_64 and aarch64) with Node.js 24.

5. Artifact Upload: Each build job uploads its artifacts (Node.js native addons) to GitHub Actions.

6. Publication: If all builds and tests are successful, the final step publishes the package to the npm registry.

Contributing to Releases

To contribute to a release:

1. Make your changes in a feature branch.
2. Use Conventional Commits syntax for your commit messages. This helps Release Please determine the next version number and generate the changelog.
3. Create a pull request to merge your changes into the main branch.
4. Once your PR is approved and merged, Release Please will include your changes in the next release PR.

Manual Releases

While the process is automated, manual intervention may sometimes be necessary:

• You can manually trigger the release workflow from the GitHub Actions tab if needed.
• If you need to make changes to the release PR created by Release Please, you can do so before merging it.

Remember, all releases are automatically published to the npm registry. Ensure you have thoroughly tested your changes before merging to main.

API Reference

ProsodyClient

• ProsodyClient.create(config: Configuration): Promise<ProsodyClient>: Initialize a client without blocking the Node.js event loop.
• send<P>(topic: string, key: string, payload: P & JsonCompatible<P>, signal?: AbortSignal): Promise<void>: Send a statically checked JSON-compatible message to a specified topic.
• excise(topic: string, key: string, signal?: AbortSignal): Promise<void>: Send an excise record for a key.
• request<R>(topic, key, payload: JsonValue, options): Promise<ReadonlyMap<string, Outcome<R>>>: Return one outcome for each subsystem.
• requestExcise<R>(topic, key, options): Promise<ReadonlyMap<string, Outcome<R>>>: Return one excise outcome for each subsystem.
• consumerState(): Promise<ConsumerState>: Get the current state of the consumer.
• assignedPartitionCount(): Promise<number>: Get the assigned partition count.
• isStalled(): Promise<boolean>: Test whether the consumer is stalled.
• sourceSystem: string: Get the source system identifier configured for the client.
• state<T>(subsystem: string, definition: ValueDefinition<T>): Promise<PublishedValue<T>>: Open a read-only published value.
• state<V>(subsystem: string, definition: MapDefinition<V>): Promise<PublishedMap<V>>: Open a read-only published map.
• state<T>(subsystem: string, definition: DequeDefinition<T>): Promise<PublishedDeque<T>>: Open a read-only published deque.
• subscribe<P = JsonValue, R = JsonValue>(eventHandler: EventHandler<P, R>): Promise<void>: Subscribe with typed payload and response values.
• unsubscribe(): Promise<void>: Stop the consumer. You can subscribe again later.
• shutdown(): Promise<void>: Stop all client services. Concurrent and repeated calls await the same operation.

AdminClient

• new AdminClient(bootstrapServers): Create an admin client for the specified Kafka servers.
• createTopic(name, partitions, replicationFactor): Create a Kafka topic.
• deleteTopic(name): Delete a Kafka topic.

EventHandler

Interface for handling messages and timers:

• EventHandler<P = JsonValue, R = JsonValue> carries the payload and response types through each callback.
• onMessage: (context, message, signal) => MaybePromise<R & JsonCompatible<R>>: Handle incoming messages.
• onExcise: (context, message, signal) => MaybePromise<R & JsonCompatible<R>>: Handle excise records.
• onTimer: (context, timer, signal) => MaybePromise<void>: Handle timer events.

Message

Represents a Kafka message with the following properties:

• topic: string: The name of the topic.
• partition: number: The partition number.
• offset: bigint: The message offset within the partition.
• timestamp: Date: The timestamp when the message was created or sent.
• key: string: The message key.
• payload: P: The statically typed message payload.

Message takes an optional payload type parameter, Message<P>, used by handlers and message-backed state collections to type payload. Unparameterized Message is Message<JsonValue>, preserving useful JSON safety without requiring an application-specific payload type.

JsonValue describes arbitrary JSON data. JsonCompatible<T> checks a known application type recursively.

Ordinary interfaces work with send(). TypeScript rejects functions, undefined, Date, symbols, bigints, and invalid nested fields before serialization.

ExciseMessage

An ExciseMessage has topic, partition, offset, timestamp, and key properties. It has no payload property.

Context

Represents the current event context:

• onCancel(): Promise<void>: A method that resolves when the context is cancelled.
• shouldCancel: boolean: A property indicating whether the context has been cancelled.

Timer scheduling methods:

• schedule(time: Date): Promise<void>: Schedules a timer to fire at the specified time
• clearAndSchedule(time: Date): Promise<void>: Clears all timers and schedules a new one
• unschedule(time: Date): Promise<void>: Removes a timer scheduled for the specified time
• clearScheduled(): Promise<void>: Removes all scheduled timers
• scheduled(): Promise<Date[]>: Returns an array of all scheduled timer times

Keyed-state binding:

• state(definition): ValueState<T> | MapState<V> | DequeState<T>: Bind a registered collection for the current attempt. Message definitions return handles that contain Message<P>. An unregistered or mismatched definition throws PermanentStateError. See Keyed State.

Timer

Represents a timer that has fired, provided to the onTimer method:

• key: string: The entity key identifying what this timer belongs to
• time: Date: The time when this timer was scheduled to fire

Requests

• RequestOptions: Contains subsystems, timeoutMs, and an optional signal.
• Outcome<T>: A Success<T> or Failure result for one subsystem.
• Success<T>: Contains ok: true and value: T.
• Failure: Contains ok: false and a ResponseError.
• ResponseError: A handler, timeout, format-mismatch, or malformed-response error.

Configuration types

• Configuration: Contains the client settings. See Configuration.
• ConsumerState: Identifies the consumer lifecycle state.
• Mode: Selects the client processing mode.
• ReadCacheConfiguration: Configures the default published-state cache.
• ReadCacheOptions: Overrides the cache for one published collection.

Payload types

• JsonPrimitive: A JSON null, boolean, number, or string.
• JsonValue: Any recursively JSON-compatible value.
• JsonCompatible<T>: Rejects non-JSON members in a known application type.
• MaybePromise<T>: A value or a promise-like value.

Keyed State

Definition constructors (each returns a frozen definition object used both in Configuration.stateCollections and with context.state()):

• value<T = JsonValue>(name: string, options?: PublishedStateDefinitionOptions): ValueDefinition<T>
• map<V = JsonValue>(name: string, options?: MapDefinitionOptions): MapDefinition<V>
• deque<T = JsonValue>(name: string, options?: DequeDefinitionOptions): DequeDefinition<T>
• messageValue<P = JsonValue>(name: string, options?: StateDefinitionOptions): MessageValueDefinition<P>
• messageMap<P = JsonValue>(name: string, options?: MessageMapDefinitionOptions): MessageMapDefinition<P>
• messageDeque<P = JsonValue>(name: string, options?: MessageDequeDefinitionOptions): MessageDequeDefinition<P>

StateDefinitionOptions: { ttlSeconds?: number; readUncommitted?: boolean }. PublishedStateDefinitionOptions adds { published?: boolean; readCache?: { ttlMs: number } | false } for JSON definitions. Map and deque option types add keysetLimit and capacity, respectively; their message equivalents omit publication options.

ValueState<T>:

• get(): Promise<T | null>
• set(value: T): Promise<void>
• clear(): Promise<void>
• commit(): Promise<void>
• rollback(): Promise<void>

MapState<V> (keys are string):

• get(key: string): Promise<V | null>
• getMany(keys: readonly string[]): Promise<(V | null)[]>
• has(key: string): Promise<boolean>
• set(key: string, value: V): Promise<void>
• delete(key: string): Promise<void>
• clear(): Promise<void>
• entries(direction?: ScanDirection): AsyncIterableIterator<[string, V]>
• keys(direction?: ScanDirection): AsyncIterableIterator<string>
• values(direction?: ScanDirection): AsyncIterableIterator<V>
• [Symbol.asyncIterator](): AsyncIterableIterator<[string, V]>
• commit(): Promise<void>
• rollback(): Promise<void>

DequeState<T>:

• push(item: T): Promise<void>
• unshift(item: T): Promise<void>
• pop(): Promise<T | null>
• shift(): Promise<T | null>
• length(): Promise<number>
• isEmpty(): Promise<boolean>
• clear(): Promise<void>
• at(index: number): Promise<T | null>
• values(direction?: ScanDirection): AsyncIterableIterator<T>
• [Symbol.asyncIterator](): AsyncIterableIterator<T>
• commit(): Promise<void>
• rollback(): Promise<void>

ScanDirection: "forward" | "backward".

Published readers take the user key as their first argument. PublishedValue<T> provides get. PublishedMap<V> provides get, getMany, has, entries, keys, and values. PublishedDeque<T> provides at, length, isEmpty, and values. The scan methods return AsyncIterableIterator directly.

StateCollectionConfig defines one stateCollections entry. It contains name, kind, payload, and the applicable collection options. Use a definition constructor to create this value.

JSON definitions also accept readCache. This option applies when the definition opens published state. It is not part of StateCollectionConfig.

The public definition types are ValueDefinition<T>, MapDefinition<V>, DequeDefinition<T>, MessageValueDefinition<P>, MessageMapDefinition<P>, and MessageDequeDefinition<P>.

All definitions expose name, kind, payload, ttlSeconds, and readUncommitted. JSON definitions also expose published and readCache. Map definitions expose keysetLimit. Deque definitions expose capacity.

Errors:

• TransientStateError extends TransientError: Reports a keyed-state error that Prosody can retry.
• PermanentStateError extends PermanentError: Reports a keyed-state error that another attempt cannot resolve.
• isStateError(error: unknown): error is PermanentStateError | TransientStateError: type-guard narrowing an error to either state error class.

Handler error types and decorators:

• EventHandlerError: Base class with an abstract isPermanent property.
• TransientError: Has isPermanent: false and marks an error as retriable.
• PermanentError: Has isPermanent: true and marks an error as final.
• transient(...errorTypes): Creates a transient-error decorator.
• permanent(...errorTypes): Creates a permanent-error decorator.

Logging and telemetry

• Logger: Provides error, warn, info, debug, and trace methods.
• initialize(): Prepare the logging and tracing system during application startup.
• loggerIsSet(): Test whether the application configured a logger.
• setLogger(logger): Replaces the logger.
• setLoggerIfUnset(logger): Sets the logger only when no logger exists.
• getCurrentLogger(): Returns the current JavaScript logger.
• flushTelemetry(): Exports pending telemetry.
• shutdownTelemetry(): Exports pending telemetry and stops its providers.

License

This project is licensed under the MIT License - see the LICENSE file for details.