Repository: numaproj/numaflow
Pull request #1494 (2024-02-07 22:56:45)

add separate server info file paths for services on client side
Implement #1456 on client side

<!--

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.

-->


KeranYang: @a3hadi , with this change, the platform stops supporting all the old UDF images (hence all e2e fail.). Users have to re-build UDF/Source/Sink images using the latest SDKs. Since server info hasn't been widely used, let's see if we can update on the platform side in a way that when a server info file is not found because of incompatible file path, the platform numa container can decide to ignore it and continue processing.
KeranYang: @a3hadi , could you also look into why the udsource e2e succeeded? Thanks!
a3hadi: > @a3hadi , with this change, the platform stops supporting all the old UDF images (hence all e2e fail.). Users have to re-build UDF/Source/Sink images using the latest SDKs. Since server info hasn't been widely used, let's see if we can update on the platform side in a way that when a server info file is not found because of incompatible file path, the platform numa container can decide to ignore it and continue processing.

I think ignoring incompatible server info path, with a message to remind to upgrade sdk is a good idea (since using the new server info paths isn't mission critical to continue processing anyways). I will focus on getting the current tests to pass first and then look into this.
a3hadi: After discussing with Keran, I think the best course of action for the "error handling" is simply, if the server info file path is not found, set it to the default path of `/var/run/numaflow/server-info`. This is because, the server info file will always be created. However there are two possible paths that it can have. If an old sdk version is being used it will be the aforementioned default path, if the sdk version is up to date it will be using the new isolated paths for each service. Thus falling back to the default path when the server info file path is not found will accommodate older sdk versions.

Patch:
diff --git a/go.mod b/go.mod
index 04216e29..be5a508a 100644
--- a/go.mod
+++ b/go.mod
@@ -30,7 +30,7 @@ require (
 	github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe
 	github.com/nats-io/nats-server/v2 v2.10.4
 	github.com/nats-io/nats.go v1.32.0
-	github.com/numaproj/numaflow-go v0.5.3-0.20231213060340-dbd9016bbcb6
+	github.com/numaproj/numaflow-go v0.6.1-0.20240212202512-715955f1e069
 	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 836b0a8b..b190e8f1 100644
--- a/go.sum
+++ b/go.sum
@@ -681,8 +681,8 @@ github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDm
 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.3-0.20231213060340-dbd9016bbcb6 h1:NZVMu8Olrlo1vXe3D0Uj4nG5zRPZOlLqSPQH4Vl14zY=
-github.com/numaproj/numaflow-go v0.5.3-0.20231213060340-dbd9016bbcb6/go.mod h1:WoMt31+h3up202zTRI8c/qe42B8UbvwLe2mJH0MAlhI=
+github.com/numaproj/numaflow-go v0.6.1-0.20240212202512-715955f1e069 h1:LBMIrUytoxuZazEnwxo+WPAxjHL1rrVhphTyByLiDYY=
+github.com/numaproj/numaflow-go v0.6.1-0.20240212202512-715955f1e069/go.mod h1:WoMt31+h3up202zTRI8c/qe42B8UbvwLe2mJH0MAlhI=
 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/sdkclient/const.go b/pkg/sdkclient/const.go
index 4862e4c9..62b10d53 100644
--- a/pkg/sdkclient/const.go
+++ b/pkg/sdkclient/const.go
@@ -2,17 +2,19 @@ package sdkclient
 
 const (
 	UDS                       = "unix"
-	MapAddr                   = "/var/run/numaflow/map.sock"
-	ReduceAddr                = "/var/run/numaflow/reduce.sock"
-	ReduceStreamAddr          = "/var/run/numaflow/reducestream.sock"
-	SessionReduceAddr         = "/var/run/numaflow/sessionreduce.sock"
-	GlobalReduceAddr          = "/var/run/numaflow/globalreduce.sock"
-	MapStreamAddr             = "/var/run/numaflow/mapstream.sock"
-	SourceAddr                = "/var/run/numaflow/source.sock"
-	SourceTransformerAddr     = "/var/run/numaflow/sourcetransform.sock"
-	SideInputAddr             = "/var/run/numaflow/sideinput.sock"
-	SinkAddr                  = "/var/run/numaflow/sink.sock"
 	WinStartTime              = "x-numaflow-win-start-time"
 	WinEndTime                = "x-numaflow-win-end-time"
 	DefaultGRPCMaxMessageSize = 64 * 1024 * 1024
+	GlobalReduceAddr          = "/var/run/numaflow/globalreduce.sock"
+
+	// Socket configs
+	MapAddr               = "/var/run/numaflow/map.sock"
+	MapStreamAddr         = "/var/run/numaflow/mapstream.sock"
+	ReduceAddr            = "/var/run/numaflow/reduce.sock"
+	ReduceStreamAddr      = "/var/run/numaflow/reducestream.sock"
+	SessionReduceAddr     = "/var/run/numaflow/sessionreduce.sock"
+	SideInputAddr         = "/var/run/numaflow/sideinput.sock"
+	SinkAddr              = "/var/run/numaflow/sink.sock"
+	SourceAddr            = "/var/run/numaflow/source.sock"
+	SourceTransformerAddr = "/var/run/numaflow/sourcetransform.sock"
 )
diff --git a/pkg/sdkclient/sinker/client.go b/pkg/sdkclient/sinker/client.go
index 9d89e201..fec321e6 100644
--- a/pkg/sdkclient/sinker/client.go
+++ b/pkg/sdkclient/sinker/client.go
@@ -23,10 +23,10 @@ import (
 	sinkpb "github.com/numaproj/numaflow-go/pkg/apis/proto/sink/v1"
 	"github.com/numaproj/numaflow-go/pkg/info"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/protobuf/types/known/emptypb"
 
 	"github.com/numaproj/numaflow/pkg/sdkclient"
+	"github.com/numaproj/numaflow/pkg/shared/util"
 )
 
 // client contains the grpc connection and the grpc client.
@@ -37,21 +37,19 @@ type client struct {
 
 var _ Client = (*client)(nil)
 
-// New creates a new client object. Sinker client doesn't require server info to start ATM.
-func New(_ *info.ServerInfo, inputOptions ...sdkclient.Option) (Client, error) {
+func New(serverInfo *info.ServerInfo, inputOptions ...sdkclient.Option) (Client, error) {
 	var opts = sdkclient.DefaultOptions(sdkclient.SinkAddr)
+
 	for _, inputOption := range inputOptions {
 		inputOption(opts)
 	}
 
-	// connect to the server
-	c := new(client)
-	sockAddr := fmt.Sprintf("%s:%s", sdkclient.UDS, opts.UdsSockAddr())
-	conn, err := grpc.Dial(sockAddr, grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(opts.MaxMessageSize()), grpc.MaxCallSendMsgSize(opts.MaxMessageSize())))
+	// Connect to the server
+	conn, err := util.ConnectToServer(opts.UdsSockAddr(), serverInfo, opts.MaxMessageSize())
 	if err != nil {
-		return nil, fmt.Errorf("failed to execute grpc.Dial(%q): %w", sockAddr, err)
+		return nil, err
 	}
+	c := new(client)
 
 	c.conn = conn
 	c.grpcClt = sinkpb.NewSinkClient(conn)
diff --git a/pkg/sdkclient/source/client/client.go b/pkg/sdkclient/source/client.go
similarity index 82%
rename from pkg/sdkclient/source/client/client.go
rename to pkg/sdkclient/source/client.go
index 30e90c3b..b71cc600 100644
--- a/pkg/sdkclient/source/client/client.go
+++ b/pkg/sdkclient/source/client.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package client
+package source
 
 import (
 	"context"
@@ -24,10 +24,10 @@ import (
 	sourcepb "github.com/numaproj/numaflow-go/pkg/apis/proto/source/v1"
 	"github.com/numaproj/numaflow-go/pkg/info"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/protobuf/types/known/emptypb"
 
 	"github.com/numaproj/numaflow/pkg/sdkclient"
+	"github.com/numaproj/numaflow/pkg/shared/util"
 )
 
 // client contains the grpc connection and the grpc client.
@@ -38,24 +38,20 @@ type client struct {
 
 var _ Client = (*client)(nil)
 
-// New creates a new client object. Source client doesn't require server info to start ATM.
-func New(_ *info.ServerInfo, inputOptions ...sdkclient.Option) (Client, error) {
+func New(serverInfo *info.ServerInfo, inputOptions ...sdkclient.Option) (Client, error) {
 	var opts = sdkclient.DefaultOptions(sdkclient.SourceAddr)
 
 	for _, inputOption := range inputOptions {
 		inputOption(opts)
 	}
 
-	// connect to the grpc server
-	c := new(client)
-	sockAddr := fmt.Sprintf("%s:%s", sdkclient.UDS, opts.UdsSockAddr())
-	conn, err := grpc.Dial(sockAddr,
-		grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(opts.MaxMessageSize()),
-			grpc.MaxCallSendMsgSize(opts.MaxMessageSize())))
+	// Connect to the server
+	conn, err := util.ConnectToServer(opts.UdsSockAddr(), serverInfo, opts.MaxMessageSize())
 	if err != nil {
-		return nil, fmt.Errorf("failed to execute grpc.Dial(%q): %w", sockAddr, err)
+		return nil, err
 	}
+	c := new(client)
+
 	c.conn = conn
 	c.grpcClt = sourcepb.NewSourceClient(conn)
 	return c, nil
diff --git a/pkg/sdkclient/source/client/interface.go b/pkg/sdkclient/source/interface.go
similarity index 99%
rename from pkg/sdkclient/source/client/interface.go
rename to pkg/sdkclient/source/interface.go
index 9a9c9815..ea897b82 100644
--- a/pkg/sdkclient/source/client/interface.go
+++ b/pkg/sdkclient/source/interface.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package client
+package source
 
 import (
 	"context"
diff --git a/pkg/sdkserverinfo/const.go b/pkg/sdkserverinfo/const.go
index b6a24f4e..9c7523a9 100644
--- a/pkg/sdkserverinfo/const.go
+++ b/pkg/sdkserverinfo/const.go
@@ -1,3 +1,14 @@
 package sdkserverinfo
 
-const ServerInfoFilePath = "/var/run/numaflow/server-info"
+const (
+	// Server information file configs
+	MapServerInfoFile               = "/var/run/numaflow/mapper-server-info"
+	MapStreamServerInfoFile         = "/var/run/numaflow/mapstreamer-server-info"
+	ReduceServerInfoFile            = "/var/run/numaflow/reducer-server-info"
+	ReduceStreamServerInfoFile      = "/var/run/numaflow/reducestreamer-server-info"
+	SessionReduceServerInfoFile     = "/var/run/numaflow/sessionreducer-server-info"
+	SideInputServerInfoFile         = "/var/run/numaflow/sideinput-server-info"
+	SinkServerInfoFile              = "/var/run/numaflow/sinker-server-info"
+	SourceServerInfoFile            = "/var/run/numaflow/sourcer-server-info"
+	SourceTransformerServerInfoFile = "/var/run/numaflow/sourcetransformer-server-info"
+)
diff --git a/pkg/sdkserverinfo/options.go b/pkg/sdkserverinfo/options.go
index 4b05de4b..1e153275 100644
--- a/pkg/sdkserverinfo/options.go
+++ b/pkg/sdkserverinfo/options.go
@@ -20,7 +20,6 @@ func (o *Options) ServerInfoReadinessTimeout() time.Duration {
 // DefaultOptions returns the default options.
 func DefaultOptions() *Options {
 	return &Options{
-		serverInfoFilePath:         ServerInfoFilePath,
 		serverInfoReadinessTimeout: 120 * time.Second, // Default timeout is 120 seconds
 	}
 }
diff --git a/pkg/sideinputs/manager/manager.go b/pkg/sideinputs/manager/manager.go
index bcbfcd70..421eb394 100644
--- a/pkg/sideinputs/manager/manager.go
+++ b/pkg/sideinputs/manager/manager.go
@@ -83,7 +83,7 @@ func (sim *sideInputsManager) Start(ctx context.Context) error {
 	}
 
 	// Wait for server info to be ready
-	serverInfo, err := sdkserverinfo.SDKServerInfo()
+	serverInfo, err := sdkserverinfo.SDKServerInfo(sdkserverinfo.WithServerInfoFilePath(sdkserverinfo.SideInputServerInfoFile))
 	if err != nil {
 		return err
 	}
diff --git a/pkg/sinks/sink.go b/pkg/sinks/sink.go
index 8a8b60bf..f43fa727 100644
--- a/pkg/sinks/sink.go
+++ b/pkg/sinks/sink.go
@@ -142,7 +142,7 @@ func (u *SinkProcessor) Start(ctx context.Context) error {
 	maxMessageSize := sharedutil.LookupEnvIntOr(dfv1.EnvGRPCMaxMessageSize, sdkclient.DefaultGRPCMaxMessageSize)
 	if udSink := u.VertexInstance.Vertex.Spec.Sink.UDSink; udSink != nil {
 		// Wait for server info to be ready
-		serverInfo, err := sdkserverinfo.SDKServerInfo()
+		serverInfo, err := sdkserverinfo.SDKServerInfo(sdkserverinfo.WithServerInfoFilePath(sdkserverinfo.SinkServerInfoFile))
 		if err != nil {
 			return err
 		}
diff --git a/pkg/sources/source.go b/pkg/sources/source.go
index cd5358a5..222dfa3d 100644
--- a/pkg/sources/source.go
+++ b/pkg/sources/source.go
@@ -31,7 +31,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/isbsvc"
 	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/sdkclient"
-	sourceclient "github.com/numaproj/numaflow/pkg/sdkclient/source/client"
+	sourceclient "github.com/numaproj/numaflow/pkg/sdkclient/source"
 	"github.com/numaproj/numaflow/pkg/sdkclient/sourcetransformer"
 	"github.com/numaproj/numaflow/pkg/sdkserverinfo"
 	jsclient "github.com/numaproj/numaflow/pkg/shared/clients/nats"
@@ -185,7 +185,7 @@ func (sp *SourceProcessor) Start(ctx context.Context) error {
 	var udsGRPCClient *udsource.GRPCBasedUDSource
 	if sp.VertexInstance.Vertex.IsUDSource() {
 		// Wait for server info to be ready
-		serverInfo, err := sdkserverinfo.SDKServerInfo()
+		serverInfo, err := sdkserverinfo.SDKServerInfo(sdkserverinfo.WithServerInfoFilePath(sdkserverinfo.SourceServerInfoFile))
 		if err != nil {
 			return err
 		}
@@ -214,7 +214,7 @@ func (sp *SourceProcessor) Start(ctx context.Context) error {
 	maxMessageSize := sharedutil.LookupEnvIntOr(dfv1.EnvGRPCMaxMessageSize, sdkclient.DefaultGRPCMaxMessageSize)
 	if sp.VertexInstance.Vertex.HasUDTransformer() {
 		// Wait for server info to be ready
-		serverInfo, err := sdkserverinfo.SDKServerInfo()
+		serverInfo, err := sdkserverinfo.SDKServerInfo(sdkserverinfo.WithServerInfoFilePath(sdkserverinfo.SourceTransformerServerInfoFile))
 		if err != nil {
 			return err
 		}
diff --git a/pkg/sources/udsource/grpc_udsource.go b/pkg/sources/udsource/grpc_udsource.go
index b87ede0a..7f4daf40 100644
--- a/pkg/sources/udsource/grpc_udsource.go
+++ b/pkg/sources/udsource/grpc_udsource.go
@@ -26,7 +26,7 @@ import (
 	"google.golang.org/protobuf/types/known/emptypb"
 
 	"github.com/numaproj/numaflow/pkg/isb"
-	sourceclient "github.com/numaproj/numaflow/pkg/sdkclient/source/client"
+	sourceclient "github.com/numaproj/numaflow/pkg/sdkclient/source"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"github.com/numaproj/numaflow/pkg/sources/udsource/utils"
 )
diff --git a/pkg/udf/map_udf.go b/pkg/udf/map_udf.go
index 50d6651e..bf511095 100644
--- a/pkg/udf/map_udf.go
+++ b/pkg/udf/map_udf.go
@@ -131,7 +131,7 @@ func (u *MapUDFProcessor) Start(ctx context.Context) error {
 	maxMessageSize := sharedutil.LookupEnvIntOr(dfv1.EnvGRPCMaxMessageSize, sdkclient.DefaultGRPCMaxMessageSize)
 	if enableMapUdfStream {
 		// Wait for server info to be ready
-		serverInfo, err := sdkserverinfo.SDKServerInfo()
+		serverInfo, err := sdkserverinfo.SDKServerInfo(sdkserverinfo.WithServerInfoFilePath(sdkserverinfo.MapStreamServerInfoFile))
 		if err != nil {
 			return err
 		}
@@ -155,7 +155,7 @@ func (u *MapUDFProcessor) Start(ctx context.Context) error {
 
 	} else {
 		// Wait for server info to be ready
-		serverInfo, err := sdkserverinfo.SDKServerInfo()
+		serverInfo, err := sdkserverinfo.SDKServerInfo(sdkserverinfo.WithServerInfoFilePath(sdkserverinfo.MapServerInfoFile))
 		if err != nil {
 			return err
 		}
diff --git a/pkg/udf/reduce_udf.go b/pkg/udf/reduce_udf.go
index d2f44539..c53fc32d 100644
--- a/pkg/udf/reduce_udf.go
+++ b/pkg/udf/reduce_udf.go
@@ -22,6 +22,7 @@ import (
 	"strings"
 	"sync"
 
+	"github.com/numaproj/numaflow-go/pkg/info"
 	"go.uber.org/zap"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
@@ -91,17 +92,22 @@ func (u *ReduceUDFProcessor) Start(ctx context.Context) error {
 
 	// create udf handler and wait until it is ready
 	if windowType.Fixed != nil || windowType.Sliding != nil {
-		// Wait for server info to be ready
-		serverInfo, err := sdkserverinfo.SDKServerInfo()
-		if err != nil {
-			return err
-		}
-
+		var serverInfo *info.ServerInfo
 		var client reducer.Client
 		// if streaming is enabled, use the reduceStreaming address
 		if (windowType.Fixed != nil && windowType.Fixed.Streaming) || (windowType.Sliding != nil && windowType.Sliding.Streaming) {
+			// Wait for server info to be ready
+			serverInfo, err = sdkserverinfo.SDKServerInfo(sdkserverinfo.WithServerInfoFilePath(sdkserverinfo.ReduceStreamServerInfoFile))
+			if err != nil {
+				return err
+			}
 			client, err = reducer.New(serverInfo, sdkclient.WithMaxMessageSize(maxMessageSize), sdkclient.WithUdsSockAddr(sdkclient.ReduceStreamAddr))
 		} else {
+			// Wait for server info to be ready
+			serverInfo, err = sdkserverinfo.SDKServerInfo(sdkserverinfo.WithServerInfoFilePath(sdkserverinfo.ReduceServerInfoFile))
+			if err != nil {
+				return err
+			}
 			client, err = reducer.New(serverInfo, sdkclient.WithMaxMessageSize(maxMessageSize))
 		}
 		if err != nil {
@@ -124,7 +130,7 @@ func (u *ReduceUDFProcessor) Start(ctx context.Context) error {
 		healthChecker = reduceHandler
 	} else if windowType.Session != nil {
 		// Wait for server info to be ready
-		serverInfo, err := sdkserverinfo.SDKServerInfo()
+		serverInfo, err := sdkserverinfo.SDKServerInfo(sdkserverinfo.WithServerInfoFilePath(sdkserverinfo.SessionReduceServerInfoFile))
 		if err != nil {
 			return err
 		}


Test patch:
diff --git a/pkg/sdkclient/source/client/client_test.go b/pkg/sdkclient/source/client_test.go
similarity index 99%
rename from pkg/sdkclient/source/client/client_test.go
rename to pkg/sdkclient/source/client_test.go
index c3619708..450394b8 100644
--- a/pkg/sdkclient/source/client/client_test.go
+++ b/pkg/sdkclient/source/client_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package client
+package source
 
 import (
 	"context"
diff --git a/pkg/sources/udsource/grpc_udsource_test.go b/pkg/sources/udsource/grpc_udsource_test.go
index fca52d46..e41521ef 100644
--- a/pkg/sources/udsource/grpc_udsource_test.go
+++ b/pkg/sources/udsource/grpc_udsource_test.go
@@ -36,7 +36,7 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/numaproj/numaflow/pkg/isb"
-	sourceclient "github.com/numaproj/numaflow/pkg/sdkclient/source/client"
+	sourceclient "github.com/numaproj/numaflow/pkg/sdkclient/source"
 	"github.com/numaproj/numaflow/pkg/sources/udsource/utils"
 )
 
diff --git a/test/diamond-e2e/testdata/cycle-backward.yaml b/test/diamond-e2e/testdata/cycle-backward.yaml
index d2cc7f0d..2126013a 100644
--- a/test/diamond-e2e/testdata/cycle-backward.yaml
+++ b/test/diamond-e2e/testdata/cycle-backward.yaml
@@ -17,13 +17,13 @@ spec:
       udf:
         container:
           # This will try each message up to 3 times before continuing, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/retry
-          image: quay.io/numaio/numaflow-go/map-retry:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-retry:v0.6.1
     - name: out
       sink:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: cat
diff --git a/test/diamond-e2e/testdata/cycle-to-self.yaml b/test/diamond-e2e/testdata/cycle-to-self.yaml
index 78c9bd90..ec02e380 100644
--- a/test/diamond-e2e/testdata/cycle-to-self.yaml
+++ b/test/diamond-e2e/testdata/cycle-to-self.yaml
@@ -13,13 +13,13 @@ spec:
       udf:
         container:
           # This will try each message up to 3 times before continuing, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/retry
-          image: quay.io/numaio/numaflow-go/map-retry:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-retry:v0.6.1
     - name: out
       sink:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: retry
diff --git a/test/diamond-e2e/testdata/join-on-map-pipeline.yaml b/test/diamond-e2e/testdata/join-on-map-pipeline.yaml
index a8d9ac5e..e2fa7bf7 100644
--- a/test/diamond-e2e/testdata/join-on-map-pipeline.yaml
+++ b/test/diamond-e2e/testdata/join-on-map-pipeline.yaml
@@ -30,7 +30,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in-0
       to: cat
diff --git a/test/diamond-e2e/testdata/join-on-reduce-pipeline.yaml b/test/diamond-e2e/testdata/join-on-reduce-pipeline.yaml
index 16b5bd11..5b9cabbe 100644
--- a/test/diamond-e2e/testdata/join-on-reduce-pipeline.yaml
+++ b/test/diamond-e2e/testdata/join-on-reduce-pipeline.yaml
@@ -17,7 +17,7 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.1
     - name: atoi-1
       partitions: 2
       scale:
@@ -25,7 +25,7 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.1
     - name: compute-sum
       partitions: 2
       udf:
@@ -48,7 +48,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: atoi-0
diff --git a/test/diamond-e2e/testdata/join-on-sink.yaml b/test/diamond-e2e/testdata/join-on-sink.yaml
index 0226cd63..6601a146 100644
--- a/test/diamond-e2e/testdata/join-on-sink.yaml
+++ b/test/diamond-e2e/testdata/join-on-sink.yaml
@@ -11,7 +11,7 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.1
     - name: even-cat
       udf:
         builtin:
@@ -25,7 +25,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
     
   edges:
     - from: in
diff --git a/test/e2e/testdata/even-odd.yaml b/test/e2e/testdata/even-odd.yaml
index ed8d99cd..23ebfb28 100644
--- a/test/e2e/testdata/even-odd.yaml
+++ b/test/e2e/testdata/even-odd.yaml
@@ -12,24 +12,24 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.1
     - name: even-sink
       partitions: 2
       sink:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
     - name: odd-sink
       sink:
         udsink:
           container:
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
     - name: number-sink
       sink:
         udsink:
           container:
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: even-or-odd
diff --git a/test/e2e/testdata/udf-filtering.yaml b/test/e2e/testdata/udf-filtering.yaml
index c74824fe..eb8bae1d 100644
--- a/test/e2e/testdata/udf-filtering.yaml
+++ b/test/e2e/testdata/udf-filtering.yaml
@@ -18,7 +18,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: p1
diff --git a/test/http-e2e/testdata/http-source-with-auth.yaml b/test/http-e2e/testdata/http-source-with-auth.yaml
index c87c2806..79d3149a 100644
--- a/test/http-e2e/testdata/http-source-with-auth.yaml
+++ b/test/http-e2e/testdata/http-source-with-auth.yaml
@@ -16,7 +16,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: out
diff --git a/test/http-e2e/testdata/http-source.yaml b/test/http-e2e/testdata/http-source.yaml
index e807c9fc..eb89ef44 100644
--- a/test/http-e2e/testdata/http-source.yaml
+++ b/test/http-e2e/testdata/http-source.yaml
@@ -15,7 +15,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: out
diff --git a/test/idle-source-e2e/testdata/idle-source-reduce-pipeline.yaml b/test/idle-source-e2e/testdata/idle-source-reduce-pipeline.yaml
index 4343469c..d6ea1044 100644
--- a/test/idle-source-e2e/testdata/idle-source-reduce-pipeline.yaml
+++ b/test/idle-source-e2e/testdata/idle-source-reduce-pipeline.yaml
@@ -22,7 +22,7 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.1
     - name: compute-sum
       partitions: 2
       udf:
@@ -45,7 +45,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: atoi
diff --git a/test/nats-e2e/testdata/nats-source-pipeline.yaml b/test/nats-e2e/testdata/nats-source-pipeline.yaml
index f6bb200a..54be8826 100644
--- a/test/nats-e2e/testdata/nats-source-pipeline.yaml
+++ b/test/nats-e2e/testdata/nats-source-pipeline.yaml
@@ -25,7 +25,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: p1
diff --git a/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml b/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml
index b67b227b..825b0bfc 100644
--- a/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml
+++ b/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml
@@ -17,7 +17,7 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.1
     - name: keyed-fixed-sum
       partitions: 1
       udf:
@@ -50,7 +50,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: atoi
diff --git a/test/reduce-e2e/testdata/complex-sliding-window-pipeline.yaml b/test/reduce-e2e/testdata/complex-sliding-window-pipeline.yaml
index 775c9d72..c311d166 100644
--- a/test/reduce-e2e/testdata/complex-sliding-window-pipeline.yaml
+++ b/test/reduce-e2e/testdata/complex-sliding-window-pipeline.yaml
@@ -17,7 +17,7 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.1
     - name: keyed-fixed-sum
       partitions: 2
       udf:
@@ -63,7 +63,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: atoi
diff --git a/test/reduce-e2e/testdata/simple-keyed-reduce-pipeline.yaml b/test/reduce-e2e/testdata/simple-keyed-reduce-pipeline.yaml
index f15d5ff4..db6555f9 100644
--- a/test/reduce-e2e/testdata/simple-keyed-reduce-pipeline.yaml
+++ b/test/reduce-e2e/testdata/simple-keyed-reduce-pipeline.yaml
@@ -17,7 +17,7 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.1
     - name: compute-sum
       partitions: 1
       udf:
@@ -40,7 +40,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: atoi
diff --git a/test/reduce-e2e/testdata/simple-non-keyed-reduce-pipeline.yaml b/test/reduce-e2e/testdata/simple-non-keyed-reduce-pipeline.yaml
index 78b18a9b..0a125aeb 100644
--- a/test/reduce-e2e/testdata/simple-non-keyed-reduce-pipeline.yaml
+++ b/test/reduce-e2e/testdata/simple-non-keyed-reduce-pipeline.yaml
@@ -17,7 +17,7 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.1
     - name: compute-sum
       udf:
         container:
@@ -39,7 +39,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: atoi
diff --git a/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml b/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml
index dcbae89e..1330f2a8 100644
--- a/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml
+++ b/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml
@@ -15,7 +15,7 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.1
     - name: compute-sum
       partitions: 2
       udf:
@@ -38,7 +38,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: atoi
diff --git a/test/reduce-e2e/testdata/simple-session-reduce-pipeline.yaml b/test/reduce-e2e/testdata/simple-session-reduce-pipeline.yaml
index 67b30aea..62d1b778 100644
--- a/test/reduce-e2e/testdata/simple-session-reduce-pipeline.yaml
+++ b/test/reduce-e2e/testdata/simple-session-reduce-pipeline.yaml
@@ -13,7 +13,7 @@ spec:
       partitions: 1
       udf:
         container:
-          # Compute the sum, see https://github.com/numaproj/numaflow-go/tree/main/pkg/reducer/examples/sum
+          # see https://github.com/numaproj/numaflow-go/tree/main/pkg/sessionreducer/examples/counter
           image: quay.io/numaio/numaflow-go/session-counter:v0.6.1
           imagePullPolicy: Always
         groupBy:
@@ -32,7 +32,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: compute-sum
diff --git a/test/reduce-sdk-e2e/testdata/reduce-stream/reduce-stream-go.yaml b/test/reduce-sdk-e2e/testdata/reduce-stream/reduce-stream-go.yaml
index b7107a0a..a8cd2f28 100644
--- a/test/reduce-sdk-e2e/testdata/reduce-stream/reduce-stream-go.yaml
+++ b/test/reduce-sdk-e2e/testdata/reduce-stream/reduce-stream-go.yaml
@@ -13,13 +13,13 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.1
     - name: compute-sum
       partitions: 2
       udf:
         container:
           # compute the sum, see https://github.com/numaproj/numaflow-go/tree/main/pkg/reducestreamer/examples/sum
-          image: quay.io/numaio/numaflow-go/reduce-stream-sum:v0.6.0
+          image: quay.io/numaio/numaflow-go/reduce-stream-sum:v0.6.1
         groupBy:
           window:
             fixed:
@@ -36,7 +36,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: atoi
diff --git a/test/reduce-sdk-e2e/testdata/reduce-stream/reduce-stream-java.yaml b/test/reduce-sdk-e2e/testdata/reduce-stream/reduce-stream-java.yaml
index fe6e689f..8dc2333a 100644
--- a/test/reduce-sdk-e2e/testdata/reduce-stream/reduce-stream-java.yaml
+++ b/test/reduce-sdk-e2e/testdata/reduce-stream/reduce-stream-java.yaml
@@ -13,7 +13,7 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.1
     - name: compute-sum
       partitions: 2
       udf:
@@ -36,7 +36,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: atoi
diff --git a/test/sdks-e2e/testdata/flatmap-stream.yaml b/test/sdks-e2e/testdata/flatmap-stream.yaml
index 927adc73..91341fcc 100644
--- a/test/sdks-e2e/testdata/flatmap-stream.yaml
+++ b/test/sdks-e2e/testdata/flatmap-stream.yaml
@@ -19,7 +19,7 @@ spec:
       udf:
         container:
           # Split input message into an array with comma, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapstreamer/examples/flatmap_stream
-          image: quay.io/numaio/numaflow-go/map-flatmap-stream:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-flatmap-stream:v0.6.1
     - name: go-udsink
       scale:
         min: 1
@@ -27,7 +27,7 @@ spec:
         udsink:
           container:
             # https://github.com/numaproj/numaflow-go/tree/main/pkg/sinker/examples/log
-            image: quay.io/numaio/numaflow-go/sink-log:v0.6.0
+            image: quay.io/numaio/numaflow-go/sink-log:v0.6.1
     - name: go-udsink-2
       scale:
         min: 1
@@ -35,7 +35,7 @@ spec:
         udsink:
           container:
             # https://github.com/numaproj/numaflow-go/tree/main/pkg/sinker/examples/log
-            image: quay.io/numaio/numaflow-go/sink-log:v0.6.0
+            image: quay.io/numaio/numaflow-go/sink-log:v0.6.1
 
     - name: python-split
       partitions: 3
diff --git a/test/sdks-e2e/testdata/flatmap.yaml b/test/sdks-e2e/testdata/flatmap.yaml
index d749223c..8beb512b 100644
--- a/test/sdks-e2e/testdata/flatmap.yaml
+++ b/test/sdks-e2e/testdata/flatmap.yaml
@@ -13,7 +13,7 @@ spec:
       udf:
         container:
           # Split input message into an array with comma, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/flatmap
-          image: quay.io/numaio/numaflow-go/map-flatmap:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-flatmap:v0.6.1
     - name: go-udsink
       scale:
         min: 1
@@ -21,7 +21,7 @@ spec:
         udsink:
           container:
             # https://github.com/numaproj/numaflow-go/tree/main/pkg/sinker/examples/log
-            image: quay.io/numaio/numaflow-go/sink-log:v0.6.0
+            image: quay.io/numaio/numaflow-go/sink-log:v0.6.1
 
     - name: python-split
       scale:
diff --git a/test/sdks-e2e/testdata/simple-keyed-reduce-pipeline.yaml b/test/sdks-e2e/testdata/simple-keyed-reduce-pipeline.yaml
index bf84ef1e..edf24176 100644
--- a/test/sdks-e2e/testdata/simple-keyed-reduce-pipeline.yaml
+++ b/test/sdks-e2e/testdata/simple-keyed-reduce-pipeline.yaml
@@ -19,7 +19,7 @@ spec:
       udf:
         container:
           # compute the sum, see https://github.com/numaproj/numaflow-java/tree/main/examples/src/main/java/io/numaproj/numaflow/examples/reduce/sum
-          image: quay.io/numaio/numaflow-java/reduce-sum:v0.6.1
+          image: quay.io/numaio/numaflow-java/reduce-sum:v0.6.0
         groupBy:
           window:
             fixed:
diff --git a/test/sdks-e2e/testdata/transformer/event-time-filter-go.yaml b/test/sdks-e2e/testdata/transformer/event-time-filter-go.yaml
index fabdb7ef..745dd092 100644
--- a/test/sdks-e2e/testdata/transformer/event-time-filter-go.yaml
+++ b/test/sdks-e2e/testdata/transformer/event-time-filter-go.yaml
@@ -10,7 +10,7 @@ spec:
         transformer:
           container:
             # Filter messages based on event time, see https://github.com/numaproj/numaflow-go/tree/main/pkg/sourcetransformer/examples/event_time_filter
-            image: quay.io/numaio/numaflow-go/mapt-event-time-filter:v0.6.0
+            image: quay.io/numaio/numaflow-go/mapt-event-time-filter:v0.6.1
     - name: sink-within-2022
       partitions: 2
       scale:
diff --git a/test/sideinputs-e2e/testdata/map-sideinput-pipeline.yaml b/test/sideinputs-e2e/testdata/map-sideinput-pipeline.yaml
index 121325df..f40dec5b 100644
--- a/test/sideinputs-e2e/testdata/map-sideinput-pipeline.yaml
+++ b/test/sideinputs-e2e/testdata/map-sideinput-pipeline.yaml
@@ -7,7 +7,7 @@ spec:
     - name: myticker
       container:
         # A map side input example , see https://github.com/numaproj-contrib/e2e-tests-go/tree/main/map-side-input
-        image: "quay.io/numaio/numaproj-contrib/e2e-map-sideinput-example:v0.0.3"
+        image: "quay.io/numaio/numaproj-contrib/e2e-map-sideinput-example:v0.6.1"
         imagePullPolicy: Always
       trigger:
         schedule: "@every 5s"
@@ -19,7 +19,7 @@ spec:
       udf:
         container:
           # A map side input udf , see https://github.com/numaproj-contrib/e2e-tests-go/tree/main/map-side-input/udf
-          image: "quay.io/numaio/numaproj-contrib/e2e-map-sideinput-udf:v0.0.2"
+          image: "quay.io/numaio/numaproj-contrib/e2e-map-sideinput-udf:v0.6.1"
           imagePullPolicy: Always
       sideInputs:
         - myticker
@@ -30,7 +30,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
             imagePullPolicy: Always
   edges:
     - from: in
diff --git a/test/sideinputs-e2e/testdata/reduce-sideinput-pipeline.yaml b/test/sideinputs-e2e/testdata/reduce-sideinput-pipeline.yaml
index 5b26ec61..02299b79 100644
--- a/test/sideinputs-e2e/testdata/reduce-sideinput-pipeline.yaml
+++ b/test/sideinputs-e2e/testdata/reduce-sideinput-pipeline.yaml
@@ -7,7 +7,7 @@ spec:
     - name: myticker
       container:
         # A reduce side input, see https://github.com/numaproj-contrib/e2e-tests-go/tree/main/reduce-side-input
-        image: "quay.io/numaio/numaproj-contrib/e2e-reduce-sideinput-example:v0.0.3"
+        image: "quay.io/numaio/numaproj-contrib/e2e-reduce-sideinput-example:v0.6.1"
         imagePullPolicy: Always
       trigger:
         schedule: "@every 5s"
@@ -21,13 +21,13 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.1
           imagePullPolicy: Always
     - name: si-e2e
       udf:
         container:
           # A reduce side input udf, see https://github.com/numaproj-contrib/e2e-tests-go/tree/main/reduce-side-input/udf
-          image: "quay.io/numaio/numaproj-contrib/e2e-reduce-sideinput-udf:dev"
+          image: "quay.io/numaio/numaproj-contrib/e2e-reduce-sideinput-udf:v0.6.1"
           imagePullPolicy: Always
         groupBy:
           window:
@@ -47,7 +47,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: atoi
diff --git a/test/sideinputs-e2e/testdata/sideinput_source.yaml b/test/sideinputs-e2e/testdata/sideinput_source.yaml
index bced4795..12dc52dc 100644
--- a/test/sideinputs-e2e/testdata/sideinput_source.yaml
+++ b/test/sideinputs-e2e/testdata/sideinput_source.yaml
@@ -28,7 +28,7 @@ spec:
         udsink:
           container:
             # see https://github.com/numaproj/numaflow-sinks/tree/78693ec48a5b8bc7ca7be2898dd8dd722392b53a/redis-e2e-test-sink
-            image: "quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0"
+            image: "quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest"
       sideInputs:
         - myticker
   edges:
diff --git a/test/transformer-e2e/testdata/source-filtering.yaml b/test/transformer-e2e/testdata/source-filtering.yaml
index 7f55500a..1abe8a1a 100644
--- a/test/transformer-e2e/testdata/source-filtering.yaml
+++ b/test/transformer-e2e/testdata/source-filtering.yaml
@@ -17,7 +17,7 @@ spec:
         udsink:
           container:
             # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
-            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
+            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:latest
   edges:
     - from: in
       to: out
diff --git a/test/udsource-e2e/testdata/simple-source-go.yaml b/test/udsource-e2e/testdata/simple-source-go.yaml
index 55b6de48..d65b4a70 100644
--- a/test/udsource-e2e/testdata/simple-source-go.yaml
+++ b/test/udsource-e2e/testdata/simple-source-go.yaml
@@ -10,9 +10,9 @@ 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.1
+            image: quay.io/numaio/numaflow-go/source-simple-source:v0.6.1
       limits:
-        readBatchSize: 2
+        readBatchSize: 500
     - name: out
       sink:
         log: {}
diff --git a/test/udsource-e2e/testdata/simple-source-java.yaml b/test/udsource-e2e/testdata/simple-source-java.yaml
index b8fd47e0..5d6182c1 100644
--- a/test/udsource-e2e/testdata/simple-source-java.yaml
+++ b/test/udsource-e2e/testdata/simple-source-java.yaml
@@ -12,7 +12,7 @@ spec:
             # See https://github.com/numaproj/numaflow-java/tree/main/examples/src/main/java/io/numaproj/numaflow/examples/source/simple
             image: quay.io/numaio/numaflow-java/source-simple-source:v0.6.0
       limits:
-        readBatchSize: 2
+        readBatchSize: 500
     - name: out
       sink:
         log: {}
diff --git a/test/udsource-e2e/testdata/simple-source-python.yaml b/test/udsource-e2e/testdata/simple-source-python.yaml
index e989b687..f0649d0f 100644
--- a/test/udsource-e2e/testdata/simple-source-python.yaml
+++ b/test/udsource-e2e/testdata/simple-source-python.yaml
@@ -9,11 +9,11 @@ spec:
         udsource:
           container:
             # A simple user-defined source for e2e testing
-            # See https://github.com/numaproj/numaflow-python/tree/main/examples/source
+            # See https://github.com/numaproj/numaflow-python/tree/main/examples/source/simple-source
             image: quay.io/numaio/numaflow-python/simple-source:v0.6.0
             imagePullPolicy: Always
       limits:
-        readBatchSize: 2
+        readBatchSize: 500
     - name: out
       sink:
         log: {}
diff --git a/test/udsource-e2e/udsource_test.go b/test/udsource-e2e/udsource_test.go
index f0df3df9..57b9f30e 100644
--- a/test/udsource-e2e/udsource_test.go
+++ b/test/udsource-e2e/udsource_test.go
@@ -75,11 +75,15 @@ func (s *UserDefinedSourceSuite) testSimpleSource(lang string) {
 	// such a large amount of data that the redis sink is not able to handle it, it breaks with OOM error
 
 	// the user-defined simple source sends the read index of the message as the message content
-	// verify the sink get the first batch of data
-	w.Expect().VertexPodLogContains("out", "0")
-	w.Expect().VertexPodLogContains("out", "1")
-	// verify the sink get the second batch of data
-	w.Expect().VertexPodLogContains("out", "2")
+	// verify the sink gets the first batch of data(0-499) - checking for some random numbers
+	w.Expect().VertexPodLogContains("out", "147")
+	w.Expect().VertexPodLogContains("out", "258")
+	w.Expect().VertexPodLogContains("out", "369")
+	// verify the sink get the second batch of data(500-999)
+	w.Expect().VertexPodLogContains("out", "520")
+	w.Expect().VertexPodLogContains("out", "630")
+	w.Expect().VertexPodLogContains("out", "999")
+
 }
 
 func TestUserDefinedSourceSuite(t *testing.T) {