Skip to content

feat: support subscribeMode field on pulsar #3831

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 9 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pubsub/pulsar/metadata.go
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ type pulsarMetadata struct {
ReceiverQueueSize int `mapstructure:"receiverQueueSize"`
SubscriptionType string `mapstructure:"subscribeType"`
SubscriptionInitialPosition string `mapstructure:"subscribeInitialPosition"`
SubscriptionMode string `mapstructure:"subscribeMode"`
Token string `mapstructure:"token"`
oauth2.ClientCredentialsMetadata `mapstructure:",squash"`
}
Expand Down
11 changes: 10 additions & 1 deletion pubsub/pulsar/metadata.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -201,4 +201,13 @@ metadata:
example: '"earliest"'
url:
title: "Pulsar SubscriptionInitialPosition"
url: "https://pkg.go.dev/github.com/apache/pulsar-client-go/pulsar#SubscriptionInitialPosition"
url: "https://pkg.go.dev/github.com/apache/pulsar-client-go/pulsar#SubscriptionInitialPosition"
- name: subscribeMode
type: string
description: |
Subscription mode indicates the cursor belongs to "durable" type or "non_durable" type, durable subscription retains messages and persists the current position.
default: '"durable"'
example: '"durable"'
url:
title: "Pulsar SubscriptionMode"
url: "https://pkg.go.dev/github.com/apache/pulsar-client-go/pulsar#SubscriptionMode"
31 changes: 31 additions & 0 deletions pubsub/pulsar/pulsar.go
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,11 @@ const (

subscribePositionEarliest = "earliest"
subscribePositionLatest = "latest"

subscribeMode = "subscribeMode"

subscribeModeDurable = "durable"
subscribeModeNonDurable = "non_durable"
)

type ProcessMode string
Expand Down Expand Up @@ -154,6 +159,11 @@ func parsePulsarMetadata(meta pubsub.Metadata) (*pulsarMetadata, error) {
return nil, errors.New("invalid subscription initial position. Accepted values are `latest` and `earliest`")
}

m.SubscriptionMode, err = parseSubscriptionMode(meta.Properties[subscribeMode])
if err != nil {
return nil, errors.New("invalid subscription mode")
}

for k, v := range meta.Properties {
switch {
case strings.HasSuffix(k, topicJSONSchemaIdentifier):
Expand Down Expand Up @@ -455,6 +465,26 @@ func getSubscribePosition(subsPositionStr string) pulsar.SubscriptionInitialPosi
return subsPosition
}

func parseSubscriptionMode(in string) (string, error) {
subsMode := strings.ToLower(in)
switch subsMode {
case subscribeModeDurable, subscribeModeNonDurable:
return subsMode, nil
case "":
return subscribeModeDurable, nil
default:
return "", fmt.Errorf("invalid subscription mode: %s", subsMode)
}
}

func getSubscriptionMode(subsModeStr string) pulsar.SubscriptionMode {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would like to see the addition of tests for this function to ensure future changes don't cause regressions

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@mikeee Your comment makes sense, I added a new test func TestParsePulsarMetadataSubscriptionCombination which combines all existing subscription fields to validate the regressions in bf064af. PTAL

switch subsModeStr {
case subscribeModeNonDurable:
return pulsar.NonDurable
default:
return pulsar.Durable
}
}
func (p *Pulsar) Subscribe(ctx context.Context, req pubsub.SubscribeRequest, handler pubsub.Handler) error {
if p.closed.Load() {
return errors.New("component is closed")
Expand All @@ -474,6 +504,7 @@ func (p *Pulsar) Subscribe(ctx context.Context, req pubsub.SubscribeRequest, han
SubscriptionName: p.metadata.ConsumerID,
Type: getSubscribeType(subscribeType),
SubscriptionInitialPosition: getSubscribePosition(subscribeInitialPosition),
SubscriptionMode: getSubscriptionMode(subscribeMode),
MessageChannel: channel,
NackRedeliveryDelay: p.metadata.RedeliveryDelay,
ReceiverQueueSize: p.metadata.ReceiverQueueSize,
Expand Down
156 changes: 156 additions & 0 deletions pubsub/pulsar/pulsar_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -170,6 +170,162 @@ func TestParsePulsarMetadataSubscriptionInitialPosition(t *testing.T) {
}
}

func TestParsePulsarMetadataSubscriptionMode(t *testing.T) {
tt := []struct {
name string
subscribeMode string
expected string
err bool
}{
{
name: "test valid subscribe mode - durable",
subscribeMode: "durable",
expected: "durable",
err: false,
},
{
name: "test valid subscribe mode - non_durable",
subscribeMode: "non_durable",
expected: "non_durable",
err: false,
},
{
name: "test valid subscribe mode - empty",
subscribeMode: "",
expected: "durable",
err: false,
},
{
name: "test invalid subscribe mode",
subscribeMode: "invalid",
err: true,
},
}
for _, tc := range tt {
t.Run(tc.name, func(t *testing.T) {
m := pubsub.Metadata{}

m.Properties = map[string]string{
"host": "a",
"subscribeMode": tc.subscribeMode,
}
meta, err := parsePulsarMetadata(m)

if tc.err {
require.Error(t, err)
assert.Nil(t, meta)
return
}

require.NoError(t, err)
assert.Equal(t, tc.expected, meta.SubscriptionMode)
})
}
}

func TestParsePulsarMetadataSubscriptionCombination(t *testing.T) {
tt := []struct {
name string
subscribeType string
subscribeInitialPosition string
subscribeMode string
expectedType string
expectedInitialPosition string
expectedMode string
err bool
}{
{
name: "test valid subscribe - default",
subscribeType: "",
subscribeInitialPosition: "",
subscribeMode: "",
expectedType: "shared",
expectedInitialPosition: "latest",
expectedMode: "durable",
err: false,
},
{
name: "test valid subscribe - pass case 1",
subscribeType: "key_shared",
subscribeInitialPosition: "earliest",
subscribeMode: "non_durable",
expectedType: "key_shared",
expectedInitialPosition: "earliest",
expectedMode: "non_durable",
err: false,
},
{
name: "test valid subscribe - pass case 2",
subscribeType: "exclusive",
subscribeInitialPosition: "latest",
subscribeMode: "durable",
expectedType: "exclusive",
expectedInitialPosition: "latest",
expectedMode: "durable",
err: false,
},
{
name: "test valid subscribe - pass case 3",
subscribeType: "failover",
subscribeInitialPosition: "earliest",
subscribeMode: "durable",
expectedType: "failover",
expectedInitialPosition: "earliest",
expectedMode: "durable",
err: false,
},
{
name: "test valid subscribe - pass case 4",
subscribeType: "shared",
subscribeInitialPosition: "latest",
subscribeMode: "non_durable",
expectedType: "shared",
expectedInitialPosition: "latest",
expectedMode: "non_durable",
err: false,
},
{
name: "test valid subscribe - fail case 1",
subscribeType: "invalid",
err: true,
},
{
name: "test valid subscribe - fail case 2",
subscribeInitialPosition: "invalid",
err: true,
},
{
name: "test valid subscribe - fail case 3",
subscribeMode: "invalid",
err: true,
},
}
for _, tc := range tt {
t.Run(tc.name, func(t *testing.T) {
m := pubsub.Metadata{}

m.Properties = map[string]string{
"host": "a",
"subscribeType": tc.subscribeType,
"subscribeInitialPosition": tc.subscribeInitialPosition,
"subscribeMode": tc.subscribeMode,
}
meta, err := parsePulsarMetadata(m)

if tc.err {
require.Error(t, err)
assert.Nil(t, meta)
return
}

require.NoError(t, err)
assert.Equal(t, tc.expectedType, meta.SubscriptionType)
assert.Equal(t, tc.expectedInitialPosition, meta.SubscriptionInitialPosition)
assert.Equal(t, tc.expectedMode, meta.SubscriptionMode)
})
}
}

func TestParsePulsarSchemaMetadata(t *testing.T) {
t.Run("test json", func(t *testing.T) {
m := pubsub.Metadata{}
Expand Down