Fix: Reuse kafka clients (#4049)

Signed-off-by: Javier Aliaga <[email protected]>

Author: javier-aliaga

.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

@@ -301,6 +301,9 @@ func (k *Kafka) ValidateAWS(metadata map[string]string) (awsAuth.Options, error)
 		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,
 		},

tests/certification/state/cassandra/docker-compose-cluster.yml

@@ -1,7 +1,7 @@
 version: '2'
 services:
   cassandra:
-    image: docker.io/bitnami/cassandra:4.0.1
+    image: docker.io/bitnamilegacy/cassandra:4.0.1
     ports:
       - '7000:7000'
       - '9042:9042'
@@ -14,7 +14,7 @@ services:
       - CASSANDRA_PASSWORD_SEEDER=yes
 
   cassandra2:
-    image: docker.io/bitnami/cassandra:4.0.1
+    image: docker.io/bitnamilegacy/cassandra:4.0.1
     ports:
       - 7001:7000
       - 9043:9042

tests/certification/state/cassandra/docker-compose-single.yml

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