Consumer on a non-persistent topic receives only the first message (MessageID (0,0) dedup)

On a **non-persistent** topic, a `pulsar-client-go` consumer that acknowledges each message receives only the first one and then silently drops the rest — no error is returned, the consumer just blocks in `Receive()`. Persistent topics are unaffected.

_Searched existing issues: yes — I could not find an existing report of this in `apache/pulsar-client-go`._

#### Expected behavior

Every message published to a non-persistent topic is delivered to a connected consumer.

#### Actual behavior

A consumer that acknowledges each message before receiving the next (e.g. a synchronous request-reply loop) receives **only the first** message and then blocks forever. With a producer that streams messages independently of consumption, the loss is partial but severe — a large fraction is dropped and the exact count varies with timing (cf. apache/pulsar#1967, which observed ~9 of 3000).

Either way the loss is inside the client: the broker confirms the messages *were* dispatched — the subscription's `msgOutCounter` advances and the consumer's `availablePermits` decreases for the missing messages.

#### Steps to reproduce

Run against any broker (a standalone is fine) with `pulsar-client-go` v0.20.0:

```go
package main

import (
"context"
"fmt"
"os"
"time"

"github.com/apache/pulsar-client-go/pulsar"
)

func main() {
client, err := pulsar.NewClient(pulsar.ClientOptions{URL: os.Getenv("URL")})
if err != nil {
panic(err)
}
defer client.Close()

const topic = "non-persistent://public/default/repro"

consumer, err := client.Subscribe(pulsar.ConsumerOptions{
Topic: topic,
SubscriptionName: "sub",
Type: pulsar.Shared,
})
if err != nil {
panic(err)
}
defer consumer.Close()

producer, err := client.CreateProducer(pulsar.ProducerOptions{Topic: topic})
if err != nil {
panic(err)
}
defer producer.Close()

// Synchronous: send one, receive it, ack it, then send the next.
for i := 0; i < 10; i++ {
producer.Send(context.Background(), &pulsar.ProducerMessage{
Payload: []byte(fmt.Sprintf("msg-%d", i)),
})
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
msg, err := consumer.Receive(ctx)
cancel()
if err != nil {
fmt.Printf("received %d/10 messages, then blocked\n", i)
return
}
fmt.Printf("got: %s\n", string(msg.Payload()))
consumer.Ack(msg) // acking is what makes the NEXT message look like a duplicate
}
fmt.Println("received 10/10 messages")
}
```

Expected output: `got: msg-0` … `got: msg-9` (`received 10/10`). Actual output (v0.20.0): `got: msg-0`, then `received 1/10 messages, then blocked`. (Confirmed on a Pulsar 5.0.0-M1 broker; adding `AckGroupingOptions: &pulsar.AckGroupingOptions{MaxSize: 1}` to the `ConsumerOptions` makes it print all 10 — see Suggested fix.)

#### System configuration

- **pulsar-client-go version**: v0.20.0 (also reproduced on v0.14.0)
- **Pulsar version** (broker): any; observed on 5.0.0-M1
- **Go version**: 1.26

#### Root cause

1. The broker gives every non-persistent message the same MessageID `(ledgerId=0, entryId=0)` — non-persistent messages are never stored, so there is no real ledger/entry to point at. In the broker, `NonPersistentTopic.publishMessage` builds entries via `EntryImpl.create(0L, 0L, ...)`.
2. In the client, the message-receive path dedups by MessageID: `pulsar/consumer_partition.go` — `if pc.ackGroupingTracker.isDuplicate(msgID) { skippedMessages++; continue }` (≈ line 1434). The default (timed) tracker's `isDuplicate` (`pulsar/ack_grouping_tracker.go`, `timedAckGroupingTracker.isDuplicate`) returns `true` when the `(ledgerId, entryId)` key is already present in its `pendingAcks` map.

Because every non-persistent MessageID is `(0,0)`, once the first message is acknowledged its key is in `pendingAcks`, and any later message that arrives before the tracker's flush (default 100ms) is judged a duplicate and dropped before it reaches `Receive()`. A synchronous consumer sends the next message immediately, so it is always inside that window — hence "only the first".

The MessageID-based dedup was introduced in #957 ("Support grouping ACK requests by time and size"), which ported the grouping-ack tracker from the Java client but not its non-persistent counterpart, so non-persistent topics have never been guarded against this.

#### Suggested fix

Don't apply MessageID-based dedup on non-persistent topics: they have no redelivery (so there is nothing to dedup against) and their MessageIDs are not unique. `newAckGroupingTracker` currently chooses the tracker based only on `AckGroupingOptions.MaxSize`, never on the topic's persistence domain. Selecting the no-op `immediateAckGroupingTracker` (whose `isDuplicate` already returns `false`) for `non-persistent://` topics would fix it.

The other official clients already special-case this:

- **Java** — `ConsumerImpl` selects `NonPersistentAcknowledgmentGroupingTracker` (its `isDuplicate()` is `return false;`) in the non-persistent branch of `topicName.isPersistent()`.
- **C++** — `ConsumerImpl.cc` uses the base `AckGroupingTracker` (whose `isDuplicate()` returns `false`) for non-persistent topics, rather than the `AckGroupingTrackerEnabled` it uses for persistent ones.

**Workaround** (application side, until fixed): pass `AckGroupingOptions{MaxSize: 1}` to `Subscribe` for non-persistent consumers, which selects the immediate tracker and disables the dedup. (Verified: it makes the reproducer above deliver all 10 messages.)

#### Prior art

The Java client had the same class of bug and fixed it: [apache/pulsar#1967](https://github.com/apache/pulsar/issues/1967) — "Non-persistent topic drop too many messages" (`type/bug`, `area/client`, **closed**). The diagnosis there matches this one exactly — a maintainer noted that non-persistent topics "end up using 0 as ledger id and entry id, so the [grouping-ack] optimization treats those messages as already 'delivered'." It was fixed in the Java client by [apache/pulsar#1994](https://github.com/apache/pulsar/pull/1994). The equivalent guard was never added to `pulsar-client-go`.

Start with the duplicate check in pulsar/consumer_partition.go and trace tracker selection through pulsar/ack_grouping_tracker.go, including newAckGroupingTracker and immediateAckGroupingTracker. Run the provided non-persistent-topic reproducer with the default acknowledgment grouping. Done means acknowledged messages on non-persistent topics are not dropped and the reproducer receives all 10 messages without the MaxSize: 1 workaround.

Written by the indexing model from the issue text.