Repository: numaproj/numaflow (go)
Issue: Resuming a paused pipeline sets incorrect replicas for some type of vertices
**Describe the bug**

If a pipeline with an HTTP source is paused, when resuming it, it only brings the HTTP source replica back to 1, even if its `spec.scale.min` is set to a number > 1.

When we resume a pipeline, we always set the replica number to 1, and let the autoscaling kick in to bring the number back to normal, but it does not work for sources like HTTP, because they do not support autoscaling.

---

<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->

**Message from the maintainers**:

Impacted by this bug? Give it a 👍. We often sort issues this way to know what to prioritize.

Pull request description: feat: set correct number of replices for some types of vertices after resuming pipeline
Fixes #991 

Currently if a user resumes a paused pipeline with a vertex that does not support autoscaling, the replica count for the vertex will always be 1 even if `spec.scale.min` is set greater than 1. This is because the work for scaling the vertex is expected to be done by an autoscaler. This is fixed by checking that a vertex is not scalable and has min set when resuming the pipeline to assure the correct replica count is set.

<!--

Before you push your changes:

* Run `make pre-push -B` to fix codegen and lint problems.

Then, you MUST:

* Sign-off your commit (otherwise the DCO check will fail).
* Use [a conventional commit message](https://www.conventionalcommits.org/en/v1.0.0/) (otherwise the commit message check will fail).

If you did not do this, reset all your commit and replace them with a single commit:

```
git reset HEAD~1 ;# change 1 to how many commits you made
git commit --signoff -m 'feat: my feat. Fixes #1234'
```

When creating your PR: 

* Make sure that "Fixes #" or "Closes #" is in both the PR title (for release notes) and description (to automatically link and close the issue).
* Say how you tested your changes. If you changed the UI, attach screenshots.
* Set your PR as a draft initially.
* Your PR needs to pass the required checks before it can be approved. 
* Once required tests have passed, mark your PR "Ready for review".

If changes were requested, once you've made them, you MUST dismiss the review to get it reviewed again.

-->


Patch:
diff --git a/pkg/reconciler/pipeline/controller.go b/pkg/reconciler/pipeline/controller.go
index 7e7ce324..a406aaec 100644
--- a/pkg/reconciler/pipeline/controller.go
+++ b/pkg/reconciler/pipeline/controller.go
@@ -785,7 +785,14 @@ func (r *pipelineReconciler) scaleVertex(ctx context.Context, pl *dfv1.Pipeline,
 	isVertexPatched := false
 	for _, vertex := range existingVertices {
 		if origin := *vertex.Spec.Replicas; origin != replicas && filter(vertex) {
-			vertex.Spec.Replicas = pointer.Int32(replicas)
+			scaleTo := replicas
+			// if vtx does not support autoscaling and min is set, scale up to min
+			if replicas == 1 {
+				if !vertex.Scalable() && vertex.Spec.Scale.Min != nil && *vertex.Spec.Scale.Min > 1 {
+					scaleTo = *vertex.Spec.Scale.Min
+				}
+			}
+			vertex.Spec.Replicas = pointer.Int32(scaleTo)
 			body, err := json.Marshal(vertex)
 			if err != nil {
 				return false, err
@@ -794,8 +801,8 @@ func (r *pipelineReconciler) scaleVertex(ctx context.Context, pl *dfv1.Pipeline,
 			if err != nil && !apierrors.IsNotFound(err) {
 				return false, err
 			}
-			log.Infow("Scaled vertex", zap.Int32("from", origin), zap.Int32("to", replicas), zap.String("vertex", vertex.Name))
-			r.recorder.Eventf(pl, corev1.EventTypeNormal, "ScalingVertex", "Scaled vertex %s from %d to %d replicas", vertex.Name, origin, replicas)
+			log.Infow("Scaled vertex", zap.Int32("from", origin), zap.Int32("to", scaleTo), zap.String("vertex", vertex.Name))
+			r.recorder.Eventf(pl, corev1.EventTypeNormal, "ScalingVertex", "Scaled vertex %s from %d to %d replicas", vertex.Name, origin, scaleTo)
 			isVertexPatched = true
 		}
 	}


Test patch:
diff --git a/pkg/reconciler/pipeline/controller_test.go b/pkg/reconciler/pipeline/controller_test.go
index f6bd6bbe..1dbb0e20 100644
--- a/pkg/reconciler/pipeline/controller_test.go
+++ b/pkg/reconciler/pipeline/controller_test.go
@@ -221,6 +221,39 @@ func Test_buildReducesVertices(t *testing.T) {
 	assert.Equal(t, int32(2), *r[pl.Name+"-"+pl.Spec.Vertices[1].Name].Spec.Replicas)
 }
 
+func Test_pauseAndResumePipeline(t *testing.T) {
+	cl := fake.NewClientBuilder().Build()
+	ctx := context.TODO()
+	testIsbSvc := testNativeRedisIsbSvc.DeepCopy()
+	testIsbSvc.Status.MarkConfigured()
+	testIsbSvc.Status.MarkDeployed()
+	err := cl.Create(ctx, testIsbSvc)
+	assert.Nil(t, err)
+	r := &pipelineReconciler{
+		client:   cl,
+		scheme:   scheme.Scheme,
+		config:   fakeConfig,
+		image:    testFlowImage,
+		logger:   zaptest.NewLogger(t).Sugar(),
+		recorder: record.NewFakeRecorder(64),
+	}
+	testObj := testPipeline.DeepCopy()
+	testObj.Spec.Vertices[0].Scale.Min = pointer.Int32(3)
+	_, err = r.reconcile(ctx, testObj)
+	assert.NoError(t, err)
+	_, err = r.pausePipeline(ctx, testObj)
+	assert.NoError(t, err)
+	v, err := r.findExistingVertices(ctx, testObj)
+	assert.NoError(t, err)
+	assert.Equal(t, int32(0), *v[testObj.Name+"-"+testObj.Spec.Vertices[0].Name].Spec.Replicas)
+	_, err = r.resumePipeline(ctx, testObj)
+	assert.NoError(t, err)
+	v, err = r.findExistingVertices(ctx, testObj)
+	assert.NoError(t, err)
+	assert.Equal(t, int32(3), *v[testObj.Name+"-"+testObj.Spec.Vertices[0].Name].Spec.Replicas)
+	assert.NoError(t, err)
+}
+
 func Test_copyVertexLimits(t *testing.T) {
 	pl := testPipeline.DeepCopy()
 	v := pl.Spec.Vertices[0].DeepCopy()