Starts an OffBroadway.Pulsar producer process linked to the current process.

Configuration

• :client (atom/0) - Name of the Pulsar.Client to attach to. The client must already be running; see "Starting a client" below. The default value is :default.

• :topics - Required. The Pulsar topics to consume from, as a non-empty list of strings. One consumer is started per topic.

• :subscription (String.t/0) - Required. The subscription name.

• :active_state_callback - A {module, function, extra_args} tuple that receives active/passive state changes for Failover consumers. nil disables it. See "Failover Active State" below. The default value is nil.

• :consumer_opts (keyword/0) - Options forwarded to Pulsar.Consumer.start_link/1, applied to every topic. They are documented and validated by Pulsar.Consumer, which rejects unknown keys and applies its own defaults to whatever is left out — including subscription_type: :shared.

  The keys the producer sets itself are rejected here: :client, :topic, :subscription_name, :callback_module, :init_args, :name, :consumer_count, :flow_policy, :flow_initial, :flow_threshold, :flow_refill. Use producer: [concurrency: N] for the consumer count and the :flow_* options above for flow control.

  :batch_index_ack_enabled is worth enabling for Broadway, which routinely completes messages from one batch out of order: without it a failed message takes its whole entry back, and the siblings Broadway already handled are delivered again. It needs acknowledgmentAtBatchIndexLevelEnabled on the broker, which is the Pulsar 4.2 default.

  A nacked message is redelivered only when :redelivery_interval is set. Left unset, a message the pipeline fails is not retried.

  The default value is [subscription_type: :shared].

• :flow_initial (pos_integer/0) - Permits each consumer grants when it subscribes. The default value is 100.

• :flow_threshold (pos_integer/0) - Refill once outstanding permits reach this level. Must be less than :flow_initial. The default value is 50.

• :flow_refill (pos_integer/0) - Permits granted by each refill. The default value is 50.

Options are validated in Broadway.Producer.prepare_for_start/2, which runs before any stage starts, so Broadway.start_link/2 raises the configuration error itself rather than each stage crash-looping under its supervisor. They are validated again in GenStage.init/1, which also covers starting this producer outside a Broadway pipeline.

The client is checked per stage instead, since a stage that restarts has to find it again.

Flow Control

Each topic or partition consumer has its own Pulsar permit window. Broadway demand and the message buffer are shared by all consumers owned by one producer stage, so deliveries that arrive ahead of demand wait in that buffer.

Each consumer grants :flow_initial permits when it subscribes. Permits are charged when messages leave the buffer for the processors, not when they are acknowledged. When the remaining permits reach :flow_threshold, the producer grants another :flow_refill permits.

The approximate maximum window per consumer is max(flow_initial, flow_threshold + flow_refill). Account for every topic or partition consumer and every producer stage when estimating total read-ahead. Larger windows reduce refill overhead but increase buffered and unacknowledged messages; smaller windows may limit throughput.

Consumer ownership

Each producer stage starts one linked consumer root per topic. Pulsar supervises the partition groups and workers below each root; Broadway supervises the stage. A root exit makes Broadway restart the stage and recreate all of its roots, while stopping the stage stops its linked roots. Retryable worker failures remain local to the Pulsar topology.

Consumer roots use the client's broker infrastructure but are not children of its consumer DynamicSupervisor. Consequently, Pulsar.Client.consumers/1 does not list them. The stage owns them by pid, so replacing the client's consumer Registry does not affect them, although their former names no longer resolve. Stop the Broadway pipeline to stop them permanently.

A terminal subscription error can leave a root alive with a stopped group, and a worker that exits normally is not replaced, leaving a group alive with nothing consuming. The stage detects both and restarts instead of remaining healthy with nothing to consume.

A worker that is replaced is not one of those cases. The replacement registers under a new pid and the broker returns whatever the old worker left unacknowledged to it, so messages still in the pipeline under the old pid are dropped when they come to be acknowledged.

Message Metadata

Each Broadway.Message carries the following :metadata:

• :message_id - Opaque id the acknowledger acks with. A list for a chunked message
• :message_id_string - The id as Pulsar prints it (ledgerId:entryId:partition, plus the batch index when batched), for logging and correlation
• :topic - The resolved topic; the concrete partition for a partitioned topic
• :base_topic - The configured topic. Equal to :topic unless the topic is partitioned
• :partition - The partition index, or nil when the topic is not partitioned
• :subscription - The Pulsar subscription name
• :key - The partition key, or nil
• :ordering_key - The ordering key, or nil
• :properties - User properties, as a map
• :producer_name - The producer that published the message
• :publish_time - Broker publish timestamp, in milliseconds since the epoch
• :event_time - Application-set event time, or nil when unset
• :redelivery_count - How many times the broker has redelivered the message
• :raw - The underlying protocol structs, as a map of :command, :metadata, :single_metadata and :broker_metadata. Its shape is unstable and follows the wire protocol

These normalized fields are consistent for individual, batched and chunked messages.

Two kinds of message never reach the pipeline: an incomplete chunked message, whose payload is a fragment, and one that failed validation. Both are logged, acknowledged and dropped.

Failover Active State

Consumers using a :failover subscription report broker-provided active and passive state changes through the optional :active_state_callback. Configure the callback as a {module, function, extra_args} tuple. It is invoked as apply(module, function, [metadata | extra_args]), where metadata contains:

• :active_state - Either :active or :passive
• :topic - The topic, or individual partition topic, whose state changed
• :subscription - The Pulsar subscription name
• :consumer_pid - The underlying Pulsar consumer process

The callback runs synchronously in the underlying Pulsar consumer and should return promptly. Exceptions raised by the callback propagate and crash the consumer. Reports are best-effort state observations: the same state may be reported more than once, and consumer termination does not guarantee a final :passive report. Treat reports idempotently and monitor :consumer_pid if local work must stop when the consumer exits.

This signal is scoped to an individual topic or partition. It is not a distributed lock or fencing mechanism, and it does not mean messages already delivered to Broadway have finished processing.

With Broadway producer concurrency greater than one, multiple underlying consumers can report different states for the same topic and subscription at the same time. Track concurrent ownership observations by :consumer_pid, not by topic alone.

Starting a client

Supervise the client above the pipeline, so the connection outlives any one producer stage and is shared by all of them:

children = [
  {Pulsar.Client, host: "pulsar://localhost:6650"},
  MyApp.PulsarPipeline
]

producer: [
  module: {OffBroadway.Pulsar.Producer,
    topics: ["topic-a", "topic-b"],
    subscription: "my-subscription"
  }
]

Name the client to run more than one, or to consume from more than one cluster:

children = [
  {Pulsar.Client, name: :analytics, host: "pulsar://analytics:6650"},
  MyApp.AnalyticsPipeline
]

producer: [
  module: {OffBroadway.Pulsar.Producer,
    client: :analytics,
    topics: ["my-topic"],
    subscription: "my-subscription"
  }
]