Merge main into dynamo-global-ttl branch - resolve conflicts

Author: ZeynelKoca

.build-tools/go.mod

@@ -12,7 +12,7 @@ require (
 )
 
 require (
-	github.com/dapr/kit v0.15.3-0.20250717140748-8b780b4d81c5 // indirect
+	github.com/dapr/kit v0.16.1 // indirect
 	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/iancoleman/orderedmap v0.0.0-20190318233801-ac98e3ecb4b0 // indirect

.build-tools/go.sum

@@ -1,6 +1,6 @@
 github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
-github.com/dapr/kit v0.15.3-0.20250717140748-8b780b4d81c5 h1:Q26gmPxs6WnnBYoudOlznPHsmrbTawcYEpHg4VoB7v8=
-github.com/dapr/kit v0.15.3-0.20250717140748-8b780b4d81c5/go.mod h1:40ZWs5P6xfYf7O59XgwqZkIyDldTIXlhTQhGop8QoSM=
+github.com/dapr/kit v0.16.1 h1:MqLAhHVg8trPy2WJChMZFU7ToeondvxcNHYVvMDiVf4=
+github.com/dapr/kit v0.16.1/go.mod h1:40ZWs5P6xfYf7O59XgwqZkIyDldTIXlhTQhGop8QoSM=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

.github/infrastructure/docker-compose-cassandra.yml

@@ -2,7 +2,7 @@ version: '2'
 
 services:
   cassandra:
-    image: docker.io/bitnami/cassandra:4.1
+    image: docker.io/bitnamilegacy/cassandra:4.1
     ports:
       - '7000:7000'
       - '9042:9042'

common/component/kafka/clients.go

@@ -22,6 +22,10 @@ func (k *Kafka) latestClients() (*clients, error) {
 
 	// case 1: use aws clients with refreshable tokens in the cfg
 	case k.awsConfig != nil:
+		if k.clients != nil {
+			return k.clients, nil
+		}
+
 		awsKafkaOpts := KafkaOptions{
 			Config:          k.config,
 			ConsumerGroup:   k.consumerGroup,
@@ -34,10 +38,12 @@ func (k *Kafka) latestClients() (*clients, error) {
 		if err != nil {
 			return nil, fmt.Errorf("failed to get AWS IAM Kafka clients: %w", err)
 		}
-		return &clients{
+
+		k.clients = &clients{
 			consumerGroup: awsKafkaClients.ConsumerGroup,
 			producer:      awsKafkaClients.Producer,
-		}, nil
+		}
+		return k.clients, nil
 
 	// case 2: normal static auth profile clients
 	default:

common/component/kafka/kafka.go

@@ -295,12 +295,16 @@ func (k *Kafka) ValidateAWS(metadata map[string]string) (awsAuth.Options, error)
 	}
 
 	return awsAuth.Options{
+		Logger:                k.logger,
 		Region:                region,
 		AccessKey:             accessKey,
 		SecretKey:             secretKey,
 		AssumeRoleArn:         role,
 		AssumeRoleSessionName: session,
 		SessionToken:          token,
+		TrustAnchorArn:        metadata["trustAnchorArn"],
+		TrustProfileArn:       metadata["trustProfileArn"],
+		Properties:            metadata,
 	}, nil
 }
 

common/component/kafka/kafka_test.go

@@ -520,6 +520,14 @@ func TestValidateAWS(t *testing.T) {
 				AssumeRoleArn:         "testRoleArn",
 				AssumeRoleSessionName: "testSessionName",
 				SessionToken:          "testSessionToken",
+				Properties: map[string]string{
+					"region":        "us-east-1",
+					"accessKey":     "testAccessKey",
+					"secretKey":     "testSecretKey",
+					"assumeRoleArn": "testRoleArn",
+					"sessionName":   "testSessionName",
+					"sessionToken":  "testSessionToken",
+				},
 			},
 			err: nil,
 		},
@@ -540,6 +548,14 @@ func TestValidateAWS(t *testing.T) {
 				AssumeRoleArn:         "awsRoleArn",
 				AssumeRoleSessionName: "awsSessionName",
 				SessionToken:          "awsSessionToken",
+				Properties: map[string]string{
+					"awsRegion":         "us-west-2",
+					"awsAccessKey":      "awsAccessKey",
+					"awsSecretKey":      "awsSecretKey",
+					"awsIamRoleArn":     "awsRoleArn",
+					"awsStsSessionName": "awsSessionName",
+					"awsSessionToken":   "awsSessionToken",
+				},
 			},
 			err: nil,
 		},

go.mod

@@ -64,7 +64,7 @@ require (
 	github.com/cloudwego/kitex-examples v0.1.1
 	github.com/cyphar/filepath-securejoin v0.2.4
 	github.com/dancannon/gorethink v4.0.0+incompatible
-	github.com/dapr/kit v0.15.3-0.20250717140748-8b780b4d81c5
+	github.com/dapr/kit v0.16.1
 	github.com/didip/tollbooth/v7 v7.0.1
 	github.com/eclipse/paho.mqtt.golang v1.4.3
 	github.com/fasthttp-contrib/sessions v0.0.0-20160905201309-74f6ac73d5d5

go.sum

@@ -516,8 +516,8 @@ github.com/dancannon/gorethink v4.0.0+incompatible h1:KFV7Gha3AuqT+gr0B/eKvGhbjm
 github.com/dancannon/gorethink v4.0.0+incompatible/go.mod h1:BLvkat9KmZc1efyYwhz3WnybhRZtgF1K929FD8z1avU=
 github.com/danieljoos/wincred v1.1.2 h1:QLdCxFs1/Yl4zduvBdcHB8goaYk9RARS2SgLLRuAyr0=
 github.com/danieljoos/wincred v1.1.2/go.mod h1:GijpziifJoIBfYh+S7BbkdUTU4LfM+QnGqR5Vl2tAx0=
-github.com/dapr/kit v0.15.3-0.20250717140748-8b780b4d81c5 h1:Q26gmPxs6WnnBYoudOlznPHsmrbTawcYEpHg4VoB7v8=
-github.com/dapr/kit v0.15.3-0.20250717140748-8b780b4d81c5/go.mod h1:40ZWs5P6xfYf7O59XgwqZkIyDldTIXlhTQhGop8QoSM=
+github.com/dapr/kit v0.16.1 h1:MqLAhHVg8trPy2WJChMZFU7ToeondvxcNHYVvMDiVf4=
+github.com/dapr/kit v0.16.1/go.mod h1:40ZWs5P6xfYf7O59XgwqZkIyDldTIXlhTQhGop8QoSM=
 github.com/dave/jennifer v1.4.0/go.mod h1:fIb+770HOpJ2fmN9EPPKOqm1vMGhB+TwXKMZhrIygKg=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

secretstores/aws/secretmanager/secretmanager.go

@@ -25,6 +25,7 @@ import (
 	"github.com/dapr/components-contrib/metadata"
 	"github.com/dapr/components-contrib/secretstores"
 	"github.com/dapr/kit/logger"
+	kitmd "github.com/dapr/kit/metadata"
 )
 
 const (
@@ -177,17 +178,11 @@ func (s *smSecretStore) BulkGetSecret(ctx context.Context, req secretstores.Bulk
 }
 
 func (s *smSecretStore) getSecretManagerMetadata(spec secretstores.Metadata) (*SecretManagerMetaData, error) {
-	b, err := json.Marshal(spec.Properties)
-	if err != nil {
-		return nil, err
-	}
-
 	var meta SecretManagerMetaData
-	err = json.Unmarshal(b, &meta)
+	err := kitmd.DecodeMetadata(spec.Properties, &meta)
 	if err != nil {
 		return nil, err
 	}
-
 	return &meta, nil
 }
 

secretstores/aws/secretmanager/secretmanager_test.go

@@ -499,3 +499,23 @@ func TestGetFeatures(t *testing.T) {
 		assert.Empty(t, f)
 	})
 }
+
+func TestGetSecretManagerMetadata(t *testing.T) {
+	s := &smSecretStore{
+		logger: logger.NewLogger("test"),
+	}
+
+	t.Run("parse multipleKeyValuesPerSecret as string", func(t *testing.T) {
+		metadata := secretstores.Metadata{}
+		metadata.Properties = map[string]string{
+			"region":                     "us-east-1",
+			"accessKey":                  "test",
+			"secretKey":                  "test",
+			"multipleKeyValuesPerSecret": "true",
+		}
+
+		meta, err := s.getSecretManagerMetadata(metadata)
+		require.NoError(t, err)
+		assert.True(t, meta.MultipleKeyValuesPerSecret)
+	})
+}