Merge branch '3318-RavenDB-state-store-new' of github.com-n-personal:nmalocic/components-contrib into 3318-RavenDB-state-store-new

Author: nmalocicvega

.github/pull_request_template.md

@@ -14,4 +14,7 @@ Please make sure you've completed the relevant tasks for this PR, out of the fol
 
 * [ ] Code compiles correctly
 * [ ] Created/updated tests
-* [ ] Extended the documentation / Created issue in the https://github.com/dapr/docs/ repo: dapr/docs#_[issue number]_
+* [ ] Extended the documentation
+    * [ ] Created the dapr/docs PR: <insert PR link here>
+
+**Note:** We expect contributors to open a corresponding documentation PR in the [dapr/docs](https://github.com/dapr/docs/) repository. As the implementer, you are the best person to document your work! Implementation PRs will not be merged until the documentation PR is opened and ready for review.

common/proto/state/sqlserver/test.pb.go

@@ -0,0 +1,10 @@
+syntax = "proto3";
+
+option go_package = "github.com/dapr/components-contrib/common/proto/state/sqlserver";
+
+import "google/protobuf/timestamp.proto";
+
+message TestEvent {
+  int32 eventId = 1;
+  google.protobuf.Timestamp timestamp = 2;
+}

common/proto/state/sqlserver/test.proto

@@ -39,6 +39,7 @@ type pulsarMetadata struct {
 	ReceiverQueueSize                int                       `mapstructure:"receiverQueueSize"`
 	SubscriptionType                 string                    `mapstructure:"subscribeType"`
 	SubscriptionInitialPosition      string                    `mapstructure:"subscribeInitialPosition"`
+	ReplicateSubscriptionState       bool                      `mapstructure:"replicateSubscriptionState"`
 	SubscriptionMode                 string                    `mapstructure:"subscribeMode"`
 	Token                            string                    `mapstructure:"token"`
 	oauth2.ClientCredentialsMetadata `mapstructure:",squash"`

pubsub/pulsar/metadata.go

@@ -202,6 +202,17 @@ metadata:
     url: 
       title: "Pulsar SubscriptionInitialPosition"
       url: "https://pkg.go.dev/github.com/apache/pulsar-client-go/pulsar#SubscriptionInitialPosition"
+  - name: replicateSubscriptionState
+    type: bool
+    description: |
+      Enable replication of subscription state across geo-replicated Pulsar clusters.
+      When enabled, subscription state (such as cursor positions and acknowledgments) will be replicated to other clusters in a geo-replicated setup.
+      This is useful for maintaining subscription consistency during cluster failovers.
+    default: 'false'
+    example: '"true", "false"'
+    url:
+      title: "Pulsar Geo-Replication"
+      url: "https://pulsar.apache.org/docs/administration-geo/"
   - name: subscribeMode
     type: string
     description: |
@@ -210,4 +221,4 @@ metadata:
     example: '"durable"'
     url: 
       title: "Pulsar SubscriptionMode"
-      url: "https://pkg.go.dev/github.com/apache/pulsar-client-go/pulsar#SubscriptionMode"
+      url: "https://pkg.go.dev/github.com/apache/pulsar-client-go/pulsar#SubscriptionMode"

pubsub/pulsar/metadata.yaml

@@ -509,6 +509,7 @@ func (p *Pulsar) Subscribe(ctx context.Context, req pubsub.SubscribeRequest, han
 		MessageChannel:              channel,
 		NackRedeliveryDelay:         p.metadata.RedeliveryDelay,
 		ReceiverQueueSize:           p.metadata.ReceiverQueueSize,
+		ReplicateSubscriptionState:  p.metadata.ReplicateSubscriptionState,
 	}
 
 	// Handle KeySharedPolicy for key_shared subscription type

pubsub/pulsar/pulsar.go

@@ -605,6 +605,48 @@ func TestEncryptionKeys(t *testing.T) {
 	})
 }
 
+func TestParsePulsarMetadataReplicateSubscriptionState(t *testing.T) {
+	tt := []struct {
+		name                       string
+		replicateSubscriptionState string
+		expected                   bool
+	}{
+		{
+			name:                       "test replicateSubscriptionState true",
+			replicateSubscriptionState: "true",
+			expected:                   true,
+		},
+		{
+			name:                       "test replicateSubscriptionState false",
+			replicateSubscriptionState: "false",
+			expected:                   false,
+		},
+		{
+			name:                       "test replicateSubscriptionState empty (defaults to false)",
+			replicateSubscriptionState: "",
+			expected:                   false,
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.name, func(t *testing.T) {
+			m := pubsub.Metadata{}
+			m.Properties = map[string]string{
+				"host": "a",
+			}
+
+			if tc.replicateSubscriptionState != "" {
+				m.Properties["replicateSubscriptionState"] = tc.replicateSubscriptionState
+			}
+
+			meta, err := parsePulsarMetadata(m)
+
+			require.NoError(t, err)
+			assert.Equal(t, tc.expected, meta.ReplicateSubscriptionState)
+		})
+	}
+}
+
 func TestSanitiseURL(t *testing.T) {
 	tests := []struct {
 		name     string

pubsub/pulsar/pulsar_test.go

@@ -16,6 +16,7 @@ package sqlserver
 import (
 	"context"
 	"database/sql"
+	"encoding/base64"
 	"encoding/hex"
 	"encoding/json"
 	"errors"
@@ -287,8 +288,15 @@ func (s *SQLServer) Get(ctx context.Context, req *state.GetRequest) (*state.GetR
 		}
 	}
 
+	bytes, err := base64.StdEncoding.DecodeString(data)
+	if err != nil {
+		s.logger.
+			WithFields(map[string]any{"error": err}).
+			Debug("error decoding base64 data. Fallback to []byte")
+		bytes = []byte(data)
+	}
 	return &state.GetResponse{
-		Data:     []byte(data),
+		Data:     bytes,
 		ETag:     ptr.Of(etag),
 		Metadata: metadata,
 	}, nil
@@ -305,16 +313,23 @@ type dbExecutor interface {
 }
 
 func (s *SQLServer) executeSet(ctx context.Context, db dbExecutor, req *state.SetRequest) error {
-	var err error
-	var bytes []byte
-	bytes, err = utils.Marshal(req.Value, json.Marshal)
-	if err != nil {
-		return err
+	var reqValue string
+
+	bytes, ok := req.Value.([]byte)
+	if !ok {
+		bt, err := json.Marshal(req.Value)
+		if err != nil {
+			return err
+		}
+		reqValue = string(bt)
+	} else {
+		reqValue = base64.StdEncoding.EncodeToString(bytes)
 	}
+
 	etag := sql.Named(rowVersionColumnName, nil)
 	if req.HasETag() {
 		var b []byte
-		b, err = hex.DecodeString(*req.ETag)
+		b, err := hex.DecodeString(*req.ETag)
 		if err != nil {
 			return state.NewETagError(state.ETagInvalid, err)
 		}
@@ -327,13 +342,14 @@ func (s *SQLServer) executeSet(ctx context.Context, db dbExecutor, req *state.Se
 	}
 
 	var res sql.Result
+	var err error
 	if req.Options.Concurrency == state.FirstWrite {
 		res, err = db.ExecContext(ctx, s.upsertCommand, sql.Named(keyColumnName, req.Key),
-			sql.Named("Data", string(bytes)), etag,
+			sql.Named("Data", reqValue), etag,
 			sql.Named("FirstWrite", 1), sql.Named("TTL", ttl))
 	} else {
 		res, err = db.ExecContext(ctx, s.upsertCommand, sql.Named(keyColumnName, req.Key),
-			sql.Named("Data", string(bytes)), etag,
+			sql.Named("Data", reqValue), etag,
 			sql.Named("FirstWrite", 0), sql.Named("TTL", ttl))
 	}