apache / apache/pulsar

Golang pulsar function can not subscribe to topic using regex

Open
#16,478 3 comments 0 reactions 0 assignees View on GitHub
Stale type/bug
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.