apache / apache/pulsar-client-go
Closing consumer can leak a consumer when the connection is lost
- Dominant language
- Go
- Stars
- 745
- Forks
- 389
- Avg merge
- 3d 20h
- Merged PRs (30d)
- 3
Description
#### Expected behavior
When calling the `Close` function on a consumer, it should be removed from the subscription in the broker eventually.
#### Actual behavior
When calling the `Close` function on a consumer and when the connection is interrupted during the RPC, the consumer ends up in a closed state as the point of view of the client, but is still active in the broker (and active so blocking in failover mode for instance).
#### Steps to reproduce
Hard to reproduce as it happens when closing a consumer and the connection is lost at the same time. It is however clear in the code:
```Go
_, err := pc.client.rpcClient.RequestOnCnx(pc._getConn(), requestID, pb.BaseCommand_CLOSE_CONSUMER, cmdClose)
if err != nil {
pc.log.WithError(err).Warn("Failed to close consumer")
} else {
pc.log.Info("Closed consumer")
}
```
Here it only logs that closing has failed but it then continues and stops everything else resulting in a consumer in the broker like this:
```
"activeConsumerName" : "1b81691e",
"consumers" : [ {
"appId" : "***",
"msgRateOut" : 0.0,
"msgThroughputOut" : 0.0,
"bytesOutCounter" : 0,
"msgOutCounter" : 0,
"msgRateRedeliver" : 0.0,
"messageAckRate" : 0.0,
"chunkedMessageRate" : 0.0,
"consumerName" : "1b81691e",
"availablePermits" : 0,
"unackedMessages" : 0,
"avgMessagesPerEntry" : 0,
"blockedConsumerOnUnackedMsgs" : false,
"drainingHashesCount" : 0,
"drainingHashesClearedTotal" : 0,
"drainingHashesUnackedMessages" : 0,
"address" : "/10.128.15.78:35170",
"connectedSince" : "2025-01-23T13:28:58.27512622Z",
"clientVersion" : "Pulsar Go v0.14.0",
"lastAckedTimestamp" : 0,
"lastConsumedTimestamp" : 0,
"lastConsumedFlowTimestamp" : 0,
"metadata" : { },
"lastAckedTime" : "1970-01-01T00:00:00Z",
"lastConsumedTime" : "1970-01-01T00:00:00Z"
}
```
#### System configuration
**Pulsar version**: v4.0.1
**Pulsar client**: v0.14.0
Contributor guide
Research direction
Start with the consumer Close path around RequestOnCnx and the logged failure handling shown in the issue. Trace what cleanup occurs when the connection is interrupted during the close RPC, then verify against broker consumer state. Done means a failed close does not leave the consumer active in the broker, including the connection-loss case.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go
- Domain
- backend, distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 42/100