Repository: kedacore/keda

Issue:
Kafka Partitions Per Pod Scaling
### Proposal

Enhance the Kafka Scaler or create a new KEDA Scaler to support scaling Kafka by partitions per pod as opposed to simple increase in pod count.

### Use-Case

Say I have 12 partitions for my topic.
At 3 pods I will have four partitions per pod:
`4 + 4 + 4 = 12`
At 4 pods I will have three partitions per pod:
`3 + 3 + 3 + 3 = 12`
If I naively scale to 5 pods, I will end up with 3 pods with 2 partitions each and 2 pods with still 3 partitions each:
`2 + 2 + 2 + 3 + 3 = 12`
To scale correctly I need to jump from 4 pods directly to 6 pods:
`2 + 2 + 2 + 2 + 2 + 2 = 12`
And instead of scaling to 7, 8, 9, 10, or 11, it should scale directly from 6 to 12.
`1 + 1 + 1 + 1 + 1 + 1 + 1 + 1 + 1 + 1 + 1 + 1 = 12`

As such the list of valid scale targets is not a simple sequence from 0 to 12. It is rather a sequence of the divisors of the partition count:
`0, 1, 2, 3, 4, 6, 12`

### Anything else?

_No response_

Discussion:
@gabrieljones: Here is a list of pod scaling sequences based on partition count:
```
  4: 1, 2, 4
  6: 1, 2, 3, 6
  8: 1, 2, 4, 8
 10: 1, 2, 5, 10
 12: 1, 2, 3, 4, 6, 12
 18: 1, 2, 3, 6, 9, 18
 20: 1, 2, 4, 5, 10, 20
 24: 1, 2, 3, 4, 6, 8, 12, 24
 30: 1, 2, 3, 5, 6, 10, 15, 30
 36: 1, 2, 3, 4, 6, 9, 12, 18, 36
 48: 1, 2, 3, 4, 6, 8, 12, 16, 24, 48
 60: 1, 2, 3, 4, 5, 6, 10, 12, 15, 20, 30, 60
 72: 1, 2, 3, 4, 6, 8, 9, 12, 18, 24, 36, 72
 84: 1, 2, 3, 4, 6, 7, 12, 14, 21, 28, 42, 84
 90: 1, 2, 3, 5, 6, 9, 10, 15, 18, 30, 45, 90
 96: 1, 2, 3, 4, 6, 8, 12, 16, 24, 32, 48, 96
108: 1, 2, 3, 4, 6, 9, 12, 18, 27, 36, 54, 108
120: 1, 2, 3, 4, 5, 6, 8, 10, 12, 15, 20, 24, 30, 40, 60, 120
168: 1, 2, 3, 4, 6, 7, 8, 12, 14, 21, 24, 28, 42, 56, 84, 168
180: 1, 2, 3, 4, 5, 6, 9, 10, 12, 15, 18, 20, 30, 36, 45, 60, 90, 180
240: 1, 2, 3, 4, 5, 6, 8, 10, 12, 15, 16, 20, 24, 30, 40, 48, 60, 80, 120, 240
336: 1, 2, 3, 4, 6, 7, 8, 12, 14, 16, 21, 24, 28, 42, 48, 56, 84, 112, 168, 336
360: 1, 2, 3, 4, 5, 6, 8, 9, 10, 12, 15, 18, 20, 24, 30, 36, 40, 45, 60, 72, 90, 120, 180, 360
```

Here is the Scala code that generated the above list:
```
def divs(x: Int): Seq[Int] = (1 to x).filter(x % _ == 0)

val res = (1 to 360).map(x => x -> divs(x))

var c = 0

val resF = for {
  i <- 4 to 360
  ds = divs(i)
  if ds.size >= c
  _ = c = ds.size
} yield i -> ds

for {
  (i, ds) <- resF
} {
  println(f"$i%3d: ${ds.mkString(", ")}")
}
```@rwkarg: This would be great for ensuring that only impactful scaling happens with Kafka consumers, especially since _all_ values between p/2 and p-1 are essentially noop for scaling (they all have at least one consumer with 2 partitions).@rwkarg: the Kafka scaler already fakes the metrics to cap the instance count at the partition count.
https://github.com/kedacore/keda/blob/28d0b69b124ab6a28c7fe736b47aa3246b4f37ee/pkg/scalers/kafka_scaler.go#L452-L457

Could similarly fake the metric value to result in the desired instance count. Not saying if that's a _good_ idea or not, but I think it would work.@zroubalik: That approach just caps the max reported metrics, not replicas. But yeah, you are right, it could be done somehow, but it would need to mimic HPA algorithm and report fake metrics, that's not ideal imho.

But happy to see some POC :)@rwkarg: It's essentially `totalLag = desiredInstances * s.metadata.lagThreshold`

The existing code do a max where `desiredInstances` == the partition count, but if we figure out that the break points are for the topic's partition count (following logic similar to @gabrieljones above), then we can compute with the naïve instance count would be (`totalLag / s.metadata.lagThreshold`), slide up to the next highest break point to get the desired instances, and the compute the fake `totalLag = desiredInstances * s.metadata.lagThreshold`@zroubalik: In theory this could work, I am curious how precise the calculation will be and if we are able to really convince the HPA to scale accordingly.

@rwkarg keen to give it a try?@stale: This issue has been automatically marked as stale because it has not had recent activity. It will be closed in 7 days if no further activity occurs. Thank you for your contributions.
@rwkarg: Haven't had any time to look at this, but it would be a nice improvement.@stale: This issue has been automatically marked as stale because it has not had recent activity. It will be closed in 7 days if no further activity occurs. Thank you for your contributions.
@stale: This issue has been automatically closed due to inactivity.
@mwarkentin: Generally this has been how we've scaled our Kafka consumers when managing replica count manually, so some way to do so with autoscaling would be great as well to help guarantee that we have an even distribution of work (assuming relatively even distribution of messages across partitions) for our consumers.@zroubalik: @mwarkentin are you willing to contribute this? :)@stale: This issue has been automatically marked as stale because it has not had recent activity. It will be closed in 7 days if no further activity occurs. Thank you for your contributions.
@mwarkentin: Looks like stalebot is ignoring the `stale-bot-ignore` label?@gabrieljones: I need to revise my sequence algorithm. It is only considering scale steps where all pods get the same number of partitions.

For example at 8 total partitions scaling from 2 to 3 is impactful as you go from 4,4 to 3,3,2.@zroubalik: @gabrieljones ack, are you willing to contribute this once you are okay with the algorithm?@patelvp: @zroubalik, I would like to take it on. Would be a nice starter project for me.@zroubalik: @patelvp great!@patelvp: @zroubalik I have most of the pieces here https://github.com/kedacore/keda/pull/6343.
I will complete the PR description and tests once I have direction. The only question I have is that I would need to know the current number of replicas for the HPA [here](https://github.com/kedacore/keda/blob/210c5f49083963b2c44fa5a49af603d5c64bff74/pkg/scalers/kafka_scaler.go#L984). This would essentially help determine what the next step should be. Not sure if there is a clean way to know the HPA replica count in the scaler code. The way I understand the architecture is that the scalers only provide metrics to the HPA to scale.@gabrieljones: Not sure if this helps. I have been working on a multi k8s cluster aware direct autoscaler. It does not use metrics and it bypasses the hpa logic altogether. It uses the hpa merely to adjust the minReplicas. It directly adjusts the `hpa.spec.minReplicas` on the hpas for the given deployment in each cluster via `io.fabric8.kubernetes.client.KubernetesClient`. It evenly distributes the desired consumer count across the multiple k8s clusters. It talks directly to the apiservers to get and set the current minReplicas on each hpa. It also talks directly to the kafka cluster via `org.apache.kafka.clients.admin.AdminClient` to get the latest offsets and consumer group offsets. It also gets the current consumer group membership and compares that with the minReplicas set on the hpa.@patelvp: > Not sure if this helps. I have been working on a multi k8s cluster aware direct autoscaler. It does not use metrics and it bypasses the hpa logic altogether. It uses the hpa merely to adjust the minReplicas. It directly adjusts the `hpa.spec.minReplicas` on the hpas for the given deployment in each cluster via `io.fabric8.kubernetes.client.KubernetesClient`. It evenly distributes the desired consumer count across the multiple k8s clusters. It talks directly to the apiservers to get and set the current minReplicas on each hpa. It also talks directly to the kafka cluster via `org.apache.kafka.clients.admin.AdminClient` to get the latest offsets and consumer group offsets. It also gets the current consumer group membership and compares that with the minReplicas set on the hpa.

@gabrieljones That is a great idea that could work. I can extend that idea to just pull the current replica count. Because that is what I want to make this work more natively.
I just want to make sure that would be an acceptable solution before investing time in working on it. I'm not sure what the maintainers think about this approach. @zroubalik: TBH I am not completely sure we should pull the actual replica count - there might be multiple scalers defined so the final replica count might be affected by another scaler. We should imho do the calculation similar way like the fallback feature.

@gabrieljones this is interesting, is the scaler public? Can we check it out?@patelvp: > TBH I am not completely sure we should pull the actual replica count - there might be multiple scalers defined so the final replica count might be affected by another scaler. We should imho do the calculation similar way like the fallback feature.
> 
> [@gabrieljones](https://github.com/gabrieljones) this is interesting, is the scaler public? Can we check it out?


@zroubalik 
Multiple scalers defined is an interesting configuration that I hadn't thought about. Shouldn't those scalers also scale based on replica counts if a Kafka scaler is present? (For example, if the average CPU is higher than the threshold, we don't just want to add one pod; we want to add pods based on the partition count) If I understand the code base correctly, they work independently of the Kafka scalers.

Also when you say the calculation similar to the way the fallback feature is implemented do you mean how it's done here https://github.com/kedacore/keda/blob/d76aa177485cf21f4d5589e4d19da22de7296439/pkg/fallback/fallback.go#L109-L115@zroubalik: > > TBH I am not completely sure we should pull the actual replica count - there might be multiple scalers defined so the final replica count might be affected by another scaler. We should imho do the calculation similar way like the fallback feature.
> > [@gabrieljones](https://github.com/gabrieljones) this is interesting, is the scaler public? Can we check it out?
> 
> [@zroubalik](https://github.com/zroubalik) Multiple scalers defined is an interesting configuration that I hadn't thought about. Shouldn't those scalers also scale based on replica counts if a Kafka scaler is present? (For example, if the average CPU is higher than the threshold, we don't just want to add one pod; we want to add pods based on the partition count) If I understand the code base correctly, they work independently of the Kafka scalers.
> 
> Also when you say the calculation similar to the way the fallback feature is implemented do you mean how it's done here
> 
> [keda/pkg/fallback/fallback.go](https://github.com/kedacore/keda/blob/d76aa177485cf21f4d5589e4d19da22de7296439/pkg/fallback/fallback.go#L109-L115)
> 
> Lines 109 to 115 in [d76aa17](/kedacore/keda/commit/d76aa177485cf21f4d5589e4d19da22de7296439)
>  if !scaledObject.IsUsingModifiers() { 
>  	normalisationValue = int64(metricSpec.External.Target.AverageValue.AsApproximateFloat64()) 
>  } else { 
>  	value, _ := strconv.ParseInt(scaledObject.Spec.Advanced.ScalingModifiers.Target, 10, 64) 
>  	normalisationValue = value 
>  	metricName = kedav1alpha1.CompositeMetricName 
>  }


The flow is as follows - each trigger sends the metric to the HPA and it does the final decision on the replica count. The fallback is try to calculate the acutal replica number to be used by trying to follow the HPA algorithm@patelvp: @zroubalik https://github.com/kedacore/keda/pull/6558 Should be ready for a review

Solution patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index daa7193f5..c69304e40 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -60,7 +60,6 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 
 ## Unreleased
 
-
 ### New
 
 - **General**: Add error and event for mismatching input property ([#6721](https://github.com/kedacore/keda/issues/6721))
@@ -86,14 +85,17 @@ 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))
 - **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/rabbitmq_scaler.go b/pkg/scalers/rabbitmq_scaler.go
index b9a5f5c75..00629b852 100644
--- a/pkg/scalers/rabbitmq_scaler.go
+++ b/pkg/scalers/rabbitmq_scaler.go
@@ -9,6 +9,7 @@ import (
 	"net/url"
 	"path"
 	"regexp"
+	"slices"
 	"strings"
 	"time"
 
@@ -37,6 +38,9 @@ const (
 	rabbitModeUnknown                      = "Unknown"
 	rabbitModeQueueLength                  = "QueueLength"
 	rabbitModeMessageRate                  = "MessageRate"
+	rabbitModeDeliverGetRate               = "DeliverGetRate"
+	rabbitModePublishedToDeliveredRatio    = "PublishedToDeliveredRatio"
+	rabbitModeExpectedQueueConsumptionTime = "ExpectedQueueConsumptionTime"
 	defaultRabbitMQQueueLength             = 20
 	rabbitMetricType                       = "External"
 	rabbitRootVhostPath                    = "/%2F"
@@ -123,17 +127,17 @@ func (r *rabbitMQMetadata) Validate() error {
 	}
 
 	if r.EnableTLS != rmqTLSEnable && r.EnableTLS != rmqTLSDisable {
-		return fmt.Errorf("err incorrect value for TLS given: %s", r.EnableTLS)
+		return fmt.Errorf("incorrect value for TLS given: %s", r.EnableTLS)
 	}
 
 	certGiven := r.Cert != ""
 	keyGiven := r.Key != ""
 	if certGiven != keyGiven {
-		return fmt.Errorf("both key and cert must be provided")
+		return fmt.Errorf("both key and certificate must be provided")
 	}
 
 	if r.PageSize < 1 {
-		return fmt.Errorf("pageSize should be 1 or greater than 1")
+		return fmt.Errorf("pageSize should be 1 or greater")
 	}
 
 	if (r.Username != "" || r.Password != "") && (r.Username == "" || r.Password == "") {
@@ -152,20 +156,20 @@ func (r *rabbitMQMetadata) Validate() error {
 		case "http", "https":
 			r.Protocol = httpProtocol
 		default:
-			return fmt.Errorf("unknown host URL scheme `%s`", parsedURL.Scheme)
+			return fmt.Errorf("unknown host URL scheme: `%s`", parsedURL.Scheme)
 		}
 	}
 
 	if r.Protocol == amqpProtocol && r.WorkloadIdentityResource != "" {
-		return fmt.Errorf("workload identity is not supported for amqp protocol currently")
+		return fmt.Errorf("workload identity is not supported for the AMQP protocol at the moment")
 	}
 
 	if r.UseRegex && r.Protocol != httpProtocol {
-		return fmt.Errorf("configure only useRegex with http protocol")
+		return fmt.Errorf("configure useRegex=true only with HTTP protocol")
 	}
 
 	if r.ExcludeUnacknowledged && r.Protocol != httpProtocol {
-		return fmt.Errorf("configure excludeUnacknowledged=true with http protocol only")
+		return fmt.Errorf("configure excludeUnacknowledged=true only with HTTP protocol")
 	}
 
 	if err := r.validateTrigger(); err != nil {
@@ -176,6 +180,22 @@ func (r *rabbitMQMetadata) Validate() error {
 }
 
 func (r *rabbitMQMetadata) validateTrigger() error {
+	modes := map[string][]string{
+		"all": {
+			rabbitModeQueueLength,
+			rabbitModeMessageRate,
+			rabbitModeDeliverGetRate,
+			rabbitModePublishedToDeliveredRatio,
+			rabbitModeExpectedQueueConsumptionTime,
+		},
+		"httpOnly": {
+			rabbitModeMessageRate,
+			rabbitModeDeliverGetRate,
+			rabbitModePublishedToDeliveredRatio,
+			rabbitModeExpectedQueueConsumptionTime,
+		},
+	}
+
 	// If nothing is specified for the trigger then return the default
 	if r.QueueLength == 0 && r.Mode == rabbitModeUnknown && r.Value == 0 {
 		r.Mode = rabbitModeQueueLength
@@ -184,7 +204,7 @@ func (r *rabbitMQMetadata) validateTrigger() error {
 	}
 
 	if r.QueueLength != 0 && (r.Mode != rabbitModeUnknown || r.Value != 0) {
-		return fmt.Errorf("queueLength is deprecated; configure only %s and %s", rabbitModeTriggerConfigName, rabbitValueTriggerConfigName)
+		return fmt.Errorf("queueLength is deprecated; use %s and %s", rabbitModeTriggerConfigName, rabbitValueTriggerConfigName)
 	}
 
 	if r.QueueLength != 0 {
@@ -202,16 +222,21 @@ func (r *rabbitMQMetadata) validateTrigger() error {
 		return fmt.Errorf("%s must be specified", rabbitValueTriggerConfigName)
 	}
 
-	if r.Mode != rabbitModeQueueLength && r.Mode != rabbitModeMessageRate {
-		return fmt.Errorf("trigger mode %s must be one of %s, %s", r.Mode, rabbitModeQueueLength, rabbitModeMessageRate)
+	if !slices.Contains(modes["all"], r.Mode) {
+		return fmt.Errorf("trigger mode %s must be one of %s, %s, %s, %s or %s", r.Mode,
+			rabbitModeQueueLength,
+			rabbitModeMessageRate,
+			rabbitModeDeliverGetRate,
+			rabbitModePublishedToDeliveredRatio,
+			rabbitModeExpectedQueueConsumptionTime)
 	}
 
-	if r.Mode == rabbitModeMessageRate && r.Protocol != httpProtocol {
-		return fmt.Errorf("protocol %s not supported; must be http to use mode %s", r.Protocol, rabbitModeMessageRate)
+	if slices.Contains(modes["httpOnly"], r.Mode) && r.Protocol != httpProtocol {
+		return fmt.Errorf("protocol %s not supported; must be HTTP to use trigger mode %s", r.Protocol, r.Mode)
 	}
 
 	if r.Protocol == amqpProtocol && r.Timeout != 0 {
-		return fmt.Errorf("amqp protocol doesn't support custom timeouts: %d", r.Timeout)
+		return fmt.Errorf("AMQP protocol doesn't support custom timeouts: %d", r.Timeout)
 	}
 
 	return nil
@@ -231,13 +256,18 @@ type regexQueueInfo struct {
 }
 
 type messageStat struct {
-	PublishDetail publishDetail `json:"publish_details"`
+	PublishDetail    publishDetail    `json:"publish_details"`
+	DeliverGetDetail deliverGetDetail `json:"deliver_get_details"`
 }
 
 type publishDetail struct {
 	Rate float64 `json:"rate"`
 }
 
+type deliverGetDetail struct {
+	Rate float64 `json:"rate"`
+}
+
 // NewRabbitMQScaler creates a new rabbitMQ scaler
 func NewRabbitMQScaler(config *scalersconfig.ScalerConfig) (Scaler, error) {
 	s := &rabbitMQScaler{}
@@ -252,7 +282,7 @@ func NewRabbitMQScaler(config *scalersconfig.ScalerConfig) (Scaler, error) {
 
 	meta, err := parseRabbitMQMetadata(config)
 	if err != nil {
-		return nil, fmt.Errorf("error parsing rabbitmq metadata: %w", err)
+		return nil, fmt.Errorf("error parsing RabbitMQ metadata: %w", err)
 	}
 
 	s.metadata = meta
@@ -277,7 +307,7 @@ func NewRabbitMQScaler(config *scalersconfig.ScalerConfig) (Scaler, error) {
 		if meta.VhostName != "" || (meta.Username != "" && meta.Password != "") {
 			hostURI, err := amqp.ParseURI(host)
 			if err != nil {
-				return nil, fmt.Errorf("error parsing rabbitmq connection string: %w", err)
+				return nil, fmt.Errorf("error parsing RabbitMQ connection string: %w", err)
 			}
 			if meta.VhostName != "" {
 				hostURI.Vhost = meta.VhostName
@@ -293,7 +323,7 @@ func NewRabbitMQScaler(config *scalersconfig.ScalerConfig) (Scaler, error) {
 
 		conn, ch, err := getConnectionAndChannel(host, meta)
 		if err != nil {
-			return nil, fmt.Errorf("error establishing rabbitmq connection: %w", err)
+			return nil, fmt.Errorf("error establishing connection to RabbitMQ: %w", err)
 		}
 		s.connection = conn
 		s.channel = ch
@@ -308,7 +338,7 @@ func parseRabbitMQMetadata(config *scalersconfig.ScalerConfig) (*rabbitMQMetadat
 	}
 
 	if err := config.TypedConfig(meta); err != nil {
-		return nil, fmt.Errorf("error parsing rabbitmq metadata: %w", err)
+		return nil, fmt.Errorf("error parsing RabbitMQ metadata: %w", err)
 	}
 
 	if config.PodIdentity.Provider == v1alpha1.PodIdentityProviderAzureWorkload {
@@ -360,7 +390,7 @@ func (s *rabbitMQScaler) Close(context.Context) error {
 	if s.connection != nil {
 		err := s.connection.Close()
 		if err != nil {
-			s.logger.Error(err, "Error closing rabbitmq connection")
+			s.logger.Error(err, "Error closing RabbitMQ connection")
 			return err
 		}
 	}
@@ -370,28 +400,28 @@ func (s *rabbitMQScaler) Close(context.Context) error {
 	return nil
 }
 
-func (s *rabbitMQScaler) getQueueStatus(ctx context.Context) (int64, float64, error) {
+func (s *rabbitMQScaler) getQueueStatus(ctx context.Context) (int64, float64, float64, error) {
 	if s.metadata.Protocol == httpProtocol {
 		info, err := s.getQueueInfoViaHTTP(ctx)
 		if err != nil {
-			return -1, -1, err
+			return -1, -1, -1, err
 		}
 
 		if s.metadata.ExcludeUnacknowledged {
 			// messages count includes only ready
-			return int64(info.MessagesReady), info.MessageStat.PublishDetail.Rate, nil
+			return int64(info.MessagesReady), info.MessageStat.PublishDetail.Rate, info.MessageStat.DeliverGetDetail.Rate, nil
 		}
 		// messages count includes count of ready and unack-ed
-		return int64(info.Messages), info.MessageStat.PublishDetail.Rate, nil
+		return int64(info.Messages), info.MessageStat.PublishDetail.Rate, info.MessageStat.DeliverGetDetail.Rate, nil
 	}
 
 	// QueueDeclarePassive assumes that the queue exists and fails if it doesn't
 	items, err := s.channel.QueueDeclarePassive(s.metadata.QueueName, false, false, false, false, amqp.Table{})
 	if err != nil {
-		return -1, -1, err
+		return -1, -1, -1, err
 	}
 
-	return int64(items.Messages), 0, nil
+	return int64(items.Messages), 0, 0, nil
 }
 
 func getJSON(ctx context.Context, s *rabbitMQScaler, url string) (queueInfo, error) {
@@ -441,7 +471,7 @@ func getJSON(ctx context.Context, s *rabbitMQScaler, url string) (queueInfo, err
 	}
 
 	body, _ := io.ReadAll(r.Body)
-	return result, fmt.Errorf("error requesting rabbitMQ API status: %s, response: %s, from: %s", r.Status, body, url)
+	return result, fmt.Errorf("error requesting RabbitMQ API status: %s, response: %s, from: %s", r.Status, body, url)
 }
 
 func getVhostAndPathFromURL(rawPath, vhostName string) (resolvedVhostPath, resolvedPath string) {
@@ -515,19 +545,42 @@ func (s *rabbitMQScaler) GetMetricSpecForScaling(context.Context) []v2.MetricSpe
 
 // GetMetricsAndActivity returns value for a supported metric and an error if there is a problem getting the metric
 func (s *rabbitMQScaler) GetMetricsAndActivity(ctx context.Context, metricName string) ([]external_metrics.ExternalMetricValue, bool, error) {
-	messages, publishRate, err := s.getQueueStatus(ctx)
+	messages, publishRate, deliverGetRate, err := s.getQueueStatus(ctx)
 	if err != nil {
 		return []external_metrics.ExternalMetricValue{}, false, s.anonymizeRabbitMQError(err)
 	}
 
 	var metric external_metrics.ExternalMetricValue
 	var isActive bool
-	if s.metadata.Mode == rabbitModeQueueLength {
+
+	switch s.metadata.Mode {
+	case rabbitModeQueueLength:
 		metric = GenerateMetricInMili(metricName, float64(messages))
 		isActive = float64(messages) > s.metadata.ActivationValue
-	} else {
+	case rabbitModeMessageRate:
 		metric = GenerateMetricInMili(metricName, publishRate)
-		isActive = publishRate > s.metadata.ActivationValue || float64(messages) > s.metadata.ActivationValue
+		isActive = (publishRate > s.metadata.ActivationValue) || (float64(messages) > s.metadata.ActivationValue)
+	case rabbitModeDeliverGetRate:
+		metric = GenerateMetricInMili(metricName, deliverGetRate)
+		isActive = deliverGetRate > s.metadata.ActivationValue
+	case rabbitModePublishedToDeliveredRatio:
+		ratio := float64(0)
+		if (publishRate > 0) && (deliverGetRate == 0) {
+			ratio = float64(s.metadata.ActivationValue)
+		} else if (publishRate > 0) && (deliverGetRate > 0) {
+			ratio = float64(publishRate / deliverGetRate)
+		}
+		metric = GenerateMetricInMili(metricName, ratio)
+		isActive = (ratio > s.metadata.ActivationValue) || ((publishRate > 0) && (deliverGetRate == 0))
+	case rabbitModeExpectedQueueConsumptionTime:
+		eta := float64(0)
+		if deliverGetRate == 0 {
+			eta = float64(s.metadata.ActivationValue)
+		} else {
+			eta = ((publishRate - deliverGetRate) / deliverGetRate) + (float64(messages) / deliverGetRate)
+		}
+		metric = GenerateMetricInMili(metricName, eta)
+		isActive = (eta > s.metadata.ActivationValue) || (deliverGetRate == 0)
 	}
 
 	return []external_metrics.ExternalMetricValue{metric}, isActive, nil
@@ -535,58 +588,71 @@ func (s *rabbitMQScaler) GetMetricsAndActivity(ctx context.Context, metricName s
 
 func getComposedQueue(s *rabbitMQScaler, q []queueInfo) (queueInfo, error) {
 	var queue = queueInfo{}
+
 	queue.Name = "composed-queue"
 	queue.MessagesUnacknowledged = 0
+
 	if len(q) > 0 {
 		switch s.metadata.Operation {
 		case sumOperation:
-			sumMessages, sumReady, sumRate := getSum(q)
+			sumMessages, sumReady, sumPublishRate, sumDeliverGetRate := getSum(q)
 			queue.Messages = sumMessages
 			queue.MessagesReady = sumReady
-			queue.MessageStat.PublishDetail.Rate = sumRate
+			queue.MessageStat.PublishDetail.Rate = sumPublishRate
+			queue.MessageStat.DeliverGetDetail.Rate = sumDeliverGetRate
 		case avgOperation:
-			avgMessages, avgReady, avgRate := getAverage(q)
+			avgMessages, avgReady, avgPublishRate, avgDeliverGetRate := getAverage(q)
 			queue.Messages = avgMessages
 			queue.MessagesReady = avgReady
-			queue.MessageStat.PublishDetail.Rate = avgRate
+			queue.MessageStat.PublishDetail.Rate = avgPublishRate
+			queue.MessageStat.DeliverGetDetail.Rate = avgDeliverGetRate
 		case maxOperation:
-			maxMessages, maxReady, maxRate := getMaximum(q)
+			maxMessages, maxReady, maxPublishRate, maxDeliverGetRate := getMaximum(q)
 			queue.Messages = maxMessages
 			queue.MessagesReady = maxReady
-			queue.MessageStat.PublishDetail.Rate = maxRate
+			queue.MessageStat.PublishDetail.Rate = maxPublishRate
+			queue.MessageStat.DeliverGetDetail.Rate = maxDeliverGetRate
 		default:
-			return queue, fmt.Errorf("operation mode %s must be one of %s, %s, %s", s.metadata.Operation, sumOperation, avgOperation, maxOperation)
+			return queue, fmt.Errorf("operation mode %s must be one of %s, %s, %s", s.metadata.Operation,
+				sumOperation, avgOperation, maxOperation)
 		}
 	} else {
 		queue.Messages = 0
+		queue.MessagesReady = 0
 		queue.MessageStat.PublishDetail.Rate = 0
+		queue.MessageStat.DeliverGetDetail.Rate = 0
 	}
 
 	return queue, nil
 }
 
-func getSum(q []queueInfo) (int, int, float64) {
+func getSum(q []queueInfo) (int, int, float64, float64) {
 	var sumMessages int
 	var sumMessagesReady int
-	var sumRate float64
+	var sumPublishRate, sumDeliverGetRate float64
+
 	for _, value := range q {
 		sumMessages += value.Messages
 		sumMessagesReady += value.MessagesReady
-		sumRate += value.MessageStat.PublishDetail.Rate
+		sumPublishRate += value.MessageStat.PublishDetail.Rate
+		sumDeliverGetRate += value.MessageStat.DeliverGetDetail.Rate
 	}
-	return sumMessages, sumMessagesReady, sumRate
+
+	return sumMessages, sumMessagesReady, sumPublishRate, sumDeliverGetRate
 }
 
-func getAverage(q []queueInfo) (int, int, float64) {
-	sumMessages, sumReady, sumRate := getSum(q)
+func getAverage(q []queueInfo) (int, int, float64, float64) {
+	sumMessages, sumReady, sumPublishRate, sumDeliverGetRate := getSum(q)
 	length := len(q)
-	return sumMessages / length, sumReady / length, sumRate / float64(length)
+
+	return sumMessages / length, sumReady / length, sumPublishRate / float64(length), sumDeliverGetRate / float64(length)
 }
 
-func getMaximum(q []queueInfo) (int, int, float64) {
+func getMaximum(q []queueInfo) (int, int, float64, float64) {
 	var maxMessages int
 	var maxReady int
-	var maxRate float64
+	var maxPublishRate, maxDeliverGetRate float64
+
 	for _, value := range q {
 		if value.Messages > maxMessages {
 			maxMessages = value.Messages
@@ -594,16 +660,20 @@ func getMaximum(q []queueInfo) (int, int, float64) {
 		if value.MessagesReady > maxReady {
 			maxReady = value.MessagesReady
 		}
-		if value.MessageStat.PublishDetail.Rate > maxRate {
-			maxRate = value.MessageStat.PublishDetail.Rate
+		if value.MessageStat.PublishDetail.Rate > maxPublishRate {
+			maxPublishRate = value.MessageStat.PublishDetail.Rate
+		}
+		if value.MessageStat.DeliverGetDetail.Rate > maxDeliverGetRate {
+			maxDeliverGetRate = value.MessageStat.DeliverGetDetail.Rate
 		}
 	}
-	return maxMessages, maxReady, maxRate
+
+	return maxMessages, maxReady, maxPublishRate, maxDeliverGetRate
 }
 
 // Mask host for log purposes
 func (s *rabbitMQScaler) anonymizeRabbitMQError(err error) error {
-	errorMessage := fmt.Sprintf("error inspecting rabbitMQ: %s", err)
+	errorMessage := fmt.Sprintf("error inspecting RabbitMQ: %s", err)
 	return fmt.Errorf("%s", rabbitMQAnonymizePattern.ReplaceAllString(errorMessage, "user:password@"))
 }
 
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/rabbitmq_scaler_test.go b/pkg/scalers/rabbitmq_scaler_test.go
index 613db0f21..32a49de0c 100644
--- a/pkg/scalers/rabbitmq_scaler_test.go
+++ b/pkg/scalers/rabbitmq_scaler_test.go
@@ -105,12 +105,22 @@ var testRabbitMQMetadata = []parseRabbitMQMetadataTestData{
 	{map[string]string{"mode": "MessageRate", "value": "1000", "queueName": "sample", "host": "http://"}, false, map[string]string{}},
 	// message rate amqp
 	{map[string]string{"mode": "MessageRate", "value": "1000", "queueName": "sample", "host": "https://"}, false, map[string]string{}},
+	// deliverGet rate amqps
+	{map[string]string{"mode": "DeliverGetRate", "value": "1000", "queueName": "sample", "host": "amqps://"}, true, map[string]string{}},
+	// deliverGet rate amqp
+	{map[string]string{"mode": "DeliverGetRate", "value": "1000", "queueName": "sample", "host": "amqp://"}, true, map[string]string{}},
+	// deliverGet rate http
+	{map[string]string{"mode": "DeliverGetRate", "value": "1000", "queueName": "sample", "host": "http://"}, false, map[string]string{}},
+	// deliverGet rate https
+	{map[string]string{"mode": "DeliverGetRate", "value": "1000", "queueName": "sample", "host": "https://"}, false, map[string]string{}},
 	// amqp host and useRegex
 	{map[string]string{"queueName": "sample", "host": "amqps://", "useRegex": "true"}, true, map[string]string{}},
 	// http host and useRegex
 	{map[string]string{"queueName": "sample", "host": "http://", "useRegex": "true"}, false, map[string]string{}},
 	// message rate and useRegex
 	{map[string]string{"mode": "MessageRate", "value": "1000", "queueName": "sample", "host": "http://", "useRegex": "true"}, false, map[string]string{}},
+	// deliverGet rate and useRegex
+	{map[string]string{"mode": "DeliverGetRate", "value": "1000", "queueName": "sample", "host": "http://", "useRegex": "true"}, false, map[string]string{}},
 	// queue length and useRegex
 	{map[string]string{"mode": "QueueLength", "value": "1000", "queueName": "sample", "host": "http://", "useRegex": "true"}, false, map[string]string{}},
 	// http valid timeout
@@ -127,6 +137,12 @@ var testRabbitMQMetadata = []parseRabbitMQMetadataTestData{
 	{map[string]string{"mode": "MessageRate", "value": "1000", "queueName": "sample", "host": "http://", "useRegex": "true", "pageSize": "-1"}, true, map[string]string{}},
 	// invalid pageSize
 	{map[string]string{"mode": "MessageRate", "value": "1000", "queueName": "sample", "host": "http://", "useRegex": "true", "pageSize": "a"}, true, map[string]string{}},
+	// valid pageSize
+	{map[string]string{"mode": "DeliverGetRate", "value": "1000", "queueName": "sample", "host": "http://", "useRegex": "true", "pageSize": "100"}, false, map[string]string{}},
+	// pageSize less than 1
+	{map[string]string{"mode": "DeliverGetRate", "value": "1000", "queueName": "sample", "host": "http://", "useRegex": "true", "pageSize": "-1"}, true, map[string]string{}},
+	// invalid pageSize
+	{map[string]string{"mode": "DeliverGetRate", "value": "1000", "queueName": "sample", "host": "http://", "useRegex": "true", "pageSize": "a"}, true, map[string]string{}},
 	// activationValue passed
 	{map[string]string{"activationValue": "10", "queueLength": "20", "queueName": "sample", "hostFromEnv": host}, false, map[string]string{}},
 	// malformed activationValue
@@ -269,43 +285,53 @@ type getQueueInfoTestData struct {
 
 var testQueueInfoTestData = []getQueueInfoTestData{
 	// queueLength
-	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10"}},
-	{response: `{"messages": 1, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10"}},
-	{response: `{"messages": 1, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10"}},
-	{response: `{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10"}},
-	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10"}},
-	{response: `{"messages": 1, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10"}},
-	{response: `{"messages": 1, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 1.4}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10"}},
-	{response: `{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 1.4}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10"}},
+	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10"}},
+	{response: `{"messages": 1, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10"}},
+	{response: `{"messages": 1, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10"}},
+	{response: `{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10"}},
+	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}, "deliver_get_details": {"rate": 0.8}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10"}},
+	{response: `{"messages": 1, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}, "deliver_get_details": {"rate": 1.2}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10"}},
+	{response: `{"messages": 1, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 1.4}, "deliver_get_details": {"rate": 1.6}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10"}},
+	{response: `{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 1.4}, "deliver_get_details": {"rate": 2}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10"}},
 	// mode QueueLength
-	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
-	{response: `{"messages": 1, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
-	{response: `{"messages": 1, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
-	{response: `{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
-	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
-	{response: `{"messages": 1, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
-	{response: `{"messages": 1, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 1.4}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
-	{response: `{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 1.4}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
+	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 1.3}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
+	{response: `{"messages": 1, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 1}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
+	{response: `{"messages": 1, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 1}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
+	{response: `{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
+	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}, "deliver_get_details": {"rate": 1.3}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
+	{response: `{"messages": 1, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}, "deliver_get_details": {"rate": 1}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
+	{response: `{"messages": 1, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 1.4}, "deliver_get_details": {"rate": 1}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
+	{response: `{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 1.4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"value": "100", "mode": "QueueLength"}},
 	// mode MessageRate
-	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
-	{response: `{"messages": 1, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
-	{response: `{"messages": 1, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
-	{response: `{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
-	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
-	{response: `{"messages": 1, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
-	{response: `{"messages": 1, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 1.4}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
-	{response: `{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 1.4}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
+	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 1.3}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
+	{response: `{"messages": 1, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 1}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
+	{response: `{"messages": 1, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 1}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
+	{response: `{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
+	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}, "deliver_get_details": {"rate": 1.3}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
+	{response: `{"messages": 1, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}, "deliver_get_details": {"rate": 1}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
+	{response: `{"messages": 1, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 1.4}, "deliver_get_details": {"rate": 1}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
+	{response: `{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 1.4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "MessageRate"}},
+	// mode DeliverGetRate
+	{response: `{"messages": 30, "messages_unacknowledged": 10, "message_stats": {"publish_details": {"rate": 5}, "deliver_get_details": {"rate": 1}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "DeliverGetRate"}},
+	{response: `{"messages": 20, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 10}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "DeliverGetRate"}},
+	{response: `{"messages": 10, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 12.5}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "DeliverGetRate"}},
+	{response: `{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: false, extraMetadata: map[string]string{"value": "100", "mode": "DeliverGetRate"}},
+	{response: `{"messages": 45, "messages_unacknowledged": 5, "message_stats": {"publish_details": {"rate": 5.5}, "deliver_get_details": {"rate": 1.5}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "DeliverGetRate"}},
+	{response: `{"messages": 22, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 1.5}, "deliver_get_details": {"rate": 23}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "DeliverGetRate"}},
+	{response: `{"messages": 3, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 18.5}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"value": "100", "mode": "DeliverGetRate"}},
+	{response: `{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: false, extraMetadata: map[string]string{"value": "100", "mode": "DeliverGetRate"}},
+
 	// error response
 	{response: `Password is incorrect`, responseStatus: http.StatusUnauthorized},
 }
 
 var testQueueInfoTestDataSingleVhost = []getQueueInfoTestData{
-	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"hostFromEnv": "plainHost", "vhostName": "myhost"}, vhostPath: "/myhost"},
-	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"hostFromEnv": "plainHost", "vhostName": "/"}, vhostPath: "//"},
-	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"hostFromEnv": "plainHost", "vhostName": ""}},
-	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"hostFromEnv": "plainHost", "vhostName": "myhost"}, vhostPath: "/myhost"},
-	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"hostFromEnv": "plainHost", "vhostName": "/"}, vhostPath: rabbitRootVhostPath},
-	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"hostFromEnv": "plainHost", "vhostName": ""}, vhostPath: "/"},
+	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}, "deliver_get_details": {"rate": 1.6}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"hostFromEnv": "plainHost", "vhostName": "myhost"}, vhostPath: "/myhost"},
+	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}, "deliver_get_details": {"rate": 2}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"hostFromEnv": "plainHost", "vhostName": "/"}, vhostPath: "//"},
+	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 1.4}, "deliver_get_details": {"rate": 2.5}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"hostFromEnv": "plainHost", "vhostName": ""}},
+	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"hostFromEnv": "plainHost", "vhostName": "myhost"}, vhostPath: "/myhost"},
+	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"hostFromEnv": "plainHost", "vhostName": "/"}, vhostPath: rabbitRootVhostPath},
+	{response: `{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"hostFromEnv": "plainHost", "vhostName": ""}, vhostPath: "/"},
 }
 
 type resolvedVhostAndPathTestData struct {
@@ -438,68 +464,89 @@ func TestGetQueueInfo(t *testing.T) {
 
 var testRegexQueueInfoTestData = []getQueueInfoTestData{
 	// sum queue length
-	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum"}},
-	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum"}},
-	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum"}},
 	{response: `{"items":[]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum"}},
 	// sum queue length + ignoreUnacknowledged
-	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum", "excludeUnacknowledged": "true"}},
-	{response: `{"items":[{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum", "excludeUnacknowledged": "true"}},
-	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum", "excludeUnacknowledged": "true"}},
-	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum"}},
-	{response: `{"items":[{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum", "excludeUnacknowledged": "true"}},
+	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum", "excludeUnacknowledged": "true"}},
+	{response: `{"items":[{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum", "excludeUnacknowledged": "true"}},
+	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum", "excludeUnacknowledged": "true"}},
+	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum", "excludeUnacknowledged": "true"}},
 	{response: `{"items":[]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "sum"}},
 	// max queue length
-	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max"}},
-	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max"}},
-	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max"}},
 	{response: `{"items":[]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max"}},
 	// max queue length + excludeUnacknowledged
-	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max", "excludeUnacknowledged": "true"}},
-	{response: `{"items":[{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max", "excludeUnacknowledged": "true"}},
-	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max", "excludeUnacknowledged": "true"}},
-	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max"}},
-	{response: `{"items":[{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max", "excludeUnacknowledged": "true"}},
+	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max", "excludeUnacknowledged": "true"}},
+	{response: `{"items":[{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max", "excludeUnacknowledged": "true"}},
+	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max", "excludeUnacknowledged": "true"}},
+	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max", "excludeUnacknowledged": "true"}},
 	{response: `{"items":[]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "max"}},
 	// avg queue length
-	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg"}},
-	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg"}},
-	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg"}},
 	{response: `{"items":[]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg"}},
 	// avg queue length + excludeUnacknowledged
-	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg", "excludeUnacknowledged": "true"}},
-	{response: `{"items":[{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg", "excludeUnacknowledged": "true"}},
-	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg", "excludeUnacknowledged": "true"}},
-	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg"}},
-	{response: `{"items":[{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg", "excludeUnacknowledged": "true"}},
+	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg", "excludeUnacknowledged": "true"}},
+	{response: `{"items":[{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg", "excludeUnacknowledged": "true"}},
+	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg", "excludeUnacknowledged": "true"}},
+	{response: `{"items":[{"messages": 4, "messages_ready": 3, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_ready": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg", "excludeUnacknowledged": "true"}},
 	{response: `{"items":[]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"queueLength": "10", "useRegex": "true", "operation": "avg"}},
 	// sum message rate
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
 	{response: `{"items":[]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
 	// max message rate
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "max"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "max"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "max"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "max"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "max"}},
 	{response: `{"items":[]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "max"}},
 	// avg message rate
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
-	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"},{"messages": 0, "messages_unacknowledged": 1, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
 	{response: `{"items":[]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"mode": "MessageRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
+	// sum deliverGet rate
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 2}}, "name": "evaluate_trials"}, {"messages": 2, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 2}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}, {"messages": 4, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 4}}, "name": "evaluate_trials"}, {"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}, {"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
+	{response: `{"items":[]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "sum"}},
+	// max deliverGet rate
+	{response: `{"items":[{"messages": 8, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 2}}, "name": "evaluate_trials"}, {"messages": 4, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}, {"messages": 4, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 4}}, "name": "evaluate_trials"}, {"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}, {"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "max"}},
+	{response: `{"items":[]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "max"}},
+	// avg deliverGet rate
+	{response: `{"items":[{"messages": 8, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 4}}, "name": "evaluate_trials"}, {"messages": 4, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}, {"messages": 4, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 4}}, "name": "evaluate_trials"}, {"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[{"messages": 4, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 4}, "deliver_get_details": {"rate": 4}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, isActive: true, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[{"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trials"}, {"messages": 0, "messages_unacknowledged": 0, "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}}, "name": "evaluate_trial2"}]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
+	{response: `{"items":[]}`, responseStatus: http.StatusOK, extraMetadata: map[string]string{"mode": "DeliverGetRate", "value": "1000", "useRegex": "true", "operation": "avg"}},
 }
 
 func TestGetQueueInfoWithRegex(t *testing.T) {
@@ -679,15 +726,15 @@ type rabbitMQErrorTestData struct {
 }
 
 var anonymizeRabbitMQErrorTestData = []rabbitMQErrorTestData{
-	{fmt.Errorf("https://user1:password1@domain.com"), "error inspecting rabbitMQ: https://user:password@domain.com"},
-	{fmt.Errorf("https://fdasr345_-:password1@domain.com"), "error inspecting rabbitMQ: https://user:password@domain.com"},
-	{fmt.Errorf("https://user1:fdasr345_-@domain.com"), "error inspecting rabbitMQ: https://user:password@domain.com"},
-	{fmt.Errorf("https://fdakls_dsa:password1@domain.com"), "error inspecting rabbitMQ: https://user:password@domain.com"},
-	{fmt.Errorf("fdasr345_-:password1@domain.com"), "error inspecting rabbitMQ: user:password@domain.com"},
-	{fmt.Errorf("this user1:password1@domain.com fails"), "error inspecting rabbitMQ: this user:password@domain.com fails"},
-	{fmt.Errorf("this https://user1:password1@domain.com fails also"), "error inspecting rabbitMQ: this https://user:password@domain.com fails also"},
-	{fmt.Errorf("nothing to replace here"), "error inspecting rabbitMQ: nothing to replace here"},
-	{fmt.Errorf("the queue https://user1:fdasr345_-@domain.com/api/virtual is unavailable"), "error inspecting rabbitMQ: the queue https://user:password@domain.com/api/virtual is unavailable"},
+	{fmt.Errorf("https://user1:password1@domain.com"), "error inspecting RabbitMQ: https://user:password@domain.com"},
+	{fmt.Errorf("https://fdasr345_-:password1@domain.com"), "error inspecting RabbitMQ: https://user:password@domain.com"},
+	{fmt.Errorf("https://user1:fdasr345_-@domain.com"), "error inspecting RabbitMQ: https://user:password@domain.com"},
+	{fmt.Errorf("https://fdakls_dsa:password1@domain.com"), "error inspecting RabbitMQ: https://user:password@domain.com"},
+	{fmt.Errorf("fdasr345_-:password1@domain.com"), "error inspecting RabbitMQ: user:password@domain.com"},
+	{fmt.Errorf("this user1:password1@domain.com fails"), "error inspecting RabbitMQ: this user:password@domain.com fails"},
+	{fmt.Errorf("this https://user1:password1@domain.com fails also"), "error inspecting RabbitMQ: this https://user:password@domain.com fails also"},
+	{fmt.Errorf("nothing to replace here"), "error inspecting RabbitMQ: nothing to replace here"},
+	{fmt.Errorf("the queue https://user1:fdasr345_-@domain.com/api/virtual is unavailable"), "error inspecting RabbitMQ: the queue https://user:password@domain.com/api/virtual is unavailable"},
 }
 
 func TestRabbitMQAnonymizeRabbitMQError(t *testing.T) {
diff --git a/tests/internals/scaling_strategies/eager_scaling_strategy_test.go b/tests/internals/scaling_strategies/eager_scaling_strategy_test.go
index 222f960ef..ec81fb303 100644
--- a/tests/internals/scaling_strategies/eager_scaling_strategy_test.go
+++ b/tests/internals/scaling_strategies/eager_scaling_strategy_test.go
@@ -101,7 +101,7 @@ func TestScalingStrategy(t *testing.T) {
 
 	RMQInstall(t, kc, rmqNamespace, user, password, vhost, WithoutOAuth())
 	// Publish 0 messges but create the queue
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, 0)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, 0, 0)
 	WaitForAllJobsSuccess(t, kc, rmqNamespace, 60, 1)
 
 	CreateKubernetesResources(t, kc, testNamespace, data, templates)
@@ -123,17 +123,17 @@ func getTemplateData() (templateData, []Template) {
 
 func testEagerScaling(t *testing.T, kc *kubernetes.Clientset) {
 	iterationCount := 20
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, 4)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, 4, 0)
 	WaitForAllJobsSuccess(t, kc, rmqNamespace, 60, 1)
 	assert.True(t, WaitForScaledJobCount(t, kc, scaledJobName, testNamespace, 4, iterationCount, 1),
 		"job count should be %d after %d iterations", 4, iterationCount)
 
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, 4)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, 4, 0)
 	WaitForAllJobsSuccess(t, kc, rmqNamespace, 60, 1)
 	assert.True(t, WaitForScaledJobCount(t, kc, scaledJobName, testNamespace, 8, iterationCount, 1),
 		"job count should be %d after %d iterations", 8, iterationCount)
 
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, 8)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, 8, 0)
 	WaitForAllJobsSuccess(t, kc, rmqNamespace, 60, 1)
 	assert.True(t, WaitForScaledJobCount(t, kc, scaledJobName, testNamespace, 10, iterationCount, 1),
 		"job count should be %d after %d iterations", 10, iterationCount)
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
diff --git a/tests/scalers/rabbitmq/rabbitmq_helper.go b/tests/scalers/rabbitmq/rabbitmq_helper.go
index d0a78e821..1e777730e 100644
--- a/tests/scalers/rabbitmq/rabbitmq_helper.go
+++ b/tests/scalers/rabbitmq/rabbitmq_helper.go
@@ -27,7 +27,88 @@ spec:
       - name: rabbitmq-client
         image: ghcr.io/kedacore/tests-rabbitmq
         imagePullPolicy: Always
-        command: ["send",  "{{.Connection}}", "{{.MessageCount}}", "{{.QueueName}}"]
+        command:
+          - send
+        args:
+          - '{{.Connection}}'
+          - '{{.MessageCount}}'
+          - '{{.QueueName}}'
+          - '{{.Interval}}'
+      restartPolicy: Never
+`
+
+	consumeTemplateJobCount = 4
+	consumeTemplate         = `
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: rabbitmq-consume-0-{{.QueueName}}
+  namespace: {{.Namespace}}
+spec:
+  template:
+    spec:
+      containers:
+      - name: rabbitmq-client
+        image: ghcr.io/kedacore/tests-rabbitmq
+        imagePullPolicy: Always
+        command:
+          - receive
+        args:
+          - '{{.Connection}}'
+      restartPolicy: Never
+---
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: rabbitmq-consume-1-{{.QueueName}}
+  namespace: {{.Namespace}}
+spec:
+  template:
+    spec:
+      containers:
+      - name: rabbitmq-client
+        image: ghcr.io/kedacore/tests-rabbitmq
+        imagePullPolicy: Always
+        command:
+          - receive
+        args:
+          - '{{.Connection}}'
+      restartPolicy: Never
+---
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: rabbitmq-consume-2-{{.QueueName}}
+  namespace: {{.Namespace}}
+spec:
+  template:
+    spec:
+      containers:
+      - name: rabbitmq-client
+        image: ghcr.io/kedacore/tests-rabbitmq
+        imagePullPolicy: Always
+        command:
+          - receive
+        args:
+          - '{{.Connection}}'
+      restartPolicy: Never
+---
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: rabbitmq-consume-3-{{.QueueName}}
+  namespace: {{.Namespace}}
+spec:
+  template:
+    spec:
+      containers:
+      - name: rabbitmq-client
+        image: ghcr.io/kedacore/tests-rabbitmq
+        imagePullPolicy: Always
+        command:
+          - receive
+        args:
+          - '{{.Connection}}'
       restartPolicy: Never
 `
 
@@ -167,6 +248,48 @@ spec:
             name: {{.SecretName}}
 `
 
+	RMQPublisherTargetDeploymentTemplate = `
+apiVersion: v1
+kind: Secret
+metadata:
+  name: {{.SecretName}}
+  namespace: {{.TestNamespace}}
+data:
+  RabbitApiHost: {{.Base64Connection}}
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{.DeploymentName}}
+  namespace: {{.TestNamespace}}
+  labels:
+    app: {{.DeploymentName}}
+spec:
+  replicas: 0
+  selector:
+    matchLabels:
+      app: {{.DeploymentName}}
+  template:
+    metadata:
+      labels:
+        app: {{.DeploymentName}}
+    spec:
+      containers:
+      - name: rabbitmq-publisher
+        image: ghcr.io/kedacore/tests-rabbitmq
+        imagePullPolicy: Always
+        command:
+          - send
+        args:
+          - '{{.Connection}}'
+          - '4'
+          - '{{.QueueName}}'
+          - '1'
+        envFrom:
+        - secretRef:
+            name: {{.SecretName}}
+`
+
 	RMQTargetDeploymentWithAuthEnvTemplate = `
 apiVersion: v1
 kind: Secret
@@ -240,6 +363,7 @@ type templateData struct {
 	HostName, VHostName string
 	Username, Password  string
 	MessageCount        int
+	Interval            int
 	EnableOAuth         bool
 	OAuthClientID       string
 	OAuthScopesKey      string
@@ -280,20 +404,44 @@ func RMQUninstall(t *testing.T, namespace, user, password, vhost string, oauth R
 	helper.DeleteNamespace(t, namespace)
 }
 
-func RMQPublishMessages(t *testing.T, namespace, connectionString, queueName string, messageCount int) {
+func RMQPublishMessages(t *testing.T, namespace, connectionString, queueName string, messageCount, interval int) {
 	data := templateData{
 		Namespace:    namespace,
 		Connection:   connectionString,
 		QueueName:    queueName,
 		MessageCount: messageCount,
+		Interval:     interval,
 	}
 
-	// Before push messages remove previous jobs if any
-	_, _ = helper.ExecuteCommand(fmt.Sprintf("kubectl delete jobs/rabbitmq-publish-%s --namespace %s", queueName, namespace))
+	// Before pushing new messages, remove all previous publishing jobs, if any.
+	RMQStopPublishingMessages(namespace, queueName)
 
 	helper.KubectlApplyWithTemplate(t, data, "rmqPublishTemplate", publishTemplate)
 }
 
+func RMQStopPublishingMessages(namespace, queueName string) {
+	_, _ = helper.ExecuteCommand(fmt.Sprintf("kubectl delete jobs/rabbitmq-publish-%s --namespace %s", queueName, namespace))
+}
+
+func RMQConsumeMessages(t *testing.T, namespace, connectionString, queueName string) {
+	data := templateData{
+		Namespace:  namespace,
+		Connection: connectionString,
+		QueueName:  queueName,
+	}
+
+	// Before consuming messages, remove all previous consumer jobs, if any.
+	RMQStopConsumingMessages(namespace, queueName)
+
+	helper.KubectlApplyWithTemplate(t, data, "rmqConsumerTemplate", consumeTemplate)
+}
+
+func RMQStopConsumingMessages(namespace, queueName string) {
+	for i := 0; i < consumeTemplateJobCount; i++ {
+		_, _ = helper.ExecuteCommand(fmt.Sprintf("kubectl delete jobs/rabbitmq-consume-%d-%s --namespace %s", i, queueName, namespace))
+	}
+}
+
 func RMQCreateVHost(t *testing.T, namespace, host, user, password, vhost string) {
 	data := templateData{
 		Namespace: namespace,
diff --git a/tests/scalers/rabbitmq/rabbitmq_queue_amqp/rabbitmq_queue_amqp_test.go b/tests/scalers/rabbitmq/rabbitmq_queue_amqp/rabbitmq_queue_amqp_test.go
index a545ae3be..fd8a66843 100644
--- a/tests/scalers/rabbitmq/rabbitmq_queue_amqp/rabbitmq_queue_amqp_test.go
+++ b/tests/scalers/rabbitmq/rabbitmq_queue_amqp/rabbitmq_queue_amqp_test.go
@@ -110,7 +110,7 @@ func getTemplateData() (templateData, []Template) {
 
 func testScaling(t *testing.T, kc *kubernetes.Clientset) {
 	t.Log("--- testing scale out ---")
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount, 0)
 	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 4, 60, 3),
 		"replica count should be 4 after 3 minute")
 
@@ -122,7 +122,7 @@ func testScaling(t *testing.T, kc *kubernetes.Clientset) {
 func testActivationValue(t *testing.T, kc *kubernetes.Clientset) {
 	t.Log("--- testing activation value ---")
 	messagesToQueue := 3
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messagesToQueue)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messagesToQueue, 0)
 
 	AssertReplicaCountNotChangeDuringTimePeriod(t, kc, deploymentName, testNamespace, 0, 60)
 }
diff --git a/tests/scalers/rabbitmq/rabbitmq_queue_amqp_auth/rabbitmq_queue_amqp_auth_test.go b/tests/scalers/rabbitmq/rabbitmq_queue_amqp_auth/rabbitmq_queue_amqp_auth_test.go
index b73a7ff63..c058b8bc8 100644
--- a/tests/scalers/rabbitmq/rabbitmq_queue_amqp_auth/rabbitmq_queue_amqp_auth_test.go
+++ b/tests/scalers/rabbitmq/rabbitmq_queue_amqp_auth/rabbitmq_queue_amqp_auth_test.go
@@ -204,7 +204,7 @@ func testAuthFromSecret(t *testing.T, kc *kubernetes.Clientset, data templateDat
 	KubectlApplyWithTemplate(t, data, "triggerAuthenticationTemplate", triggerAuthenticationTemplate)
 	defer KubectlDeleteWithTemplate(t, data, "triggerAuthenticationTemplate", triggerAuthenticationTemplate)
 
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount, 0)
 	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 4, 60, 1),
 		"replica count should be 4 after 1 minute")
 
@@ -218,7 +218,7 @@ func testAuthFromEnv(t *testing.T, kc *kubernetes.Clientset, data templateData)
 	KubectlApplyWithTemplate(t, data, "scaledObjectAuthFromEnvTemplate", scaledObjectAuthFromEnvTemplate)
 	defer KubectlDeleteWithTemplate(t, data, "scaledObjectAuthFromEnvTemplate", scaledObjectAuthFromEnvTemplate)
 
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount, 0)
 	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 4, 60, 1),
 		"replica count should be 4 after 1 minute")
 
@@ -252,7 +252,7 @@ func testInvalidUsernameAndPassword(t *testing.T, kc *kubernetes.Clientset, data
 func testActivationValue(t *testing.T, kc *kubernetes.Clientset) {
 	t.Log("--- testing activation value ---")
 	messagesToQueue := 3
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messagesToQueue)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messagesToQueue, 0)
 
 	AssertReplicaCountNotChangeDuringTimePeriod(t, kc, deploymentName, testNamespace, 0, 60)
 }
diff --git a/tests/scalers/rabbitmq/rabbitmq_queue_amqp_vhost/rabbitmq_queue_amqp_vhost_test.go b/tests/scalers/rabbitmq/rabbitmq_queue_amqp_vhost/rabbitmq_queue_amqp_vhost_test.go
index 915eefa1f..d7c87a276 100644
--- a/tests/scalers/rabbitmq/rabbitmq_queue_amqp_vhost/rabbitmq_queue_amqp_vhost_test.go
+++ b/tests/scalers/rabbitmq/rabbitmq_queue_amqp_vhost/rabbitmq_queue_amqp_vhost_test.go
@@ -110,7 +110,7 @@ func getTemplateData() (templateData, []Template) {
 
 func testScaling(t *testing.T, kc *kubernetes.Clientset) {
 	t.Log("--- testing scale out ---")
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount, 0)
 	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 4, 60, 3),
 		"replica count should be 4 after 3 minute")
 
@@ -122,7 +122,7 @@ func testScaling(t *testing.T, kc *kubernetes.Clientset) {
 func testActivationValue(t *testing.T, kc *kubernetes.Clientset) {
 	t.Log("--- testing activation value ---")
 	messagesToQueue := 3
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messagesToQueue)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messagesToQueue, 0)
 
 	AssertReplicaCountNotChangeDuringTimePeriod(t, kc, deploymentName, testNamespace, 0, 60)
 }
diff --git a/tests/scalers/rabbitmq/rabbitmq_queue_http/rabbitmq_queue_http_test.go b/tests/scalers/rabbitmq/rabbitmq_queue_http/rabbitmq_queue_http_test.go
index b27e006c1..dc5621be2 100644
--- a/tests/scalers/rabbitmq/rabbitmq_queue_http/rabbitmq_queue_http_test.go
+++ b/tests/scalers/rabbitmq/rabbitmq_queue_http/rabbitmq_queue_http_test.go
@@ -109,7 +109,7 @@ func getTemplateData() (templateData, []Template) {
 
 func testScaling(t *testing.T, kc *kubernetes.Clientset) {
 	t.Log("--- testing scale out ---")
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount, 0)
 	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 4, 60, 3),
 		"replica count should be 4 after 3 minute")
 
diff --git a/tests/scalers/rabbitmq/rabbitmq_queue_http_aad_wi/rabbitmq_queue_http_aad_wi_test.go b/tests/scalers/rabbitmq/rabbitmq_queue_http_aad_wi/rabbitmq_queue_http_aad_wi_test.go
index cad930240..f83c3dc73 100644
--- a/tests/scalers/rabbitmq/rabbitmq_queue_http_aad_wi/rabbitmq_queue_http_aad_wi_test.go
+++ b/tests/scalers/rabbitmq/rabbitmq_queue_http_aad_wi/rabbitmq_queue_http_aad_wi_test.go
@@ -161,7 +161,7 @@ func getTemplateData() (templateData, []Template) {
 
 func testScaling(t *testing.T, kc *kubernetes.Clientset) {
 	t.Log("--- testing scale out ---")
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount, 0)
 	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 4, 60, 3),
 		"replica count should be 4 after 3 minute")
 
diff --git a/tests/scalers/rabbitmq/rabbitmq_queue_http_auth/rabbitmq_queue_http_auth_test.go b/tests/scalers/rabbitmq/rabbitmq_queue_http_auth/rabbitmq_queue_http_auth_test.go
index ed6949264..e811e7454 100644
--- a/tests/scalers/rabbitmq/rabbitmq_queue_http_auth/rabbitmq_queue_http_auth_test.go
+++ b/tests/scalers/rabbitmq/rabbitmq_queue_http_auth/rabbitmq_queue_http_auth_test.go
@@ -204,7 +204,7 @@ func testAuthFromSecret(t *testing.T, kc *kubernetes.Clientset, data templateDat
 	KubectlApplyWithTemplate(t, data, "triggerAuthenticationTemplate", triggerAuthenticationTemplate)
 	defer KubectlDeleteWithTemplate(t, data, "triggerAuthenticationTemplate", triggerAuthenticationTemplate)
 
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount, 0)
 	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 4, 60, 1),
 		"replica count should be 4 after 1 minute")
 
@@ -218,7 +218,7 @@ func testAuthFromEnv(t *testing.T, kc *kubernetes.Clientset, data templateData)
 	KubectlApplyWithTemplate(t, data, "scaledObjectAuthFromEnvTemplate", scaledObjectAuthFromEnvTemplate)
 	defer KubectlDeleteWithTemplate(t, data, "scaledObjectAuthFromEnvTemplate", scaledObjectAuthFromEnvTemplate)
 
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount, 0)
 	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 4, 60, 1),
 		"replica count should be 4 after 1 minute")
 
@@ -252,7 +252,7 @@ func testInvalidUsernameAndPassword(t *testing.T, kc *kubernetes.Clientset, data
 func testActivationValue(t *testing.T, kc *kubernetes.Clientset) {
 	t.Log("--- testing activation value ---")
 	messagesToQueue := 3
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messagesToQueue)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messagesToQueue, 0)
 
 	AssertReplicaCountNotChangeDuringTimePeriod(t, kc, deploymentName, testNamespace, 0, 60)
 }
diff --git a/tests/scalers/rabbitmq/rabbitmq_queue_http_dget/rabbitmq_queue_http_dget_test.go b/tests/scalers/rabbitmq/rabbitmq_queue_http_dget/rabbitmq_queue_http_dget_test.go
new file mode 100644
index 000000000..047a20cbc
--- /dev/null
+++ b/tests/scalers/rabbitmq/rabbitmq_queue_http_dget/rabbitmq_queue_http_dget_test.go
@@ -0,0 +1,127 @@
+//go:build e2e
+// +build e2e
+
+package rabbitmq_queue_http_dget_test
+
+import (
+	"encoding/base64"
+	"fmt"
+	"testing"
+
+	"github.com/joho/godotenv"
+	"github.com/stretchr/testify/assert"
+	"k8s.io/client-go/kubernetes"
+
+	. "github.com/kedacore/keda/v2/tests/helper"
+	. "github.com/kedacore/keda/v2/tests/scalers/rabbitmq"
+)
+
+// Load environment variables from .env file
+var _ = godotenv.Load("../../../.env")
+
+const (
+	testName = "rmq-queue-http-dget-test"
+)
+
+var (
+	testNamespace        = fmt.Sprintf("%s-ns", testName)
+	rmqNamespace         = fmt.Sprintf("%s-rmq", testName)
+	deploymentName       = fmt.Sprintf("%s-deployment", testName)
+	secretName           = fmt.Sprintf("%s-secret", testName)
+	scaledObjectName     = fmt.Sprintf("%s-so", testName)
+	queueName            = "hello"
+	user                 = fmt.Sprintf("%s-user", testName)
+	password             = fmt.Sprintf("%s-password", testName)
+	vhost                = "/"
+	connectionString     = fmt.Sprintf("amqp://%s:%s@rabbitmq.%s.svc.cluster.local/", user, password, rmqNamespace)
+	httpConnectionString = fmt.Sprintf("http://%s:%s@rabbitmq.%s.svc.cluster.local/", user, password, rmqNamespace)
+)
+
+const (
+	scaledObjectTemplate = `
+apiVersion: keda.sh/v1alpha1
+kind: ScaledObject
+metadata:
+  name: {{.ScaledObjectName}}
+  namespace: {{.TestNamespace}}
+spec:
+  scaleTargetRef:
+    name: {{.DeploymentName}}
+  pollingInterval: 1
+  cooldownPeriod: 5
+  minReplicaCount: 1
+  maxReplicaCount: 4
+  advanced:
+    horizontalPodAutoscalerConfig:
+      behavior:
+        scaleDown:
+          stabilizationWindowSeconds: 5
+  triggers:
+    - type: rabbitmq
+      metricType: Value
+      metadata:
+        queueName: {{.QueueName}}
+        hostFromEnv: RabbitApiHost
+        protocol: http
+        mode: DeliverGetRate
+        value: '3'
+        activationValue: '0'
+`
+)
+
+type templateData struct {
+	TestNamespace                string
+	DeploymentName               string
+	ScaledObjectName             string
+	SecretName                   string
+	QueueName                    string
+	Connection, Base64Connection string
+}
+
+// Setup environment and run tests
+func TestScaler(t *testing.T) {
+	t.Log("--- setting up ---")
+	kc := GetKubernetesClient(t)
+	data, templates := getTemplateData()
+	t.Cleanup(func() {
+		DeleteKubernetesResources(t, testNamespace, data, templates)
+		RMQUninstall(t, rmqNamespace, user, password, vhost, WithoutOAuth())
+	})
+
+	// Create Kubernetes resources
+	RMQInstall(t, kc, rmqNamespace, user, password, vhost, WithoutOAuth())
+	CreateKubernetesResources(t, kc, testNamespace, data, templates)
+
+	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 0, 60, 1),
+		"replica count should be 0 after 1 minute")
+
+	// Run tests
+	testScaling(t, kc)
+}
+
+func getTemplateData() (templateData, []Template) {
+	return templateData{
+			TestNamespace:    testNamespace,
+			DeploymentName:   deploymentName,
+			ScaledObjectName: scaledObjectName,
+			SecretName:       secretName,
+			QueueName:        queueName,
+			Connection:       connectionString,
+			Base64Connection: base64.StdEncoding.EncodeToString([]byte(httpConnectionString)),
+		}, []Template{
+			{Name: "deploymentTemplate", Config: RMQPublisherTargetDeploymentTemplate},
+			{Name: "scaledObjectTemplate", Config: scaledObjectTemplate},
+		}
+}
+
+func testScaling(t *testing.T, kc *kubernetes.Clientset) {
+	t.Log("--- testing scale out ---")
+	RMQConsumeMessages(t, rmqNamespace, connectionString, queueName)
+	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 4, 60, 3),
+		"replica count should be 4 after 3 minute")
+
+	RMQStopConsumingMessages(rmqNamespace, queueName)
+	t.Log("--- testing scale in ---")
+	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 1, 60, 1),
+		"replica count should be 1 after 1 minute")
+}
diff --git a/tests/scalers/rabbitmq/rabbitmq_queue_http_dpratio/rabbitmq_queue_http_dpratio_test.go b/tests/scalers/rabbitmq/rabbitmq_queue_http_dpratio/rabbitmq_queue_http_dpratio_test.go
new file mode 100644
index 000000000..36cd97fe8
--- /dev/null
+++ b/tests/scalers/rabbitmq/rabbitmq_queue_http_dpratio/rabbitmq_queue_http_dpratio_test.go
@@ -0,0 +1,127 @@
+//go:build e2e
+// +build e2e
+
+package rabbitmq_queue_http_dpratio_test
+
+import (
+	"encoding/base64"
+	"fmt"
+	"testing"
+
+	"github.com/joho/godotenv"
+	"github.com/stretchr/testify/assert"
+	"k8s.io/client-go/kubernetes"
+
+	. "github.com/kedacore/keda/v2/tests/helper"
+	. "github.com/kedacore/keda/v2/tests/scalers/rabbitmq"
+)
+
+// Load environment variables from .env file
+var _ = godotenv.Load("../../../.env")
+
+const (
+	testName = "rmq-queue-http-dpratio-test"
+)
+
+var (
+	testNamespace        = fmt.Sprintf("%s-ns", testName)
+	rmqNamespace         = fmt.Sprintf("%s-rmq", testName)
+	deploymentName       = fmt.Sprintf("%s-deployment", testName)
+	secretName           = fmt.Sprintf("%s-secret", testName)
+	scaledObjectName     = fmt.Sprintf("%s-so", testName)
+	queueName            = "hello"
+	user                 = fmt.Sprintf("%s-user", testName)
+	password             = fmt.Sprintf("%s-password", testName)
+	vhost                = "/"
+	connectionString     = fmt.Sprintf("amqp://%s:%s@rabbitmq.%s.svc.cluster.local/", user, password, rmqNamespace)
+	httpConnectionString = fmt.Sprintf("http://%s:%s@rabbitmq.%s.svc.cluster.local/", user, password, rmqNamespace)
+	messageCount         = 4
+)
+
+const (
+	scaledObjectTemplate = `
+apiVersion: keda.sh/v1alpha1
+kind: ScaledObject
+metadata:
+  name: {{.ScaledObjectName}}
+  namespace: {{.TestNamespace}}
+spec:
+  scaleTargetRef:
+    name: {{.DeploymentName}}
+  pollingInterval: 1
+  cooldownPeriod: 5
+  minReplicaCount: 0
+  maxReplicaCount: 4
+  advanced:
+    horizontalPodAutoscalerConfig:
+      behavior:
+        scaleDown:
+          stabilizationWindowSeconds: 5
+  triggers:
+    - type: rabbitmq
+      metricType: Value
+      metadata:
+        queueName: {{.QueueName}}
+        hostFromEnv: RabbitApiHost
+        protocol: http
+        mode: PublishedToDeliveredRatio
+        value: '1'
+        activationValue: '1'
+`
+)
+
+type templateData struct {
+	TestNamespace                string
+	DeploymentName               string
+	ScaledObjectName             string
+	SecretName                   string
+	QueueName                    string
+	Connection, Base64Connection string
+}
+
+func TestScaler(t *testing.T) {
+	// setup
+	t.Log("--- setting up ---")
+	kc := GetKubernetesClient(t)
+	data, templates := getTemplateData()
+	t.Cleanup(func() {
+		DeleteKubernetesResources(t, testNamespace, data, templates)
+		RMQUninstall(t, rmqNamespace, user, password, vhost, WithoutOAuth())
+	})
+
+	// Create kubernetes resources
+	RMQInstall(t, kc, rmqNamespace, user, password, vhost, WithoutOAuth())
+	CreateKubernetesResources(t, kc, testNamespace, data, templates)
+
+	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 0, 60, 1),
+		"replica count should be 0 after 1 minute")
+
+	testScaling(t, kc)
+}
+
+func getTemplateData() (templateData, []Template) {
+	return templateData{
+			TestNamespace:    testNamespace,
+			DeploymentName:   deploymentName,
+			ScaledObjectName: scaledObjectName,
+			SecretName:       secretName,
+			QueueName:        queueName,
+			Connection:       connectionString,
+			Base64Connection: base64.StdEncoding.EncodeToString([]byte(httpConnectionString)),
+		}, []Template{
+			{Name: "deploymentTemplate", Config: RMQTargetDeploymentTemplate},
+			{Name: "scaledObjectTemplate", Config: scaledObjectTemplate},
+		}
+}
+
+func testScaling(t *testing.T, kc *kubernetes.Clientset) {
+	t.Log("--- testing scale out ---")
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount, 1)
+	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 4, 60, 3),
+		"replica count should be 4 after 3 minute")
+
+	RMQStopPublishingMessages(rmqNamespace, queueName)
+	t.Log("--- testing scale in ---")
+	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 0, 60, 1),
+		"replica count should be 0 after 1 minute")
+}
diff --git a/tests/scalers/rabbitmq/rabbitmq_queue_http_eqct/rabbitmq_queue_http_eqct_test.go b/tests/scalers/rabbitmq/rabbitmq_queue_http_eqct/rabbitmq_queue_http_eqct_test.go
new file mode 100644
index 000000000..75681deb5
--- /dev/null
+++ b/tests/scalers/rabbitmq/rabbitmq_queue_http_eqct/rabbitmq_queue_http_eqct_test.go
@@ -0,0 +1,127 @@
+//go:build e2e
+// +build e2e
+
+package rabbitmq_queue_http_eqct_test
+
+import (
+	"encoding/base64"
+	"fmt"
+	"testing"
+
+	"github.com/joho/godotenv"
+	"github.com/stretchr/testify/assert"
+	"k8s.io/client-go/kubernetes"
+
+	. "github.com/kedacore/keda/v2/tests/helper"
+	. "github.com/kedacore/keda/v2/tests/scalers/rabbitmq"
+)
+
+// Load environment variables from .env file
+var _ = godotenv.Load("../../../.env")
+
+const (
+	testName = "rmq-queue-http-eqct-test"
+)
+
+var (
+	testNamespace        = fmt.Sprintf("%s-ns", testName)
+	rmqNamespace         = fmt.Sprintf("%s-rmq", testName)
+	deploymentName       = fmt.Sprintf("%s-deployment", testName)
+	secretName           = fmt.Sprintf("%s-secret", testName)
+	scaledObjectName     = fmt.Sprintf("%s-so", testName)
+	queueName            = "hello"
+	user                 = fmt.Sprintf("%s-user", testName)
+	password             = fmt.Sprintf("%s-password", testName)
+	vhost                = "/"
+	connectionString     = fmt.Sprintf("amqp://%s:%s@rabbitmq.%s.svc.cluster.local/", user, password, rmqNamespace)
+	httpConnectionString = fmt.Sprintf("http://%s:%s@rabbitmq.%s.svc.cluster.local/", user, password, rmqNamespace)
+	messageCount         = 3
+)
+
+const (
+	scaledObjectTemplate = `
+apiVersion: keda.sh/v1alpha1
+kind: ScaledObject
+metadata:
+  name: {{.ScaledObjectName}}
+  namespace: {{.TestNamespace}}
+spec:
+  scaleTargetRef:
+    name: {{.DeploymentName}}
+  pollingInterval: 1
+  cooldownPeriod: 5
+  minReplicaCount: 0
+  maxReplicaCount: 4
+  advanced:
+    horizontalPodAutoscalerConfig:
+      behavior:
+        scaleDown:
+          stabilizationWindowSeconds: 5
+  triggers:
+    - type: rabbitmq
+      metricType: Value
+      metadata:
+        queueName: {{.QueueName}}
+        hostFromEnv: RabbitApiHost
+        protocol: http
+        mode: ExpectedQueueConsumptionTime
+        value: '1'
+        activationValue: '1'
+`
+)
+
+type templateData struct {
+	TestNamespace                string
+	DeploymentName               string
+	ScaledObjectName             string
+	SecretName                   string
+	QueueName                    string
+	Connection, Base64Connection string
+}
+
+func TestScaler(t *testing.T) {
+	// setup
+	t.Log("--- setting up ---")
+	kc := GetKubernetesClient(t)
+	data, templates := getTemplateData()
+	t.Cleanup(func() {
+		DeleteKubernetesResources(t, testNamespace, data, templates)
+		RMQUninstall(t, rmqNamespace, user, password, vhost, WithoutOAuth())
+	})
+
+	// Create kubernetes resources
+	RMQInstall(t, kc, rmqNamespace, user, password, vhost, WithoutOAuth())
+	CreateKubernetesResources(t, kc, testNamespace, data, templates)
+
+	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 0, 60, 1),
+		"replica count should be 0 after 1 minute")
+
+	testScaling(t, kc)
+}
+
+func getTemplateData() (templateData, []Template) {
+	return templateData{
+			TestNamespace:    testNamespace,
+			DeploymentName:   deploymentName,
+			ScaledObjectName: scaledObjectName,
+			SecretName:       secretName,
+			QueueName:        queueName,
+			Connection:       connectionString,
+			Base64Connection: base64.StdEncoding.EncodeToString([]byte(httpConnectionString)),
+		}, []Template{
+			{Name: "deploymentTemplate", Config: RMQTargetDeploymentTemplate},
+			{Name: "scaledObjectTemplate", Config: scaledObjectTemplate},
+		}
+}
+
+func testScaling(t *testing.T, kc *kubernetes.Clientset) {
+	t.Log("--- testing scale out ---")
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount, 1)
+	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 4, 60, 3),
+		"replica count should be 4 after 3 minute")
+
+	RMQStopPublishingMessages(rmqNamespace, queueName)
+	t.Log("--- testing scale in ---")
+	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 0, 60, 1),
+		"replica count should be 0 after 1 minute")
+}
diff --git a/tests/scalers/rabbitmq/rabbitmq_queue_http_regex/rabbitmq_queue_http_regex_test.go b/tests/scalers/rabbitmq/rabbitmq_queue_http_regex/rabbitmq_queue_http_regex_test.go
index 0ac8c1c6f..0a15773cb 100644
--- a/tests/scalers/rabbitmq/rabbitmq_queue_http_regex/rabbitmq_queue_http_regex_test.go
+++ b/tests/scalers/rabbitmq/rabbitmq_queue_http_regex/rabbitmq_queue_http_regex_test.go
@@ -112,10 +112,10 @@ func getTemplateData() (templateData, []Template) {
 
 func testScaling(t *testing.T, kc *kubernetes.Clientset) {
 	t.Log("--- testing scale out ---")
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount, 0)
 	// dummies
-	RMQPublishMessages(t, rmqNamespace, connectionString, fmt.Sprintf("%s-1", queueName), messageCount)
-	RMQPublishMessages(t, rmqNamespace, connectionString, fmt.Sprintf("%s-%s", queueName, queueName), messageCount)
+	RMQPublishMessages(t, rmqNamespace, connectionString, fmt.Sprintf("%s-1", queueName), messageCount, 0)
+	RMQPublishMessages(t, rmqNamespace, connectionString, fmt.Sprintf("%s-%s", queueName, queueName), messageCount, 0)
 
 	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 4, 60, 2),
 		"replica count should be 4 after 2 minute")
diff --git a/tests/scalers/rabbitmq/rabbitmq_queue_http_regex_aad_wi/rabbitmq_queue_http_regex_aad_wi_test.go b/tests/scalers/rabbitmq/rabbitmq_queue_http_regex_aad_wi/rabbitmq_queue_http_regex_aad_wi_test.go
index 1faf94c67..050ce3cd0 100644
--- a/tests/scalers/rabbitmq/rabbitmq_queue_http_regex_aad_wi/rabbitmq_queue_http_regex_aad_wi_test.go
+++ b/tests/scalers/rabbitmq/rabbitmq_queue_http_regex_aad_wi/rabbitmq_queue_http_regex_aad_wi_test.go
@@ -163,10 +163,10 @@ func getTemplateData() (templateData, []Template) {
 
 func testScaling(t *testing.T, kc *kubernetes.Clientset) {
 	t.Log("--- testing scale out ---")
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount, 0)
 	// dummies
-	RMQPublishMessages(t, rmqNamespace, connectionString, fmt.Sprintf("%s-1", queueName), messageCount)
-	RMQPublishMessages(t, rmqNamespace, connectionString, fmt.Sprintf("%s-%s", queueName, queueName), messageCount)
+	RMQPublishMessages(t, rmqNamespace, connectionString, fmt.Sprintf("%s-1", queueName), messageCount, 0)
+	RMQPublishMessages(t, rmqNamespace, connectionString, fmt.Sprintf("%s-%s", queueName, queueName), messageCount, 0)
 
 	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 4, 60, 2),
 		"replica count should be 4 after 2 minute")
diff --git a/tests/scalers/rabbitmq/rabbitmq_queue_http_regex_vhost/rabbitmq_queue_http_regex_vhost_test.go b/tests/scalers/rabbitmq/rabbitmq_queue_http_regex_vhost/rabbitmq_queue_http_regex_vhost_test.go
index d633a0e0e..510114333 100644
--- a/tests/scalers/rabbitmq/rabbitmq_queue_http_regex_vhost/rabbitmq_queue_http_regex_vhost_test.go
+++ b/tests/scalers/rabbitmq/rabbitmq_queue_http_regex_vhost/rabbitmq_queue_http_regex_vhost_test.go
@@ -120,10 +120,10 @@ func getTemplateData() (templateData, []Template) {
 
 func testScaling(t *testing.T, kc *kubernetes.Clientset) {
 	t.Log("--- testing scale out ---")
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount, 0)
 	// dummies
-	RMQPublishMessages(t, rmqNamespace, dummyConnectionString1, fmt.Sprintf("%s-1", queueName), messageCount)
-	RMQPublishMessages(t, rmqNamespace, dummyConnectionString2, fmt.Sprintf("%s-%s", queueName, queueName), messageCount)
+	RMQPublishMessages(t, rmqNamespace, dummyConnectionString1, fmt.Sprintf("%s-1", queueName), messageCount, 0)
+	RMQPublishMessages(t, rmqNamespace, dummyConnectionString2, fmt.Sprintf("%s-%s", queueName, queueName), messageCount, 0)
 
 	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 4, 60, 2),
 		"replica count should be 4 after 2 minute")
diff --git a/tests/scalers/rabbitmq/rabbitmq_queue_http_vhost/rabbitmq_queue_http_vhost_test.go b/tests/scalers/rabbitmq/rabbitmq_queue_http_vhost/rabbitmq_queue_http_vhost_test.go
index 0720b0fe0..d2669f54a 100644
--- a/tests/scalers/rabbitmq/rabbitmq_queue_http_vhost/rabbitmq_queue_http_vhost_test.go
+++ b/tests/scalers/rabbitmq/rabbitmq_queue_http_vhost/rabbitmq_queue_http_vhost_test.go
@@ -109,7 +109,7 @@ func getTemplateData() (templateData, []Template) {
 
 func testScaling(t *testing.T, kc *kubernetes.Clientset) {
 	t.Log("--- testing scale out ---")
-	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount)
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messageCount, 0)
 	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 4, 60, 3),
 		"replica count should be 4 after 3 minute")