Repository: numaproj/numaflow
Pull request #1153 (2023-10-03 23:57:10)

fix: not considered as back pressured when onFull is discardLatest
Explain what this PR does.

<!--

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/go.mod b/go.mod
index 55a9b7cc..96bc3c33 100644
--- a/go.mod
+++ b/go.mod
@@ -27,7 +27,7 @@ require (
 	github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe
 	github.com/nats-io/nats-server/v2 v2.9.19
 	github.com/nats-io/nats.go v1.27.1
-	github.com/numaproj/numaflow-go v0.5.1-0.20230912211616-62600351d97f
+	github.com/numaproj/numaflow-go v0.5.1
 	github.com/prometheus/client_golang v1.14.0
 	github.com/prometheus/common v0.37.0
 	github.com/redis/go-redis/v9 v9.0.3
diff --git a/go.sum b/go.sum
index e6d65d65..ee2006f4 100644
--- a/go.sum
+++ b/go.sum
@@ -670,8 +670,8 @@ github.com/nats-io/nkeys v0.4.4/go.mod h1:XUkxdLPTufzlihbamfzQ7mw/VGx6ObUs+0bN5s
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
-github.com/numaproj/numaflow-go v0.5.1-0.20230912211616-62600351d97f h1:flJEIyki2WF5KuM4sRbvKQwkpEK6TL40lcwNTZqYfbY=
-github.com/numaproj/numaflow-go v0.5.1-0.20230912211616-62600351d97f/go.mod h1:5zwvvREIbqaCPCKsNE1MVjVToD0kvkCh2Z90Izlhw5U=
+github.com/numaproj/numaflow-go v0.5.1 h1:mvala+EmlrRtI20cr1y928zR7dO/HKUJsLai7vISHEA=
+github.com/numaproj/numaflow-go v0.5.1/go.mod h1:5zwvvREIbqaCPCKsNE1MVjVToD0kvkCh2Z90Izlhw5U=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/oklog/ulid v1.3.1 h1:EGfNDEx6MqHz8B3uNV6QAib1UR2Lm97sHi3ocA6ESJ4=
diff --git a/pkg/apis/numaflow/v1alpha1/vertex_types.go b/pkg/apis/numaflow/v1alpha1/vertex_types.go
index e807bb22..271a8871 100644
--- a/pkg/apis/numaflow/v1alpha1/vertex_types.go
+++ b/pkg/apis/numaflow/v1alpha1/vertex_types.go
@@ -109,15 +109,9 @@ func (v Vertex) Scalable() bool {
 	if v.Spec.Scale.Disabled || v.IsReduceUDF() {
 		return false
 	}
-	if v.IsASink() || v.IsMapUDF() {
+	if v.IsASink() || v.IsMapUDF() || v.IsASource() {
 		return true
 	}
-	if v.IsASource() {
-		src := v.Spec.Source
-		if src.Kafka != nil || src.UDSource != nil {
-			return true
-		}
-	}
 	return false
 }
 
diff --git a/pkg/reconciler/vertex/scaling/scaling.go b/pkg/reconciler/vertex/scaling/scaling.go
index b8c6c2c4..93dbfb25 100644
--- a/pkg/reconciler/vertex/scaling/scaling.go
+++ b/pkg/reconciler/vertex/scaling/scaling.go
@@ -166,7 +166,7 @@ func (s *Scaler) scaleOneVertex(ctx context.Context, key string, worker int) err
 		log.Debug("Vertex being deleted.")
 		return nil
 	}
-	if !vertex.Scalable() { // A vertex which is not scalable, such as HTTP source, or autoscaling disabled.
+	if !vertex.Scalable() { // A vertex which is not scalable, such as a Reducer, or autoscaling disabled.
 		s.StopWatching(key) // Remove it in case it's watched.
 		return nil
 	}
@@ -456,6 +456,10 @@ func (s *Scaler) hasBackPressure(pl dfv1.Pipeline, vertex dfv1.Vertex) (bool, bo
 	directPressure, downstreamPressure := false, false
 loop:
 	for _, e := range downstreamEdges {
+		if e.BufferFullWritingStrategy() == dfv1.DiscardLatest {
+			// If the edge is configured to discard latest on full, we don't consider it as back pressure.
+			continue
+		}
 		vertexKey := pl.Namespace + "/" + pl.Name + "-" + e.To
 		pendingVal, ok := s.vertexMetricsCache.Get(vertexKey + "/pending")
 		if !ok { // Vertex key has not been cached, skip it.
diff --git a/pkg/sources/generator/tickgen.go b/pkg/sources/generator/tickgen.go
index 202075b9..9ecb6606 100644
--- a/pkg/sources/generator/tickgen.go
+++ b/pkg/sources/generator/tickgen.go
@@ -272,6 +272,10 @@ loop:
 	return msgs, nil
 }
 
+func (mg *memgen) Pending(_ context.Context) (int64, error) {
+	return isb.PendingNotAvailable, nil
+}
+
 func (mg *memgen) PublishSourceWatermarks(msgs []*isb.ReadMessage) {
 	if len(msgs) <= 0 {
 		return
diff --git a/pkg/sources/http/http.go b/pkg/sources/http/http.go
index 85520df2..ec01b264 100644
--- a/pkg/sources/http/http.go
+++ b/pkg/sources/http/http.go
@@ -243,6 +243,10 @@ loop:
 	return msgs, nil
 }
 
+func (h *httpSource) Pending(_ context.Context) (int64, error) {
+	return isb.PendingNotAvailable, nil
+}
+
 func (h *httpSource) PublishSourceWatermarks(msgs []*isb.ReadMessage) {
 	var oldest time.Time
 	for _, m := range msgs {
diff --git a/pkg/sources/nats/nats.go b/pkg/sources/nats/nats.go
index 37c94be8..89b4ecb3 100644
--- a/pkg/sources/nats/nats.go
+++ b/pkg/sources/nats/nats.go
@@ -240,6 +240,10 @@ loop:
 	return msgs, nil
 }
 
+func (ns *natsSource) Pending(_ context.Context) (int64, error) {
+	return isb.PendingNotAvailable, nil
+}
+
 func (ns *natsSource) PublishSourceWatermarks(msgs []*isb.ReadMessage) {
 	var oldest time.Time
 	for _, m := range msgs {
diff --git a/pkg/sources/sourcer.go b/pkg/sources/sourcer.go
index b712825c..6c1b24f4 100644
--- a/pkg/sources/sourcer.go
+++ b/pkg/sources/sourcer.go
@@ -26,4 +26,6 @@ import (
 type Sourcer interface {
 	isb.BufferReader
 	forward.StarterStopper
+	isb.LagReader
+	isb.SourceWatermarkPublisher
 }
diff --git a/ui/src/App.tsx b/ui/src/App.tsx
index e037eda3..446ead89 100644
--- a/ui/src/App.tsx
+++ b/ui/src/App.tsx
@@ -57,9 +57,9 @@ function App() {
       return;
     }
     const resizeObserver = new ResizeObserver(() => {
-      setPageWidth(pageRef.current.offsetWidth);
+      setPageWidth(pageRef?.current?.offsetWidth);
     });
-    resizeObserver.observe(pageRef.current);
+    resizeObserver.observe(pageRef?.current);
     return function cleanup() {
       resizeObserver.disconnect();
     };
diff --git a/ui/src/components/common/SummaryPageLayout/index.tsx b/ui/src/components/common/SummaryPageLayout/index.tsx
index bfdc5fd5..2a6b69a9 100644
--- a/ui/src/components/common/SummaryPageLayout/index.tsx
+++ b/ui/src/components/common/SummaryPageLayout/index.tsx
@@ -200,9 +200,9 @@ export function SummaryPageLayout({
       return;
     }
     const resizeObserver = new ResizeObserver(() => {
-      setSummaryHeight(sumaryRef.current.offsetHeight);
+      setSummaryHeight(sumaryRef?.current?.offsetHeight);
     });
-    resizeObserver.observe(sumaryRef.current);
+    resizeObserver.observe(sumaryRef?.current);
     return function cleanup() {
       resizeObserver.disconnect();
     };
diff --git a/ui/src/components/pages/Namespace/partials/PipelineCard/index.tsx b/ui/src/components/pages/Namespace/partials/PipelineCard/index.tsx
index bae8b495..295e55ec 100644
--- a/ui/src/components/pages/Namespace/partials/PipelineCard/index.tsx
+++ b/ui/src/components/pages/Namespace/partials/PipelineCard/index.tsx
@@ -10,7 +10,13 @@ import {
   Select,
   SelectChangeEvent,
 } from "@mui/material";
-import {IconsStatusMap, ISBStatusString, StatusString} from "../../../../../utils";
+import {
+  GetISBType,
+  IconsStatusMap,
+  ISBStatusString,
+  StatusString,
+  UNKNOWN,
+} from "../../../../../utils";
 import { AppContextProps } from "../../../../../types/declarations/app";
 import { AppContext } from "../../../../../App";
 import { SidebarType } from "../../../../common/SlidingSidebar";
@@ -51,6 +57,11 @@ export function PipelineCard({
   const handleDeleteChange = useCallback((event: SelectChangeEvent<string>) => {
     setDeleteOption(event.target.value);
   }, []);
+
+  const isbType = GetISBType(isbData?.isbService?.spec) || UNKNOWN;
+  const isbStatus = isbData?.isbService?.status?.phase || UNKNOWN;
+  const pipelineStatus = statusData?.pipeline?.status?.phase || UNKNOWN;
+
   return (
     <>
       <Paper
@@ -150,13 +161,13 @@ export function PipelineCard({
               }}
             >
               <img
-                src={IconsStatusMap[statusData?.pipeline?.status?.phase]}
-                alt={statusData?.pipeline?.status?.phase}
+                src={IconsStatusMap[pipelineStatus]}
+                alt="Status"
                 className={"pipeline-logo"}
               />
               <img
                 src={IconsStatusMap[statusData?.status]}
-                alt={statusData?.status}
+                alt="Health"
                 className={"pipeline-logo"}
               />
             </Box>
@@ -168,7 +179,7 @@ export function PipelineCard({
                 paddingLeft: "1rem",
               }}
             >
-              <span>{StatusString[statusData?.pipeline?.status?.phase]}</span>
+              <span>{StatusString[pipelineStatus]}</span>
               <span>{StatusString[statusData?.status]}</span>
             </Box>
           </Grid>
@@ -203,12 +214,11 @@ export function PipelineCard({
               }}
             >
               <span>{isbData?.name}</span>
-              <span>{isbData?.isbService?.status?.type}</span>
+              <span>{isbType}</span>
               <span>
-                {
-                  isbData?.isbService?.spec[isbData?.isbService?.status?.type]
-                    .replicas
-                }
+                {isbType && isbData?.isbService?.spec[isbType]
+                  ? isbData?.isbService?.spec[isbType].replicas
+                  : UNKNOWN}
               </span>
             </Box>
           </Grid>
@@ -242,13 +252,13 @@ export function PipelineCard({
               }}
             >
               <img
-                src={IconsStatusMap[isbData?.isbService?.status?.phase]}
-                alt={isbData?.isbService?.status?.phase}
+                src={IconsStatusMap[isbStatus]}
+                alt="Status"
                 className={"pipeline-logo"}
               />
               <img
                 src={IconsStatusMap[isbData?.status]}
-                alt={isbData?.status}
+                alt="Health"
                 className={"pipeline-logo"}
               />
             </Box>
@@ -260,7 +270,7 @@ export function PipelineCard({
                 paddingLeft: "1rem",
               }}
             >
-              <span>{ISBStatusString[isbData?.isbService?.status?.phase]}</span>
+              <span>{ISBStatusString[isbStatus]}</span>
               <span>{ISBStatusString[isbData?.status]}</span>
             </Box>
           </Grid>
diff --git a/ui/src/components/pages/Pipeline/index.tsx b/ui/src/components/pages/Pipeline/index.tsx
index 2fad0130..a62a7470 100644
--- a/ui/src/components/pages/Pipeline/index.tsx
+++ b/ui/src/components/pages/Pipeline/index.tsx
@@ -16,6 +16,7 @@ import { PipelineISBStatus } from "./partials/PipelineISBStatus";
 import { SidebarType } from "../../common/SlidingSidebar";
 import { AppContextProps } from "../../../types/declarations/app";
 import { AppContext } from "../../../App";
+import { UNKNOWN } from "../../../utils";
 import noError from "../../../images/no-error.svg";
 
 import "./style.css";
@@ -32,6 +33,7 @@ export function Pipeline() {
     }
     const pipelineData = data?.pipelineData;
     const isbData = data?.isbData;
+    const pipelineStatus = pipelineData?.pipeline?.status?.phase || UNKNOWN;
     return [
       // pipeline collection
       {
@@ -41,7 +43,7 @@ export function Pipeline() {
             type: SummarySectionType.CUSTOM,
             customComponent: (
               <PipelineStatus
-                status={pipelineData?.pipeline?.status?.phase}
+                status={pipelineStatus}
                 healthStatus={pipelineData?.status}
                 key={"pipeline-status"}
               />
diff --git a/ui/src/components/pages/Pipeline/partials/PipelineISBStatus/index.tsx b/ui/src/components/pages/Pipeline/partials/PipelineISBStatus/index.tsx
index c3722b34..2feff768 100644
--- a/ui/src/components/pages/Pipeline/partials/PipelineISBStatus/index.tsx
+++ b/ui/src/components/pages/Pipeline/partials/PipelineISBStatus/index.tsx
@@ -1,10 +1,17 @@
 import React from "react";
 import Box from "@mui/material/Box";
-import {IconsStatusMap, ISBStatusString} from "../../../../../utils";
+import {
+  GetISBType,
+  IconsStatusMap,
+  ISBStatusString,
+  UNKNOWN,
+} from "../../../../../utils";
 
 import "./style.css";
 
 export function PipelineISBStatus({ isbData }) {
+  const isbType = GetISBType(isbData?.isbService?.spec) || UNKNOWN;
+  const isbStatus = isbData?.isbService?.status?.phase || UNKNOWN;
   return (
     <Box
       sx={{
@@ -28,13 +35,13 @@ export function PipelineISBStatus({ isbData }) {
           >
             <Box sx={{ display: "flex", flexDirection: "column" }}>
               <img
-                src={IconsStatusMap[isbData?.isbService?.status?.phase]}
-                alt={IconsStatusMap[isbData?.isbService?.status?.phase]}
+                src={IconsStatusMap[isbStatus]}
+                alt="Status"
                 className={"pipeline-logo"}
               />
               <img
                 src={IconsStatusMap[isbData?.status]}
-                alt={isbData?.status}
+                alt="Health"
                 className={"pipeline-logo"}
               />
             </Box>
@@ -46,9 +53,11 @@ export function PipelineISBStatus({ isbData }) {
               }}
             >
               <span className="pipeline-logo-text">
-                {ISBStatusString[isbData?.isbService?.status?.phase]}
+                {ISBStatusString[isbStatus]}
+              </span>
+              <span className="pipeline-logo-text">
+                {ISBStatusString[isbData?.status]}
               </span>
-              <span className="pipeline-logo-text">{ISBStatusString[isbData?.status]}</span>
             </Box>
           </Box>
         </Box>
@@ -86,18 +95,14 @@ export function PipelineISBStatus({ isbData }) {
                 <span>{isbData?.name}</span>
               </div>
               <div className="isb-status-text">
-                <span>{isbData?.isbService?.status?.type}</span>
-                {/*<Chip*/}
-                {/*  label={isbData?.isbService?.status?.type}*/}
-                {/*  sx={{ height: "20px" }}*/}
-                {/*/>*/}
+                <span>{isbType}</span>
               </div>
             </Box>
             <Box
               sx={{
                 display: "flex",
                 flexDirection: "column",
-                marginLeft: "2rem"
+                marginLeft: "2rem",
               }}
             >
               <div className="pipeline-summary-text">
@@ -105,11 +110,9 @@ export function PipelineISBStatus({ isbData }) {
                   <div className="pipeline-summary-text">
                     <span className="pipeline-summary-subtitle">Size: </span>
                     <span>
-                      {
-                        isbData?.isbService?.spec[
-                          isbData?.isbService?.status?.type
-                        ]?.replicas
-                      }
+                      {isbType && isbData?.isbService?.spec[isbType]
+                        ? isbData?.isbService?.spec[isbType].replicas
+                        : UNKNOWN}
                     </span>
                   </div>
                 </span>
diff --git a/ui/src/components/pages/Pipeline/partials/PipelineStatus/index.tsx b/ui/src/components/pages/Pipeline/partials/PipelineStatus/index.tsx
index 2bad988b..e83d31a5 100644
--- a/ui/src/components/pages/Pipeline/partials/PipelineStatus/index.tsx
+++ b/ui/src/components/pages/Pipeline/partials/PipelineStatus/index.tsx
@@ -1,6 +1,6 @@
 import React from "react";
 import Box from "@mui/material/Box";
-import {IconsStatusMap, StatusString} from "../../../../../utils";
+import { IconsStatusMap, StatusString } from "../../../../../utils";
 
 import "./style.css";
 
diff --git a/ui/src/types/declarations/pipeline.d.ts b/ui/src/types/declarations/pipeline.d.ts
index 4d976912..9f2b2521 100644
--- a/ui/src/types/declarations/pipeline.d.ts
+++ b/ui/src/types/declarations/pipeline.d.ts
@@ -87,6 +87,7 @@ export interface IsbServiceStatus {
 
 export interface IsbServiceSpec {
   jetstream?: Jetstream;
+  redis?: Redis;
 }
 
 export interface PipelineSummary {
diff --git a/ui/src/utils/fetchWrappers/namespaceSummaryFetch.ts b/ui/src/utils/fetchWrappers/namespaceSummaryFetch.ts
index b91f5a8e..0ec9a1e9 100644
--- a/ui/src/utils/fetchWrappers/namespaceSummaryFetch.ts
+++ b/ui/src/utils/fetchWrappers/namespaceSummaryFetch.ts
@@ -116,11 +116,7 @@ export const useNamespaceSummaryFetch = ({
     data: pipelineData,
     loading: pipelineLoading,
     error: pipelineError,
-  } = useFetch(
-    `/api/v1/namespaces/${namespace}/pipelines`,
-    undefined,
-    options
-  );
+  } = useFetch(`/api/v1/namespaces/${namespace}/pipelines`, undefined, options);
   const {
     data: isbData,
     loading: isbLoading,
@@ -169,11 +165,11 @@ export const useNamespaceSummaryFetch = ({
       return;
     }
     if (pipelineData && isbData) {
-      const pipeLineMap = pipelineData.data.reduce((map: any, obj: any) => {
+      const pipeLineMap = pipelineData?.data?.reduce((map: any, obj: any) => {
         map[obj.name] = obj;
         return map;
       }, {});
-      const isbMap = isbData.data.reduce((map: any, obj: any) => {
+      const isbMap = isbData?.data?.reduce((map: any, obj: any) => {
         map[obj.name] = obj;
         return map;
       }, {});
@@ -183,8 +179,8 @@ export const useNamespaceSummaryFetch = ({
       //   MOCK_ISB_DATA
       // );
       const nsSummary = rawDataToNamespaceSummary(
-        pipelineData.data,
-        isbData.data
+        pipelineData?.data,
+        isbData?.data
       );
       setResults({
         data: nsSummary,
diff --git a/ui/src/utils/index.ts b/ui/src/utils/index.ts
index a7285f9c..bdcb9694 100644
--- a/ui/src/utils/index.ts
+++ b/ui/src/utils/index.ts
@@ -5,6 +5,7 @@ import heartFill from "../../src/images/heart-fill.png";
 import warning from "../../src/images/warning-circle.png";
 import critical from "../../src/images/critical.png";
 import moment from "moment";
+import { IsbServiceSpec } from "../types/declarations/pipeline";
 
 // global constants
 export const RUNNING = "Running";
@@ -19,6 +20,11 @@ export const PENDING = "Pending";
 export const PAUSING = "Pausing";
 export const PAUSED = "Paused";
 export const DELETING = "Deleting";
+export const UNKNOWN = "Unknown";
+
+// ISB types
+export const JETSTREAM = "jetstream";
+export const REDIS = "redis";
 
 export function getBaseHref(): string {
   if (window.__RUNTIME_CONFIG__?.BASE_HREF) {
@@ -179,6 +185,7 @@ export const IconsStatusMap = {
   [HEALTHY]: heartFill,
   [WARNING]: warning,
   [CRITICAL]: critical,
+  [UNKNOWN]: circleDash,
 };
 
 interface StatusStringType {
@@ -198,6 +205,7 @@ export const StatusString: StatusStringType = {
   [HEALTHY]: "Healthy",
   [WARNING]: "Warning",
   [CRITICAL]: "Critical",
+  [UNKNOWN]: "Unknown",
 };
 
 export const ISBStatusString: StatusStringType = {
@@ -207,6 +215,7 @@ export const ISBStatusString: StatusStringType = {
   [HEALTHY]: "Healthy",
   [WARNING]: "Warning",
   [CRITICAL]: "Critical",
+  [UNKNOWN]: "Unknown",
 };
 
 // returns the duration string in the format of 1d 2hr 3min 4sec 5ms
@@ -236,3 +245,12 @@ export const DurationString = (duration: number): string => {
     return `${milliseconds}ms`;
   }
 };
+
+export const GetISBType = (spec: IsbServiceSpec): string | null => {
+  if (spec?.jetstream) {
+    return JETSTREAM;
+  } else if (spec?.redis) {
+    return REDIS;
+  }
+  return null;
+};


Test patch:
diff --git a/pkg/apis/numaflow/v1alpha1/vertex_types_test.go b/pkg/apis/numaflow/v1alpha1/vertex_types_test.go
index 5a879ef4..96232dd3 100644
--- a/pkg/apis/numaflow/v1alpha1/vertex_types_test.go
+++ b/pkg/apis/numaflow/v1alpha1/vertex_types_test.go
@@ -526,11 +526,15 @@ func TestScalable(t *testing.T) {
 	v.Spec.Sink = nil
 	v.Spec.UDF = &UDF{}
 	assert.True(t, v.Scalable())
+	v.Spec.UDF = &UDF{
+		GroupBy: &GroupBy{},
+	}
+	assert.False(t, v.Scalable())
 	v.Spec.UDF = nil
 	v.Spec.Source = &Source{
 		HTTP: &HTTPSource{},
 	}
-	assert.False(t, v.Scalable())
+	assert.True(t, v.Scalable())
 	v.Spec.Source = &Source{
 		Kafka: &KafkaSource{},
 	}
diff --git a/pkg/reconciler/pipeline/controller_test.go b/pkg/reconciler/pipeline/controller_test.go
index 1dbb0e20..2a30e7ca 100644
--- a/pkg/reconciler/pipeline/controller_test.go
+++ b/pkg/reconciler/pipeline/controller_test.go
@@ -21,8 +21,6 @@ import (
 	"testing"
 
 	"github.com/goccy/go-json"
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/reconciler"
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap/zaptest"
 	appv1 "k8s.io/api/apps/v1"
@@ -37,6 +35,9 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/reconciler"
 )
 
 const (
@@ -250,7 +251,9 @@ func Test_pauseAndResumePipeline(t *testing.T) {
 	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)
+	// when auto-scaling is enabled, while resuming the pipeline, instead of setting the replicas to Scale.Min,
+	// we set it to one and let auto-scaling to scale up
+	assert.Equal(t, int32(1), *v[testObj.Name+"-"+testObj.Spec.Vertices[0].Name].Spec.Replicas)
 	assert.NoError(t, err)
 }
 
diff --git a/test/udsource-e2e/testdata/simple-source-go.yaml b/test/udsource-e2e/testdata/simple-source-go.yaml
index 41abaa6f..b7a51e3d 100644
--- a/test/udsource-e2e/testdata/simple-source-go.yaml
+++ b/test/udsource-e2e/testdata/simple-source-go.yaml
@@ -10,7 +10,7 @@ spec:
           container:
             # A simple user-defined source for e2e testing
             # See https://github.com/numaproj/numaflow-go/tree/main/pkg/sourcer/examples/simple_source
-            image: quay.io/numaio/numaflow-go/source-simple-source:v0.5.0
+            image: quay.io/numaio/numaflow-go/source-simple-source:v0.5.1
       limits:
         readBatchSize: 2
     - name: out