feat: support subscribeMode field on pulsar (#3831)

Signed-off-by: Eric Shen <[email protected]>
Signed-off-by: Eric Shen <[email protected]>
Co-authored-by: Mike Nguyen <[email protected]>
Co-authored-by: Loong Dai <[email protected]>
Co-authored-by: Nelson Parente <[email protected]>
Co-authored-by: Eric Shen <[email protected]>
Co-authored-by: Yaron Schneider <[email protected]>

Author: 6 people

pubsub/pulsar/metadata.go

@@ -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"`
 }

pubsub/pulsar/metadata.yaml

@@ -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"

pubsub/pulsar/pulsar.go

@@ -99,6 +99,11 @@ const (
 
 	subscribePositionEarliest = "earliest"
 	subscribePositionLatest   = "latest"
+
+	subscribeMode = "subscribeMode"
+
+	subscribeModeDurable    = "durable"
+	subscribeModeNonDurable = "non_durable"
 )
 
 type ProcessMode string
@@ -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):
@@ -455,6 +465,27 @@ 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 {
+	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")
@@ -474,6 +505,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,

pubsub/pulsar/pulsar_test.go

@@ -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{}