xk6-pulsar

A k6 extension for load testing Apache Pulsar with support for producers, consumers, and admin operations via the Pulsar REST Admin API.

Features

• Producer — send messages to any Pulsar topic
• Consumer — receive messages with automatic acknowledgment; supports exclusive, shared, and failover subscription types
• AdminClient — create, delete, and list topics via the Pulsar Admin REST API
• Compression — LZ4, Zstd, and Snappy compression for producers
• Metrics — built-in k6 counters for produced/consumed messages and errors

Prerequisites

• Go 1.26.4 or later
• xk6 — go install go.k6.io/xk6/cmd/xk6@latest
• A running Apache Pulsar broker

Build from Source

xk6 build --with github.com/jonasclaes/xk6-pulsar@latest

For local development:

xk6 build --with github.com/jonasclaes/xk6-pulsar=.

Quick Start

Start Pulsar locally

docker-compose up -d
# or directly:
docker run -it -p 6650:6650 -p 8080:8080 apachepulsar/pulsar:latest bin/pulsar standalone

Producer

import { Producer } from "k6/x/pulsar";

const producer = new Producer({
  serviceURL: "pulsar://localhost:6650",
  topic: "persistent://public/default/my-topic",
});

export default function () {
  producer.produce({
    messages: [{ data: "Hello, Pulsar!" }, { data: "Another message" }],
  });
}

export function teardown() {
  producer.close();
}

Consumer

import { Consumer } from "k6/x/pulsar";

const consumer = new Consumer({
  serviceURL: "pulsar://localhost:6650",
  topic: "persistent://public/default/my-topic",
  subscription: "my-subscription",
});

export default function () {
  const messages = consumer.consume({ maxMessages: 10, timeoutMs: 5000 });
  console.log(`Received ${messages.length} messages`);
}

export function teardown() {
  consumer.close();
}

AdminClient

The AdminClient calls the Pulsar Admin REST API directly. By default the web URL is inferred from serviceURL by replacing pulsar:// with http:// and port 6650 with 8080.

import { AdminClient } from "k6/x/pulsar";

const admin = new AdminClient({
  serviceURL: "pulsar://localhost:6650",
  // webURL: 'http://localhost:8080', // optional — inferred automatically
});

export function setup() {
  admin.createTopic({
    topic: "persistent://public/default/my-topic",
    numPartitions: 3,
  });
}

export default function () {
  const topics = admin.listTopics();
  console.log(`Total topics: ${topics.length}`);
}

export function teardown() {
  admin.deleteTopic("persistent://public/default/my-topic");
  admin.close();
}

API Reference

Producer

new Producer(config);

Field	Type	Required	Default	Description
serviceURL	string	Yes	—	Pulsar broker URL, e.g. pulsar://localhost:6650
topic	string	Yes	—	Fully-qualified topic name
name	string	No	—	Producer name
compressionType	string	No	"none"	"lz4", "zstd", or "snappy"
maxPendingMessages	number	No	1000	Max messages queued before blocking
batchingDelayMs	number	No	10	Time to wait before flushing a batch

Methods:

• produce({ messages: [{ key, value }] }) — send an array of messages
• close() — flush and close the producer and client

Consumer

new Consumer(config);

Field	Type	Required	Default	Description
serviceURL	string	Yes	—	Pulsar broker URL
topic	string	Yes	—	Topic to subscribe to
subscription	string	Yes	—	Subscription name
type	string	No	"exclusive"	"exclusive", "shared", "key_shared", or "failover"
initialPosition	string	No	"latest"	"latest" or "earliest"

Methods:

• consume({ maxMessages?, timeoutMs? }) — returns an array of { key, value, topic } objects; defaults: maxMessages=1, timeoutMs=1000
• close() — close the consumer and client

AdminClient

new AdminClient(config);

Field	Type	Required	Default	Description
serviceURL	string	Yes	—	Pulsar broker URL (used to derive webURL)
webURL	string	No	derived	Admin REST API base URL, e.g. http://localhost:8080
tenant	string	No	"public"	Pulsar tenant for listTopics
namespace	string	No	"default"	Pulsar namespace for listTopics

Methods:

• createTopic({ topic, numPartitions?, replicationFactor? }) — creates a non-partitioned topic (numPartitions=1) or a partitioned topic (numPartitions>1)
• deleteTopic(topic) — deletes the topic
• listTopics() — returns [{ topic, partitions }] for the configured tenant/namespace
• close() — no-op (HTTP client holds no persistent connections)

Constants

Constant	Value
TLS_1_0 / TLS_1_1 / TLS_1_2 / TLS_1_3	TLS version strings
SASL_NONE / SASL_PLAIN	SASL mechanism strings
SCHEMA_TYPE_STRING / SCHEMA_TYPE_BYTES / SCHEMA_TYPE_JSON	Schema type strings
ROUTING_KEY / ROUTING_SINGLE_PARTITION / ROUTING_ROUND_ROBIN / ROUTING_CUSTOM	Routing mode strings
KEY / VALUE	Element type strings
NANOSECOND … HOUR	Duration constants in nanoseconds

Metrics

Metric	Type	Description
pulsar_produced_messages	Counter	Total messages sent
pulsar_consumed_messages	Counter	Total messages received
pulsar_producer_errors	Counter	Total producer errors
pulsar_consumer_errors	Counter	Total consumer errors

Contributing

See CONTRIBUTING.md.

License

Apache License, Version 2.0 — see LICENSE.