Upgrading to 3.0

This guide is a code-migration checklist for applications upgrading from 2.x. For details about new features and internal design, follow the links in Where to look next.

First, update the dependency:

{:pulsar, "~> 3.0", hex: :pulsar_elixir}

Most required changes produce a compiler warning or a startup error. Four do not:

• config :pulsar is no longer read. Move it into your supervision tree.
• Consumer callback init/1 became init/2. A stale init/1 compiles but is never called.
• Partition keys use a different hash by default. Plan the routing change before deployment.
• Compressed chunk framing changed. Drain compressed chunked messages produced by 2.x before upgrading.

1. Move configuration into your supervision tree

3.0 does not start a client, consumer, or producer from application configuration. Replace config :pulsar with a Pulsar.Client child in your application supervisor:

# 2.x - config/config.exs
config :pulsar,
  host: "pulsar://localhost:6650",
  consumers: [
    orders: [
      topic: "persistent://public/default/orders",
      subscription_name: "order-service",
      callback_module: MyApp.OrderHandler
    ]
  ],
  producers: [
    audit: [topic: "persistent://public/default/audit"]
  ]

# 3.x - your application supervisor
children = [
  {Pulsar.Client,
   host: "pulsar://localhost:6650",
   consumers: [
     [topic: "persistent://public/default/orders",
      subscription_name: "order-service",
      callback_module: MyApp.OrderHandler,
      name: :orders]
   ],
   producers: [
     [topic: "persistent://public/default/audit", name: :audit]
   ]}
]

Supervisor.start_link(children, strategy: :one_for_one)

:consumers and :producers are now lists of keyword lists. Move the old keyword-list key into each resource's :name option.

Names are exact registry keys in 3.0. If existing code looks up "orders", configure name: "orders"; name: :orders is a different key. An unnamed client is registered as :default, which is also the default :client used by consumers and producers.

For multiple clusters, add one named client child per cluster:

children = [
  {Pulsar.Client, name: :east, host: "pulsar://east.example.com:6650"},
  {Pulsar.Client, name: :west, host: "pulsar://west.example.com:6650"}
]

Move the remaining application settings to the resource they configure:

2. Replace calls to the Pulsar module

The public API is split between Pulsar.Client, Pulsar.Consumer, and Pulsar.Producer:

For example:

# 2.x
Pulsar.start_consumer(topic, subscription, MyApp.Handler, subscription_type: :Shared)
Pulsar.send(:audit, payload)

# 3.x
Pulsar.Consumer.start(topic, subscription, MyApp.Handler, subscription_type: :shared)
Pulsar.Producer.send(:audit, payload)

ack/2 and nack/2 take the worker pid that delivered the message, not the logical consumer name. If acknowledgement happens elsewhere, pass self() and message.message_id out of the callback:

def handle_message(message, state) do
  MyApp.Jobs.enqueue(message.payload, ack: {self(), message.message_id})
  {:noreply, state}
end

# In another process:
Pulsar.Consumer.ack(worker, message_id)

Do not call ack(self(), ...) synchronously from the callback: the callback runs in the worker, so that would be a GenServer.call to itself.

Pulsar.Client.consumers/1 and producers/1 return one topology root per logical resource, rather than listing its individual workers.

3. Update option values

Enum option values are lowercase and snake_case in 3.0:

Options are validated at startup, so old or misspelled values now raise.

Changes used by some applications

• Remove :producer_count. A producer now has one worker per partition. :consumer_count remains supported.

• Remove :host from Pulsar.Reader.stream/2. Start a client first and use :client to select it when necessary:

  {:ok, _client} = Pulsar.Client.start_link(host: "pulsar://localhost:6650")
  Pulsar.Reader.stream(topic, start_position: :earliest)

• If you used flow_initial: 0 for fully manual flow control, add a custom :flow_policy. flow_initial: 0 with the default :auto policy is rejected:

  Pulsar.Consumer.start(topic, subscription, MyApp.Handler,
    flow_initial: 0,
    flow_policy: {MyApp.Flow, :decide, []}
  )
  
  defmodule MyApp.Flow do
    def decide(_flow), do: :ok
  end

  Grant permits from another process with Pulsar.Consumer.send_flow/3. See Pulsar.Consumer for custom policy details.

4. Update consumer callbacks

Change callback init/1 to init/2. The second argument contains the resolved topic, partition, subscription, and consumer identity:

# 2.x
def init(opts) do
  {:ok, %{max: Keyword.get(opts, :max, 1000)}}
end

# 3.0
def init(opts, context) do
  {:ok, %{max: Keyword.get(opts, :max, 1000), partition: context.partition}}
end

Search every callback module for def init(. Because use Pulsar.Consumer.Callback supplies a default init/2, leaving an old init/1 in place compiles but silently uses the default state instead.

Invalid messages now go to the optional handle_invalid_message/2 callback. Implement it if your application must record or divert checksum, decompression, or framing failures:

def handle_invalid_message(message, state) do
  MyApp.InvalidMessages.record(message.validation_error, message.raw)
  {:ok, state}
end

The default implementation logs and acknowledges the invalid message.

5. Update message access

Pulsar.Message no longer exposes wire-protocol fields at its top level:

Use accessors for values that should work across normal, batched, and chunked messages:

# 2.x
key = message.single_metadata && message.single_metadata.partition_key
producer = message.metadata.producer_name

# 3.0
key = Pulsar.Message.key(message)
producer = Pulsar.Message.producer_name(message)

Other accessors include publish_time/1, event_time/1, ordering_key/1, properties/1, redelivery_count/1, message_id_string/1, chunked?/1, complete?/1, and valid?/1.

Treat message.message_id as opaque and pass it directly to ack/2 or nack/2. It can be a list for a chunked message.

6. Update startup and send-result handling

Consumer and producer startup is asynchronous. If the next operation requires a ready resource, wait explicitly:

{:ok, _producer} = Pulsar.Producer.start(topic: topic, name: :audit)
:ok = Pulsar.Producer.await_ready(:audit, timeout: 10_000)

Until initialization finishes, operations can return {:error, :not_ready}.

Update producer result matches for these 3.0 return values:

case Pulsar.Producer.send(:audit, payload) do
  {:ok, :deduplicated} ->
    :ok

  {:ok, message_id} ->
    {:published, message_id}

  {:error, :producer_queue_full} ->
    {:retry, :overloaded}

  {:error, :send_timeout} ->
    {:retry, :unknown_publish_outcome}
end

:send_timeout does not prove that the broker rejected the message; it means no acknowledgement arrived before the deadline. A missing named producer now returns {:error, :not_found} instead of {:error, :producer_not_found}.

7. Plan deployment compatibility

These changes may require rollout coordination even after the code compiles.

Partition-key routing

3.0 changes the default partition-key hash from :erlang.phash2 to :murmur3_32. This affects partitioned topics produced with :partition_key and can temporarily break per-key ordering during a mixed rollout.

Either drain the topic before switching, or preserve the 2.x routing during the rollout:

Pulsar.Producer.start(topic: topic, name: :orders, hashing_scheme: :phash2_legacy)

Move to :murmur3_32 after the old messages are drained. Partition keys must be binaries in 3.0.

Compressed chunked messages

If both :chunking_enabled and :compression are enabled, drain messages produced by 2.x and upgrade producers and consumers together. 3.0 does not read the old compressed chunk framing. Uncompressed chunked messages remain compatible.

:batch_enabled and :chunking_enabled can no longer be combined. Choose one; 2.x silently ignored chunking when both were enabled.

Startup delay

:startup_delay_ms and :startup_jitter_ms now default to 0 instead of 1000. Set them explicitly if your deployment relied on staggered consumer or producer startup.

Final checklist

• Move all config :pulsar values into supervised clients and resources.
• Replace calls to the old Pulsar API.
• Lowercase enum option values.
• Change every callback init/1 to init/2.
• Replace direct message metadata access with Pulsar.Message accessors.
• Remove :producer_count and reader :host options where used.
• Update readiness, send-result, and missing-producer matches.
• Plan partition-key and compressed-chunk migrations before deployment.

Where to look next

• Pulsar.Client, Pulsar.Consumer, and Pulsar.Producer document their options and API.
• Pulsar.Consumer.Callback documents callback return values and lifecycle events.
• The architecture guide covers ownership, startup, and recovery.
• The acknowledgement and redelivery, batching, chunking, schemas, and dead letter policies guides cover those features in depth.
• Broadway users should follow the off_broadway_pulsar 2.0 upgrade guide.