[Bug] Transactional ACKs omit batch size, causing Shared consumers to stop receiving

#### Expected behavior

After a Shared consumer acknowledges every message in a producer batch with `AckWithTxn` and successfully commits the transaction, the broker's consumer `unackedMessages` count should return to zero. Subsequent messages should continue to be dispatched.

#### Actual behavior

With `EnableBatchIndexAcknowledgment=false`, successful transactional acknowledgments of batched messages leave an increasing unacknowledged-message count. The consumer eventually reaches `maxUnackedMessagesPerConsumer` and stops receiving, although the connection remains open and previous transactions committed successfully.

I reproduced this with the unmodified `github.com/apache/pulsar-client-go v0.21.0` and Pulsar 4.0.3. With three messages per producer batch and an unacked limit of 20, the count increases by two per committed batch. The next receive times out after ten committed batches:

```text
committed producer batch 0
broker subscription: {UnackedMessages:2 Consumers:[{Blocked:false}]}
committed producer batch 1
broker subscription: {UnackedMessages:4 Consumers:[{Blocked:false}]}
...
committed producer batch 9
broker subscription: {UnackedMessages:20 Consumers:[{Blocked:true}]}
broker subscription: {UnackedMessages:20 Consumers:[{Blocked:true}]}
panic: receive producer batch 10: context deadline exceeded
```

The batch numbers above are zero-based. Only one producer batch is published and fully acknowledged per transaction before the next batch is sent; this is not a large number of legitimately in-flight transactions.

#### Steps to reproduce

1. Start a disposable local broker. The low limit makes the failure quick to reproduce; transactions are enabled, and broker-side batch-index acknowledgment is disabled:

```bash
docker run -d --name pulsar-ack-repro -p 127.0.0.1:6650:6650 -p 127.0.0.1:8080:8080 -e PULSAR_MEM='-Xms512m -Xmx1g -XX:MaxDirectMemorySize=512m' -e PULSAR_PREFIX_transactionCoordinatorEnabled=true -e PULSAR_PREFIX_acknowledgmentAtBatchIndexLevelEnabled=false -e PULSAR_PREFIX_maxUnackedMessagesPerConsumer=20 apachepulsar/pulsar:4.0.3 sh -c 'bin/apply-config-from-env.py conf/standalone.conf && bin/pulsar standalone -nfw --advertised-address localhost'
```

Wait for standalone initialization to finish (`curl -fsS http://localhost:8080/admin/v2/brokers/health` returns `ok`).

2. In an empty directory, save the program below as `main.go`, then run:

```bash
go mod init example.com/pulsar-ack-repro
go get github.com/apache/pulsar-client-go@v0.21.0
go mod tidy
go run .
```

The program explicitly flushes each three-message producer batch and verifies the received batch size. It uses non-transactional publication and transactional acknowledgment only, so transactional publication is not required to reproduce the failure.

Self-contained reproduction (main.go)

```go
package main

import (
"context"
"encoding/json"
"fmt"
"net/http"
"time"

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

func check(err error) {
if err != nil {
panic(err)
}
}

func stats(topic string) {
httpClient := &http.Client{Timeout: 5 * time.Second}
resp, err := httpClient.Get("http://localhost:8080/admin/v2/persistent/public/default/" + topic + "/stats")
check(err)
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
panic(resp.Status)
}
var result struct {
Subscriptions map[string]struct {
UnackedMessages int `json:"unackedMessages"`
Consumers []struct {
Blocked bool `json:"blockedConsumerOnUnackedMsgs"`
} `json:"consumers"`
} `json:"subscriptions"`
}
check(json.NewDecoder(resp.Body).Decode(&result))
fmt.Printf("broker subscription: %+v\n", result.Subscriptions["repro"])
}

func main() {
client, err := pulsar.NewClient(pulsar.ClientOptions{URL: "pulsar://localhost:6650", EnableTransaction: true})
check(err)
defer client.Close()
topic := fmt.Sprintf("txn-batch-ack-%d", time.Now().UnixNano())
consumer, err := client.Subscribe(pulsar.ConsumerOptions{
Topic: topic, SubscriptionName: "repro", Type: pulsar.Shared,
EnableBatchIndexAcknowledgment: false,
})
check(err)
defer consumer.Close()
producer, err := client.CreateProducer(pulsar.ProducerOptions{
Topic: topic, BatchingMaxPublishDelay: time.Hour,
})
check(err)
defer producer.Close()
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
for round := 0; round < 24; round++ {
sent := make(chan error, 3)
for i := 0; i < 3; i++ {
producer.SendAsync(ctx, &pulsar.ProducerMessage{Payload: []byte("hello")},
func(_ pulsar.MessageID, _ *pulsar.ProducerMessage, err error) { sent <- err })
}
check(producer.FlushWithCtx(ctx))
for i := 0; i < 3; i++ {
select {
case err := <-sent:
check(err)
case <-ctx.Done():
panic(ctx.Err())
}
}
txn, err := client.NewTransaction(time.Minute)
check(err)
for i := 0; i < 3; i++ {
msg, err := consumer.Receive(ctx)
if err != nil {
stats(topic)
panic(fmt.Errorf("receive producer batch %d: %w", round, err))
}
if msg.ID().BatchSize() != 3 {
panic("expected an actual three-message producer batch")
}
check(consumer.AckWithTxn(msg, txn))
}
check(txn.Commit(ctx))
fmt.Printf("committed producer batch %d\n", round)
stats(topic)
}
}
```

3. Observe the growing `unackedMessages` count and eventual receive timeout shown above. After testing, remove the disposable broker with `docker rm -f -v pulsar-ack-repro`.

#### System configuration

- **Pulsar broker:** 4.0.3, official `apachepulsar/pulsar:4.0.3` standalone image.
- **Go client:** unmodified v0.21.0, the latest published release at the time of this report.
- **Go:** go1.27.0 darwin/arm64; broker running in Docker.
- **Subscription:** Shared; one non-partitioned topic is sufficient.
- **Consumer:** `EnableBatchIndexAcknowledgment=false`.
- **Broker:** `transactionCoordinatorEnabled=true`, `acknowledgmentAtBatchIndexLevelEnabled=false`, `maxUnackedMessagesPerConsumer=20`.

#### Source analysis and local verification

The following appears to explain the counter discrepancy:

- [`ackIDCommon`](https://github.com/apache/pulsar-client-go/blob/61d7a95e66cddf329adc6d925a8bd60ccd26eda5/pulsar/consumer_partition.go#L563) replaces a fully acknowledged batch with an entry-level `messageID`, retaining only ledger and entry IDs and losing `batchSize`.
- [`internalAckWithTxn`](https://github.com/apache/pulsar-client-go/blob/61d7a95e66cddf329adc6d925a8bd60ccd26eda5/pulsar/consumer_partition.go#L607) does not populate `MessageIdData.BatchSize` in the transaction ACK.
- In [Pulsar 4.0.3 `Consumer.individualAckWithTransaction`](https://github.com/apache/pulsar/blob/v4.0.3/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java#L631), an ACK without `BatchSize` decrements the unacked count by one. Dispatch increments it by the number of logical messages in the batch. For a three-message batch, this leaves two counted as unacknowledged.

The client links are pinned to the inspected master revision `61d7a95e66cddf329adc6d925a8bd60ccd26eda5`, which still contains this path. Runtime reproduction above was performed on v0.21.0; master was inspected, not independently run.

I also ran the same program against a local candidate fix that preserves batch size and sends the existing protobuf field. All 24 batches completed, with `unackedMessages=0` and `blockedConsumerOnUnackedMsgs=false`. Whole-entry IDs need to remain identifiable as whole-entry acknowledgments when preserving batch size, so that ordinary ACK grouping is not changed into partial batch acknowledgment.

I checked related reports including #993 and #1019. This reproduction uses `AckWithTxn` with successful commits, rather than ordinary ACK grouping or deliberately leaving messages unacknowledged.

Start in pulsar/consumer_partition.go at ackIDCommon and internalAckWithTxn, then run the provided Docker Pulsar 4.0.3 reproduction with main.go. Verify transactional ACKs preserve the producer batch size without changing whole-entry ACK behavior. Done means all 24 batches complete, unackedMessages remains zero, and the consumer is not blocked.

Written by the indexing model from the issue text.