Rebase from master

Signed-off-by: dttung2905 <[email protected]>

pkg/scalers/apache_kafka_scaler.go

@@ -189,6 +189,7 @@ func parseApacheKafkaAuthParams(config *ScalerConfig, meta *apacheKafkaMetadata)
 			if err != nil {
 				return fmt.Errorf("%w. No awsRegion given", err)
 			}
+			meta.awsRegion = awsRegion.(string)
 			auth, err := awsutils.GetAwsAuthorization(config.TriggerUniqueKey, config.PodIdentity, config.TriggerMetadata, config.AuthParams, config.ResolvedEnv)
 			if err != nil {
 				return err

pkg/scalers/apache_kafka_scaler_test.go

@@ -288,7 +288,6 @@ func getBrokerApacheKafkaTestBase(t *testing.T, meta apacheKafkaMetadata, testDa
 	if er != nil {
 		t.Errorf("Unable to convert test data lagThreshold %s to string", testData.metadata["lagThreshold"])
 	}
-
 	if meta.lagThreshold != expectedLagThreshold && meta.lagThreshold != defaultKafkaLagThreshold {
 		t.Errorf("Expected lagThreshold to be either %v or %v got %v ", meta.lagThreshold, defaultKafkaLagThreshold, expectedLagThreshold)
 	}

pkg/scalers/kafka_scaler.go

@@ -419,7 +419,7 @@ func parseKafkaMetadata(config *ScalerConfig, logger logr.Logger) (kafkaMetadata
 		}
 		meta.offsetResetPolicy = policy
 	}
-
+	meta.lagThreshold = defaultKafkaLagThreshold
 	lagThreshold, err := getParameterFromConfigV2(config, lagThresholdMetricName, true, false, false, true, defaultKafkaLagThreshold, reflect.TypeOf(64))
 	if err != nil {
 		return meta, err
@@ -482,6 +482,7 @@ func parseKafkaMetadata(config *ScalerConfig, logger logr.Logger) (kafkaMetadata
 	if err != nil {
 		return meta, fmt.Errorf("error parsing kafka version: %w", err)
 	}
+	meta.version = version
 	meta.triggerIndex = config.TriggerIndex
 	return meta, nil
 }

pkg/scalers/kafka_scaler_test.go

@@ -317,7 +317,7 @@ var kafkaMetricIdentifiers = []kafkaMetricIdentifier{
 }
 
 func TestGetBrokers(t *testing.T) {
-	for idx, testData := range parseKafkaMetadataTestDataset {
+	for _, testData := range parseKafkaMetadataTestDataset {
 		meta, err := parseKafkaMetadata(&ScalerConfig{TriggerMetadata: testData.metadata, AuthParams: validWithAuthParams}, logr.Discard())
 		getBrokerTestBase(t, meta, testData, err)