chore: partitions for source should always be 1 (#811)

Signed-off-by: Derek Wang <[email protected]>
numaproj · Jun 28, 2023 · 10cec59 · 10cec59
1 parent dd060cb
commit 10cec59
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 4 deletions.
diff --git a/pkg/apis/numaflow/v1alpha1/vertex_types.go b/pkg/apis/numaflow/v1alpha1/vertex_types.go
@@ -438,7 +438,7 @@ func (av AbstractVertex) GetPartitionCount() int {
 	if av.Partitions == nil || *av.Partitions < 1 {
 		return 1
 	}
-	if av.IsReduceUDF() && !av.UDF.GroupBy.Keyed {
+	if av.IsASource() || (av.IsReduceUDF() && !av.UDF.GroupBy.Keyed) {
 		return 1
 	}
 	return int(*av.Partitions)

diff --git a/pkg/reconciler/vertex/scaling/scaling.go b/pkg/reconciler/vertex/scaling/scaling.go
@@ -268,16 +268,16 @@ func (s *Scaler) scaleOneVertex(ctx context.Context, key string, worker int) err
 	}
 	current := int32(vertex.GetReplicas())
 	desired := s.desiredReplicas(ctx, vertex, totalRate, totalPending, totalBufferLength, targetAvailableBufferLength)
-	log.Debugf("Calculated desired replica number of vertex %q is: %t", vertex.Name, desired)
+	log.Debugf("Calculated desired replica number of vertex %q is: %d", vertex.Name, desired)
 	max := vertex.Spec.Scale.GetMaxReplicas()
 	min := vertex.Spec.Scale.GetMinReplicas()
 	if desired > max {
 		desired = max
-		log.Debugf("Calculated desired replica number %t of vertex %q is greater than max, using max %t", vertex.Name, desired, max)
+		log.Debugf("Calculated desired replica number %d of vertex %q is greater than max, using max %d", vertex.Name, desired, max)
 	}
 	if desired < min {
 		desired = min
-		log.Debugf("Calculated desired replica number %t of vertex %q is smaller than min, using min %t", vertex.Name, desired, min)
+		log.Debugf("Calculated desired replica number %d of vertex %q is smaller than min, using min %d", vertex.Name, desired, min)
 	}
 	if current > max || current < min { // Someone might have manually scaled up/down the vertex
 		return s.patchVertexReplicas(ctx, vertex, desired)