apache / apache/pulsar-client-go

regexp subscription is recreating deleted topics when it shouldn't

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

Description

#### Steps to reproduce

Initialize a standalone cluster and create some entities:

```
$ pulsarctl tenant create -c standalone testtenant
$ pulsarctl namespace create testtenant/logs

```

Create a client with the following code (I have it in a directory called `rexconsumer`):
```go
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"os"
"os/signal"
"time"
"github.com/apache/pulsar-client-go/pulsar"
)
// TopicWildcardName returns a fully-qualified pulsar topic string
func TopicWildcardName(tenant, namespace string) string {
return fmt.Sprintf("persistent://%s/%s/*", tenant, namespace)
}
func main() {
log.SetFlags(0)
client, err := pulsar.NewClient(pulsar.ClientOptions{
URL: "pulsar://localhost:6650",
})
if err != nil {
log.Fatal(err)
}
defer client.Close()
topicpattern := TopicWildcardName(os.Args[1], os.Args[2])
log.Println(topicpattern)
consumer, err := client.Subscribe(pulsar.ConsumerOptions{
TopicsPattern: topicpattern,
SubscriptionName: "regexconsumer",
Type: pulsar.Exclusive,
SubscriptionInitialPosition: pulsar.SubscriptionPositionEarliest,
AutoDiscoveryPeriod: time.Second,
})
ctx, cFunc := context.WithCancel(context.Background())
sig := make(chan os.Signal, 1)
signal.Notify(sig, os.Interrupt, os.Kill)
go func() {
<-sig
cFunc()
}()
MainLoop:
for {
select {
case <-ctx.Done():
log.Println("closing gracefully")
consumer.Close()
break MainLoop
default:
// get message
}
msg, err := consumer.Receive(ctx)
if err != nil {
log.Println("consumer receive error:", err)
continue
}
log.Printf("Payload:\n%s",string(msg.Payload()))
consumer.Ack(msg)
}
}
```

start the consumer:
```
$ ./rexconsumer testtenant logs
persistent://testtenant/logs/*
INFO[0000] [Connecting to broker] remote_addr="pulsar://localhost:6650"
INFO[0000] [TCP connection established] local_addr="[::1]:63062" remote_addr="pulsar://localhost:6650"
INFO[0000] [Connection is ready]
```

Create some topics and verify they are there:

```
$ pulsarctl topics create testtenant/logs/topic1 0
Create topic persistent://testtenant/logs/topic1 with 0 partitions successfully
$ pulsarctl topics create testtenant/logs/topic2 0
Create topic persistent://testtenant/logs/topic2 with 0 partitions successfully
$ pulsarctl topics list testtenant/logs
+-------------------------------------+---------------+
| TOPIC NAME | PARTITIONED ? |
+-------------------------------------+---------------+
| persistent://testtenant/logs/topic1 | N |
| persistent://testtenant/logs/topic2 | N |
+-------------------------------------+---------------+
```

Now force-delete one of the topics:

```
pulsarctl topics delete -f -n persistent://testtenant/logs/topic1
Delete topic persistent://testtenant/logs/topic1 successfully
```

In the consumer you will see:
```
INFO[0430] Broker notification of Closed consumer: [1] local_addr="[::1]:63062" remote_addr="pulsar://localhost:6650"
INFO[0430] [Reconnecting to broker in 100ms] consumerID=1 name=wfidv subscription=regexconsumer topic="persistent://testtenant/logs/topic1"
INFO[0430] [Connected consumer] consumerID=1 name=wfidv subscription=regexconsumer topic="persistent://testtenant/logs/topic1"
INFO[0430] [Reconnected consumer to broker] consumerID=1 name=wfidv subscription=regexconsumer topic="persistent://testtenant/logs/topic1"
```

And the topic is still there:
```
$ pulsarctl topics list testtenant/logs
+-------------------------------------+---------------+
| TOPIC NAME | PARTITIONED ? |
+-------------------------------------+---------------+
| persistent://testtenant/logs/topic1 | N |
| persistent://testtenant/logs/topic2 | N |
+-------------------------------------+---------------+

```

#### Expected behavior

My understanding from #226 is that regex/pattern consumers should be applying this flag so that they will not do this. The topic should not be re-created.

#### Actual behavior

As above. Topic is re-created.

#### System configuration

**Pulsar version**: 2.8.0-SNAPSHOT (head)

Contributor guide

Open the contributing guide

Research direction

Start at the Go client.Subscribe entry point and inspect how ConsumerOptions.TopicsPattern, AutoDiscoveryPeriod, and reconnect handling manage deleted topics. Reproduce the standalone-cluster sequence with pulsarctl, then verify that a force-deleted topic is not recreated while the pattern consumer continues discovering other topics.

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
Clearly specified
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.