Repository: kedacore/keda

Issue:
Solace Scaler: Enhancement to support a hostlist of Solace brokers
### Proposal

The issue I am reporting is the current Solace scaler does not support Solace Disaster Recovery (DR). Solace DR introduces an additional Solace Broker and operates in an active standby mode, so the scaler needs to query two different brokers with different endpoints, verify connectivity status as well as determine which of the brokers are active, once that is deteremined, the scaler operates as it did before.

I've already made the necessary code cahge

### Use-Case

Use case is to support Solace deployments with Disaster Recovery enabled.

### Is this a feature you are interested in implementing yourself?

Yes

### Anything else?

_No response_

Hints from issue comments:

Solution patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ffc0985f8..56945df74 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -85,15 +85,18 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 - **General**: Allow excluding labels from being propagated from ScaledObject and ScaledJob to generated HPA and Job objects ([#6849](https://github.com/kedacore/keda/issues/6849))
 - **General**: chore: only add webhook DNS names when webhook patching is enabled ([#7002](https://github.com/kedacore/keda/issues/7002))
 - **General**: Improve Events emitted from ScaledObject controller ([#6802](https://github.com/kedacore/keda/issues/6802))
+- **Apache Kafka Scaler**: Add support for even distribution of partitions to consumers ([#2581](https://github.com/kedacore/keda/issues/2581))
 - **Azure Pipelines Scaler**: Add new `fetchUnfinishedJobsOnly` property to fetch only unfinished pipeline jobs for a pool ([#6819](https://github.com/kedacore/keda/issues/6819))
 - **Datadog Scaler**: Add a specific timeout configuration parameter for the Datadog trigger ([#6999](https://github.com/kedacore/keda/pull/6999))
 - **Datadog Scaler**: Fix bug with datadogNamespace config ([#6828](https://github.com/kedacore/keda/pull/6828))
 - **Datadog Scaler**: Improve Datadog scaler error messages ([#6999](https://github.com/kedacore/keda/pull/6999))
 - **Github Scaler**: Add support to control unlabeled job/runner matching ([#6900](https://github.com/kedacore/keda/issues/6900))
 - **InfluxDB Scaler**: Add support for InfluxDB v3 ([#6981](https://github.com/kedacore/keda/issues/6981))
+- **Kafka Scaler**: Add support for even distribution of partitions to consumers ([#2581](https://github.com/kedacore/keda/issues/2581))
 - **Metrics API Scaler**: Support AuthParams for authMode ([#6939](https://github.com/kedacore/keda/issues/6939))
 - **Metrics API Scaler**: Support multiple auth methods simultaneously ([#6642](https://github.com/kedacore/keda/issues/6642))
 - **RabbitMQ Scaler**: add `DeliverGetRate`, `PublishedToDeliveredRatio` and `ExpectedQueueConsumptionTime` trigger modes to RabbitMQ scaler ([#7071](https://github.com/kedacore/keda/issues/7071))
+- **Solace Scaler**: Add hostlist support for Solace brokers ([#7090](https://github.com/kedacore/keda/issues/7090))
 - **Temporal Scaler**: Always set `temporal-namespace` header on requests([#7079](https://github.com/kedacore/keda/issues/7079))
 - **Temporal Scaler**: Support custom tlsServerName ([#6820](https://github.com/kedacore/keda/pull/6820))
 
diff --git a/pkg/scalers/apache_kafka_scaler.go b/pkg/scalers/apache_kafka_scaler.go
index 101e33120..d304dbd1a 100644
--- a/pkg/scalers/apache_kafka_scaler.go
+++ b/pkg/scalers/apache_kafka_scaler.go
@@ -60,8 +60,9 @@ type apacheKafkaMetadata struct {
 
 	// If an invalid offset is found, whether to scale to 1 (false - the default) so consumption can
 	// occur or scale to 0 (true). See discussion in https://github.com/kedacore/keda/issues/2612
-	ScaleToZeroOnInvalidOffset bool `keda:"name=scaleToZeroOnInvalidOffset, order=triggerMetadata, optional"`
-	LimitToPartitionsWithLag   bool `keda:"name=limitToPartitionsWithLag,   order=triggerMetadata, optional"`
+	ScaleToZeroOnInvalidOffset         bool `keda:"name=scaleToZeroOnInvalidOffset,         order=triggerMetadata, optional"`
+	LimitToPartitionsWithLag           bool `keda:"name=limitToPartitionsWithLag,           order=triggerMetadata, optional"`
+	EnsureEvenDistributionOfPartitions bool `keda:"name=ensureEvenDistributionOfPartitions, order=triggerMetadata, optional"`
 
 	// SASL
 	SASLType kafkaSaslType `keda:"name=sasl,     order=triggerMetadata;authParams, enum=none;plaintext;scram_sha256;scram_sha512;gssapi;aws_msk_iam, default=none"`
@@ -505,9 +506,15 @@ func (s *apacheKafkaScaler) getTotalLag(ctx context.Context) (int64, int64, erro
 
 	s.logger.V(1).Info(fmt.Sprintf("Kafka scaler: Consumer offsets %v, producer offsets %v", consumerOffsets, producerOffsets))
 
-	if !s.metadata.AllowIdleConsumers || s.metadata.LimitToPartitionsWithLag {
+	if !s.metadata.AllowIdleConsumers || s.metadata.LimitToPartitionsWithLag || s.metadata.EnsureEvenDistributionOfPartitions {
 		// don't scale out beyond the number of topicPartitions or partitionsWithLag depending on settings
 		upperBound := totalTopicPartitions
+		// Ensure that the number of partitions is evenly distributed across the number of consumers
+		if s.metadata.EnsureEvenDistributionOfPartitions {
+			nextFactor := getNextFactorThatBalancesConsumersToTopicPartitions(totalLag, totalTopicPartitions, s.metadata.LagThreshold)
+			s.logger.V(1).Info(fmt.Sprintf("Kafka scaler: Providing metrics to ensure even distribution of partitions on totalLag %v, topicPartitions %v, evenPartitions %v", totalLag, totalTopicPartitions, nextFactor))
+			totalLag = nextFactor * s.metadata.LagThreshold
+		}
 		if s.metadata.LimitToPartitionsWithLag {
 			upperBound = partitionsWithLag
 		}
diff --git a/pkg/scalers/kafka_scaler.go b/pkg/scalers/kafka_scaler.go
index fbe935291..6495d2bf9 100644
--- a/pkg/scalers/kafka_scaler.go
+++ b/pkg/scalers/kafka_scaler.go
@@ -23,7 +23,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math"
 	"os"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -68,8 +70,9 @@ type kafkaMetadata struct {
 
 	// If an invalid offset is found, whether to scale to 1 (false - the default) so consumption can
 	// occur or scale to 0 (true). See discussion in https://github.com/kedacore/keda/issues/2612
-	scaleToZeroOnInvalidOffset bool
-	limitToPartitionsWithLag   bool
+	scaleToZeroOnInvalidOffset         bool
+	limitToPartitionsWithLag           bool
+	ensureEvenDistributionOfPartitions bool
 
 	// SASL
 	saslType kafkaSaslType
@@ -594,6 +597,22 @@ func parseKafkaMetadata(config *scalersconfig.ScalerConfig, logger logr.Logger)
 		}
 	}
 
+	meta.ensureEvenDistributionOfPartitions = false
+	if val, ok := config.TriggerMetadata["ensureEvenDistributionOfPartitions"]; ok {
+		t, err := strconv.ParseBool(val)
+		if err != nil {
+			return meta, fmt.Errorf("error parsing ensureEvenDistributionOfPartitions: %w", err)
+		}
+		meta.ensureEvenDistributionOfPartitions = t
+
+		if meta.limitToPartitionsWithLag && meta.ensureEvenDistributionOfPartitions {
+			return meta, fmt.Errorf("limitToPartitionsWithLag and ensureEvenDistributionOfPartitions cannot be set simultaneously")
+		}
+		if len(meta.topic) == 0 && meta.ensureEvenDistributionOfPartitions {
+			return meta, fmt.Errorf("topic must be specified when using ensureEvenDistributionOfPartitions")
+		}
+	}
+
 	meta.version = sarama.V1_0_0_0
 	if val, ok := config.TriggerMetadata["version"]; ok {
 		val = strings.TrimSpace(val)
@@ -972,9 +991,15 @@ func (s *kafkaScaler) getTotalLag() (int64, int64, error) {
 	}
 	s.logger.V(1).Info(fmt.Sprintf("Kafka scaler: Providing metrics based on totalLag %v, topicPartitions %v, threshold %v", totalLag, len(topicPartitions), s.metadata.lagThreshold))
 
-	if !s.metadata.allowIdleConsumers || s.metadata.limitToPartitionsWithLag {
+	if !s.metadata.allowIdleConsumers || s.metadata.limitToPartitionsWithLag || s.metadata.ensureEvenDistributionOfPartitions {
 		// don't scale out beyond the number of topicPartitions or partitionsWithLag depending on settings
 		upperBound := totalTopicPartitions
+		// Ensure that the number of partitions is evenly distributed across the number of consumers
+		if s.metadata.ensureEvenDistributionOfPartitions {
+			nextFactor := getNextFactorThatBalancesConsumersToTopicPartitions(totalLag, totalTopicPartitions, s.metadata.lagThreshold)
+			s.logger.V(1).Info(fmt.Sprintf("Kafka scaler: Providing metrics to ensure even distribution of partitions on totalLag %v, topicPartitions %v, evenPartitions %v", totalLag, totalTopicPartitions, nextFactor))
+			totalLag = nextFactor * s.metadata.lagThreshold
+		}
 		if s.metadata.limitToPartitionsWithLag {
 			upperBound = partitionsWithLag
 		}
@@ -986,6 +1011,16 @@ func (s *kafkaScaler) getTotalLag() (int64, int64, error) {
 	return totalLag, totalLagWithPersistent, nil
 }
 
+func getNextFactorThatBalancesConsumersToTopicPartitions(totalLag int64, totalTopicPartitions int64, lagThreshold int64) int64 {
+	factors := FindFactors(totalTopicPartitions)
+	for _, factor := range factors {
+		if factor*lagThreshold >= totalLag {
+			return factor
+		}
+	}
+	return totalTopicPartitions
+}
+
 type brokerOffsetResult struct {
 	offsetResp *sarama.OffsetResponse
 	err        error
@@ -1051,3 +1086,28 @@ func (s *kafkaScaler) getProducerOffsets(topicPartitions map[string][]int32) (ma
 
 	return topicPartitionsOffsets, nil
 }
+
+// FindFactors returns all factors of a given number
+func FindFactors(n int64) []int64 {
+	if n < 1 {
+		return nil
+	}
+
+	var factors []int64
+	sqrtN := int64(math.Sqrt(float64(n)))
+
+	for i := int64(1); i <= sqrtN; i++ {
+		if n%i == 0 {
+			factors = append(factors, i)
+			if i != n/i {
+				factors = append(factors, n/i)
+			}
+		}
+	}
+
+	sort.Slice(factors, func(i, j int) bool {
+		return factors[i] < factors[j]
+	})
+
+	return factors
+}
diff --git a/pkg/scalers/solace_scaler.go b/pkg/scalers/solace_scaler.go
index 40df8985f..d4a675059 100644
--- a/pkg/scalers/solace_scaler.go
+++ b/pkg/scalers/solace_scaler.go
@@ -24,6 +24,7 @@ const (
 	// REST ENDPOINT String Patterns
 	solaceSempQueryFieldURLSuffix = "?select=msgs,msgSpoolUsage,averageRxMsgRate"
 	solaceSempEndpointURLTemplate = "%s/%s/%s/monitor/msgVpns/%s/%ss/%s" + solaceSempQueryFieldURLSuffix
+	solaceSempVpnStateURLTemplate = "%s/%s/%s/monitor/msgVpns/%s" + "?select=state"
 
 	// SEMP REST API Context
 	solaceAPIName            = "SEMP"
@@ -75,6 +76,7 @@ type SolaceScaler struct {
 	metadata   *SolaceMetadata
 	httpClient *http.Client
 	logger     logr.Logger
+	curHostIdx int
 }
 
 type SolaceMetadata struct {
@@ -84,7 +86,8 @@ type SolaceMetadata struct {
 	SolaceMetaSempBaseURL string `keda:"name=solaceSempBaseURL,  order=triggerMetadata"`
 
 	// Full SEMP URL to target queue (CONSTRUCTED IN CODE)
-	EndpointURL string
+	endpointURLsList []string
+	vpnStateURLsList []string
 
 	// Solace Message VPN
 	MessageVpn string `keda:"name=messageVpn,   order=triggerMetadata"`
@@ -119,20 +122,31 @@ func (s *SolaceMetadata) Validate() error {
 	return nil
 }
 
-// SEMP API Response Root Struct
-type solaceSEMPResponse struct {
-	Collections solaceSEMPCollections `json:"collections"`
-	Data        solaceSEMPData        `json:"data"`
-	Meta        solaceSEMPMetadata    `json:"meta"`
+// SEMP API Response Root Struct for VPNs
+type solaceVpnSEMPResponse struct {
+	Data solaceVpnSEMPData  `json:"data"`
+	Meta solaceSEMPMetadata `json:"meta"`
+}
+
+// SEMP API Response Queue Data Struct
+type solaceVpnSEMPData struct {
+	State string `json:"state"`
+}
+
+// SEMP API Response Root Struct for Queues
+type solaceQueueSEMPResponse struct {
+	Collections solaceQueueSEMPCollections `json:"collections"`
+	Data        solaceQueueSEMPData        `json:"data"`
+	Meta        solaceSEMPMetadata         `json:"meta"`
 }
 
 // SEMP API Response Collections Struct
-type solaceSEMPCollections struct {
+type solaceQueueSEMPCollections struct {
 	Msgs solaceSEMPMessages `json:"msgs"`
 }
 
 // SEMP API Response Queue Data Struct
-type solaceSEMPData struct {
+type solaceQueueSEMPData struct {
 	MsgSpoolUsage int `json:"msgSpoolUsage"`
 	MsgRcvRate    int `json:"averageRxMsgRate"`
 }
@@ -171,6 +185,7 @@ func NewSolaceScaler(config *scalersconfig.ScalerConfig) (Scaler, error) {
 		metadata:   solaceMetadata,
 		httpClient: httpClient,
 		logger:     logger,
+		curHostIdx: 0,
 	}, nil
 }
 
@@ -183,15 +198,26 @@ func parseSolaceMetadata(config *scalersconfig.ScalerConfig) (*SolaceMetadata, e
 	meta.triggerIndex = config.TriggerIndex
 
 	// Format Solace SEMP Queue Endpoint (REST URL)
-	meta.EndpointURL = fmt.Sprintf(
-		solaceSempEndpointURLTemplate,
-		meta.SolaceMetaSempBaseURL,
-		solaceAPIName,
-		solaceAPIVersion,
-		meta.MessageVpn,
-		solaceAPIObjectTypeQueue,
-		url.QueryEscape(meta.QueueName),
-	)
+	sempURLs := strings.Split(meta.SolaceMetaSempBaseURL, ",")
+	for i := 0; i < len(sempURLs); i++ {
+		sempURL := strings.TrimSpace(sempURLs[i])
+		meta.endpointURLsList = append(meta.endpointURLsList, fmt.Sprintf(
+			solaceSempEndpointURLTemplate,
+			sempURL,
+			solaceAPIName,
+			solaceAPIVersion,
+			meta.MessageVpn,
+			solaceAPIObjectTypeQueue,
+			url.QueryEscape(meta.QueueName),
+		))
+		meta.vpnStateURLsList = append(meta.vpnStateURLsList, fmt.Sprintf(
+			solaceSempVpnStateURLTemplate,
+			sempURL,
+			solaceAPIName,
+			solaceAPIVersion,
+			url.QueryEscape(meta.MessageVpn),
+		))
+	}
 
 	return meta, nil
 }
@@ -245,52 +271,126 @@ func (s *SolaceScaler) GetMetricSpecForScaling(context.Context) []v2.MetricSpec
 	return metricSpecList
 }
 
-// returns SolaceMetricValues struct populated from broker  SEMP endpoint
-func (s *SolaceScaler) getSolaceQueueMetricsFromSEMP(ctx context.Context) (SolaceMetricValues, error) {
-	var scaledMetricEndpointURL = s.metadata.EndpointURL
+func (s *SolaceScaler) getVpnState(ctx context.Context, sempURL string) (string, error) {
+	var httpClient = s.httpClient
+	var sempResponse solaceVpnSEMPResponse
+
+	// Define HTTP Request
+	request, err := http.NewRequestWithContext(ctx, "GET", sempURL, nil)
+	if err != nil {
+		return "", fmt.Errorf("failed attempting request to solace semp api: %s url: %s", err.Error(), sempURL)
+	}
+
+	// Add HTTP Auth and Headers
+	request.SetBasicAuth(s.metadata.Username, s.metadata.Password)
+	request.Header.Set("Content-Type", "application/json")
+
+	// Call Solace SEMP API
+	response, err := httpClient.Do(request)
+	if err != nil {
+		return "", fmt.Errorf("call to solace semp api failed: %s, url: %s", err.Error(), sempURL)
+	}
+	defer response.Body.Close()
+
+	// Check HTTP Status Code
+	if response.StatusCode < 200 || response.StatusCode > 299 {
+		return "", fmt.Errorf("semp request http status code: %d - %s, url: %s", response.StatusCode, response.Status, sempURL)
+	}
+
+	// Decode SEMP Response and Test
+	if err := json.NewDecoder(response.Body).Decode(&sempResponse); err != nil {
+		return "", fmt.Errorf("failed to read semp response body: %s, url: %s", err.Error(), sempURL)
+	}
+
+	if sempResponse.Meta.ResponseCode < 200 || sempResponse.Meta.ResponseCode > 299 {
+		return "", fmt.Errorf("solace semp api returned error status: %d, URL: %s", sempResponse.Meta.ResponseCode, sempURL)
+	}
+
+	return sempResponse.Data.State, nil
+}
+
+func (s *SolaceScaler) getQueueMetrics(ctx context.Context, sempURL string) (SolaceMetricValues, error) {
 	var httpClient = s.httpClient
-	var sempResponse solaceSEMPResponse
+	var sempResponse solaceQueueSEMPResponse
 	var metricValues SolaceMetricValues
 
-	//	RETRIEVE METRICS FROM SOLACE SEMP API
-	//	Define HTTP Request
-	request, err := http.NewRequestWithContext(ctx, "GET", scaledMetricEndpointURL, nil)
+	// Define HTTP Request
+	request, err := http.NewRequestWithContext(ctx, "GET", sempURL, nil)
 	if err != nil {
-		return SolaceMetricValues{}, fmt.Errorf("failed attempting request to solace semp api: %w", err)
+		return SolaceMetricValues{}, fmt.Errorf("failed attempting request to solace semp api: %s, url: %s", err.Error(), sempURL)
 	}
 
-	//	Add HTTP Auth and Headers
+	// Add HTTP Auth and Headers
 	request.SetBasicAuth(s.metadata.Username, s.metadata.Password)
 	request.Header.Set("Content-Type", "application/json")
 
-	//	Call Solace SEMP API
+	// Call Solace SEMP API
 	response, err := httpClient.Do(request)
 	if err != nil {
-		return SolaceMetricValues{}, fmt.Errorf("call to solace semp api failed: %w", err)
+		return SolaceMetricValues{}, fmt.Errorf("call to solace semp api failed: %s, url: %s", err.Error(), sempURL)
 	}
 	defer response.Body.Close()
 
 	// Check HTTP Status Code
 	if response.StatusCode < 200 || response.StatusCode > 299 {
-		sempError := fmt.Errorf("semp request http status code: %s - %s", strconv.Itoa(response.StatusCode), response.Status)
-		return SolaceMetricValues{}, sempError
+		return SolaceMetricValues{}, fmt.Errorf("semp request http status code: %d - %s, url: %s", response.StatusCode, response.Status, sempURL)
 	}
 
 	// Decode SEMP Response and Test
 	if err := json.NewDecoder(response.Body).Decode(&sempResponse); err != nil {
-		return SolaceMetricValues{}, fmt.Errorf("failed to read semp response body: %w", err)
+		return SolaceMetricValues{}, fmt.Errorf("failed to read semp response body: %s, url: %s", err.Error(), sempURL)
 	}
+
 	if sempResponse.Meta.ResponseCode < 200 || sempResponse.Meta.ResponseCode > 299 {
-		return SolaceMetricValues{}, fmt.Errorf("solace semp api returned error status: %d", sempResponse.Meta.ResponseCode)
+		return SolaceMetricValues{}, fmt.Errorf("solace semp api returned error status: %d, url: %s", sempResponse.Meta.ResponseCode, sempURL)
 	}
 
-	// Set Return Values
 	metricValues.msgCount = sempResponse.Collections.Msgs.Count
 	metricValues.msgSpoolUsage = sempResponse.Data.MsgSpoolUsage
 	metricValues.msgRcvRate = sempResponse.Data.MsgRcvRate
+
 	return metricValues, nil
 }
 
+// returns SolaceMetricValues struct populated from broker  SEMP endpoint
+func (s *SolaceScaler) getSolaceQueueMetricsFromSEMP(ctx context.Context) (SolaceMetricValues, error) {
+	var metricValues SolaceMetricValues
+	var errorList []string
+
+	// RETRIEVE METRICS FROM SOLACE SEMP API
+	for i := 0; i < len(s.metadata.endpointURLsList); i++ {
+		idx := (s.curHostIdx + i) % len(s.metadata.endpointURLsList)
+		sempQueueURL := s.metadata.endpointURLsList[idx]
+		sempVpnStateURL := s.metadata.vpnStateURLsList[idx]
+
+		vpnState, err := s.getVpnState(ctx, sempVpnStateURL)
+		if err != nil {
+			errorList = append(errorList, "Host "+strconv.Itoa(idx+1)+" Error: "+err.Error())
+			continue
+		}
+
+		if vpnState != "up" {
+			errorList = append(errorList, "Host "+strconv.Itoa(idx+1)+" Error: Message vpn is not up ("+vpnState+") url: "+sempVpnStateURL)
+			continue
+		}
+
+		metricValues, err = s.getQueueMetrics(ctx, sempQueueURL)
+		if err != nil {
+			errorList = append(errorList, "Host "+strconv.Itoa(idx+1)+" Error: "+err.Error())
+			continue
+		}
+
+		// Found active broker, update currHostIdx if it has changed
+		if idx != s.curHostIdx {
+			s.curHostIdx = idx
+		}
+
+		return metricValues, nil
+	}
+
+	return SolaceMetricValues{}, fmt.Errorf("unable to collect metrics, error(s): %s", strings.Join(errorList, "\n  "))
+}
+
 // INTERFACE METHOD
 // Call SEMP API to retrieve metrics
 // returns value for named metric
@@ -298,8 +398,8 @@ func (s *SolaceScaler) getSolaceQueueMetricsFromSEMP(ctx context.Context) (Solac
 func (s *SolaceScaler) GetMetricsAndActivity(ctx context.Context, metricName string) ([]external_metrics.ExternalMetricValue, bool, error) {
 	var metricValues, mv SolaceMetricValues
 	var mve error
+
 	if mv, mve = s.getSolaceQueueMetricsFromSEMP(ctx); mve != nil {
-		s.logger.Error(mve, "call to semp endpoint failed")
 		return []external_metrics.ExternalMetricValue{}, false, mve
 	}
 	metricValues = mv
diff --git a/schema/generated/scalers-metadata-schema.json b/schema/generated/scalers-metadata-schema.json
index 1238db78f..c234488d3 100644
--- a/schema/generated/scalers-metadata-schema.json
+++ b/schema/generated/scalers-metadata-schema.json
@@ -138,6 +138,12 @@
                     "optional": true,
                     "metadataVariableReadable": true
                 },
+                {
+                    "name": "ensureEvenDistributionOfPartitions",
+                    "type": "string",
+                    "optional": true,
+                    "metadataVariableReadable": true
+                },
                 {
                     "name": "sasl",
                     "type": "string",
diff --git a/schema/generated/scalers-metadata-schema.yaml b/schema/generated/scalers-metadata-schema.yaml
index 46081dc94..86b8424ef 100644
--- a/schema/generated/scalers-metadata-schema.yaml
+++ b/schema/generated/scalers-metadata-schema.yaml
@@ -92,6 +92,10 @@ scalers:
           type: string
           optional: true
           metadataVariableReadable: true
+        - name: ensureEvenDistributionOfPartitions
+          type: string
+          optional: true
+          metadataVariableReadable: true
         - name: sasl
           type: string
           default: none



Test patch:
diff --git a/pkg/scalers/apache_kafka_scaler_test.go b/pkg/scalers/apache_kafka_scaler_test.go
index 511077d22..d6ea6ec28 100644
--- a/pkg/scalers/apache_kafka_scaler_test.go
+++ b/pkg/scalers/apache_kafka_scaler_test.go
@@ -13,17 +13,18 @@ import (
 )
 
 type parseApacheKafkaMetadataTestData struct {
-	metadata                 map[string]string
-	isError                  bool
-	numBrokers               int
-	brokers                  []string
-	group                    string
-	topic                    []string
-	partitionLimitation      []int
-	offsetResetPolicy        offsetResetPolicy
-	allowIdleConsumers       bool
-	excludePersistentLag     bool
-	limitToPartitionsWithLag bool
+	metadata                           map[string]string
+	isError                            bool
+	numBrokers                         int
+	brokers                            []string
+	group                              string
+	topic                              []string
+	partitionLimitation                []int
+	offsetResetPolicy                  offsetResetPolicy
+	allowIdleConsumers                 bool
+	excludePersistentLag               bool
+	limitToPartitionsWithLag           bool
+	ensureEvenDistributionOfPartitions bool
 }
 
 type parseApacheKafkaAuthParamsTestData struct {
@@ -66,61 +67,65 @@ var validApacheKafkaWithoutAuthParams = map[string]string{}
 
 var parseApacheKafkaMetadataTestDataset = []parseApacheKafkaMetadataTestData{
 	// failure, no consumer group
-	{map[string]string{"bootstrapServers": "foobar:9092"}, true, 1, []string{"foobar:9092"}, "", nil, nil, "latest", false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092"}, true, 1, []string{"foobar:9092"}, "", nil, nil, "latest", false, false, false, false},
 	// success, no topics
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group"}, false, 1, []string{"foobar:9092"}, "my-group", nil, nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group"}, false, 1, []string{"foobar:9092"}, "my-group", nil, nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, ignore partitionLimitation if no topics
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "partitionLimitation": "1,2,3,4,5,6"}, false, 1, []string{"foobar:9092"}, "my-group", nil, nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "partitionLimitation": "1,2,3,4,5,6"}, false, 1, []string{"foobar:9092"}, "my-group", nil, nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, no limitation with whitespaced limitation value
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "partitionLimitation": "           "}, false, 1, []string{"foobar:9092"}, "my-group", nil, nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "partitionLimitation": "           "}, false, 1, []string{"foobar:9092"}, "my-group", nil, nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, no limitation
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "partitionLimitation": ""}, false, 1, []string{"foobar:9092"}, "my-group", nil, nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "partitionLimitation": ""}, false, 1, []string{"foobar:9092"}, "my-group", nil, nil, offsetResetPolicy("latest"), false, false, false, false},
 	// failure, lagThreshold is negative value
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "lagThreshold": "-1"}, true, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "lagThreshold": "-1"}, true, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false, false},
 	// failure, lagThreshold is 0
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "lagThreshold": "0"}, true, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "lagThreshold": "0"}, true, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, LagThreshold is 1000000
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "lagThreshold": "1000000", "activationLagThreshold": "0"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "lagThreshold": "1000000", "activationLagThreshold": "0"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, activationLagThreshold is 0
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "lagThreshold": "10", "activationLagThreshold": "0"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "lagThreshold": "10", "activationLagThreshold": "0"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, partitionLimitation as list
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "partitionLimitation": "1,2,3,4"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, []int{1, 2, 3, 4}, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "partitionLimitation": "1,2,3,4"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, []int{1, 2, 3, 4}, offsetResetPolicy("latest"), false, false, false, false},
 	// success, partitionLimitation as range
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "partitionLimitation": "1-4"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, []int{1, 2, 3, 4}, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "partitionLimitation": "1-4"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, []int{1, 2, 3, 4}, offsetResetPolicy("latest"), false, false, false, false},
 	// success, partitionLimitation mixed list + ranges
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "partitionLimitation": "1-4,8,10-12"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, []int{1, 2, 3, 4, 8, 10, 11, 12}, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "partitionLimitation": "1-4,8,10-12"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, []int{1, 2, 3, 4, 8, 10, 11, 12}, offsetResetPolicy("latest"), false, false, false, false},
 	// failure, partitionLimitation wrong data type
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "partitionLimitation": "a,b,c,d"}, true, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "partitionLimitation": "a,b,c,d"}, true, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, more brokers
-	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topics"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topics"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, offsetResetPolicy policy latest
-	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topics", "offsetResetPolicy": "latest"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topics", "offsetResetPolicy": "latest"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false, false},
 	// failure, offsetResetPolicy policy wrong
-	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topics", "offsetResetPolicy": "foo"}, true, 2, []string{"foo:9092", "bar:9092"}, "my-group", []string{"my-topics"}, nil, "", false, false, false},
+	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topics", "offsetResetPolicy": "foo"}, true, 2, []string{"foo:9092", "bar:9092"}, "my-group", []string{"my-topics"}, nil, "", false, false, false, false},
 	// success, offsetResetPolicy policy earliest
-	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topics", "offsetResetPolicy": "earliest"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("earliest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topics", "offsetResetPolicy": "earliest"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("earliest"), false, false, false, false},
 	// failure, allowIdleConsumers malformed
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "allowIdleConsumers": "notvalid"}, true, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "allowIdleConsumers": "notvalid"}, true, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, allowIdleConsumers is true
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "allowIdleConsumers": "true"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), true, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "allowIdleConsumers": "true"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), true, false, false, false},
 	// failure, excludePersistentLag is malformed
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "excludePersistentLag": "notvalid"}, true, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "excludePersistentLag": "notvalid"}, true, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, excludePersistentLag is true
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "excludePersistentLag": "true"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, true, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "excludePersistentLag": "true"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, true, false, false},
 	// success, version supported
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "allowIdleConsumers": "true"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), true, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "allowIdleConsumers": "true"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), true, false, false, false},
 	// success, limitToPartitionsWithLag is true
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "limitToPartitionsWithLag": "true"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, true},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "limitToPartitionsWithLag": "true"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, true, false},
 	// failure, limitToPartitionsWithLag is malformed
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "limitToPartitionsWithLag": "notvalid"}, true, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "limitToPartitionsWithLag": "notvalid"}, true, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), false, false, false, false},
 	// failure, allowIdleConsumers and limitToPartitionsWithLag cannot be set to true simultaneously
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "allowIdleConsumers": "true", "limitToPartitionsWithLag": "true"}, true, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), true, false, true},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "allowIdleConsumers": "true", "limitToPartitionsWithLag": "true"}, true, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), true, false, true, false},
 	// success, allowIdleConsumers can be set when limitToPartitionsWithLag is false
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "allowIdleConsumers": "true", "limitToPartitionsWithLag": "false"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), true, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topics", "allowIdleConsumers": "true", "limitToPartitionsWithLag": "false"}, false, 1, []string{"foobar:9092"}, "my-group", []string{"my-topics"}, nil, offsetResetPolicy("latest"), true, false, false, false},
 	// failure, topic must be specified when limitToPartitionsWithLag is true
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "limitToPartitionsWithLag": "true"}, true, 1, []string{"foobar:9092"}, "my-group", nil, nil, offsetResetPolicy("latest"), false, false, true},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "limitToPartitionsWithLag": "true"}, true, 1, []string{"foobar:9092"}, "my-group", nil, nil, offsetResetPolicy("latest"), false, false, true, false},
+	// success, ensureEvenDistributionOfPartitions is true
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "ensureEvenDistributionOfPartitions": "true"}, false, 1, []string{"foobar:9092"}, "my-group", nil, nil, offsetResetPolicy("latest"), false, false, false, true},
+	// failure, limitToPartitionsWithLag and ensureEvenDistributionOfPartitions cannot be set to true simultaneously
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "ensureEvenDistributionOfPartitions": "true", "limitToPartitionsWithLag": "true"}, true, 1, []string{"foobar:9092"}, "my-group", nil, nil, offsetResetPolicy("latest"), false, false, true, true},
 }
 
 var parseApacheKafkaAuthParamsTestDataset = []parseApacheKafkaAuthParamsTestData{
@@ -285,6 +290,9 @@ func getBrokerApacheKafkaTestBase(t *testing.T, meta apacheKafkaMetadata, testDa
 	if err == nil && meta.LimitToPartitionsWithLag != testData.limitToPartitionsWithLag {
 		t.Errorf("Expected limitToPartitionsWithLag %t but got %t\n", testData.limitToPartitionsWithLag, meta.LimitToPartitionsWithLag)
 	}
+	if err == nil && meta.EnsureEvenDistributionOfPartitions != testData.ensureEvenDistributionOfPartitions {
+		t.Errorf("Expected ensureEvenDistributionOfPartitions %t but got %t\n", testData.ensureEvenDistributionOfPartitions, meta.EnsureEvenDistributionOfPartitions)
+	}
 
 	expectedLagThreshold, er := parseExpectedLagThreshold(testData.metadata)
 	if er != nil {
diff --git a/pkg/scalers/kafka_scaler_test.go b/pkg/scalers/kafka_scaler_test.go
index 782a99b18..b39568f86 100644
--- a/pkg/scalers/kafka_scaler_test.go
+++ b/pkg/scalers/kafka_scaler_test.go
@@ -17,17 +17,18 @@ import (
 )
 
 type parseKafkaMetadataTestData struct {
-	metadata                 map[string]string
-	isError                  bool
-	numBrokers               int
-	brokers                  []string
-	group                    string
-	topic                    string
-	partitionLimitation      []int32
-	offsetResetPolicy        offsetResetPolicy
-	allowIdleConsumers       bool
-	excludePersistentLag     bool
-	limitToPartitionsWithLag bool
+	metadata                           map[string]string
+	isError                            bool
+	numBrokers                         int
+	brokers                            []string
+	group                              string
+	topic                              string
+	partitionLimitation                []int32
+	offsetResetPolicy                  offsetResetPolicy
+	allowIdleConsumers                 bool
+	excludePersistentLag               bool
+	limitToPartitionsWithLag           bool
+	ensureEvenDistributionOfPartitions bool
 }
 
 type parseKafkaAuthParamsTestData struct {
@@ -77,67 +78,71 @@ var validWithoutAuthParams = map[string]string{}
 
 var parseKafkaMetadataTestDataset = []parseKafkaMetadataTestData{
 	// failure, no bootstrapServers
-	{map[string]string{}, true, 0, nil, "", "", nil, "", false, false, false},
+	{map[string]string{}, true, 0, nil, "", "", nil, "", false, false, false, false},
 	// failure, no consumer group
-	{map[string]string{"bootstrapServers": "foobar:9092"}, true, 1, []string{"foobar:9092"}, "", "", nil, "latest", false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092"}, true, 1, []string{"foobar:9092"}, "", "", nil, "latest", false, false, false, false},
 	// success, no topic
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group"}, false, 1, []string{"foobar:9092"}, "my-group", "", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group"}, false, 1, []string{"foobar:9092"}, "my-group", "", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, ignore partitionLimitation if no topic
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "partitionLimitation": "1,2,3,4,5,6"}, false, 1, []string{"foobar:9092"}, "my-group", "", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "partitionLimitation": "1,2,3,4,5,6"}, false, 1, []string{"foobar:9092"}, "my-group", "", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, no limitation with whitespaced limitation value
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "partitionLimitation": "           "}, false, 1, []string{"foobar:9092"}, "my-group", "", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "partitionLimitation": "           "}, false, 1, []string{"foobar:9092"}, "my-group", "", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, no limitation
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "partitionLimitation": ""}, false, 1, []string{"foobar:9092"}, "my-group", "", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "partitionLimitation": ""}, false, 1, []string{"foobar:9092"}, "my-group", "", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// failure, version not supported
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "version": "1.2.3.4"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "version": "1.2.3.4"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// failure, lagThreshold is negative value
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "lagThreshold": "-1"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "lagThreshold": "-1"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// failure, lagThreshold is 0
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "lagThreshold": "0"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "lagThreshold": "0"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, lagThreshold is 1000000
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "lagThreshold": "1000000", "activationLagThreshold": "0"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "lagThreshold": "1000000", "activationLagThreshold": "0"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// failure, activationLagThreshold is not int
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "lagThreshold": "10", "activationLagThreshold": "AA"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "lagThreshold": "10", "activationLagThreshold": "AA"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, activationLagThreshold is 0
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "lagThreshold": "10", "activationLagThreshold": "0"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "lagThreshold": "10", "activationLagThreshold": "0"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, partitionLimitation as list
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "partitionLimitation": "1,2,3,4"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", []int32{1, 2, 3, 4}, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "partitionLimitation": "1,2,3,4"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", []int32{1, 2, 3, 4}, offsetResetPolicy("latest"), false, false, false, false},
 	// success, partitionLimitation as range
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "partitionLimitation": "1-4"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", []int32{1, 2, 3, 4}, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "partitionLimitation": "1-4"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", []int32{1, 2, 3, 4}, offsetResetPolicy("latest"), false, false, false, false},
 	// success, partitionLimitation mixed list + ranges
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "partitionLimitation": "1-4,8,10-12"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", []int32{1, 2, 3, 4, 8, 10, 11, 12}, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "partitionLimitation": "1-4,8,10-12"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", []int32{1, 2, 3, 4, 8, 10, 11, 12}, offsetResetPolicy("latest"), false, false, false, false},
 	// failure, partitionLimitation wrong data type
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "partitionLimitation": "a,b,c,d"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "partitionLimitation": "a,b,c,d"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, more brokers
-	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topic"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topic"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, offsetResetPolicy policy latest
-	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topic", "offsetResetPolicy": "latest"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topic", "offsetResetPolicy": "latest"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// failure, offsetResetPolicy policy wrong
-	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topic", "offsetResetPolicy": "foo"}, true, 2, []string{"foo:9092", "bar:9092"}, "my-group", "my-topic", nil, "", false, false, false},
+	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topic", "offsetResetPolicy": "foo"}, true, 2, []string{"foo:9092", "bar:9092"}, "my-group", "my-topic", nil, "", false, false, false, false},
 	// success, offsetResetPolicy policy earliest
-	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topic", "offsetResetPolicy": "earliest"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("earliest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topic", "offsetResetPolicy": "earliest"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("earliest"), false, false, false, false},
 	// failure, allowIdleConsumers malformed
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "allowIdleConsumers": "notvalid"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "allowIdleConsumers": "notvalid"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, allowIdleConsumers is true
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "allowIdleConsumers": "true"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), true, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "allowIdleConsumers": "true"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), true, false, false, false},
 	// failure, excludePersistentLag is malformed
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "excludePersistentLag": "notvalid"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "excludePersistentLag": "notvalid"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// success, excludePersistentLag is true
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "excludePersistentLag": "true"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, true, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "excludePersistentLag": "true"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, true, false, false},
 	// success, version supported
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "allowIdleConsumers": "true", "version": "1.0.0"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), true, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "allowIdleConsumers": "true", "version": "1.0.0"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), true, false, false, false},
 	// success, limitToPartitionsWithLag is true
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "limitToPartitionsWithLag": "true"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, true},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "limitToPartitionsWithLag": "true"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, true, false},
 	// failure, limitToPartitionsWithLag is malformed
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "limitToPartitionsWithLag": "notvalid"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "limitToPartitionsWithLag": "notvalid"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false, false},
 	// failure, allowIdleConsumers and limitToPartitionsWithLag cannot be set to true simultaneously
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "allowIdleConsumers": "true", "limitToPartitionsWithLag": "true"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), true, false, true},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "allowIdleConsumers": "true", "limitToPartitionsWithLag": "true"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), true, false, true, false},
 	// success, allowIdleConsumers can be set when limitToPartitionsWithLag is false
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "allowIdleConsumers": "true", "limitToPartitionsWithLag": "false"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), true, false, false},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "allowIdleConsumers": "true", "limitToPartitionsWithLag": "false"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), true, false, false, false},
 	// failure, topic must be specified when limitToPartitionsWithLag is true
-	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "limitToPartitionsWithLag": "true"}, true, 1, []string{"foobar:9092"}, "my-group", "", nil, offsetResetPolicy("latest"), false, false, true},
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "limitToPartitionsWithLag": "true"}, true, 1, []string{"foobar:9092"}, "my-group", "", nil, offsetResetPolicy("latest"), false, false, true, false},
+	// success, ensureEvenDistributionOfPartitions is true
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "ensureEvenDistributionOfPartitions": "true"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, false, true},
+	// failure, limitToPartitionsWithLag and ensureEvenDistributionOfPartitions cannot be set to true simultaneously
+	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "ensureEvenDistributionOfPartitions": "true", "limitToPartitionsWithLag": "true"}, true, 1, []string{"foobar:9092"}, "my-group", "my-topic", nil, offsetResetPolicy("latest"), false, false, true, true},
 }
 
 var parseKafkaAuthParamsTestDataset = []parseKafkaAuthParamsTestData{
@@ -392,6 +397,9 @@ func getBrokerTestBase(t *testing.T, meta kafkaMetadata, testData parseKafkaMeta
 	if err == nil && meta.limitToPartitionsWithLag != testData.limitToPartitionsWithLag {
 		t.Errorf("Expected limitToPartitionsWithLag %t but got %t\n", testData.limitToPartitionsWithLag, meta.limitToPartitionsWithLag)
 	}
+	if err == nil && meta.ensureEvenDistributionOfPartitions != testData.ensureEvenDistributionOfPartitions {
+		t.Errorf("Expected ensureEvenDistributionOfPartitions %t but got %t\n", testData.ensureEvenDistributionOfPartitions, meta.ensureEvenDistributionOfPartitions)
+	}
 	expectedLagThreshold, er := parseExpectedLagThreshold(testData.metadata)
 	if er != nil {
 		t.Errorf("Unable to convert test data lagThreshold %s to string", testData.metadata["lagThreshold"])
@@ -654,6 +662,75 @@ func TestGetTopicPartitions(t *testing.T) {
 	}
 }
 
+func TestFindFactors(t *testing.T) {
+	factors := FindFactors(10)
+	if !reflect.DeepEqual(factors, []int64{1, 2, 5, 10}) {
+		t.Errorf("Expected factors to be %v but got %v", []int64{1, 2, 5, 10}, factors)
+	}
+
+	factors = FindFactors(13)
+	if !reflect.DeepEqual(factors, []int64{1, 13}) {
+		t.Errorf("Expected factors to be %v but got %v", []int64{1, 13}, factors)
+	}
+
+	factors = FindFactors(17)
+	if !reflect.DeepEqual(factors, []int64{1, 17}) {
+		t.Errorf("Expected factors to be %v but got %v", []int64{1, 17}, factors)
+	}
+
+	factors = FindFactors(25)
+	if !reflect.DeepEqual(factors, []int64{1, 5, 25}) {
+		t.Errorf("Expected factors to be %v but got %v", []int64{1, 5, 25}, factors)
+	}
+
+	factors = FindFactors(50)
+	if !reflect.DeepEqual(factors, []int64{1, 2, 5, 10, 25, 50}) {
+		t.Errorf("Expected factors to be %v but got %v", []int64{1, 2, 5, 10, 25, 50}, factors)
+	}
+
+	factors = FindFactors(100)
+	if !reflect.DeepEqual(factors, []int64{1, 2, 4, 5, 10, 20, 25, 50, 100}) {
+		t.Errorf("Expected factors to be %v but got %v", []int64{1, 2, 4, 5, 10, 20, 25, 50, 100}, factors)
+	}
+}
+
+func TestGetNextFactor(t *testing.T) {
+	factor := getNextFactorThatBalancesConsumersToTopicPartitions(10, 100, 10)
+	if factor != 1 {
+		t.Errorf("Expected factor to be %v but got %v", 1, factor)
+	}
+
+	factor = getNextFactorThatBalancesConsumersToTopicPartitions(20, 100, 10)
+	if factor != 2 {
+		t.Errorf("Expected factor to be %v but got %v", 2, factor)
+	}
+
+	factor = getNextFactorThatBalancesConsumersToTopicPartitions(40, 100, 10)
+	if factor != 4 {
+		t.Errorf("Expected factor to be %v but got %v", 4, factor)
+	}
+
+	factor = getNextFactorThatBalancesConsumersToTopicPartitions(50, 100, 10)
+	if factor != 5 {
+		t.Errorf("Expected factor to be %v but got %v", 5, factor)
+	}
+
+	factor = getNextFactorThatBalancesConsumersToTopicPartitions(99, 100, 10)
+	if factor != 10 {
+		t.Errorf("Expected factor to be %v but got %v", 10, factor)
+	}
+
+	factor = getNextFactorThatBalancesConsumersToTopicPartitions(100, 100, 10)
+	if factor != 10 {
+		t.Errorf("Expected factor to be %v but got %v", 10, factor)
+	}
+
+	factor = getNextFactorThatBalancesConsumersToTopicPartitions(101, 100, 10)
+	if factor != 20 {
+		t.Errorf("Expected factor to be %v but got %v", 20, factor)
+	}
+}
+
 var _ sarama.ClusterAdmin = (*MockClusterAdmin)(nil)
 
 type MockClusterAdmin struct {
diff --git a/pkg/scalers/solace_scaler_test.go b/pkg/scalers/solace_scaler_test.go
index 23e79245d..cb02b1099 100644
--- a/pkg/scalers/solace_scaler_test.go
+++ b/pkg/scalers/solace_scaler_test.go
@@ -4,8 +4,6 @@ import (
 	"context"
 	"fmt"
 	"net/http"
-	"net/url"
-	"strings"
 	"testing"
 
 	v2 "k8s.io/api/autoscaling/v2"
@@ -534,7 +532,7 @@ var testSolaceExpectedMetricNames = map[string]string{
 func TestSolaceParseSolaceMetadata(t *testing.T) {
 	for _, testData := range testParseSolaceMetadata {
 		fmt.Print(testData.testID)
-		meta, err := parseSolaceMetadata(&scalersconfig.ScalerConfig{ResolvedEnv: nil, TriggerMetadata: testData.metadata, AuthParams: nil, TriggerIndex: testData.triggerIndex})
+		_, err := parseSolaceMetadata(&scalersconfig.ScalerConfig{ResolvedEnv: nil, TriggerMetadata: testData.metadata, AuthParams: nil, TriggerIndex: testData.triggerIndex})
 		switch {
 		case err != nil && !testData.isError:
 			t.Error("expected success but got error: ", err)
@@ -545,10 +543,6 @@ func TestSolaceParseSolaceMetadata(t *testing.T) {
 		default:
 			fmt.Println(" --> PASS")
 		}
-		if !testData.isError && strings.Contains(testData.metadata["queueName"], "/") && !strings.Contains(meta.EndpointURL, url.QueryEscape(testData.metadata["queueName"])) {
-			t.Error("expected endpointURL to query escape special characters in the URL but got:", meta.EndpointURL)
-			fmt.Println(" --> FAIL")
-		}
 	}
 	for _, testData := range testSolaceEnvCreds {
 		fmt.Print(testData.testID)
diff --git a/tests/scalers/apache_kafka/apache_kafka_test.go b/tests/scalers/apache_kafka/apache_kafka_test.go
index da1083687..01cbc4615 100644
--- a/tests/scalers/apache_kafka/apache_kafka_test.go
+++ b/tests/scalers/apache_kafka/apache_kafka_test.go
@@ -24,46 +24,50 @@ const (
 )
 
 var (
-	testNamespace                  = fmt.Sprintf("%s-ns", testName)
-	deploymentName                 = fmt.Sprintf("%s-deployment", testName)
-	kafkaName                      = fmt.Sprintf("%s-kafka", testName)
-	kafkaClientName                = fmt.Sprintf("%s-client", testName)
-	scaledObjectName               = fmt.Sprintf("%s-so", testName)
-	bootstrapServer                = fmt.Sprintf("%s-kafka-bootstrap.%s:9092", kafkaName, testNamespace)
-	topic1                         = "kafka-topic"
-	topic2                         = "kafka-topic2"
-	zeroInvalidOffsetTopic         = "kafka-topic-zero-invalid-offset"
-	zeroInvalidOffsetEarliestGroup = "invalidOffsetZeroEarliest"
-	zeroInvalidOffsetLatestGroup   = "invalidOffsetZeroLatest"
-	oneInvalidOffsetTopic          = "kafka-topic-one-invalid-offset"
-	oneInvalidOffsetEarliestGroup  = "invalidOffsetOneEarliest"
-	oneInvalidOffsetLatestGroup    = "invalidOffsetOneLatest"
-	persistentLagTopic             = "kafka-topic-persistent-lag"
-	persistentLagGroup             = "persistentLag"
-	persistentLagDeploymentGroup   = "persistentLagDeploymentGroup"
-	limitToPartitionsWithLagTopic  = "limit-to-partitions-with-lag"
-	limitToPartitionsWithLagGroup  = "limitToPartitionsWithLag"
-	topicPartitions                = 3
+	testNamespace                                     = fmt.Sprintf("%s-ns", testName)
+	deploymentName                                    = fmt.Sprintf("%s-deployment", testName)
+	kafkaName                                         = fmt.Sprintf("%s-kafka", testName)
+	kafkaClientName                                   = fmt.Sprintf("%s-client", testName)
+	scaledObjectName                                  = fmt.Sprintf("%s-so", testName)
+	bootstrapServer                                   = fmt.Sprintf("%s-kafka-bootstrap.%s:9092", kafkaName, testNamespace)
+	topic1                                            = "kafka-topic"
+	topic2                                            = "kafka-topic2"
+	zeroInvalidOffsetTopic                            = "kafka-topic-zero-invalid-offset"
+	zeroInvalidOffsetEarliestGroup                    = "invalidOffsetZeroEarliest"
+	zeroInvalidOffsetLatestGroup                      = "invalidOffsetZeroLatest"
+	oneInvalidOffsetTopic                             = "kafka-topic-one-invalid-offset"
+	oneInvalidOffsetEarliestGroup                     = "invalidOffsetOneEarliest"
+	oneInvalidOffsetLatestGroup                       = "invalidOffsetOneLatest"
+	persistentLagTopic                                = "kafka-topic-persistent-lag"
+	persistentLagGroup                                = "persistentLag"
+	persistentLagDeploymentGroup                      = "persistentLagDeploymentGroup"
+	limitToPartitionsWithLagTopic                     = "limit-to-partitions-with-lag"
+	limitToPartitionsWithLagGroup                     = "limitToPartitionsWithLag"
+	ensureEvenDistributionOfPartitionsTopic           = "ensure-even-distribution-of-partitions"
+	ensureEvenDistributionOfPartitionsGroup           = "ensureEvenDistributionOfPartitions"
+	ensureEvenDistributionOfPartitionsTopicPartitions = 10
+	topicPartitions                                   = 3
 )
 
 type templateData struct {
-	TestNamespace            string
-	DeploymentName           string
-	ScaledObjectName         string
-	KafkaName                string
-	KafkaTopicName           string
-	KafkaTopicPartitions     int
-	KafkaClientName          string
-	TopicName                string
-	Topic1Name               string
-	Topic2Name               string
-	BootstrapServer          string
-	ResetPolicy              string
-	Params                   string
-	Commit                   string
-	ScaleToZeroOnInvalid     string
-	ExcludePersistentLag     string
-	LimitToPartitionsWithLag string
+	TestNamespace                      string
+	DeploymentName                     string
+	ScaledObjectName                   string
+	KafkaName                          string
+	KafkaTopicName                     string
+	KafkaTopicPartitions               int
+	KafkaClientName                    string
+	TopicName                          string
+	Topic1Name                         string
+	Topic2Name                         string
+	BootstrapServer                    string
+	ResetPolicy                        string
+	Params                             string
+	Commit                             string
+	ScaleToZeroOnInvalid               string
+	ExcludePersistentLag               string
+	LimitToPartitionsWithLag           string
+	EnsureEvenDistributionOfPartitions string
 }
 
 const (
@@ -361,6 +365,45 @@ spec:
       activationLagThreshold: '1'
       limitToPartitionsWithLag: '{{.LimitToPartitionsWithLag}}'`
 
+	ensureEvenDistributionOfPartitionsObjectTemplate = `
+apiVersion: keda.sh/v1alpha1
+kind: ScaledObject
+metadata:
+  name: {{.ScaledObjectName}}
+  namespace: {{.TestNamespace}}
+  labels:
+    app: {{.DeploymentName}}
+spec:
+  pollingInterval: 5
+  cooldownPeriod: 0
+  scaleTargetRef:
+    name: {{.DeploymentName}}
+  advanced:
+    horizontalPodAutoscalerConfig:
+      behavior:
+        scaleUp:
+          stabilizationWindowSeconds: 0
+          policies:
+          - type: Percent
+            value: 100
+            periodSeconds: 15
+        scaleDown:
+          stabilizationWindowSeconds: 0
+          policies:
+          - type: Percent
+            value: 100
+            periodSeconds: 15
+  triggers:
+  - type: apache-kafka
+    metadata:
+      topic: {{.TopicName}}
+      bootstrapServers: {{.BootstrapServer}}
+      consumerGroup:  {{.ResetPolicy}}
+      offsetResetPolicy: 'earliest'
+      lagThreshold: '1'
+      activationLagThreshold: '1'
+      ensureEvenDistributionOfPartitions: '{{.EnsureEvenDistributionOfPartitions}}'`
+
 	kafkaClusterTemplate = `apiVersion: kafka.strimzi.io/v1beta2
 kind: Kafka
 metadata:
@@ -466,6 +509,7 @@ func TestScaler(t *testing.T) {
 	addTopic(t, data, oneInvalidOffsetTopic, 1)
 	addTopic(t, data, persistentLagTopic, topicPartitions)
 	addTopic(t, data, limitToPartitionsWithLagTopic, topicPartitions)
+	addTopic(t, data, ensureEvenDistributionOfPartitionsTopic, ensureEvenDistributionOfPartitionsTopicPartitions)
 
 	// test scaling
 	testEarliestPolicy(t, kc, data)
@@ -477,6 +521,7 @@ func TestScaler(t *testing.T) {
 	testOneOnInvalidOffsetWithEarliestOffsetResetPolicy(t, kc, data)
 	testPersistentLag(t, kc, data)
 	testScalingOnlyPartitionsWithLag(t, kc, data)
+	testScalingEnsureEvenDistributionOfPartitions(t, kc, data)
 }
 
 func testEarliestPolicy(t *testing.T, kc *kubernetes.Clientset, data templateData) {
@@ -771,6 +816,53 @@ func testScalingOnlyPartitionsWithLag(t *testing.T, kc *kubernetes.Clientset, da
 	AssertReplicaCountNotChangeDuringTimePeriod(t, kc, deploymentName, testNamespace, 2, 60)
 }
 
+func testScalingEnsureEvenDistributionOfPartitions(t *testing.T, kc *kubernetes.Clientset, data templateData) {
+	t.Log("--- testing  ensureEvenDistributionOfPartitions: no scale out ---")
+
+	// Simulate Consumption from topic by consumer group
+	// To avoid edge case where scaling could be effectively disabled (Consumer never makes a commit)
+	commitPartition(t, ensureEvenDistributionOfPartitionsTopic, "latest")
+
+	data.Params = fmt.Sprintf("--topic %s --group %s", ensureEvenDistributionOfPartitionsTopic, ensureEvenDistributionOfPartitionsGroup)
+	data.Commit = StringFalse
+	data.TopicName = ensureEvenDistributionOfPartitionsTopic
+	data.EnsureEvenDistributionOfPartitions = StringTrue
+	data.ResetPolicy = "latest"
+
+	KubectlApplyWithTemplate(t, data, "singleDeploymentTemplate", singleDeploymentTemplate)
+	defer KubectlDeleteWithTemplate(t, data, "singleDeploymentTemplate", singleDeploymentTemplate)
+	KubectlApplyWithTemplate(t, data, "ensureEvenDistributionOfPartitionsObjectTemplate", ensureEvenDistributionOfPartitionsObjectTemplate)
+	defer KubectlDeleteWithTemplate(t, data, "ensureEvenDistributionOfPartitionsObjectTemplate", ensureEvenDistributionOfPartitionsObjectTemplate)
+
+	// Shouldn't scale pods applying latest policy
+	AssertReplicaCountNotChangeDuringTimePeriod(t, kc, deploymentName, testNamespace, 0, 30)
+
+	// Shouldn't scale pods with only 1 message due to activation value
+	publishMessage(t, ensureEvenDistributionOfPartitionsTopic)
+	AssertReplicaCountNotChangeDuringTimePeriod(t, kc, deploymentName, testNamespace, 0, 30)
+
+	// Publish 4 messages to increase the lag to 5
+	messages := 4
+	for i := 0; i < messages; i++ {
+		publishMessage(t, ensureEvenDistributionOfPartitionsTopic)
+	}
+
+	// Total lag is 5, we have 10 paritions we should scale to 5 pods
+	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 5, 60, 2),
+		"replica count should be %d after 2 minute", 5)
+
+	// Publish 3 messages to increase the lag to 8
+	messages = 3
+	for i := 0; i < messages; i++ {
+		publishMessage(t, ensureEvenDistributionOfPartitionsTopic)
+	}
+
+	// Total lag is 8, we have 10 paritions we should scale to 8 pods but since ensureEvenDistributionOfPartitions is true
+	// we should scale to 10 pods
+	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 10, 60, 2),
+		"replica count should be %d after 2 minute", 10)
+}
+
 func addTopic(t *testing.T, data templateData, name string, partitions int) {
 	t.Log("--- adding kafka topic" + name + " and partitions " + strconv.Itoa(partitions) + " ---")
 	data.KafkaTopicName = name
diff --git a/tests/scalers/kafka/kafka_test.go b/tests/scalers/kafka/kafka_test.go
index 899b3bfbf..77e7d8af9 100644
--- a/tests/scalers/kafka/kafka_test.go
+++ b/tests/scalers/kafka/kafka_test.go
@@ -24,46 +24,50 @@ const (
 )
 
 var (
-	testNamespace                  = fmt.Sprintf("%s-ns", testName)
-	deploymentName                 = fmt.Sprintf("%s-deployment", testName)
-	kafkaName                      = fmt.Sprintf("%s-kafka", testName)
-	kafkaClientName                = fmt.Sprintf("%s-client", testName)
-	scaledObjectName               = fmt.Sprintf("%s-so", testName)
-	bootstrapServer                = fmt.Sprintf("%s-kafka-bootstrap.%s:9092", kafkaName, testNamespace)
-	topic1                         = "kafka-topic"
-	topic2                         = "kafka-topic2"
-	zeroInvalidOffsetTopic         = "kafka-topic-zero-invalid-offset"
-	zeroInvalidOffsetEarliestGroup = "invalidOffsetZeroEarliest"
-	zeroInvalidOffsetLatestGroup   = "invalidOffsetZeroLatest"
-	oneInvalidOffsetTopic          = "kafka-topic-one-invalid-offset"
-	oneInvalidOffsetEarliestGroup  = "invalidOffsetOneEarliest"
-	oneInvalidOffsetLatestGroup    = "invalidOffsetOneLatest"
-	persistentLagTopic             = "kafka-topic-persistent-lag"
-	persistentLagGroup             = "persistentLag"
-	persistentLagDeploymentGroup   = "persistentLagDeploymentGroup"
-	limitToPartitionsWithLagTopic  = "limit-to-partitions-with-lag"
-	limitToPartitionsWithLagGroup  = "limitToPartitionsWithLag"
-	topicPartitions                = 3
+	testNamespace                                     = fmt.Sprintf("%s-ns", testName)
+	deploymentName                                    = fmt.Sprintf("%s-deployment", testName)
+	kafkaName                                         = fmt.Sprintf("%s-kafka", testName)
+	kafkaClientName                                   = fmt.Sprintf("%s-client", testName)
+	scaledObjectName                                  = fmt.Sprintf("%s-so", testName)
+	bootstrapServer                                   = fmt.Sprintf("%s-kafka-bootstrap.%s:9092", kafkaName, testNamespace)
+	topic1                                            = "kafka-topic"
+	topic2                                            = "kafka-topic2"
+	zeroInvalidOffsetTopic                            = "kafka-topic-zero-invalid-offset"
+	zeroInvalidOffsetEarliestGroup                    = "invalidOffsetZeroEarliest"
+	zeroInvalidOffsetLatestGroup                      = "invalidOffsetZeroLatest"
+	oneInvalidOffsetTopic                             = "kafka-topic-one-invalid-offset"
+	oneInvalidOffsetEarliestGroup                     = "invalidOffsetOneEarliest"
+	oneInvalidOffsetLatestGroup                       = "invalidOffsetOneLatest"
+	persistentLagTopic                                = "kafka-topic-persistent-lag"
+	persistentLagGroup                                = "persistentLag"
+	persistentLagDeploymentGroup                      = "persistentLagDeploymentGroup"
+	limitToPartitionsWithLagTopic                     = "limit-to-partitions-with-lag"
+	limitToPartitionsWithLagGroup                     = "limitToPartitionsWithLag"
+	ensureEvenDistributionOfPartitionsTopic           = "ensure-even-distribution-of-partitions"
+	ensureEvenDistributionOfPartitionsGroup           = "ensureEvenDistributionOfPartitions"
+	ensureEvenDistributionOfPartitionsTopicPartitions = 10
+	topicPartitions                                   = 3
 )
 
 type templateData struct {
-	TestNamespace            string
-	DeploymentName           string
-	ScaledObjectName         string
-	KafkaName                string
-	KafkaTopicName           string
-	KafkaTopicPartitions     int
-	KafkaClientName          string
-	TopicName                string
-	Topic1Name               string
-	Topic2Name               string
-	BootstrapServer          string
-	ResetPolicy              string
-	Params                   string
-	Commit                   string
-	ScaleToZeroOnInvalid     string
-	ExcludePersistentLag     string
-	LimitToPartitionsWithLag string
+	TestNamespace                      string
+	DeploymentName                     string
+	ScaledObjectName                   string
+	KafkaName                          string
+	KafkaTopicName                     string
+	KafkaTopicPartitions               int
+	KafkaClientName                    string
+	TopicName                          string
+	Topic1Name                         string
+	Topic2Name                         string
+	BootstrapServer                    string
+	ResetPolicy                        string
+	Params                             string
+	Commit                             string
+	ScaleToZeroOnInvalid               string
+	ExcludePersistentLag               string
+	LimitToPartitionsWithLag           string
+	EnsureEvenDistributionOfPartitions string
 }
 
 const (
@@ -360,6 +364,45 @@ spec:
       activationLagThreshold: '1'
       limitToPartitionsWithLag: '{{.LimitToPartitionsWithLag}}'`
 
+	ensureEvenDistributionOfPartitionsObjectTemplate = `
+apiVersion: keda.sh/v1alpha1
+kind: ScaledObject
+metadata:
+  name: {{.ScaledObjectName}}
+  namespace: {{.TestNamespace}}
+  labels:
+    app: {{.DeploymentName}}
+spec:
+  pollingInterval: 5
+  cooldownPeriod: 0
+  scaleTargetRef:
+    name: {{.DeploymentName}}
+  advanced:
+    horizontalPodAutoscalerConfig:
+      behavior:
+        scaleUp:
+          stabilizationWindowSeconds: 0
+          policies:
+          - type: Percent
+            value: 100
+            periodSeconds: 15
+        scaleDown:
+          stabilizationWindowSeconds: 0
+          policies:
+          - type: Percent
+            value: 100
+            periodSeconds: 15
+  triggers:
+  - type: kafka
+    metadata:
+      topic: {{.TopicName}}
+      bootstrapServers: {{.BootstrapServer}}
+      consumerGroup:  {{.ResetPolicy}}
+      offsetResetPolicy: 'earliest'
+      lagThreshold: '1'
+      activationLagThreshold: '1'
+      ensureEvenDistributionOfPartitions: '{{.EnsureEvenDistributionOfPartitions}}'`
+
 	kafkaClusterTemplate = `apiVersion: kafka.strimzi.io/v1beta2
 kind: Kafka
 metadata:
@@ -465,6 +508,7 @@ func TestScaler(t *testing.T) {
 	addTopic(t, data, oneInvalidOffsetTopic, 1)
 	addTopic(t, data, persistentLagTopic, topicPartitions)
 	addTopic(t, data, limitToPartitionsWithLagTopic, topicPartitions)
+	addTopic(t, data, ensureEvenDistributionOfPartitionsTopic, ensureEvenDistributionOfPartitionsTopicPartitions)
 
 	// test scaling
 	testEarliestPolicy(t, kc, data)
@@ -476,6 +520,7 @@ func TestScaler(t *testing.T) {
 	testOneOnInvalidOffsetWithEarliestOffsetResetPolicy(t, kc, data)
 	testPersistentLag(t, kc, data)
 	testScalingOnlyPartitionsWithLag(t, kc, data)
+	testScalingEnsureEvenDistributionOfPartitions(t, kc, data)
 }
 
 func testEarliestPolicy(t *testing.T, kc *kubernetes.Clientset, data templateData) {
@@ -769,6 +814,53 @@ func testScalingOnlyPartitionsWithLag(t *testing.T, kc *kubernetes.Clientset, da
 	AssertReplicaCountNotChangeDuringTimePeriod(t, kc, deploymentName, testNamespace, 2, 60)
 }
 
+func testScalingEnsureEvenDistributionOfPartitions(t *testing.T, kc *kubernetes.Clientset, data templateData) {
+	t.Log("--- testing  ensureEvenDistributionOfPartitions: no scale out ---")
+
+	// Simulate Consumption from topic by consumer group
+	// To avoid edge case where scaling could be effectively disabled (Consumer never makes a commit)
+	commitPartition(t, ensureEvenDistributionOfPartitionsTopic, "latest")
+
+	data.Params = fmt.Sprintf("--topic %s --group %s", ensureEvenDistributionOfPartitionsTopic, ensureEvenDistributionOfPartitionsGroup)
+	data.Commit = StringFalse
+	data.TopicName = ensureEvenDistributionOfPartitionsTopic
+	data.EnsureEvenDistributionOfPartitions = StringTrue
+	data.ResetPolicy = "latest"
+
+	KubectlApplyWithTemplate(t, data, "singleDeploymentTemplate", singleDeploymentTemplate)
+	defer KubectlDeleteWithTemplate(t, data, "singleDeploymentTemplate", singleDeploymentTemplate)
+	KubectlApplyWithTemplate(t, data, "ensureEvenDistributionOfPartitionsObjectTemplate", ensureEvenDistributionOfPartitionsObjectTemplate)
+	defer KubectlDeleteWithTemplate(t, data, "ensureEvenDistributionOfPartitionsObjectTemplate", ensureEvenDistributionOfPartitionsObjectTemplate)
+
+	// Shouldn't scale pods applying latest policy
+	AssertReplicaCountNotChangeDuringTimePeriod(t, kc, deploymentName, testNamespace, 0, 30)
+
+	// Shouldn't scale pods with only 1 message due to activation value
+	publishMessage(t, ensureEvenDistributionOfPartitionsTopic)
+	AssertReplicaCountNotChangeDuringTimePeriod(t, kc, deploymentName, testNamespace, 0, 30)
+
+	// Publish 4 messages to increase the lag to 5
+	messages := 4
+	for i := 0; i < messages; i++ {
+		publishMessage(t, ensureEvenDistributionOfPartitionsTopic)
+	}
+
+	// Total lag is 5, we have 10 paritions we should scale to 5 pods
+	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 5, 60, 2),
+		"replica count should be %d after 2 minute", 5)
+
+	// Publish 3 messages to increase the lag to 8
+	messages = 3
+	for i := 0; i < messages; i++ {
+		publishMessage(t, ensureEvenDistributionOfPartitionsTopic)
+	}
+
+	// Total lag is 8, we have 10 paritions we should scale to 8 pods but since ensureEvenDistributionOfPartitions is true
+	// we should scale to 10 pods
+	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 10, 60, 2),
+		"replica count should be %d after 2 minute", 10)
+}
+
 func addTopic(t *testing.T, data templateData, name string, partitions int) {
 	t.Log("--- adding kafka topic" + name + " and partitions " + strconv.Itoa(partitions) + " ---")
 	data.KafkaTopicName = name