MCP Tools for Apache Pulsar — subscription types, backlog metrics, and topic stats

Pulsar's hierarchy: tenant, namespace, topic

Pulsar organizes topics in a three-level hierarchy. The tenant is the top-level organizational unit (analogous to an organization or team). The namespace is a configuration boundary within a tenant — it holds policies for message retention, replication, quota limits, authentication, and encryption. The topic is the message stream within a namespace.

Topic names follow this pattern: persistent://tenant/namespace/topic or non-persistent://tenant/namespace/topic. Persistent topics write to BookKeeper and survive broker restarts. Non-persistent topics are held in broker memory only and lose messages if the broker restarts or if the consumer is too slow (no buffering on the broker side).

const PULSAR_ADMIN = process.env.PULSAR_ADMIN_URL ?? 'http://localhost:8080';

async function pulsarAdminRequest(path, method = 'GET', body = null) {
  const res = await fetch(`${PULSAR_ADMIN}/admin/v2${path}`, {
    method,
    headers: {
      'Content-Type': 'application/json',
      // If auth enabled: 'Authorization': `Bearer ${token}`
    },
    body: body ? JSON.stringify(body) : undefined
  });

  if (!res.ok) {
    const text = await res.text();
    throw new Error(`Pulsar Admin API ${res.status} on ${method} ${path}: ${text}`);
  }
  if (res.status === 204) return null; // No content
  return res.json();
}

// List all topics in a namespace
async function listTopics(tenant, namespace) {
  return pulsarAdminRequest(`/persistent/${tenant}/${namespace}`);
}

// Get topic statistics (subscriptions, consumers, backlog)
async function getTopicStats(tenant, namespace, topic) {
  return pulsarAdminRequest(`/persistent/${tenant}/${namespace}/${topic}/stats`);
}

// Get partitioned topic stats
async function getPartitionedTopicStats(tenant, namespace, topic) {
  return pulsarAdminRequest(`/persistent/${tenant}/${namespace}/${topic}/partitioned-stats`);
}

The topic stats response includes a subscriptions object where each key is a subscription name and the value contains consumer health metrics. This is the primary health data structure for MCP monitoring tools.

Four subscription types: delivery semantics and failure modes

Pulsar subscriptions are named — unlike Kafka consumer groups (identified by group ID in consumer configuration), Pulsar subscription names are part of the topic's persistent state on the broker. A subscription continues to accumulate messages even when no consumers are connected.

The four subscription types determine how messages are distributed across the consumers attached to a subscription:

• Exclusive — only one consumer can be connected at a time. A second consumer attempting to subscribe receives an error (ConsumerBusyException). Ordered delivery. Use for sequential processors where correctness requires exactly-one active consumer.
• Shared — multiple consumers, round-robin delivery. Any consumer can acknowledge any message — ack order is independent of receive order. No ordering guarantee. Maximum throughput. Use for parallel workers where message order does not matter.
• Failover — multiple consumers, but only the active consumer receives messages. Other consumers are standbys that take over if the active consumer disconnects. Ordered delivery within each consumer's active window. Use for HA exclusive consumers.
• Key_Shared — multiple consumers, but messages with the same key (the orderingKey property) are always delivered to the same consumer. Ordered delivery per key. Like Kafka partition assignment without fixed partition count. Use for stateful stream processing where all messages for one entity must go to one consumer.

async function analyzeSubscriptionHealth(tenant, namespace, topic) {
  const stats = await getTopicStats(tenant, namespace, topic);

  const subscriptionHealth = Object.entries(stats.subscriptions).map(([name, sub]) => {
    return {
      name,
      type: sub.type, // 'Exclusive', 'Shared', 'Failover', 'Key_Shared'
      consumer_count: sub.consumers?.length ?? 0,
      backlog_messages: sub.msgBacklog,
      backlog_bytes: sub.backlogSize,
      msg_rate_out: sub.msgRateOut,   // messages per second being delivered
      msg_rate_redeliver: sub.msgRateRedeliver, // redelivery rate (nacked or expired)
      unacked_messages: sub.unackedMessages,
      // Critical: subscription with backlog but no consumers
      consumer_starvation: sub.msgBacklog > 0 && (sub.consumers?.length ?? 0) === 0,
      // Warning: high redeliver rate indicates consumer failures
      high_redeliver: sub.msgRateRedeliver > sub.msgRateOut * 0.1, // >10% redeliver rate
      consumers: (sub.consumers ?? []).map(c => ({
        consumer_name: c.consumerName,
        unacked_messages: c.unackedMessages,
        msg_rate_out: c.msgRateOut,
        available_permits: c.availablePermits, // flow control: 0 = consumer at capacity
        connected_since: c.connectedSince,
        address: c.address,
      }))
    };
  });

  return subscriptionHealth;
}

Backlog: bytes vs messages as consumer lag signals

Pulsar's topic stats report consumer lag in two units: msgBacklog (message count) and backlogSize (bytes). For homogeneous message payloads, message count is intuitive. For heterogeneous payloads (variable-size events, JSON blobs of different complexity), byte count is more stable as a lag signal because it correlates with actual storage consumption and processing time.

Backlog accumulation rate (how fast backlog grows) is more useful than absolute backlog size. Combine consecutive snapshots to compute the net accumulation rate:

async function computeBacklogRate(tenant, namespace, topic, subscriptionName, intervalMs = 10000) {
  const before = await getTopicStats(tenant, namespace, topic);
  const beforeSub = before.subscriptions[subscriptionName];
  if (!beforeSub) throw new Error(`Subscription '${subscriptionName}' not found`);

  await new Promise(r => setTimeout(r, intervalMs));

  const after = await getTopicStats(tenant, namespace, topic);
  const afterSub = after.subscriptions[subscriptionName];

  const deltaMessages = afterSub.msgBacklog - beforeSub.msgBacklog;
  const deltaBytes = afterSub.backlogSize - beforeSub.backlogSize;
  const intervalSeconds = intervalMs / 1000;

  return {
    subscription: subscriptionName,
    msg_backlog_current: afterSub.msgBacklog,
    bytes_backlog_current: afterSub.backlogSize,
    msg_rate_growth_per_sec: deltaMessages / intervalSeconds,
    byte_rate_growth_per_sec: deltaBytes / intervalSeconds,
    // Negative growth = consuming faster than producing (catching up)
    // Positive growth = producing faster than consuming (falling behind)
    trending: deltaMessages > 0 ? 'growing' : deltaMessages < 0 ? 'draining' : 'stable'
  };
}

// Skip messages to recover from poison pill (message that always fails)
async function skipMessages(tenant, namespace, topic, subscriptionName, numMessages) {
  return pulsarAdminRequest(
    `/persistent/${tenant}/${namespace}/${topic}/subscription/${subscriptionName}/skip/${numMessages}`,
    'POST'
  );
}

// Reset subscription position to a specific timestamp for replay
async function resetSubscriptionToTimestamp(tenant, namespace, topic, subscriptionName, timestampMs) {
  return pulsarAdminRequest(
    `/persistent/${tenant}/${namespace}/${topic}/subscription/${subscriptionName}/resetcursor/${timestampMs}`,
    'POST'
  );
}

The msgRateRedeliver metric is a strong health signal that message count alone misses. High redeliver rate means consumers are repeatedly nacking messages (processing failures) or ack deadlines are expiring (consumers too slow or stuck). A subscription with zero msgRateOut but positive msgRateRedeliver is thrashing — receiving and failing messages in a retry loop without making progress.

Partitioned topics: parallelism and per-partition health

A Pulsar partitioned topic is an abstraction over multiple internal topics, each called a partition. Publishing to the partitioned topic dispatutes messages across partitions based on the message key's hash (or round-robin for keyless messages). Each partition is an independent topic with its own broker assignment, ledger, and subscription cursor.

async function analyzePartitionedTopic(tenant, namespace, topic) {
  // Check if this is a partitioned topic
  let partitionMetadata;
  try {
    partitionMetadata = await pulsarAdminRequest(
      `/persistent/${tenant}/${namespace}/${topic}/partitions`
    );
  } catch {
    // Not partitioned (or doesn't exist)
    return { is_partitioned: false };
  }

  const partitionCount = partitionMetadata.partitions;

  // Partitioned stats aggregates across all partitions
  const stats = await getPartitionedTopicStats(tenant, namespace, topic);

  // Per-partition breakdown
  const perPartitionStats = await Promise.all(
    Array.from({ length: partitionCount }, (_, i) =>
      getTopicStats(tenant, namespace, `${topic}-partition-${i}`)
        .then(s => ({ partition: i, stats: s }))
        .catch(e => ({ partition: i, error: e.message }))
    )
  );

  // Find skewed partitions (one partition has disproportionate backlog)
  const backlogs = perPartitionStats
    .filter(p => !p.error)
    .map(p => ({
      partition: p.partition,
      backlog: Object.values(p.stats.subscriptions ?? {})
        .reduce((sum, sub) => sum + (sub.msgBacklog ?? 0), 0)
    }));

  const maxBacklog = Math.max(...backlogs.map(b => b.backlog));
  const avgBacklog = backlogs.reduce((s, b) => s + b.backlog, 0) / backlogs.length;
  const skewed = backlogs.filter(b => b.backlog > avgBacklog * 3);

  return {
    is_partitioned: true,
    partition_count: partitionCount,
    total_backlog: stats.msgBacklog,
    total_backlog_bytes: stats.backlogSize,
    producer_count: stats.publishers?.length ?? 0,
    partitions: backlogs,
    skewed_partitions: skewed, // partitions with 3x average backlog
  };
}

Partition skew — where one partition accumulates a large backlog while others are empty — is a common problem when producers use deterministic key hashing. If one key dominates the message stream, its hash maps to one partition, starving the others. An MCP tool that only checks aggregate backlog would miss this: total backlog could be moderate while one partition's consumers are overwhelmed.

Tiered storage and retention: what MCP tools need to know

Pulsar stores messages in BookKeeper ledgers. When a namespace retention policy allows it, old ledgers can be offloaded to cloud object storage (S3, GCS, Azure Blob) via tiered storage. From the consumer perspective, tiered messages are transparent — the Pulsar client fetches them on demand. From the MCP tool perspective, tiered storage affects what the Admin API reports:

async function getStorageBreakdown(tenant, namespace, topic) {
  const stats = await getTopicStats(tenant, namespace, topic);

  return {
    // Storage currently in BookKeeper (hot storage)
    storage_size_bytes: stats.storageSize,
    // Total storage across all tiers (BookKeeper + object storage)
    // Only available with offload metadata
    offloaded_size_bytes: stats.offloadedStorageSize ?? 0,

    // Message retention configuration is on the namespace, not the topic
    // Query it separately:
    // GET /admin/v2/namespaces/{tenant}/{namespace}/retention
    // Returns: { retentionTimeInMinutes, retentionSizeInMB }
    // -1 = infinite retention

    // Backlog quota violation
    backlog_quota_exceeded: stats.backlogQuotaExceeded ?? false,
  };
}

// Get namespace retention policy
async function getNamespaceRetention(tenant, namespace) {
  const policy = await pulsarAdminRequest(`/namespaces/${tenant}/${namespace}/retention`);
  return {
    retention_time_minutes: policy.retentionTimeInMinutes, // -1 = infinite
    retention_size_mb: policy.retentionSizeInMB,           // -1 = infinite
    retention_time_hours: policy.retentionTimeInMinutes === -1
      ? 'infinite'
      : (policy.retentionTimeInMinutes / 60).toFixed(1)
  };
}

An important retention vs backlog interaction: Pulsar only deletes messages from BookKeeper when all subscriptions have acknowledged them AND the retention policy allows deletion. If any subscription has a non-zero backlog, those messages stay in BookKeeper regardless of retention policy. A stale subscription (a subscription that was created but whose consumer stopped) that accumulates unbounded backlog can prevent Pulsar from reclaiming storage. An MCP tool that audits subscription age and backlog for orphaned subscriptions is essential in shared Pulsar clusters.

Further reading

• MCP Tools for Apache Kafka — consumer groups, offsets, and lag monitoring
• MCP Tools for RabbitMQ — Management API, queue depth, and dead-letter exchange
• MCP Tools for AWS SQS — visibility timeout, long polling, and DLQ redrive
• MCP Tools for NATS and JetStream — subjects, streams, and consumer pull delivery
• MCP Server Message Queue — patterns for async task processing
• MCP Server Health Checks — implementing and monitoring protocol probes
• MCP Tools for DevOps — patterns across messaging platforms