# `OffBroadway.Pulsar.Producer`
[🔗](https://github.com/efcasado/off_broadway_pulsar/blob/main/lib/off_broadway/pulsar/producer.ex#L1)

A Broadway producer for Apache Pulsar.

This producer receives messages from Pulsar topics and forwards them to the
Broadway pipeline. It implements flow control using Pulsar's permit window
mechanism, which proactively requests batches of messages rather than
requesting per-message.

The connection belongs to your application: supervise a `Pulsar.Client` and this
producer attaches its consumers to it.

Each producer stage owns the consumers it starts. See "Consumer ownership" in
`start_link/1`.

See `start_link/1` for detailed configuration options.

# `start_link`

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

## Configuration

* `:client` (`t: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` (`t: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` (`t: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` (`t:pos_integer/0`) - Permits each consumer grants when it subscribes. The default value is `100`.

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

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

Options are validated in `c: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
`c: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"
      }
    ]

---

*Consult [api-reference.md](api-reference.md) for complete listing*