You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "PavelZeger (via GitHub)" <gi...@apache.org> on 2026/05/21 14:54:45 UTC
[I] [PIP] Add `Pause()` / `Resume()` to `Consumer` [pulsar-client-go]
PavelZeger opened a new issue, #1504:
URL: https://github.com/apache/pulsar-client-go/issues/1504
## Gap
Java exposes:
```java
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
```go
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:
```go
// 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.
```go
// 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`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [PIP] Add `Pause()` / `Resume()` to `Consumer` [pulsar-client-go]
Posted by "crossoverJie (via GitHub)" <gi...@apache.org>.
crossoverJie commented on issue #1504:
URL: https://github.com/apache/pulsar-client-go/issues/1504#issuecomment-4514807845
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [PIP] Add `Pause()` / `Resume()` to `Consumer` [pulsar-client-go]
Posted by "PavelZeger (via GitHub)" <gi...@apache.org>.
PavelZeger commented on issue #1504:
URL: https://github.com/apache/pulsar-client-go/issues/1504#issuecomment-4518838790
Thanks @nodece and @crossoverJie for the proposal's review. I'll prepare the PR with mentioned details above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [PIP] Add `Pause()` / `Resume()` to `Consumer` [pulsar-client-go]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on issue #1504:
URL: https://github.com/apache/pulsar-client-go/issues/1504#issuecomment-4517244885
Good feature, +1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [PIP] Add `Pause()` / `Resume()` to `Consumer` [pulsar-client-go]
Posted by "PavelZeger (via GitHub)" <gi...@apache.org>.
PavelZeger commented on issue #1504:
URL: https://github.com/apache/pulsar-client-go/issues/1504#issuecomment-4524921989
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:
```go
// 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:
```go
// 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:
```go
// 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [PIP] Add `Pause()` / `Resume()` to `Consumer` [pulsar-client-go]
Posted by "crossoverJie (via GitHub)" <gi...@apache.org>.
crossoverJie commented on issue #1504:
URL: https://github.com/apache/pulsar-client-go/issues/1504#issuecomment-4533415764
@PavelZeger Thanks LGTM.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [PIP] Add `Pause()` / `Resume()` to `Consumer` [pulsar-client-go]
Posted by "PavelZeger (via GitHub)" <gi...@apache.org>.
PavelZeger commented on issue #1504:
URL: https://github.com/apache/pulsar-client-go/issues/1504#issuecomment-4509566826
Hi @nodece , @crossoverJie , @shibd , @freeznet , @RobertIndie ! Appreciate for a review of this PIP.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org