Merge branch 'main' into add-akeyless-secretstore

Author: kgal-akl

.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

@@ -181,6 +181,12 @@ func (k *Kafka) Init(ctx context.Context, metadata map[string]string) error {
 		if err != nil {
 			return err
 		}
+	case oidcPrivateKeyJWTAuthType:
+		k.logger.Info("Configuring SASL OAuth2/OIDC authentication with private key JWT")
+		err = updateOidcPrivateKeyJWTAuthInfo(config, meta)
+		if err != nil {
+			return err
+		}
 	case passwordAuthType:
 		k.logger.Info("Configuring SASL Password authentication")
 		k.saslUsername = meta.SaslUsername
@@ -295,12 +301,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,
 		},

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

@@ -561,3 +561,9 @@ func BenchmarkGetKeyVersion(b *testing.B) {
 		}
 	}
 }
+
+func Test_KeyList(t *testing.T) {
+	s := NewRedisStateStore(logger.NewLogger("test"))
+	_, ok := s.(state.KeysLiker)
+	require.True(t, ok)
+}