ISSUE-14878: The reader of __transaction_buffer_snapshot did not been closed property

[sijie]

Original Issue: apache#14878

---

Describe the bug

"subscriptions" : {
    "__compaction" : {
      "msgRateOut" : 0.0,
      "msgThroughputOut" : 0.0,
      "bytesOutCounter" : 3068776,
      "msgOutCounter" : 10729,
      "msgRateRedeliver" : 0.0,
      "chunkedMessageRate" : 0,
      "msgBacklog" : 96,
      "backlogSize" : 0,
      "msgBacklogNoDelayed" : 96,
      "blockedSubscriptionOnUnackedMsgs" : false,
      "msgDelayed" : 0,
      "unackedMessages" : 0,
      "type" : "Exclusive",
      "msgRateExpired" : 0.0,
      "totalMsgExpired" : 0,
      "lastExpireTimestamp" : 0,
      "lastConsumedFlowTimestamp" : 1648218295485,
      "lastConsumedTimestamp" : 0,
      "lastAckedTimestamp" : 0,
      "lastMarkDeleteAdvancedTimestamp" : 0,
      "consumers" : [ ],
      "isDurable" : true,
      "isReplicated" : false,
      "allowOutOfOrderDelivery" : false,
      "consumersAfterMarkDeletePosition" : { },
      "nonContiguousDeletedMessagesRanges" : 0,
      "nonContiguousDeletedMessagesRangesSerializedSize" : 0,
      "replicated" : false,
      "durable" : true
    },
    "reader-52dbece9b6" : {
      "msgRateOut" : 0.0,
      "msgThroughputOut" : 0.0,
      "bytesOutCounter" : 0,
      "msgOutCounter" : 0,
      "msgRateRedeliver" : 0.0,
      "chunkedMessageRate" : 0,
      "msgBacklog" : 4768,
      "backlogSize" : 0,
      "msgBacklogNoDelayed" : 4768,
      "blockedSubscriptionOnUnackedMsgs" : false,
      "msgDelayed" : 0,
      "unackedMessages" : 0,
      "type" : "Exclusive",
      "activeConsumerName" : "fa67a",
      "msgRateExpired" : 0.0,
      "totalMsgExpired" : 0,
      "lastExpireTimestamp" : 0,
      "lastConsumedFlowTimestamp" : 0,
      "lastConsumedTimestamp" : 0,
      "lastAckedTimestamp" : 0,
      "lastMarkDeleteAdvancedTimestamp" : 0,
      "consumers" : [ {
        "msgRateOut" : 0.0,
        "msgThroughputOut" : 0.0,
        "bytesOutCounter" : 0,
        "msgOutCounter" : 0,
        "msgRateRedeliver" : 0.0,
        "chunkedMessageRate" : 0.0,
        "consumerName" : "fa67a",
        "availablePermits" : 0,
        "unackedMessages" : 0,
        "avgMessagesPerEntry" : 0,
        "blockedConsumerOnUnackedMsgs" : false,
        "lastAckedTimestamp" : 0,
        "lastConsumedTimestamp" : 0,
        "metadata" : { },
        "connectedSince" : "2022-03-25T22:13:36.114886+08:00",
        "clientVersion" : "2.9.2",
        "address" : "/127.0.0.1:51161"
      } ],
      "isDurable" : false,
      "isReplicated" : false,
      "allowOutOfOrderDelivery" : false,
      "consumersAfterMarkDeletePosition" : { },
      "nonContiguousDeletedMessagesRanges" : 0,
      "nonContiguousDeletedMessagesRangesSerializedSize" : 0,
      "replicated" : false,
      "durable" : false
    }
  }

From the broker logs:

~/Downloads/geo-cluster (branch-0.8.1*) » cat cluster-a*/logs/* | grep -v "Handle read response:" | grep "52dbece9b6"                                                                                      lipenghui@lipenghuideMacBook-Pro-2
22:13:04.659 [pulsar-io-16-2] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:51161] Subscribing on topic persistent://public/default/__transaction_buffer_snapshot / reader-52dbece9b6
22:13:35.732 [ForkJoinPool.commonPool-worker-31] INFO  org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://public/default/__transaction_buffer_snapshot][reader-52dbece9b6] Creating non-durable subscription at msg id -1:-1:-1:-1
22:13:36.248 [ForkJoinPool.commonPool-worker-31] INFO  org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://public/default/__transaction_buffer_snapshot][reader-52dbece9b6] Created new subscription for 10
22:13:36.248 [ForkJoinPool.commonPool-worker-31] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:51161] Created subscription on topic persistent://public/default/__transaction_buffer_snapshot / reader-52dbece9b6
22:12:52.577 [pulsar-io-16-3] INFO  org.apache.pulsar.client.impl.ConsumerImpl - [persistent://public/default/__transaction_buffer_snapshot][reader-52dbece9b6] Subscribing to topic on cnx [id: 0x7ecdb373, L:/127.0.0.1:51161 - R:/127.0.0.1:6652], consumerId 10
22:13:28.365 [pulsar-io-16-18] WARN  org.apache.pulsar.client.impl.ConsumerImpl - [persistent://public/default/__transaction_buffer_snapshot][reader-52dbece9b6] Failed to subscribe to topic on /127.0.0.1:6652

Looks like related to the 22:13:28.365 [pulsar-io-16-18] WARN org.apache.pulsar.client.impl.ConsumerImpl - [persistent://public/default/__transaction_buffer_snapshot][reader-52dbece9b6] Failed to subscribe to topic on /127.0.0.1:6652

Because after checking the normal reader cleanup in the broker logs

22:20:06.971 [pulsar-io-16-2] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:51161] Subscribing on topic persistent://public/default/__transaction_buffer_snapshot / reader-c065e64d8a
22:20:07.001 [ForkJoinPool.commonPool-worker-7] INFO  org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://public/default/__transaction_buffer_snapshot][reader-c065e64d8a] Creating non-durable subscription at msg id -1:-1:-1:-1
22:20:07.002 [ForkJoinPool.commonPool-worker-7] INFO  org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://public/default/__transaction_buffer_snapshot][reader-c065e64d8a] Created new subscription for 82
22:20:07.003 [ForkJoinPool.commonPool-worker-7] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:51161] Created subscription on topic persistent://public/default/__transaction_buffer_snapshot / reader-c065e64d8a
22:20:07.378 [pulsar-io-16-2] INFO  org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer - Removing consumer Consumer{subscription=PersistentSubscription{topic=persistent://public/default/__transaction_buffer_snapshot, name=reader-c065e64d8a}, consumerId=82, consumerName=9a4b0, address=/127.0.0.1:51161}
22:20:07.378 [pulsar-io-16-2] INFO  org.apache.pulsar.broker.service.persistent.PersistentSubscription - [persistent://public/default/__transaction_buffer_snapshot][reader-c065e64d8a] Successfully closed subscription [NonDurableCursorImpl{ledger=public/default/persistent/__transaction_buffer_snapshot, ackPos=49:4007, readPos=49:4008}]
22:20:07.379 [pulsar-io-16-2] INFO  org.apache.pulsar.broker.service.persistent.PersistentSubscription - [persistent://public/default/__transaction_buffer_snapshot][reader-c065e64d8a] Successfully closed dispatcher for reader
22:20:06.970 [pulsar-io-16-3] INFO  org.apache.pulsar.client.impl.ConsumerImpl - [persistent://public/default/__transaction_buffer_snapshot][reader-c065e64d8a] Subscribing to topic on cnx [id: 0x7ecdb373, L:/127.0.0.1:51161 - R:/127.0.0.1:6652], consumerId 82
22:20:07.016 [pulsar-io-16-17] INFO  org.apache.pulsar.client.impl.ConsumerImpl - [persistent://public/default/__transaction_buffer_snapshot][reader-c065e64d8a] Subscribed to topic on /127.0.0.1:6652 -- consumer: 82
22:20:07.086 [pulsar-transaction-executor-13-1] INFO  org.apache.pulsar.client.impl.ConsumerImpl - [persistent://public/default/__transaction_buffer_snapshot][reader-c065e64d8a] Get topic last message Id
22:20:07.089 [pulsar-io-16-17] INFO  org.apache.pulsar.client.impl.ConsumerImpl - [persistent://public/default/__transaction_buffer_snapshot][reader-c065e64d8a] Successfully getLastMessageId 49:4007
22:20:07.373 [pulsar-transaction-executor-13-1] INFO  org.apache.pulsar.client.impl.ConsumerImpl - [persistent://public/default/__transaction_buffer_snapshot][reader-c065e64d8a] Get topic last message Id
22:20:07.378 [pulsar-io-16-17] INFO  org.apache.pulsar.client.impl.ConsumerImpl - [persistent://public/default/__transaction_buffer_snapshot][reader-c065e64d8a] Successfully getLastMessageId 49:4007
22:20:07.379 [pulsar-io-16-17] INFO  org.apache.pulsar.client.impl.ConsumerImpl - [persistent://public/default/__transaction_buffer_snapshot] [reader-c065e64d8a] Closed consumer

To Reproduce
Don't have a stable way to reproduce the issue.

Expected behavior
The reader should be closed property to avoid reader instance leak

Screenshots

Additional context
last branch-2.9 apache@f0a2171