apache / apache/pulsar-client-go

[Bug]Partition consumer not recreated after error and closure

Open
#1,426 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Go
Stars
745
Forks
389
Avg merge
3d 20h
Merged PRs (30d)
3

Description

#### Expected behavior

When an error occurs and the partition consumer is closed, subsequent retries or partition expansion logic should recreate the closed partition consumer to ensure no partitions are permanently skipped.

#### Actual behavior

Once a partition consumer is closed due to an error, it is never recreated. This results in missed consumption on that partition.

#### Steps to reproduce

1. Create a consumer on a partitioned topic.
2. Trigger an error at [consumer_impl.go#L418](https://github.com/apache/pulsar-client-go/blob/v0.16.0/pulsar/consumer_impl.go#L418).
3. Observe that the partition consumer is closed.
4. Wait for retry or trigger partition expansion.
5. Verify that the closed consumer is not recreated, and messages on that partition are no longer consumed.

#### System configuration
**Pulsar version**: 0.14.0, 0.15.1, 0.16.0

#### Relevant Code Snippets with Bug Notes

- [consumer_impl.go#L309-L333](https://github.com/apache/pulsar-client-go/blob/v0.16.0/pulsar/consumer_impl.go#L309-L333) ingores the error from method `internalTopicSubscribeToPartitions` and calls it on ticker
```go
func (c *consumer) runBackgroundPartitionDiscovery(period time.Duration) (cancel func()) {
var wg sync.WaitGroup
stopDiscoveryCh := make(chan struct{})
ticker := time.NewTicker(period)

wg.Add(1)
go func() {
defer wg.Done()
for {
select {
case <-stopDiscoveryCh:
return
case <-ticker.C:
c.log.Debug("Auto discovering new partitions")
c.internalTopicSubscribeToPartitions() // Uncaught error
}
}
}()

return func() {
ticker.Stop()
close(stopDiscoveryCh)
wg.Wait()
}
}
```

- [consumer_impl.go#L418-L427](https://github.com/apache/pulsar-client-go/blob/v0.16.0/pulsar/consumer_impl.go#L418-L427) closes all partition consumer when encounting error
```go
if err != nil {
// Since there were some failures,
// cleanup all the partitions that succeeded in creating the consumer
for _, c := range c.consumers {
if c != nil {
c.Close()
}
}
return err
}
```

- [consumer_impl.go#L351-L354](https://github.com/apache/pulsar-client-go/blob/v0.16.0/pulsar/consumer_impl.go#L351-L354) returns fastly when the number of partitions has not changed even though the old consumers are closed.
```go
if oldNumPartitions == newNumPartitions {
c.log.Debug("Number of partitions in topic has not changed")
return nil
}
```

- [consumer_impl.go#L363-L370](https://github.com/apache/pulsar-client-go/blob/v0.16.0/pulsar/consumer_impl.go#L363-L370) reuses the existing closed consumer instances.
```go
// When for some reason (eg: forced deletion of sub partition) causes oldNumPartitions> newNumPartitions,
// we need to rebuild the cache of new consumers, otherwise the array will be out of bounds.
if oldConsumers != nil && oldNumPartitions < newNumPartitions {
// Copy over the existing consumer instances
for i := 0; i < oldNumPartitions; i++ {
c.consumers[i] = oldConsumers[i]
}
}
```

Contributor guide

Open the contributing guide

Research direction

Start in pulsar/consumer_impl.go around lines 309-333, 351-370, and 418-427, tracing how partition discovery handles errors, closed consumers, and unchanged partition counts. Reproduce the documented retry or partition-expansion scenario; done means a closed partition consumer is recreated and messages on that partition are consumed again.

Written by the indexing model from the issue text.

Assessment

Tech stack
go
Domain
distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
55/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.