[feat][pip][consumer] Add Pause/Resume/Paused to Consumer

[PavelZeger]

Fixes #1504

Motivation

The Java client exposes pause() / resume() on every Consumer; the Go client has no equivalent. While paused, a consumer should stop issuing flow permits to the broker so no new
messages are delivered, while already-buffered, in-flight, and previously-granted messages remain available via Receive() / Chan() and ack/nack keep working.

Pause/resume is the standard backpressure pattern for when a downstream is temporarily unavailable (database failover, a "halt ingestion" feature flag, orderly shutdown that finishes in-flight work before closing). The only workaround today is closing and re-subscribing, which is heavyweight (re-resolves the topic, redelivers unacked messages, and can rebalance other consumers in KeyShared/Failover).

The semantics were reviewed and agreed in #1504.

Modifications

• Added Pause(), Resume(), and Paused() to the Consumer interface pulsar/consumer.go), with GoDoc matching the Java semantics.
• partitionConsumer (pulsar/consumer_partition.go):
• Added a paused flag and gated permit flow in flowIfNeed — the single place that returns permits to the broker.
• When paused, the dispatcher connect/reconnect branch stashes the initial grant into availablePermits instead of sending it, so reconnecting while paused does not resume delivery.
• resume() flushes only the permits actually owed (via flowIfNeed), mirroring Java's increaseAvailablePermits(cnx, 0), rather than granting a fresh full batch (which would over-grant after a mid-stream pause).
• Wired the API through all four Consumer implementations: consumer, zeroQueueConsumer, multiTopicConsumer, and regexConsumer. Paused() is reported at the top level ("Pause() called more recently than Resume()").
• Propagated the paused state to dynamically created children: new partitions (internalTopicSubscribeToPartitions) and newly discovered regex topics (regexConsumer.subscribe) inherit the parent's paused state at creation.
• For the zero-queue consumer, Receive() is unchanged: the existing permit is gated while paused, so Receive(ctx) blocks until Resume() (or the context is cancelled).

Deviations from the original PIP

The observable behavior agreed in #1504 is unchanged; four implementation details differ from the issue's wording. Resume() flushes only the permits actually owed (and the dispatcher stashes the initial grant while paused) instead of sending a full currentQueueSize batch, which would over-grant after a mid-stream pause. The zero-queue consumer reuses the same flow gate instead of a new resumedCh. The getter is named Paused() rather than IsPaused(). BatchReceive is out of scope, since the Go client doesn't have that method yet.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Added TestConsumerPauseResume, TestPartitionedConsumerPauseResume, TestZeroQueueConsumerPauseResume, and TestRegexConsumerPauseResumeInheritsNewTopics in pulsar/consumer_test.go. They verify that a paused consumer stops delivery before draining the full backlog, that Resume() delivers the remainder with no message loss, that Pause()/Resume() are idempotent, that Paused() reflects state, that a newly discovered regex topic inherits the paused state, and that a zero-queue Receive() blocks while paused and returns after resume.
• Verified locally against Apache Pulsar 4.0.3 standalone (all four tests pass, repeated runs stable);

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API: yes — adds Pause(), Resume(), and Paused() to the Consumer interface
• The schema: no
• The default values of configurations: no
• The wire protocol: no (uses the existing CommandFlow; pausing simply withholds it)

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? GoDocs (doc comments on the new Consumer interface methods)

[PavelZeger]

Hi @nodece and @crossoverJie . Appreciate for the review

[Copilot]

Pull request overview

This PR adds a pause/resume backpressure API to the Go Consumer to match the Java client’s semantics: when paused, the consumer stops issuing flow permits so the broker stops delivering new messages while already-buffered/in-flight messages remain consumable and ack/nack continues to work.

Changes:

• Added Pause(), Resume(), and Paused() to the public Consumer interface with detailed GoDoc describing semantics across subscription types.
• Implemented paused state propagation across consumer implementations, and gated flow-permit issuance in partitionConsumer (including reconnect initial permit handling).
• Added integration tests covering pause/resume behavior for standard, partitioned, zero-queue, and regex consumers.

Reviewed changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
pulsar/consumer.go	Extends the public Consumer interface with Pause/Resume/Paused and documents expected semantics.
pulsar/consumer_impl.go	Wires Pause/Resume/Paused through the main consumer and ensures newly discovered partitions inherit paused state.
pulsar/consumer_partition.go	Adds per-partition paused flag and gates flow-permit issuance (including reconnect initial grants).
pulsar/consumer_multitopic.go	Propagates Pause/Resume/Paused across multi-topic consumers.
pulsar/consumer_regex.go	Adds pause state to regex consumer and ensures newly discovered topics inherit paused state.
pulsar/consumer_zero_queue.go	Implements Pause/Resume/Paused for zero-queue consumer via its underlying partitionConsumer.
pulsar/consumer_test.go	Adds pause/resume tests and helper drain functions used by the new test cases.
pulsar/internal/pulsartracing/consumer_interceptor_test.go	Updates a mock consumer to satisfy the expanded Consumer interface.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated 3 comments.

[PavelZeger]

Additional note: there is a known edge case in the partitioned/regex path. If a new partition (or regex topic) is discovered at the exact moment someone calls Resume(), that new child can end up paused while Paused() says false. It fixes itself on the next Pause()/Resume(), and the timing window is tiny.
I'm leaving it as-is here on purpose. #1508 (StartPaused) fixes it at the root by creating new children already paused instead of pausing them after the fact - same as the Java client. A lock here would just get undone once #1508 will be executed.