feat: terminate reduce vertex pods when pausing pipeline

[dpadhiar]

Explain what this PR does.

Fixes #990.

When pausing a pipeline currently, all vertex pods will be determined as replicas is set to 0 for each. However, this isn't the case for any reduce vertices as they use partitions to determine replicas for the vertex.

To remedy this, we change the logic for checking replicas of a reduce vertex to respect replicas only if the value is 0. The only case it will be 0 is when the pipeline is paused, otherwise it will respect the partitions value which defaults to 1.

if v.IsReduceUDF() {
	// Replicas will be 0 only when pausing a pipeline
	if v.Spec.Replicas != nil && int(*v.Spec.Replicas) == 0 {
		return 0
	}
	// Replica of a reduce vertex is determined by the partitions.
	return v.GetPartitionCount()
}

[KeranYang]

Suggested change

	// if vtx does not support autoscaling and min is set, scale up to min
	// if replicas equals to 1, it means we are resuming a paused pipeline.
	// in this case, if a vertex doesn't support auto-scaling, we scale up based on the vertex's configuration:
	// for a reducer, we scale up to the partition count.
	// for a non-reducer, if min is set, we scale up to min.

[KeranYang]

Suggested change

// reduce UDF uses partition count to determine replica count