Pulsar.Consumer (Pulsar v3.0.0)

Copy Markdown View Source

A consumer subscribes to a topic and hands each message to a callback module.

This module is how you add, inspect and stop consumers. To declare them on a client instead, so they start and restart with it, see Pulsar.Client. The callback module they dispatch into is Pulsar.Consumer.Callback.

start/1 adds a consumer to a running client and stop/2 removes it. Operations target the logical consumer by its stable root or registered name without exposing its partition workers. await_ready/2 waits for its topology and configured workers when an operation must not observe asynchronous startup.

ack/2 and nack/2 acknowledge manually, from a process other than the worker that delivered the message. send_flow/3 grants permits to a worker or every worker behind the consumer root.

Options

  • :topic (String.t/0) - Required. Topic to subscribe to.

  • :subscription_name (String.t/0) - Required. Name of the subscription.

  • :callback_module (atom/0) - Required. Module implementing Pulsar.Consumer.Callback.

  • :client (atom/0) - Client the consumer belongs to. The default value is :default.

  • :name - Name the consumer group is registered under. Defaults to "<topic>-<subscription_name>". Consumers within it are named after the group and their index on the broker.

  • :subscription_type - How the subscription is shared between consumers. The default value is :shared.

  • :consumer_count (pos_integer/0) - Number of consumer processes to start for the topic, or for each partition. The default value is 1.

  • :init_args (term/0) - Passed to the callback module's init/1. The default value is [].

  • :flow_initial (non_neg_integer/0) - Permits granted to the broker on subscribe. 0 disables automatic flow control, leaving it to Pulsar.Consumer.send_flow/2. Permits belong to a worker instance, so replacement workers also start with 0 and must be granted permits again. The default value is 100.

  • :flow_threshold (non_neg_integer/0) - Outstanding permits at which more are requested. Ignored when :flow_initial is 0. The default value is 50.

  • :flow_refill (non_neg_integer/0) - Permits requested on each refill. Ignored when :flow_initial is 0. The default value is 50.

  • :initial_position - Where a new subscription starts reading. The default value is :latest.

  • :start_message_id (tuple of non_neg_integer/0, non_neg_integer/0 values) - Seek to a {ledger_id, entry_id} before reading.

  • :start_timestamp (non_neg_integer/0) - Seek to a publish time, in milliseconds since the epoch, before reading.

  • :durable (boolean/0) - Whether the broker persists the subscription's position. The default value is true.

  • :read_compacted (boolean/0) - Read only the latest value per key from a compacted topic. When enabled, messages compaction has replaced are filtered out of a batch rather than delivered. When disabled, the consumer can receive the original history, including superseded values. The default value is false.

  • :force_create_topic (boolean/0) - Create the topic if it does not exist. The default value is true.

  • :batch_index_ack_enabled (boolean/0) - Tell the broker which messages of a batch an ack was for, so that a nack redelivers only the rest of the entry instead of all of it. Costs one ack command per message rather than one per entry.

    Requires acknowledgmentAtBatchIndexLevelEnabled=true on the broker. Without it the broker ignores the set and acknowledges the whole entry, losing the messages batched alongside the acked one. Nothing in the protocol reports the setting, so this cannot be detected: Pulsar's shipped broker.conf enables it, standalone.conf does not.

    The default value is false.

  • :redelivery_interval (pos_integer/0) - Milliseconds between redelivery requests for negatively acknowledged messages. Absent by default, in which case they are not redelivered and a nacked message from a batch leaves its entry unacknowledged until it is acked or the consumer restarts.

  • :dead_letter_policy (keyword/0) - Diverts a message to another topic once it has been redelivered too often. Omit it entirely for no dead letter topic.

    The producer this needs runs under the consumer, so it restarts on its own and a dead letter topic that is unavailable leaves the message nacked rather than disturbing the subscription. A partitioned consumer diverts every partition into one dead letter topic.

    A diverted message keeps its key, ordering key, properties and event time, and gains REAL_TOPIC and ORIGIN_MESSAGE_ID properties naming where it came from.

    A batch diverts as one entry, so if any message in it fails to publish the entry is redelivered and the rest are diverted a second time. Deduplicate on ORIGIN_MESSAGE_ID if that matters.

    Diverting replaces delivery rather than accompanying it, so neither Pulsar.Consumer.Callback.handle_message/2 nor Pulsar.Consumer.Callback.handle_invalid_message/2 is called for a message that reaches the threshold.

    • :max_redelivery (pos_integer/0) - Required. Deliveries to attempt before diverting the message.

    • :topic (String.t/0) - Topic to divert to. Defaults to "<topic>-<subscription>-DLQ".

    • :producer - Options for the producer that publishes to the dead letter topic, such as :compression or :batch_enabled. Takes Pulsar.Producer's defaults otherwise. Its :topic, :client and :name come from the consumer and cannot be set here.

  • :max_pending_chunked_messages (pos_integer/0) - Incomplete chunked messages to hold before evicting the oldest. The default value is 10.

  • :expire_incomplete_chunked_message_after (pos_integer/0) - Milliseconds before an incomplete chunked message is given up on. The default value is 60000.

  • :chunk_cleanup_interval - Milliseconds between sweeps for expired chunked messages. false disables the sweep, leaving incomplete chunks to accumulate until the consumer restarts; nil is accepted as an alias for false. The default value is 30000.

  • :schema (keyword/0) - Schema to register with the subscription, as [type: atom, definition: term]. See Pulsar.Schema.

  • :partition_discovery_interval_ms - For a partitioned topic, how often to look for partitions added since startup. false disables later metadata checks, but not initial topic discovery or local recovery of groups that have stopped. The default value is 60000.

  • :startup_delay_ms (non_neg_integer/0) - Delay before a consumer subscribes. A broker that is not connected yet is retried, so this is only needed to stagger a large number of restarts. The default value is 0.

  • :startup_jitter_ms (non_neg_integer/0) - Random extra delay on top of :startup_delay_ms, to spread out restarts. The default value is 0.

Summary

Functions

ack(consumer, message_ids)

Acknowledges one or more messages, marking them as processed.

await_ready(consumer, opts \\ [])

Waits for a consumer and all its configured workers to be ready.

nack(consumer, message_ids)

Negatively acknowledges one or more messages, asking the broker to redeliver them.

send_flow(consumer, permits, opts \\ [])

Grants a consumer more flow permits.

start(opts)

Adds a consumer to a running client.

start(topic, subscription_name, callback_module, opts \\ [])

Same as start/1, with the three required options given positionally.

start_link(opts)

Starts a consumer, linked to the calling process.

stop(consumer, opts \\ [])

Stops a consumer, given its pid or its name.

topic(consumer)

Returns the topic a consumer is subscribed to.

Functions

ack(consumer, message_ids)

@spec ack(
  pid(),
  Pulsar.Protocol.Binary.Pulsar.Proto.MessageIdData.t()
  | [Pulsar.Protocol.Binary.Pulsar.Proto.MessageIdData.t()]
) :: :ok | {:error, term()}

Acknowledges one or more messages, marking them as processed.

Takes the pid of the worker that delivered them. Not a group or a name: an acknowledgement carries the consumer id of the worker the broker sent the message to, so no other worker can answer for it.

Manual acknowledgement is for handing a message to whatever actually processes it and acknowledging once that work is done. Return {:noreply, state} from Pulsar.Consumer.Callback.handle_message/2 to leave the message unacknowledged, passing along the worker and the message id:

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

The job calls Pulsar.Consumer.ack(consumer, message_id) when it finishes. It has to be that process and not the callback: every callback function runs inside its worker, so ack(self(), ...) is a GenServer call a process makes to itself, which exits with :calling_self and takes the consumer down.

Batched messages

The broker acknowledges entries, not the messages inside them, so acking a batched message only counts it off: its entry is acknowledged once every message in it has been acked. The call is unchanged, but a message left unacked holds the ones batched with it, and a nack brings the whole entry back — including messages already acked from it.

:batch_index_ack_enabled narrows that to just the unacked messages, on brokers configured for it.

Every message must be acked eventually, or nacked with a :redelivery_interval configured to bring it back. One that is neither holds its entry's bookkeeping for the life of the consumer.

await_ready(consumer, opts \\ [])

@spec await_ready(
  pid() | String.t() | atom(),
  keyword()
) :: :ok | {:error, :not_found | :timeout}

Waits for a consumer and all its configured workers to be ready.

Takes the stable root returned by start/1 or its registered name. A named consumer is resolved repeatedly, so the wait also tolerates its client or resource branch restarting.

Readiness means initial topic discovery and topology construction have completed, and every configured worker has subscribed and initialized its callback. A worker that repeatedly fails initialization causes the wait to time out. Readiness is a snapshot: it does not guarantee continued broker availability or prevent a worker from restarting immediately afterward.

Options:

  • :timeout - maximum time to wait in milliseconds, or :infinity; defaults to 5 seconds
  • :client - client name or pid used to resolve a consumer name; defaults to :default

nack(consumer, message_ids)

@spec nack(
  pid(),
  Pulsar.Protocol.Binary.Pulsar.Proto.MessageIdData.t()
  | [Pulsar.Protocol.Binary.Pulsar.Proto.MessageIdData.t()]
) :: :ok | {:error, term()}

Negatively acknowledges one or more messages, asking the broker to redeliver them.

Takes the pid of the worker that delivered them, on the same terms as ack/2.

Redelivered messages that exceed :max_redelivery go to the dead letter topic when :dead_letter_policy is configured, whether they were acknowledged manually or not.

send_flow(consumer, permits, opts \\ [])

@spec send_flow(pid() | String.t() | atom(), pos_integer(), keyword()) ::
  :ok | {:error, term()}

Grants a consumer more flow permits.

Only needed when :flow_initial is 0, which turns off automatic flow control.

Takes the stable consumer root, one of its worker pids, or its name. Every live worker behind a root is granted the permits, and the first refusal is returned — retrying by name is safe, since a worker that already holds permits is only over-credited, and a worker that refused has usually been replaced by one with a different pid. If a configured group has no live worker, the other groups are granted permits before {:error, :no_consumers_available} is returned.

A consumer with no workers is an error rather than a silent success: nothing was granted, so nothing will be delivered. Permits belong to individual worker instances; a replacement starts with the configured :flow_initial and therefore needs another grant when that value is 0.

start(opts)

@spec start(keyword()) :: DynamicSupervisor.on_start_child()

Adds a consumer to a running client.

For consumers whose set is only known at runtime. Prefer the client's :consumers for ones known up front: a consumer added here is not recreated if the client restarts.

Returns once the stable consumer supervisor has been registered. Topic discovery and worker initialization continue asynchronously; worker-dependent operations return {:error, :not_ready} until discovery completes.

start(topic, subscription_name, callback_module, opts \\ [])

@spec start(String.t(), String.t(), module(), keyword()) ::
  DynamicSupervisor.on_start_child()

Same as start/1, with the three required options given positionally.

start_link(opts)

@spec start_link(keyword()) :: Supervisor.on_start()

Starts a consumer, linked to the calling process.

Returns the stable consumer root. See the module documentation for the options.

stop(consumer, opts \\ [])

@spec stop(
  pid() | String.t() | atom(),
  keyword()
) :: :ok | {:error, :not_found}

Stops a consumer, given its pid or its name.

A pid must be the stable root returned by start/1 or start_link/1. Worker pids used for acknowledgement are not consumer roots and return {:error, :not_found} here.

A root started as a static child will be restarted by its supervisor; remove that child from the supervision tree instead.

topic(consumer)

@spec topic(pid()) :: String.t() | {:error, :not_found}

Returns the topic a consumer is subscribed to.

A stable root returns the topic it was configured with. This remains the exact topic for a consumer started on a concrete partition such as topic-partition-3; a root that discovered a partitioned base topic returns that base topic.

A worker returns its resolved topic, which is the concrete partition for a partitioned consumer.