Fix tests

internal/controller/eventing/controller.go

@@ -233,14 +233,11 @@ func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error {
 		Owns(&v1.Deployment{}, builder.WithPredicates(
 			predicate.Funcs{
 				UpdateFunc: func(e event.UpdateEvent) bool {
+					// Ignore the replicas equality check because the HPA changes it in the EPP deployment.
 					o := e.ObjectOld.(*v1.Deployment).DeepCopy()
 					n := e.ObjectNew.(*v1.Deployment).DeepCopy()
-
-					// Ignore the replicas during equality check
-					// because the HPA changes it in the EPP deployment.
 					o.Spec.Replicas = nil
 					n.Spec.Replicas = nil
-
 					return !reflect.DeepEqual(o, n)
 				},
 			},

pkg/eventing/deployment.go

@@ -50,7 +50,9 @@ var (
 func newNATSPublisherDeployment(
 	eventing *v1alpha1.Eventing,
 	natsConfig env.NATSConfig,
-	publisherConfig env.PublisherConfig) *appsv1.Deployment {
+	publisherConfig env.PublisherConfig,
+	replicas *int32,
+) *appsv1.Deployment {
 	return newDeployment(
 		eventing,
 		publisherConfig,
@@ -60,12 +62,15 @@ func newNATSPublisherDeployment(
 		WithLogEnvVars(publisherConfig, eventing),
 		WithAffinity(GetPublisherDeploymentName(*eventing)),
 		WithPriorityClassName(PriorityClassName),
+		WithSpecReplicas(replicas),
 	)
 }
 
 func newEventMeshPublisherDeployment(
 	eventing *v1alpha1.Eventing,
-	publisherConfig env.PublisherConfig) *appsv1.Deployment {
+	publisherConfig env.PublisherConfig,
+	replicas *int32,
+) *appsv1.Deployment {
 	return newDeployment(
 		eventing,
 		publisherConfig,
@@ -74,6 +79,7 @@ func newEventMeshPublisherDeployment(
 		WithBEBEnvVars(GetPublisherDeploymentName(*eventing), publisherConfig, eventing),
 		WithLogEnvVars(publisherConfig, eventing),
 		WithPriorityClassName(PriorityClassName),
+		WithSpecReplicas(replicas),
 	)
 }
 
@@ -373,3 +379,9 @@ func getEventMeshEnvVars(publisherName string, publisherConfig env.PublisherConf
 		},
 	}
 }
+
+func WithSpecReplicas(replicas *int32) DeployOpt {
+	return func(deployment *appsv1.Deployment) {
+		deployment.Spec.Replicas = replicas
+	}
+}

pkg/eventing/deployment_test.go

@@ -66,13 +66,13 @@ func TestNewDeployment(t *testing.T) {
 					testutils.WithEventingCRName(tc.givenPublisherName),
 					testutils.WithEventingCRNamespace(publisherNamespace),
 					testutils.WithEventingEventTypePrefix(eventTypePrefix),
-				), natsConfig, publisherConfig)
+				), natsConfig, publisherConfig, nil)
 			case "EventMesh":
 				deployment = newEventMeshPublisherDeployment(testutils.NewEventingCR(
 					testutils.WithEventingCRName(tc.givenPublisherName),
 					testutils.WithEventingCRNamespace(publisherNamespace),
 					testutils.WithEventMeshBackend("test-namespace/test-name"),
-				), publisherConfig)
+				), publisherConfig, nil)
 			default:
 				t.Errorf("Invalid backend!")
 			}

pkg/eventing/manager.go

@@ -7,6 +7,7 @@ import (
 	ecv1alpha1 "github.com/kyma-project/kyma/components/eventing-controller/api/v1alpha1"
 	appsv1 "k8s.io/api/apps/v1"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/utils/pointer"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
@@ -90,11 +91,19 @@ func (em *EventingManager) applyPublisherProxyDeployment(
 	backendType v1alpha1.BackendType) (*appsv1.Deployment, error) {
 	var desiredPublisher *appsv1.Deployment
 
+	var replicas *int32
+	hpa, err := em.kubeClient.GetHPA(ctx, GetPublisherDeploymentName(*eventing), eventing.Namespace)
+	if err == nil && hpa != nil {
+		replicas = hpa.Spec.MinReplicas
+	} else {
+		replicas = pointer.Int32(int32(eventing.Spec.Publisher.Replicas.Min))
+	}
+
 	switch backendType {
 	case v1alpha1.NatsBackendType:
-		desiredPublisher = newNATSPublisherDeployment(eventing, *natsConfig, em.backendConfig.PublisherConfig)
+		desiredPublisher = newNATSPublisherDeployment(eventing, *natsConfig, em.backendConfig.PublisherConfig, replicas)
 	case v1alpha1.EventMeshBackendType:
-		desiredPublisher = newEventMeshPublisherDeployment(eventing, em.backendConfig.PublisherConfig)
+		desiredPublisher = newEventMeshPublisherDeployment(eventing, em.backendConfig.PublisherConfig, replicas)
 	default:
 		return nil, fmt.Errorf("unknown EventingBackend type %q", backendType)
 	}

pkg/eventing/manager_test.go

@@ -108,6 +108,7 @@ func Test_ApplyPublisherProxyDeployment(t *testing.T) {
 			kubeClient.On("UpdateDeployment", ctx, mock.Anything).Return(nil)
 			kubeClient.On("Create", ctx, mock.Anything).Return(nil)
 			kubeClient.On("PatchApply", ctx, mock.Anything).Return(tc.patchApplyErr)
+			kubeClient.On("GetHPA", ctx, mock.Anything, mock.Anything).Return(nil, nil)
 
 			mockClient := fake.NewClientBuilder().WithScheme(newScheme).WithObjects().Build()
 

pkg/k8s/client.go

@@ -5,24 +5,24 @@ import (
 	"errors"
 	"strings"
 
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-
-	eventingv1alpha2 "github.com/kyma-project/kyma/components/eventing-controller/api/v1alpha2"
-
-	natsv1alpha1 "github.com/kyma-project/nats-manager/api/v1alpha1"
 	istiosec "istio.io/client-go/pkg/apis/security/v1beta1"
 	admissionv1 "k8s.io/api/admissionregistration/v1"
 	v1 "k8s.io/api/apps/v1"
+	v2 "k8s.io/api/autoscaling/v2"
 	corev1 "k8s.io/api/core/v1"
 	rbac "k8s.io/api/rbac/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	k8sclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	eventingv1alpha2 "github.com/kyma-project/kyma/components/eventing-controller/api/v1alpha2"
+	natsv1alpha1 "github.com/kyma-project/nats-manager/api/v1alpha1"
 )
 
 var NatsGVK = schema.GroupVersionResource{
@@ -52,6 +52,7 @@ type Client interface {
 	GetSubscriptions(ctx context.Context) (*eventingv1alpha2.SubscriptionList, error)
 	GetConfigMap(ctx context.Context, name, namespace string) (*corev1.ConfigMap, error)
 	PatchApplyPeerAuthentication(ctx context.Context, authentication *istiosec.PeerAuthentication) error
+	GetHPA(ctx context.Context, name, namespace string) (*v2.HorizontalPodAutoscaler, error)
 }
 
 type KubeClient struct {
@@ -271,3 +272,11 @@ func (c *KubeClient) GetConfigMap(ctx context.Context, name, namespace string) (
 	}
 	return cm, nil
 }
+
+func (c *KubeClient) GetHPA(ctx context.Context, name, namespace string) (*v2.HorizontalPodAutoscaler, error) {
+	hpa := &v2.HorizontalPodAutoscaler{}
+	if err := c.client.Get(ctx, client.ObjectKey{Name: name, Namespace: namespace}, hpa); err != nil {
+		return nil, err
+	}
+	return hpa, nil
+}

pkg/object/equality.go

@@ -99,8 +99,7 @@ func eventingBackendEqual(b1, b2 *eventingv1alpha1.EventingBackend) bool {
 }
 
 // publisherProxyDeploymentEqual asserts the equality of two Deployment objects
-// for event publisher proxy deployments. It ignores checking the equality of the
-// spec replicas because it is managed by the HPA.
+// for event publisher proxy deployments.
 func publisherProxyDeploymentEqual(d1, d2 *appsv1.Deployment) bool {
 	if d1 == nil || d2 == nil {
 		return false
@@ -111,6 +110,9 @@ func publisherProxyDeploymentEqual(d1, d2 *appsv1.Deployment) bool {
 	if !reflect.DeepEqual(d1.Labels, d2.Labels) {
 		return false
 	}
+	if !reflect.DeepEqual(d1.Spec.Replicas, d2.Spec.Replicas) {
+		return false
+	}
 
 	cst1 := d1.Spec.Template
 	cst2 := d2.Spec.Template

pkg/object/equality_test.go

@@ -531,6 +531,33 @@ func TestPublisherProxyDeploymentEqual(t *testing.T) {
 			},
 			expectedResult: false,
 		},
+		"should be unequal if replicas changes": {
+			getPublisher1: func() *appsv1.Deployment {
+				replicas := int32(2)
+				p := defaultNATSPublisher.DeepCopy()
+				p.Spec.Replicas = &replicas
+				return p
+			},
+			getPublisher2: func() *appsv1.Deployment {
+				return defaultNATSPublisher.DeepCopy()
+			},
+			expectedResult: false,
+		},
+		"should be equal if replicas are the same": {
+			getPublisher1: func() *appsv1.Deployment {
+				replicas := int32(2)
+				p := defaultNATSPublisher.DeepCopy()
+				p.Spec.Replicas = &replicas
+				return p
+			},
+			getPublisher2: func() *appsv1.Deployment {
+				replicas := int32(2)
+				p := defaultNATSPublisher.DeepCopy()
+				p.Spec.Replicas = &replicas
+				return p
+			},
+			expectedResult: true,
+		},
 		"should be equal if spec annotations are nil and empty": {
 			getPublisher1: func() *appsv1.Deployment {
 				p := defaultNATSPublisher.DeepCopy()