[PIP] Add `Pause()` / `Resume()` to `Consumer`

[PavelZeger]

Gap

Java exposes:

void pause();
void resume();

on every Consumer. While paused, the consumer stops issuing flow permits to the broker, so no new messages are delivered. Already-delivered (in-flight) messages are unaffected.
The Go client has no equivalent. The closest workaround is closing and re-subscribing, which is heavyweight (re-resolves the topic, redelivers any unacked, may rebalance other consumers in KeyShared/Failover).

Why it's important

Pause/resume is the standard pattern for backpressure when downstream is temporarily unavailable: a database is failing over, a feature flag is in the "halt ingestion" state, etc. It is also useful for orderly shutdown - "stop accepting new work, finish what's in flight, then close."

Proposed Go API

type Consumer interface {
  // ... existing methods ...

  // Pause stops the consumer from requesting more messages from the broker.
  // In-flight messages already delivered to this consumer remain available
  // via Receive() / Chan(); ack/nack still work. Idempotent.
  Pause()

  // Resume reverses Pause. Idempotent.
  Resume()
}

A small accessor is also useful for tests and for users who want to expose the state via metrics:

// IsPaused reports whether Pause() has been called more recently than Resume().
IsPaused() bool

(Java doesn't expose an isPaused getter, but it's cheap and harmless and helps reduce flaky tests.)

Implementation

The Go consumer maintains an availablePermits counter on each partitionConsumer that gets refilled when the application drains the queue. Pausing means: stop sending CommandFlow until resumed.

// partitionConsumer fields:
paused atomic.Bool

// In the place that decides whether to send a flow update:
if pc.paused.Load() {
  return
}

// Public API (consumer_impl.go):
func (c *consumer) Pause() {
  for _, pc := range c.consumers {
      pc.paused.Store(true)
  }
}

func (c *consumer) Resume() {
  for _, pc := range c.consumers {
      if pc.paused.CompareAndSwap(true, false) {
          pc.requestMorePermits() // a small helper that re-checks thresholds
      }
  }
}

Pause must be applied per partition consumer. The multi-topic and regex consumers must propagate to all underlying partition consumers.

Edge cases

• Pausing right before a reconnect: when the partition consumer reconnects, it must re-check paused and not blindly send a fresh flow command.
• KeyShared semantics: pausing a single consumer in a key-shared subscription will cause its keys to be redistributed to other consumers only if it disconnects. Pause does not change membership, so it just builds a backlog on that key range. This should be documented as well.
• Interaction with BatchReceive (suggestion 01): a paused consumer should return an empty batch on timeout rather than blocking forever.

References

• Java: Consumer.pause() / Consumer.resume() in pulsar-client-api/.../api/Consumer.java.

Are you willing to submit a PR?

• I'm willing to submit a PR!

[PavelZeger]

Hi @nodece , @crossoverJie , @shibd , @freeznet , @RobertIndie ! Appreciate for a review of this PIP.

[crossoverJie]

Thanks for the proposal. I think adding Pause() / Resume() to the Go consumer makes sense and would close a useful parity gap with the Java client.

A few details may need to be clarified before implementation:

1. Pause() should probably be documented as “stop issuing new flow permits”, not “stop receiving immediately”. The client may
   already have locally buffered messages, broker-dispatched messages in flight, or previously granted permits. So users may still
   observe messages from Receive() / Chan() after calling Pause() until those are drained.
2. Reconnect needs explicit handling. In the current Go implementation, the dispatcher sends initial permits when it receives a new
   connection event. If the consumer is paused, reconnect should not blindly send those initial permits. Otherwise a paused consumer can
   start receiving again after reconnect.
3. EnableZeroQueueConsumer should be covered. In zero-queue mode, Receive() itself triggers one permit. The proposal should
   define the expected behavior when paused, for example: Receive(ctx) blocks until Resume() or until the context is cancelled.
4. Regex consumers and dynamic topic discovery need to inherit the paused state. If a regex consumer is paused and later discovers/
   subscribes to a new topic, the newly created internal consumer should also start paused.
5. IsPaused() should probably be defined at the top-level consumer as “whether Pause() was called more recently than Resume()”.
   Deriving it from all partition consumers may be ambiguous, especially with regex/multi-topic consumers and dynamically added
   partitions.
6. The subscription-type behavior should be documented a bit more broadly. For example, in KeyShared and Failover, pause does not
   change membership or trigger reassignment/failover because the consumer remains connected. It only affects flow permits. In Shared,
   dispatching should naturally move away from the paused consumer after its permits are exhausted.
   Overall, I support the API, but I think the proposal should make the exact semantics around permits, reconnect, zero-queue consumers,
   and dynamic consumers explicit. That would make the implementation easier to review and prevent surprising behavior.

[nodece]

Good feature, +1

[PavelZeger]

Thanks @nodece and @crossoverJie. Let me go through them and pin down the semantics before I open the PR.

1. Pause() semantics

Agreed. I'll reword the API doc to match Java: pausing only stops sending CommandFlow. Locally buffered messages, in-flight messages, and previously granted permits are all unaffected and still come out of Receive() / Chan().

2. Reconnect must respect paused

Right - the dispatcher sends initial permits on every new connection (https://github.com/apache/pulsar-client-go/blob/master/pulsar/consumer_partition.go#L1872-L1885). I'll put the paused check in flowIfNeed and at the reconnect branch so every flow path - initial connect, reconnect, dispatcher refill, ack refill - goes through the same gate:

// partitionConsumer
paused atomic.Bool

// flowIfNeed (consumer_partition.go:308)
func (p *availablePermits) flowIfNeed() {
    if p.pc.paused.Load() {
        return
    }
    // existing threshold logic
}

// dispatcher reconnect branch (consumer_partition.go:1882)
if !pc.paused.Load() {
    if err := pc.internalFlow(initialPermits); err != nil && !pc.options.enableZeroQueueConsumer {
        pc.log.WithError(err).Error("unable to send initial permits to broker")
    }
}

A Pause() before a reconnect means no flow command goes out on the new connection. Resume() then sends a fresh batch sized to currentQueueSize.

3. Zero-queue consumer

Good point - Receive() itself triggers the one permit (https://github.com/apache/pulsar-client-go/blob/master/pulsar/consumer_zero_queue.go#L132). When paused, Receive(ctx) will block
until Resume() is called or ctx is cancelled. I'll use a resumedCh that Pause() creates and Resume() closes:

// zeroQueueConsumer
func (z *zeroQueueConsumer) Receive(ctx context.Context) (Message, error) {
    // ... existing state/lock checks ...
    if z.pc.paused.Load() {
        select {
        case <-z.pc.resumedCh: // closed by Resume()
        case <-ctx.Done():
            return nil, ctx.Err()
        case <-z.closeCh:
            return nil, newError(ConsumerClosed, "consumer closed")
        }
    }
    z.pc.availablePermits.inc()
    // existing receive loop
}

The partition consumer's paused field stays the single source of truth.

4. Regex / multi-topic - newly discovered topics inherit paused

Agreed. Both regexConsumer (https://github.com/apache/pulsar-client-go/blob/master/pulsar/consumer_regex.go) and multiTopicConsumer
(https://github.com/apache/pulsar-client-go/blob/master/pulsar/consumer_multitopic.go) discover topics over time. I'll track paused on the parent and apply it to each new child under the same consumersLock already guarding the map:

// regexConsumer.subscribe (consumer_regex.go:412)
func (c *regexConsumer) subscribe(topics []string, dlq *dlqRouter, rlq *retryRouter) {
    consumers := make(map[string]Consumer, len(topics))
    for ce := range subscriber(c.client, topics, c.options, c.messageCh, dlq, rlq) {
        if ce.err != nil {
            c.log.Warnf("Failed to subscribe to topic=%s", ce.topic)
            continue
        }
        consumers[ce.topic] = ce.consumer
    }

    c.consumersLock.Lock()
    defer c.consumersLock.Unlock()
    for t, child := range consumers {
        if c.paused.Load() {
            child.Pause()
        }
        c.consumers[t] = child
    }
}

Same shape for multiTopicConsumer. Dynamic partitions on a single consumer go through consumers atomic.Value in consumer_impl.go and pick up paused the same way.

5. IsPaused()

Top-level only, defined as "Pause() called more recently than Resume()". Internally just return c.paused.Load() on the parent. No attempt to derive it from child consumers — that would be racy with discovery.

6. Subscription-type behavior

I'll add this to the "Edge cases" section:

• Shared: dispatch moves off the paused consumer once its granted permits run out. No reassignment - it stays connected.
• KeyShared: pause does not reassign keys. The paused consumer's keys build up in the broker backlog until Resume().
• Failover: pause does not trigger failover. The consumer stays active; backlog grows until resumed.
• Exclusive: backlog grows until resumed.

---

PIP updates I can make based on this exchange:

• Reword Pause() docstring to include "previously granted permits"
• Add a zero-queue subsection under "Edge cases"
• Make the regex/multi-topic rule explicit for dynamic discovery, not just existing children
• Move paused to the top-level consumer struct (the current sketch only puts it on partitionConsumer)
• Expand the subscription-type notes beyond KeyShared as above

One more thing not raised in the review but worth flagging for the PR: BatchReceive interaction. A paused consumer should return an empty batch on the configured timeout rather than blocking forever. I'll cover this in the implementation but it doesn't need to change the API.

---

I'll open the PR after your confirmation of the proposed and discussed changes.

[crossoverJie]

@PavelZeger Thanks LGTM.