You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "BewareMyPower (via GitHub)" <gi...@apache.org> on 2026/05/19 15:20:01 UTC
[PR] fix: deadlock when increasing partitioned consumers [pulsar-client-go]
BewareMyPower opened a new pull request, #1500:
URL: https://github.com/apache/pulsar-client-go/pull/1500
### Motivation
#1494 introduces a possible deadlock after ensuring the thread safety.
To fix the thread safety issue, every APIs that need to find a specific sub-consumer, like `Ack` and `Seek`, will now require locking the parent consumer. However, in `internalTopicSubscribeToPartitions`, it will wait all sub-consumers' creations are done:
https://github.com/apache/pulsar-client-go/blob/87ce8f90d0e2df9402133ebc52560638bb09778d/pulsar/consumer_impl.go#L414-L416
https://github.com/apache/pulsar-client-go/blob/87ce8f90d0e2df9402133ebc52560638bb09778d/pulsar/consumer_impl.go#L428
`range ch` will wait the `err` returned by `newPartitionConsumer` is sent to the `ch`.
However, it could be blocked by `grabConn`: https://github.com/apache/pulsar-client-go/blob/87ce8f90d0e2df9402133ebc52560638bb09778d/pulsar/consumer_partition.go#L434
this could start a Subscribe RPC in the connection that might receive user's Ack or Seek requests.
### Modifications
- Add `TestInternalTopicSubscribeToPartitionsDoesNotBlockExistingPartitionLookup` to reproduce the deadlock issue.
- Adopt the lock-free implementation to manage consumers, the test above will pass after test
- Delay the dispatching logic after adding new sub-consumers to `c.consumers`, otherwise, a message could be received from a new sub-consumer and acknowledged before that consumer is added to `c.consumers`.
- Add `TestInternalTopicSubscribeToPartitionsPublishesConsumersBeforeDispatchingMessages` to verify it works
--
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
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: deadlock when increasing partitioned consumers [pulsar-client-go]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on PR #1500:
URL: https://github.com/apache/pulsar-client-go/pull/1500#issuecomment-4504296859
Great fix on the deadlock path from #1494 — the lock-free `partitionConsumers()` snapshot + delayed `startDispatcher()` is a solid direction for the original blocking issue.
I think there is still a new concurrency window introduced by removing the old mutex guarding `consumers`.
In the new code:
- `internalTopicSubscribeToPartitions()` builds `newConsumers`, then publishes via `c.consumers.Store(...)`, then starts dispatchers for new partitions.
- `closeWithCause()` closes only one snapshot (`consumers := c.partitionConsumers()`) and no longer serializes with partition expansion.
So this interleaving seems possible:
1. Goroutine A enters `internalTopicSubscribeToPartitions()` and starts creating new partition consumers.
2. Goroutine B calls `closeWithCause()`, takes an old snapshot, closes those consumers, and continues close flow.
3. Goroutine A continues and still executes `c.consumers.Store(...)` + `startDispatcher()` for newly created consumers.
Result: newly added partition consumers may be published/started after close has already closed only the old snapshot.
Before this PR, the parent lock serialized expansion and close-related operations over `c.consumers`; removing that lock fixes deadlock, but also removes that lifecycle serialization.
Could we gate expansion with context cancellation or a closing flag (checked before create and before Store/startDispatcher), so in-flight expansion stops/cleans up once close starts?
--
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
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: deadlock when increasing partitioned consumers [pulsar-client-go]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #1500:
URL: https://github.com/apache/pulsar-client-go/pull/1500#issuecomment-4498290379
@RobertIndie addressed, PTAL again
--
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
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: deadlock when increasing partitioned consumers [pulsar-client-go]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1500:
URL: https://github.com/apache/pulsar-client-go/pull/1500#discussion_r3271909299
##########
pulsar/consumer_impl.go:
##########
@@ -782,35 +771,29 @@ func (c *consumer) SeekByTime(time time.Time) error {
return errs
}
-func (c *consumer) findPartitionConsumer(msgID MessageID) (*partitionConsumer, error) {
- c.Lock()
- defer c.Unlock()
- return c.unsafeFindPartitionConsumer(msgID)
-}
-
-// NOTE: This method must be called when c.Lock is held
-func (c *consumer) unsafeFindPartitionConsumer(msgID MessageID) (*partitionConsumer, error) {
+func findPartitionConsumer(consumers []*partitionConsumer, msgID MessageID) (*partitionConsumer, error) {
partition := int(msgID.PartitionIdx())
- if partition < 0 || partition >= len(c.consumers) {
- c.log.Errorf("invalid partition index %d expected a partition between [0-%d]",
- partition, len(c.consumers))
+ if partition < 0 || partition >= len(consumers) {
return nil, fmt.Errorf("invalid partition index %d expected a partition between [0-%d]",
- partition, len(c.consumers))
+ partition, len(consumers))
+ }
+ return consumers[partition], nil
+}
+
+func (c *consumer) partitionConsumers() []*partitionConsumer {
+ v := c.consumers.Load()
+ if v == nil {
+ return nil
}
- return c.consumers[partition], nil
+ consumers := v.([]*partitionConsumer)
+ return append([]*partitionConsumer(nil), consumers...)
Review Comment:
partitionConsumers() clones the slice on every call, which adds allocations and O(n) copying on hot paths (Ack/AckID/Nack/etc.). Since updates are already copy-on-write via atomic.Value, consider returning the stored slice directly (treating it as immutable, like producer.getProducers()), or providing a separate explicit snapshot/copy helper only where needed.
##########
pulsar/consumer_impl.go:
##########
@@ -782,35 +771,29 @@ func (c *consumer) SeekByTime(time time.Time) error {
return errs
}
-func (c *consumer) findPartitionConsumer(msgID MessageID) (*partitionConsumer, error) {
- c.Lock()
- defer c.Unlock()
- return c.unsafeFindPartitionConsumer(msgID)
-}
-
-// NOTE: This method must be called when c.Lock is held
-func (c *consumer) unsafeFindPartitionConsumer(msgID MessageID) (*partitionConsumer, error) {
+func findPartitionConsumer(consumers []*partitionConsumer, msgID MessageID) (*partitionConsumer, error) {
partition := int(msgID.PartitionIdx())
- if partition < 0 || partition >= len(c.consumers) {
- c.log.Errorf("invalid partition index %d expected a partition between [0-%d]",
- partition, len(c.consumers))
+ if partition < 0 || partition >= len(consumers) {
return nil, fmt.Errorf("invalid partition index %d expected a partition between [0-%d]",
- partition, len(c.consumers))
+ partition, len(consumers))
+ }
+ return consumers[partition], nil
Review Comment:
The error message range is off by one: valid partition indices are 0..len(consumers)-1, but the message prints "[0-%d]" with len(consumers). Adjust the upper bound to len(consumers)-1 (and consider including the actual len for clarity).
##########
pulsar/consumer_test.go:
##########
@@ -5927,3 +5929,380 @@ func TestSelectConnectionForSameConsumer(t *testing.T) {
"The consumer uses a different connection when reconnecting")
}
}
+
+func TestInternalTopicSubscribeToPartitionsDoesNotBlockExistingPartitionLookup(t *testing.T) {
+ lookupURL, err := url.Parse("pulsar://localhost:6650")
+ require.NoError(t, err)
+
+ allowSubscribe := make(chan struct{})
+ subscribeStarted := make(chan struct{})
+ var releaseSubscribe sync.Once
+
+ logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
+ slog.SetDefault(logger)
Review Comment:
This test sets a process-wide default slog logger via slog.SetDefault, which can affect logging behavior of other tests running in the same package/process and make failures harder to interpret. Prefer using a local logger instance (and passing it where needed) or restoring the previous default in a defer.
--
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
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: deadlock when increasing partitioned consumers [pulsar-client-go]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #1500:
URL: https://github.com/apache/pulsar-client-go/pull/1500#issuecomment-4505509116
@nodece `closeWithCause` calls `c.stopDiscovery()`, which cancels the timer and waits for the previous `internalTopicSubscribeToPartitions` is done by `wg.Wait()`.
https://github.com/apache/pulsar-client-go/blob/a1765ef9c22a4cbd7b7a4704b6136e96860b3815/pulsar/consumer_impl.go#L346-L349
--
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
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: deadlock when increasing partitioned consumers [pulsar-client-go]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower merged PR #1500:
URL: https://github.com/apache/pulsar-client-go/pull/1500
--
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
For queries about this service, please contact Infrastructure at:
users@infra.apache.org