Golang pulsar function can not subscribe to topic using regex
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
**Describe the bug**
I am trying to create a pulsar function using golang that subscribes to a topic with a regex for example "persistent://payments/payment-instruction/.*.created". The consumer logs a message that the subscription was created, when messages are produced matching the pattern the consumer does not receive them.
**To Reproduce**
Steps to reproduce the behavior:
1. Create a Go function
```
func echoFunc(ctx context.Context, in []byte) error {
if fc, ok := pf.FromContext(ctx); ok {
logutil.Infof("This input has a length of &d", len(in))
logutil.Infof("Fully-qualified function name is: %s\\%s\\%s\\\n", fc.GetFuncTenant(), fc.GetFuncNamespace(), fc.GetFuncName())
data := PaymentInstruction{}
jsonErr := json.Unmarshal(in, &data)
if jsonErr != nil {
return jsonErr
}
logutil.Infof("Payment Instruction: %+v", data)
if data.Amount%2 == 0 {
msgId, err := fc.NewOutputMessage("public/default/accepted").Send(ctx, &pulsar.ProducerMessage{Payload: in})
if err != nil {
return err
}
logutil.Infof("msgId: %s", msgId)
} else {
msgId, err := fc.NewOutputMessage("public/default/rejected").Send(ctx, &pulsar.ProducerMessage{Payload: in})
if err != nil {
return err
}
logutil.Infof("msgId: %s", msgId)
}
}
return nil
}
func main() {
pf.Start(echoFunc)
}
```
2. Configure function using a yml config file
```
tenant: payment
namespace: payment-instruction
name: payment-instruction-validation
go: /pulsar/payment-instruction-validation-policy
inputSpecs:
persistent://public/default/created.*:
isRegexPattern: true
schemaProperties: {}
```
3. create pulsar function
`./bin/pulsar-admin functions localrun --function-config-file ./functions-config.yml`
4. create producer
```
client, err := pulsar.NewClient(pulsar.ClientOptions{
URL: os.Getenv("PULSAR_TENANT_URL"),
ConnectionTimeout: 30 * time.Second,
OperationTimeout: 30 * time.Second,
})
if err != nil {
log.Println("Error creating pulsar client: " + err.Error())
return
}
defer client.Close()
properties := make(map[string]string)
properties["pulsar"] = "hello"
producer, err := client.CreateProducer(pulsar.ProducerOptions{
Topic: "payments/payment-instruction/" + "created" + "someinstanceid",
})
if err != nil {
log.Println("error creating pulsar producer: " + err.Error())
return
}
defer producer.Close()
payload, _ := json.Marshal(&PaymentInstruction{
SourceEndpoint: PaymentEndpoint{
RoutingInfo: map[string]string{
"accountNumber": randomNumber(8),
"sortCode": randomNumber(6),
},
PaymentRail: "Bank",
AccountIdentifier: uuid.New().String(),
},
TargetEndpoint: PaymentEndpoint{
RoutingInfo: map[string]string{
"accountNumber": randomNumber(8),
"sortCode": randomNumber(6),
},
PaymentRail: "Bank",
AccountIdentifier: uuid.New().String(),
},
Amount: randomPrice(99999),
})
msgId, err := producer.Send(context.Background(), &pulsar.ProducerMessage{
Payload: []byte(payload),
})
if err != nil {
log.Println("error producing message: " + err.Error())
return
}
log.Printf("msgId: %s", msgId)
```
5. compile and run producer
**Expected behavior**
function should receive and log the event and republish either public/default/accepted or public/default/rejected
**Desktop (please complete the following information):**
- OS: Windows - pulsar on docker
**Additional context**
When creating the function there is a log message that states that regexPatternSubscription is set to false even when input specs sets isRegexPattern to true
The same problem exists when using topicsPattern in the config file
Contributor guide
Research direction
Start with the `pulsar-admin functions localrun --function-config-file` path and the `inputSpecs` handling shown in the report; verify why `isRegexPattern` and `topicsPattern` are not reaching the subscription. Reproduce with the provided Go function, configuration, and producer, then confirm the function receives matching topics and republishes to the accepted or rejected output.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go, java
- Domain
- distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 30/100