Understanding the Apache Pulsar protocol – a minimalist Go broker as an analysis tool

How does the Pulsar protocol work on the wire?

A rather fundamental question has come up again and again in technical discussions recently. Partly while troubleshooting, partly out of curiosity - or out of the feeling that you rely heavily on abstractions in everyday life:

What actually happens on the line at Pulsar?

The client libraries make many things pleasantly easy. They encapsulate the protocol, hide state machines, handle retry logic, flow control and error handling. This is exactly right for productive use - and usually completely sufficient.

At the same time, a large part of the system remains invisible. Anyone who wants to build a real understanding of the system beyond APIs and configuration options will quickly reach a limit: the crucial assumptions lie beneath the libraries, in the protocol itself.

Admittedly, for me too, until recently, part of it was more perceived knowledge than real understanding. I wanted to change that.

The idea: A minimal broker as an analysis tool

When looking at the documentation you will notice that central parts of the protocol are openly defined via Protobuf. Structures like BaseCommand, CommandSend, CommandSubscribe or MessageMetadata describe very precisely what information is exchanged between client and broker.

This gave rise to the idea of ​​building a minimalist broker based on this specification that correctly implements basic publish/subscribe patterns so that standard Pulsar clients can talk to them - ideally without realizing that they are not currently connected to a full Apache Pulsar broker.

Not as a replacement for Pulsar and not with any claim to completeness, but as an experiment. A broker that understands just enough of the protocol to keep real clients happy is a great way to expose the true invariants of the protocol.

Pulsar’s Protobuf specification: rough orientation

The Pulsar protocol description is available as Protobuf definition in proto2 format.
This file is not an implementation, but a formal contract that specifies exactly which data structures may be exchanged between client and broker.

The header of the file already makes some key design decisions visible:

• syntax = "proto2" Pulsar deliberately uses Protobuf 2 with required fields. Missing mandatory fields are hard protocol errors and in practice lead to immediate connection terminations.
• package pulsar.proto A clear, stable namespace for all messages.
• Language-specific options (java_package, LITE_RUNTIME) show that this specification is intended directly for code generation - not just documentation.

Messages as data contracts

Each message describes a clearly defined data structure:

• MessageMetadata bundles everything that describes a message: producer identity, sequence numbers, properties, compression, schema information, transactions.
• MessageIdData defines the logical identity of a message (Ledger, Entry, Partition, Batch).
• Schema does not describe the transport, but the meaning of the payload.

What is important is that these messages say nothing about transport, sequence or state machines. They only define the form of the data.

Enums as hard state spaces

Enums like CompressionType, ProducerAccessMode, ServerError or ProtocolVersion explicitly define permitted states.

For a minimalist broker this is instructive:

• new enum values appear with new versions
• Clients often expect tolerance for unknown values
• Real incompatibilities usually show up not in the schema, but in the behavior

ProtocolVersion is particularly crucial: clients expect the broker to accept their version - or reject it explicitly and comprehensibly.

Commands and the central node: BaseCommand

The actual core of the specification is BaseCommand.
It acts as an envelope for all protocol operations:

• type determines which command is present (CONNECT, SEND, ACK, FLOW, …)
• exactly one associated sub-message field is set

This results in a clear basic pattern for the broker:

1. Read BaseCommand
2. Evaluate type
3. Process appropriate sub-message

Everything else - framing, checksums, payload parsing - lies outside Protobuf and is regulated separately by the Pulsar Binary protocol.

Implicit complexity

What the specification makes seem sober is actually highly condensed:

• optional fields, default values and feature flags encode years of evolution
• many fields only make sense in certain combinations
• Clients rely on specific responses in a specific order even though this is not explicitly stated in the schema

This is exactly where a minimalist broker reveals its value: it forces a distinction between formally specified and implicitly expected.

Implementation: AI-generated code, human feedback

All of the code in this prototype was generated by an AI. The human role was

• provide the Pulsar Protobuf specification
• clearly formulate which protocol paths need to be supported
• test real Pulsar clients against the broker
• Systematically translate deviations back into more precise requirements

The code was not created from a pre-designed architectural model, but from iterative cross-checking of real client behavior.

Network and frame handling

The broker’s entry point is deliberately unspectacular: a TCP listener on port 6650. Each incoming connection is handled in its own goroutine. The real core lies in frame handling.

Pulsar uses clear length framing. Each frame begins with a uint32 totalSize, followed by an inner frame, which in turn begins with the length of the Protobuf command. Only after reading the complete frame is the actual Protobuf command extracted and deserialized. The inner frame begins again with a length field that specifies the size of the Protobuf command:

// Gesamten Frame lesen
frame := make([]byte, totalSize)
if _, err := io.ReadFull(conn, frame); err != nil {
    return err
}

r := bytes.NewReader(frame)

// Länge des Protobuf-Commands
var cmdSizeBuf [4]byte
if _, err := io.ReadFull(r, cmdSizeBuf[:]); err != nil {
    return fmt.Errorf("read command size: %w", err)
}
cmdSize := binary.BigEndian.Uint32(cmdSizeBuf[:])
if cmdSize == 0 || int(cmdSize) > r.Len() {
    return fmt.Errorf("invalid command size %d", cmdSize)
}

// Protobuf-Command lesen
cmdBytes := make([]byte, cmdSize)
if _, err := io.ReadFull(r, cmdBytes); err != nil {
    return fmt.Errorf("read command: %w", err)
}

// BaseCommand deserialisieren
var base pulsar.BaseCommand
if err := proto.Unmarshal(cmdBytes, &base); err != nil {
    return fmt.Errorf("unmarshal BaseCommand: %w", err)
}

Only after successful unmarshalling is it clear which command type it is. From this point on, the code leaves the transport layer and switches to the actual protocol logic, which is controlled via the BaseCommand.Type.

Command dispatch via BaseCommand

After successful unmarshalling, the command type is evaluated and passed on to the responsible handler. The dispatch takes place exclusively via the type field of the BaseCommand:

switch base.GetType() {

case pulsar.BaseCommand_CONNECT:
    return b.handleConnect(conn, &base)

case pulsar.BaseCommand_PRODUCER:
    return b.handleProducer(conn, &base)

case pulsar.BaseCommand_SUBSCRIBE:
    return b.handleSubscribe(conn, &base)

case pulsar.BaseCommand_SEND:
    // SEND ist ein Sonderfall: auf den Command folgt ein Message-Frame
    payloadSection, _ := io.ReadAll(r)
    return b.handleSend(conn, &base, payloadSection)

case pulsar.BaseCommand_FLOW:
    return b.handleFlow(conn, &base)

case pulsar.BaseCommand_ACK:
    return b.handleAck(conn, &base)

case pulsar.BaseCommand_PING:
    return b.handlePing(conn, &base)

default:
    // Unbekannte Commands werden toleriert, aber nicht verarbeitet
    log.Printf("unhandled command type: %v", base.GetType())
    return nil
}

An important special case is CONNECT. The broker responds immediately with a CONNECTED, which explicitly reflects the protocol version used by the client:

func (b *broker) handleConnect(conn net.Conn, base *pulsar.BaseCommand) error {
    cmd := base.GetConnect()
    if cmd == nil {
        return fmt.Errorf("CONNECT without payload")
    }

    resp := &pulsar.BaseCommand{
        Type: pulsar.BaseCommand_CONNECTED.Enum(),
        Connected: &pulsar.CommandConnected{
            ServerVersion:   proto.String("minipulsar-0.1"),
            ProtocolVersion: proto.Int32(cmd.GetProtocolVersion()),
            MaxMessageSize:  proto.Int32(5 * 1024 * 1024),
        },
    }

    return b.writeCommand(conn, resp)
}

PRODUCER and SUBSCRIBE: State per connection

Producers and consumers are deliberately managed per connection in the prototype. Keys are made up of the TCP connection and the ID assigned by the client.

// Verbindungsspezifische Schlüssel verhindern ID-Kollisionen
type producerKey struct {
    conn net.Conn
    id   uint64
}

type consumerKey struct {
    conn net.Conn
    id   uint64
}

Producers and consumers are then always addressed via these keys:

type broker struct {
    producers map[producerKey]*producer
    consumers map[consumerKey]*consumer
}

When creating a producer, the key is explicitly taken from the current one connection and the ID assigned by the client:

func (b *broker) handleProducer(conn net.Conn, base *pulsar.BaseCommand) error {
    cmd := base.GetProducer()
    producerID := cmd.GetProducerId()

    key := producerKey{
        conn: conn,
        id:   producerID,
    }

    b.mu.Lock()
    b.producers[key] = &producer{
        id:    producerID,
        topic: cmd.GetTopic(),
        conn:  conn,
    }
    b.mu.Unlock()

    // PRODUCER_SUCCESS antworten …
    return nil
}

The same principle applies to consumers with SUBSCRIBE:

func (b *broker) handleSubscribe(conn net.Conn, base *pulsar.BaseCommand) error {
    cmd := base.GetSubscribe()
    consumerID := cmd.GetConsumerId()

    key := consumerKey{
        conn: conn,
        id:   consumerID,
    }

    c := &consumer{
        id:           consumerID,
        topic:        cmd.GetTopic(),
        subscription: cmd.GetSubscription(),
        conn:         conn,
    }

    b.mu.Lock()
    b.consumers[key] = c
    b.mu.Unlock()

    // SUCCESS antworten …
    return nil
}

The background is inconspicuous, but critical: Pulsar IDs are only unique within a connection. Two clients can both create a producer with ID 1 without influencing each other. Without this separation, subtle errors arise that clients quickly reveal.

SEND: Message frames, magic and CRC

In the Pulsar protocol, the SEND command is followed by no further Protobuf object, but rather a binary message frame with a fixed structure. The broker must interpret this frame with byte precision. First, the magic bytes that mark the beginning of a message frame are checked:

var magicBuf [2]byte
if _, err := io.ReadFull(r, magicBuf[:]); err != nil {
    return fmt.Errorf("read magic: %w", err)
}

magic := binary.BigEndian.Uint16(magicBuf[:])
if magic != magicMessageFormat {
    return fmt.Errorf("unexpected magic 0x%x", magic)
}

The CRC32C checksum follows immediately afterwards. It is not calculated over the entire frame, but only over precisely defined bytes:

// CRC32C (Castagnoli) – vom Protokoll vorgegeben
var checksumBuf [4]byte
if _, err := io.ReadFull(r, checksumBuf[:]); err != nil {
    return fmt.Errorf("read checksum: %w", err)
}

The length of the message metadata is then read and its Protobuf payload is extracted:

var metaSizeBuf [4]byte
if _, err := io.ReadFull(r, metaSizeBuf[:]); err != nil {
    return fmt.Errorf("read metadata size: %w", err)
}
metaSize := binary.BigEndian.Uint32(metaSizeBuf[:])
if metaSize == 0 || int(metaSize) > r.Len() {
    return fmt.Errorf("invalid metadata size %d", metaSize)
}

metaBytes := make([]byte, metaSize)
if _, err := io.ReadFull(r, metaBytes); err != nil {
    return fmt.Errorf("read metadata: %w", err)
}

var meta pulsar.MessageMetadata
if err := proto.Unmarshal(metaBytes, &meta); err != nil {
    return fmt.Errorf("unmarshal metadata: %w", err)
}

The remaining portion of the frame is the payload, whose bytes are also included in the CRC calculation:

payload, err := io.ReadAll(r)
if err != nil {
    return fmt.Errorf("read payload: %w", err)
}

The validation of the checksum is carried out exactly according to the Pulsar specification: CRC32C via Metadata length field + Metadata + Payload - not via Magic or the CRC field itself.

crc := crc32.New(crc32.MakeTable(crc32.Castagnoli))
crc.Write(metaSizeBuf[:])
crc.Write(metaBytes)
crc.Write(payload)

if crc.Sum32() != binary.BigEndian.Uint32(checksumBuf[:]) {
    return fmt.Errorf("checksum mismatch")
}

Persistence: Make state visible

Once messages are processed, the broker needs state.
The prototype uses SQLite – not for performance reasons, but for state to be kept transparent and comprehensible.

Messages are saved append-only:

CREATE TABLE messages (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  topic TEXT NOT NULL,
  payload BLOB NOT NULL,
  publish_time INTEGER NOT NULL,
  sequence_id INTEGER NOT NULL
);

For each subscription there is a dispatch cursor that indicates which one Message ID messages are actually eligible for delivery:

CREATE TABLE subscription_cursor (
  topic TEXT NOT NULL,
  name  TEXT NOT NULL,
  next_message_id INTEGER NOT NULL,
  PRIMARY KEY (topic, name)
);

Messages that have already been delivered but not yet confirmed will be sent separately listed as Pending:

CREATE TABLE subscription_pending (
  topic TEXT NOT NULL,
  name  TEXT NOT NULL,
  message_id INTEGER NOT NULL,
  consumer_id INTEGER NOT NULL,
  delivered_at INTEGER NOT NULL,
  PRIMARY KEY (topic, name, message_id)
);

Claiming moves the cursor

The central point lies in the claim process. In a single transaction:

1. the current cursor read
2. deliverable messages selected
3. Pending entries created
4. the cursor moves monotonously forward

// Cursor lesen
var cur int64
err := tx.QueryRow(
    "SELECT next_message_id FROM subscription_cursor WHERE topic=? AND name=?",
    topic, sub,
).Scan(&cur)

// Pending-Einträge anlegen (Claim)
_, err = tx.Exec(
    "INSERT INTO subscription_pending(topic, name, message_id, consumer_id, delivered_at) VALUES(?,?,?,?,?)",
    topic, sub, m.id, consumerUID, now,
)

// Cursor NUR hier bewegen – nicht beim ACK
_, err = tx.Exec(
    "UPDATE subscription_cursor SET next_message_id=? WHERE topic=? AND name=?",
    lastID+1, topic, sub,
)

Only after a successful commit is a message permanently saved as marked “claimed”.

ACKs only clear pending

ACKs do not access the cursor.
You only remove pending entries for the respective consumer:

_, err := tx.Exec(
    "DELETE FROM subscription_pending WHERE topic=? AND name=? AND message_id=? AND consumer_id=?",
    topic, sub, msgID, c.uid,
)

This makes it clear:

• Claim decides who gets a message
• ACK only confirms processing

Technical assessment

• Cursor movement on the claim prevents duplicates
• ACKs are idempotent and local to the consumer
• Crashed consumers leave pending entries but do not permanently block the cursor
• The model is simple but correct for shared subscriptions

This design is not an optimization, but a consistent implementation of the Protocol semantics**:
Delivery and confirmation are separate states - and are also persisted separately.

FLOW and delivery loop

The delivery is strictly based on the FLOW commands of the clients. Consumers signal how many messages they can currently accept. The prototype manages these permits explicitly and only delivers messages if if permits are available.

A FLOW command increases the available permits of a consumer and triggers delivery if necessary:

func (b *broker) handleFlow(conn net.Conn, base *pulsar.BaseCommand) error {
    cmd := base.GetFlow()
    key := consumerKey{conn: conn, id: cmd.GetConsumerId()}

    b.mu.RLock()
    c := b.consumers[key]
    b.mu.RUnlock()
    if c == nil {
        return nil
    }

    c.mu.Lock()
    c.permits += int(cmd.GetMessagePermits())
    c.mu.Unlock()

    s := b.getOrCreateSubState(c.topic, c.subscription)
    b.maybeStartSubDelivery(s)
    return nil
}

Delivery will only start if no loop is already running and at least one consumer has free permits:

func (b *broker) maybeStartSubDelivery(s *subState) {
    s.mu.Lock()
    if s.delivering {
        s.mu.Unlock()
        return
    }
    s.delivering = true
    s.mu.Unlock()

    go b.deliveryLoopShared(s)
}

The actual delivery logic consists of a simple loop per subscription. There is no redelivery logic, no timeouts and no prioritization:

func (b *broker) deliveryLoopShared(s *subState) {
    defer func() {
        s.mu.Lock()
        s.delivering = false
        s.mu.Unlock()
    }()

    for {
        c := s.nextConsumerWithPermits()
        if c == nil {
            return
        }

        msgs, _ := b.dbClaimBatch(s.key.topic, s.key.name, c.uid, c.permits)
        if len(msgs) == 0 {
            return
        }

        for _, m := range msgs {
            _ = b.writeMsgFrame(c.conn, c.id, &m)
            c.mu.Lock()
            c.permits--
            c.mu.Unlock()
        }
    }
}

Technical classification

• FLOW is the only trigger for delivery
• Permits fully map the backpressure mechanism
• The broker does not make its own scheduling decisions

The conscious reduction of the delivery logic makes this pure Pulsar protocol behavior visible: Delivery only takes place upon explicit request from the client.

Scope and limitations of the prototype

The result is a process that speaks the Pulsar binary protocol on port 6650 and is accepted by standard clients. Producers can publish messages, Consumers receive and confirm basic shared subscription scenarios work as expected.

At the same time, almost everything that Apache Pulsar does as a distributed system is deliberately missing distinguishes itself: no partitioning, no ledger or segment model, none Authentication or authorization, no retention, no compaction. The code is not ready for production and expressly does not make this claim.

Outlook: What else the prototype can be used for

Even if this prototype is explicitly not a product, it opens up an interesting space for thought.
A broker that speaks the Pulsar Binary protocol correctly, but deliberately remains minimal internally, is more than a gimmick - it can serve as a tool for very different questions.

1. Pulsar protocol as a frontend for other brokers

An obvious application is to use the Pulsar protocol implemented in Go as a protocol frontend for other message brokers that are also implemented in Go - such as NATS or related systems.

The Pulsar protocol would serve exclusively as a compatibility layer:
Pulsar clients talk to the broker as is, while the actual persistence, fan-out logic or replication is handled by another backend.

This opens up interesting potential:

• existing Pulsar clients can access alternative backends without being customized
• Operating models can be decoupled: Pulsar externally, lean broker internally
• Comparing and evaluating different messaging backends becomes easier because client behavior remains constant

2. A deliberately reduced but “real” Pulsar broker

A second path would be further development towards a full-fledged Pulsar broker with a greatly reduced feature set. Not as a competition to the official project, but as a specialized variant.

The focus would not be on completeness, but on:

• clearly defined protocol subset
• strict correctness to the protocol
• high performance through targeted simplification

Such a broker could be particularly interesting for edge scenarios: low-performance hardware, little memory, limited dependencies - but still native Pulsar communication.

As a link between edge and cloud-based Pulsar clusters, a coupling could be implemented without protocol breaks.

3. Protocol-based testing and fuzzing environment

A minimalist broker is a great testing tool. Instead of exclusively testing clients against a complex reference stack, you can specifically check:

• which protocol variants clients actually send
• how they react to unusual but formally correct answers
• where there are implicit assumptions about order, timing or states

This makes the broker a tool for systematic protocol testing, fuzzing and regression analysis - especially for client upgrades.

4. Distributed systems teaching and analysis tool

The prototype is small enough to be fully understood, but real enough to speak to real clients. It is precisely this combination that makes it a suitable didactic tool.

It is suitable for:

• explain the Pulsar protocol in a tangible way
• Make state machines and flow control visible
• Discuss differences between formal specification and lived practice

Not as tutorial code, but as a comprehensible study object.

Conclusion

The code of this prototype is visibly raw. It is neither modular in structure nor designed for long-term maintainability, but rather the result of a quick, targeted exploration.

This is not an oversight, but rather an expression of the way we work. The prototype is not an elaboration, but a working sketch. He shows how far you can get there in a short time if you have logical thinking and an understanding of protocols and clear targets form the guardrails - and large parts of them Implementation work can be delegated to an AI.

The actual engineering process would only begin after: Establish structure, separate responsibilities, add tests, Establish maintainability. For the purpose of this experiment – understanding of the Pulsar Binary Protocol – this state is sufficient and even helpful because it makes the mechanics clearly visible.

Image credit: The Pulsar logo is a registered trademark of the Apache Software Foundation. Use in editorial reporting in accordance with official guidelines. Thumbnail illustration inspired by the Go Gopher, designed by Renée French. Used under the CC BY 3.0 license.