Chunking

What is Chunking?

Chunking is a feature in Apache Pulsar that allows large messages to be split into smaller pieces (chunks) for transmission. This is particularly useful when your messages exceed the broker's maximum message size limit (typically 5MB by default), and you want to handle large payloads without hitting broker or network constraints.

When a producer sends a large message with chunking enabled, it automatically splits the message into multiple chunks. The consumer then reassembles these chunks back into the original message before delivering it to the application layer.

How Chunking Works

Producer Side

Pulsar.Producer.start/2 adds a producer to a running client, so start one first — in your supervision tree, or directly in a script:

{:ok, _pid} = Pulsar.Client.start_link(host: "pulsar://localhost:6650")

When a producer is configured with chunking enabled:

{:ok, producer} = Pulsar.Producer.start(
  "my-topic",
  chunking_enabled: true,
  max_message_size: 1024  # Split messages larger than 1KB
)

Large messages are automatically:

1. Compressed as a whole, when :compression is set
2. Split into chunks of max_message_size bytes, capped so that a chunk plus the metadata travelling with it stays inside the limit the broker advertised at connect time
3. Each chunk is assigned a unique UUID and sequence number
4. Chunks are sent to the broker individually
5. Each chunk consumes one flow control permit

Compression runs before the split, so a chunk carries a slice of the compressed message rather than being compressed on its own. This is how the Java client frames chunks, so a compressed chunked message can cross between the two. It also means a payload that compresses to under max_message_size is sent whole and never chunked.

Consumer Side

The consumer automatically handles chunk assembly:

1. Chunk Reception: Each chunk arrives as a separate broker message
2. Buffering: Chunks are buffered in memory until all chunks for a message arrive
3. Assembly: Once complete, chunks are reassembled into the original payload
4. Delivery: The complete message is delivered to your handle_message/2 callback

def handle_message(%Pulsar.Message{} = message, state) do
  # message.payload contains the complete, reassembled payload
  # message.chunk_metadata indicates if this was a chunked message

  case message.chunk_metadata do
    %{chunked: true, complete: true, num_chunks: n} ->
      IO.puts("Received complete chunked message with #{n} chunks")

    %{chunked: true, complete: false, error: reason} ->
      IO.puts("Received incomplete chunked message: #{reason}")

    nil ->
      IO.puts("Received non-chunked message")
  end

  {:ok, state}
end

Chunked Message Metadata

The Pulsar.Message struct provides information about chunked messages:

• chunk_metadata - Contains chunking information:
  • chunked: true - Indicates this was a chunked message
  • complete: true|false - Whether all chunks were received
  • uuid - Unique identifier for the chunked message
  • num_chunks - Total number of chunks (for complete messages)
  • received_chunks - Number of chunks received (for incomplete messages)
  • error - Reason for incompleteness (if incomplete)

A chunked message is assembled before the callback sees it, so payload is the complete payload and message_id covers every chunk: acknowledging it acknowledges them all.

Pulsar.Message's accessors — producer_name/1, key/1, properties/1 and the rest — answer the same way for a chunked message as for any other. Only the raw field reflects the split, holding a list of protocol structs per chunk rather than a single one.

Configuration Options

Producer Configuration

{Pulsar.Client,
 host: "pulsar://localhost:6650",
 producers: [
   [topic: "my-topic",
    name: :my_producer,
    chunking_enabled: true,        # Enable chunking (default: false)
    max_message_size: 1024 * 1024  # Split messages larger than 1MB (default: 5MB)
   ]
 ]}

Consumer Configuration

{Pulsar.Client,
 host: "pulsar://localhost:6650",
 consumers: [
   [topic: "my-topic",
    subscription_name: "my-sub",
    callback_module: MyConsumer,

    # Chunking-related options:
    max_pending_chunked_messages: 10,                        # Max concurrent chunked messages (default: 10)
    expire_incomplete_chunked_message_after: 60_000,         # Timeout in ms (default: 60s)
    chunk_cleanup_interval: 30_000                           # Cleanup check interval in ms (default: 30s)
   ]
 ]}

Configuration Details

• max_pending_chunked_messages: Maximum number of incomplete chunked messages to buffer simultaneously. If this limit is reached and a new chunked message arrives, the oldest incomplete message is evicted and delivered as incomplete with error: :queue_full.

• expire_incomplete_chunked_message_after: How long to wait for all chunks before timing out. Expired messages are delivered as incomplete with error: :expired.

• chunk_cleanup_interval: How often to check for and clean up expired chunked messages. Set to false to disable automatic cleanup (not recommended for production); nil is accepted as an alias. Keep it below expire_incomplete_chunked_message_after, since an expired chunk is only released on the next sweep.

Handling Incomplete Chunks

Chunks may not complete for several reasons:

1. Expiration: Not all chunks arrived within the timeout period
2. Queue overflow: Too many concurrent chunked messages

Incomplete chunks are delivered to your callback with complete: false. Their payload is whatever chunks did arrive, concatenated, and under :compression those are still compressed: a message only decompresses once all of its chunks are back together, so a partial one cannot be decompressed at all. Treat the payload of an incomplete message as opaque.

def handle_message(%Pulsar.Message{chunk_metadata: %{complete: false, error: reason, received_chunks: n}}, state) do
  Logger.warning("Incomplete chunk: #{reason}, received #{n} chunks")

  # Return error to trigger redelivery
  {:error, :incomplete_chunk, state}
end

Flow Control and Permits

Flow control permits are only decremented when messages are assembled and delivered to your callback:

• Individual chunks arriving: No permits are decremented yet
• Chunked message completed: Decrements N permits (where N = number of chunks)
• Chunked message expired/evicted: Decrements M permits (where M = number of chunks received)
• Non-chunked message: Decrements 1 permit

The Pulsar.Message.num_broker_messages/1 helper returns the correct permit count:

# Non-chunked message
Pulsar.Message.num_broker_messages(message) # => 1

# Complete chunked message with 3 chunks
Pulsar.Message.num_broker_messages(message) # => 3

# Incomplete chunked message with 2 out of 3 chunks received
Pulsar.Message.num_broker_messages(message) # => 2

This ensures that flow control accurately reflects the number of broker messages consumed, regardless of whether messages are chunked or not.

Helper Functions

The Pulsar.Message module provides helpers for working with chunked messages:

# Check if message is chunked
Pulsar.Message.chunked?(message) # => true for chunked, false otherwise

# Check if chunked message is complete
Pulsar.Message.complete?(message) # => true if complete, false if incomplete

# Get maximum redelivery count (max across all chunks for chunked messages)
redelivery_count = Pulsar.Message.redelivery_count(message)

# Get number of broker messages consumed (for flow control)
num_permits = Pulsar.Message.num_broker_messages(message)

Example: Complete Chunked Message Flow

{:ok, _pid} = Pulsar.Client.start_link(host: "pulsar://localhost:6650")

# Producer sends large message
{:ok, producer} = Pulsar.Producer.start(
  "large-files",
  chunking_enabled: true,
  max_message_size: 1024 * 1024  # 1MB chunks
)

# Send 5MB file
large_payload = File.read!("large_file.dat")  # 5MB
{:ok, _msg_id} = Pulsar.Producer.send(producer, large_payload)
# Producer automatically splits into 5 chunks

# Consumer receives and assembles
defmodule MyConsumer do
  use Pulsar.Consumer.Callback

  def handle_message(%Pulsar.Message{} = message, state) do
    if Pulsar.Message.chunked?(message) and Pulsar.Message.complete?(message) do
      # message.payload contains complete 5MB file
      num_chunks = message.chunk_metadata.num_chunks
      IO.puts("Received complete file in #{num_chunks} chunks")
      process_file(message.payload)
      {:ok, state}
    else
      # Regular non-chunked message or incomplete chunked message
      {:ok, state}
    end
  end
end

Telemetry Events

The consumer emits telemetry events for chunk lifecycle:

Each of these carries uuid alongside the topic, base_topic, partition, subscription_name and consumer_id that every consumer event carries, and the two that give up on a message also carry reason. received_chunks against num_chunks says how much of the message had arrived before it was dropped, and age_ms on :complete is how long assembly took, which is what :expire_incomplete_chunked_message_after should be set against.

The producer emits [:pulsar, :producer, :chunk, :start], :sent and :complete, all carrying uuid alongside the topic, base_topic, partition, producer_id and producer_name that every producer event carries. Their total_size and chunk_size count the bytes actually sent, so with :compression set they describe the compressed message rather than the payload handed to Pulsar.Producer.send/3. On the consumer side :complete's total_size is the reassembled message after decompression, so the two do not line up when compression is on.

:topic names a single partition and :base_topic the topic it belongs to, so one set of events both aggregates over a partitioned topic and breaks down by partition. They are equal, and :partition is nil, when the topic is not partitioned.

See the Telemetry documentation for more details on monitoring chunked messages.