What Makes Apache Pulsar a Next‑Gen Cloud‑Native Messaging Platform?

What is Pulsar?

Apache Pulsar is an Apache top‑level project that provides a cloud‑native, distributed messaging and streaming platform with integrated storage, lightweight serverless compute, multi‑tenant support, persistent storage, cross‑region replication, strong consistency, high throughput, and low latency.

Pulsar Key Features

Pulsar instances natively support multiple clusters and seamless cross‑datacenter replication.

Extremely low publish and end‑to‑end latency.

Scales to over one million topics.

Simple client APIs for Java, Go, Python, and C++.

Supports exclusive, shared, failover, and key‑shared subscription modes.

Persistent message storage via Apache BookKeeper.

Serverless compute with Pulsar Functions.

Serverless connectors (Pulsar IO) for easy data ingress and egress.

Tiered storage that offloads cold data to S3, GCS, etc.

Pulsar vs Kafka

Benchmark reports show Pulsar can achieve 605 MB/s publish and end‑to‑end throughput and 3.5 GB/s catch‑up read throughput, with latency remaining under 15 ms (P99) compared to Kafka’s higher latency under similar conditions.

Feature Comparison

Multi‑language clients (Java, Go, Python, C++).

Management tools (Pulsar Manager vs Kafka Manager).

Built‑in stream processing (Pulsar Functions vs Kafka Streams).

Rich integrations via Pulsar Connectors.

Exactly‑once processing support.

Log compression.

Multi‑tenant capabilities.

Security management.

Architecture Design

Pulsar separates storage and compute. Brokers handle message serving without storing data, while BookKeeper provides a distributed log storage layer. This separation enables higher availability, flexible scaling, and eliminates the need for data rebalancing.

Persistent Storage

Pulsar uses BookKeeper’s distributed log as the storage backend. Messages are stored in ledgers, which are immutable logs composed of fragments replicated across BookKeeper nodes for durability and performance.

# Server parameters
journalDirectories=/data/appData/pulsar/bookkeeper/journal
# Ledger storage settings
ledgerDirectories=/data/appData/pulsar/bookkeeper/ledgers

# Managed ledger defaults
managedLedgerDefaultEnsembleSize=2
managedLedgerDefaultWriteQuorum=2
managedLedgerDefaultAckQuorum=2

Metadata Storage

Both Pulsar and BookKeeper use Apache Zookeeper for metadata and health monitoring.

$PULSAR_HOME/bin/pulsar zookeeper-shell
> ls /

Message Mechanism

Pulsar follows a publish‑subscribe model where producers publish to topics and consumers subscribe, with messages retained until acknowledged.

Topic Types

{persistent|non-persistent}://tenant/namespace/topic

Non‑Partitioned Topics

$PULSAR_HOME/bin/pulsar-admin topics list public/default
$PULSAR_HOME/bin/pulsar-admin topics create persistent://public/default/input-seed-avro-topic

Partitioned Topics

$PULSAR_HOME/bin/pulsar-admin topics create-partitioned-topic persistent://public/default/output-seed-avro-topic --partitions 2

Message Structure

public interface Message<T> {
  Map<String,String> getProperties();
  boolean hasProperty(String var1);
  String getProperty(String var1);
  byte[] getData();
  T getValue();
  MessageId getMessageId();
  long getPublishTime();
  long getEventTime();
  long getSequenceId();
  String getProducerName();
  boolean hasKey();
  String getKey();
  byte[] getKeyBytes();
  boolean hasOrderingKey();
  byte[] getOrderingKey();
  String getTopicName();
  Optional<EncryptionContext> getEncryptionCtx();
  int getRedeliveryCount();
  byte[] getSchemaVersion();
  boolean isReplicated();
  String getReplicatedFrom();
}

Producer Example

public void send() throws PulsarClientException {
  final String serviceUrl = "pulsar://server-100:6650";
  PulsarClient client = PulsarClient.builder()
      .serviceUrl(serviceUrl)
      .connectionTimeout(10000, TimeUnit.MILLISECONDS)
      .build();
  final String topic = "persistent://public/default/topic-sensor-temp";
  Producer<byte[]> producer = client.newProducer()
      .producerName("sensor-temp")
      .topic(topic)
      .compressionType(CompressionType.LZ4)
      .enableChunking(true)
      .enableBatching(true)
      .batchingMaxBytes(1024)
      .batchingMaxMessages(10)
      .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
      .blockIfQueueFull(true)
      .maxPendingMessages(512)
      .sendTimeout(1, TimeUnit.SECONDS)
      .create();
  MessageId mid = producer.send("sensor-temp".getBytes());
  System.out.printf("
message with ID %s successfully sent", mid);
  producer.close();
  client.close();
}

Consumer Example

public void consume() throws PulsarClientException {
  final String serviceUrl = "pulsar://server-101:6650";
  final String topic = "input-seed-avro-topic";
  PulsarClient client = PulsarClient.builder()
      .serviceUrl(serviceUrl)
      .enableTcpNoDelay(true)
      .build();
  Consumer<byte[]> consumer = client.newConsumer()
      .consumerName("seed-avro-consumer")
      .subscriptionName("seed-avro-subscription")
      .subscriptionType(SubscriptionType.Exclusive)
      .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
      .topic(topic)
      .receiverQueueSize(10)
      .subscribe();
  while (true) {
    try {
      Message<byte[]> msg = consumer.receive();
      LOG.info("Received message: {}", new String(msg.getData()));
      consumer.acknowledge(msg);
    } catch (PulsarClientException e) {
      consumer.negativeAcknowledge(msg);
    }
  }
}

Subscription Types

Exclusive – only one consumer can consume.

Failover – one active consumer with backups.

Shared – multiple consumers share messages.

Key_Shared – messages with the same key go to the same consumer.

Ordering Guarantees

Exclusive and Failover subscriptions guarantee order; Shared subscriptions provide parallelism, while Key_Shared maintains order per key.

Multi‑Topic Subscriptions

Pattern examples: persistent://public/default/.* and

persistent://public/default/foo.*

Reader API

public void read() throws IOException {
  final String serviceUrl = "pulsar://server-101:6650";
  PulsarClient client = PulsarClient.builder()
      .serviceUrl(serviceUrl)
      .build();
  Reader<byte[]> reader = client.newReader()
      .topic("my-topic")
      .startMessageId(MessageId.earliest())
      .create();
  while (true) {
    Message<byte[]> message = reader.readNext();
    System.out.println(new String(message.getData()));
  }
}

Partitioned Topics

Partitioned topics can be created and managed via pulsar-admin topics create-partitioned-topic and related commands.

Message Retention and Expiry

If no retention policy is set, messages are deleted once all subscriptions have acknowledged them. Retention policies can be configured per namespace for size and time.

# Default retention time (minutes)
defaultRetentionTimeInMinutes=4320
# Default retention size (MB)
defaultRetentionSizeInMB=10240

Pulsar Schema

Schemas provide type safety and preserve data semantics across systems.

Primitive types (e.g., STRING).

Complex types such as KeyValue and Avro/JSON/Protobuf structs.

Producer<String> producer = client.newProducer(Schema.STRING).create();
producer.newMessage().value("Hello Pulsar!").send();
Consumer<String> consumer = client.newConsumer(Schema.STRING).subscribe();
consumer.receive();

Schema Management

$PULSAR_HOME/bin/pulsar-admin schemas get persistent://public/default/spirit-avro-topic
$PULSAR_HOME/bin/pulsar-admin schemas upload persistent://public/default/test-topic --filename $PULSAR_HOME/connectors/json-schema.json
$PULSAR_HOME/bin/pulsar-admin schemas delete persistent://public/default/spirit-avro-topic

Pulsar Functions

Functions enable lightweight stream processing with a simple programming model.

public class WordCountWindowFunction implements WindowFunction<String,Void> {
  @Override
  public Void process(Collection<Record<String>> inputs, WindowContext context) {
    // processing logic
    return null;
  }
}

Functions can be deployed via configuration files and the Pulsar admin CLI.

Pulsar Connectors

Built‑in source connectors include Canal, File, Flume, Kafka, RabbitMQ; sink connectors include Elasticsearch, HBase, HDFS, InfluxDB, JDBC, MongoDB, Redis, etc. Processing guarantees: at‑most‑once, at‑least‑once, effectively‑once.

JDBC Sink Example (ClickHouse)

CREATE DATABASE IF NOT EXISTS monitor;
CREATE TABLE IF NOT EXISTS monitor.pulsar_clickhouse_jdbc_sink (
  id UInt32,
  name String
) ENGINE = TinyLog;
INSERT INTO monitor.pulsar_clickhouse_jdbc_sink (id, name) VALUES (1, 'tmp');

$PULSAR_HOME/bin/pulsar-admin sinks create \
  --tenant public \
  --namespace default \
  --name pulsar-clickhouse-jdbc-sink \
  --inputs pulsar-clickhouse-jdbc-sink-topic \
  --sink-config-file $PULSAR_HOME/connectors/pulsar-clickhouse-jdbc-sink.yaml \
  --archive $PULSAR_HOME/connectors/pulsar-io-jdbc-clickhouse-2.6.2.nar \
  --processing-guarantees EFFECTIVELY_ONCE \
  --parallelism 1

Pulsar Deployment

Directory layout includes bin, conf, lib, and example scripts. Standalone mode can be started with bin/pulsar standalone or as a daemon. Cluster deployment involves ZooKeeper, BookKeeper, and multiple brokers.

Client Usage

# Consumer
$PULSAR_HOME/bin/pulsar-client consume persistent://public/default/seed-avro-topic \
  --subscription-name cli-pack-avro-subscription \
  --subscription-type Exclusive \
  --subscription-position Latest \
  --num-messages 0
# Producer
$PULSAR_HOME/bin/pulsar-client produce persistent://public/default/seed-avro-topic \
  --num-produce 100 \
  --messages "Hello Pulsar" \
  --separator ","

Pulsar Admin

Administration can be performed via pulsar-admin CLI or REST API, covering clusters, tenants, brokers, namespaces, topics, schemas, and functions.

$PULSAR_HOME/bin/pulsar-admin clusters
$PULSAR_HOME/bin/pulsar-admin tenants
$PULSAR_HOME/bin/pulsar-admin brokers
$PULSAR_HOME/bin/pulsar-admin namespaces
$PULSAR_HOME/bin/pulsar-admin topics

Pulsar Manager

Pulsar Manager provides a Web UI for managing environments, clusters, tenants, namespaces, and topics.

Pulsar Flink Integration

The pulsar-flink connector enables Flink jobs to read from and write to Pulsar with exactly‑once semantics.

public class PulsarSinkJob {
  public static FlinkPulsarSink<SeedEvent> getPulsarSink(ParameterTool params) {
    // configuration and sink creation
  }
  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // checkpointing, state backend, source, processing, sink
    env.execute("PulsarSinkJob");
  }
}

public class PulsarSourceJob {
  public static FlinkPulsarSource<SeedEvent> getPulsarSource(ParameterTool params) {
    // configuration and source creation
  }
  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // checkpointing, source, map, sink
    env.execute("PulsarSourceJob");
  }
}

“The instructor is Larry Zhang, a Cloud Wisdom Service Engineering Operations Developer focusing on open‑source OMP platform development and PaaS design.”