kafka - don't scale beyond number of partitions

Former-commit-id: 3656c06
asoundarya96 · May 6, 2019 · 93a600e · 93a600e
1 parent fb9366c
commit 93a600e
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/pkg/scalers/kafka_scaler.go b/pkg/scalers/kafka_scaler.go
@@ -227,7 +227,11 @@ func (s *kafkaScaler) GetMetrics(ctx context.Context, merticName string, metricS
 		totalLag += lag
 	}
 
-	// We should find a way to not scale beyond the number of partitions
+	// don't scale out beyond the number of partitions
+	if (totalLag / s.metadata.lagThreshold) > int64(len(partitions)) {
+		totalLag = int64(len(partitions)) * s.metadata.lagThreshold
+	}
+
 	metric := external_metrics.ExternalMetricValue{
 		MetricName: merticName,
 		Value:      *resource.NewQuantity(int64(totalLag), resource.DecimalSI),