[I] [Bug] Transactional ACKs omit batch size, causing Shared consumers to stop receiving [pulsar-client-go]

["Ebispongebob (via GitHub)" <gi...@apache.org>]

Ebispongebob opened a new issue, #1534:
URL: https://github.com/apache/pulsar-client-go/issues/1534

   #### 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.
   
   <details>
   <summary>Self-contained reproduction (main.go)</summary>
   
   ```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)
   	}
   }
   ```
   
   </details>
   
   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.
   


-- 
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