Repository: numaproj/numaflow
Pull request #1683 (2024-04-17 18:23:05)

chore: add version compatibility check
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 f5f99281..0660c908 100644
--- a/go.mod
+++ b/go.mod
@@ -4,9 +4,11 @@ go 1.21
 
 require (
 	github.com/IBM/sarama v1.43.0
+	github.com/Masterminds/semver/v3 v3.1.1
 	github.com/Masterminds/sprig/v3 v3.2.2
 	github.com/ahmetb/gen-crd-api-reference-docs v0.3.0
 	github.com/antonmedv/expr v1.9.0
+	github.com/aquasecurity/go-pep440-version v0.0.0-20210121094942-22b2f8951d46
 	github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de
 	github.com/casbin/casbin/v2 v2.77.2
 	github.com/coreos/go-oidc/v3 v3.6.0
@@ -31,7 +33,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.33.1
-	github.com/numaproj/numaflow-go v0.7.0-rc1.0.20240415065440-ea7f6923f367
+	github.com/numaproj/numaflow-go v0.7.0-rc2
 	github.com/prometheus/client_golang v1.18.0
 	github.com/prometheus/common v0.45.0
 	github.com/redis/go-redis/v9 v9.0.3
@@ -69,10 +71,10 @@ require (
 require (
 	github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible // indirect
 	github.com/Masterminds/goutils v1.1.1 // indirect
-	github.com/Masterminds/semver/v3 v3.1.1 // indirect
 	github.com/TylerBrock/colorjson v0.0.0-20200706003622-8a50f05110d2 // indirect
 	github.com/ajg/form v1.5.1 // indirect
 	github.com/andybalholm/brotli v1.0.4 // indirect
+	github.com/aquasecurity/go-version v0.0.0-20210121072130-637058cfe492 // indirect
 	github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bytedance/sonic v1.11.3 // indirect
@@ -202,6 +204,7 @@ require (
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
 	golang.org/x/tools v0.16.1 // indirect
+	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/genproto v0.0.0-20230803162519-f966b187b2e5 // indirect
diff --git a/go.sum b/go.sum
index 874fa646..12ce7c08 100644
--- a/go.sum
+++ b/go.sum
@@ -80,6 +80,10 @@ github.com/andybalholm/brotli v1.0.4/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHG
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/antonmedv/expr v1.9.0 h1:j4HI3NHEdgDnN9p6oI6Ndr0G5QryMY0FNxT4ONrFDGU=
 github.com/antonmedv/expr v1.9.0/go.mod h1:5qsM3oLGDND7sDmQGDXHkYfkjYMUX14qsgqmHhwGEk8=
+github.com/aquasecurity/go-pep440-version v0.0.0-20210121094942-22b2f8951d46 h1:vmXNl+HDfqqXgr0uY1UgK1GAhps8nbAAtqHNBcgyf+4=
+github.com/aquasecurity/go-pep440-version v0.0.0-20210121094942-22b2f8951d46/go.mod h1:olhPNdiiAAMiSujemd1O/sc6GcyePr23f/6uGKtthNg=
+github.com/aquasecurity/go-version v0.0.0-20210121072130-637058cfe492 h1:rcEG5HI490FF0a7zuvxOxen52ddygCfNVjP0XOCMl+M=
+github.com/aquasecurity/go-version v0.0.0-20210121072130-637058cfe492/go.mod h1:9Beu8XsUNNfzml7WBf3QmyPToP1wm1Gj/Vc5UJKqTzU=
 github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de h1:FxWPpzIjnTlhPwqqXc4/vE0f7GvRjuAsbW+HOIe8KnA=
 github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de/go.mod h1:DCaWoUhZrYW9p1lxo/cm8EmUOOzAPSEZNGF2DK1dJgw=
 github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
@@ -132,6 +136,7 @@ github.com/coreos/go-oidc/v3 v3.6.0 h1:AKVxfYw1Gmkn/w96z0DbT/B/xFnzTd3MkZvWLjF4n
 github.com/coreos/go-oidc/v3 v3.6.0/go.mod h1:ZpHUsHBucTUj6WOkrP4E20UPynbLZzhTQ1XKCXkxyPc=
 github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
 github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
+github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
 github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v0.0.0-20161028175848-04cdfd42973b/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -677,8 +682,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.7.0-rc1.0.20240415065440-ea7f6923f367 h1:JNyNFp/HNH0wHyq49Eq/d2cOdOHoSrs3NOX6zNG4hKk=
-github.com/numaproj/numaflow-go v0.7.0-rc1.0.20240415065440-ea7f6923f367/go.mod h1:WoMt31+h3up202zTRI8c/qe42B8UbvwLe2mJH0MAlhI=
+github.com/numaproj/numaflow-go v0.7.0-rc2 h1:oLP0uM9ToFEtiSyzX46cFOlr7ctgcH02cY/XFA1qF8E=
+github.com/numaproj/numaflow-go v0.7.0-rc2/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/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=
@@ -826,6 +831,7 @@ github.com/ugorji/go v1.2.7/go.mod h1:nF9osbDWLy6bDVv/Rtoh6QgnvNDpmCalQV5urGCCS6
 github.com/ugorji/go/codec v1.2.7/go.mod h1:WGN1fab3R1fzQlVQTkfxVtIBhWDRqOviHU95kRgeqEY=
 github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE=
 github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
+github.com/urfave/cli/v2 v2.3.0/go.mod h1:LJmUH05zAU44vOAcrfzZQKsZbVcdbOG8rtL3/XcUArI=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
 github.com/valyala/fasthttp v1.37.0 h1:7WHCyI7EAkQMVmrfBhWTCOaeROb1aCBiTopx63LkMbE=
@@ -1243,6 +1249,8 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 h1:H2TDz8ibqkAF6YGhCdN3jS9O0/s90v0rJh3X/OLHEUk=
+golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2/go.mod h1:K8+ghG5WaK9qNqU5K3HdILfMLy1f3aNYFI/wnl100a8=
 gomodules.xyz/jsonpatch/v2 v2.4.0 h1:Ci3iUJyx9UeRx7CeFN8ARgGbkESwJK+KB9lLcWxY/Zw=
 gomodules.xyz/jsonpatch/v2 v2.4.0/go.mod h1:AH3dM2RI6uoBZxn3LVrfvJ3E0/9dG4cSrbuBJT4moAY=
 google.golang.org/api v0.4.0/go.mod h1:8k5glujaEP+g9n7WNsDg8QP6cUVNI86fCNMcbazEtwE=
diff --git a/pkg/shared/util/grpc_utils.go b/pkg/shared/util/grpc_utils.go
index 0f189e0a..2c9a5891 100644
--- a/pkg/shared/util/grpc_utils.go
+++ b/pkg/shared/util/grpc_utils.go
@@ -20,18 +20,88 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"strings"
 	"time"
 
+	"github.com/Masterminds/semver/v3"
+	pep440 "github.com/aquasecurity/go-pep440-version"
 	"github.com/numaproj/numaflow-go/pkg/info"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/status"
 
+	"github.com/numaproj/numaflow"
 	sdkerr "github.com/numaproj/numaflow/pkg/sdkclient/error"
 	resolver "github.com/numaproj/numaflow/pkg/sdkclient/grpc_resolver"
 )
 
+func checkConstraint(version *semver.Version, constraint string) error {
+	if c, err := semver.NewConstraint(constraint); err != nil {
+		return fmt.Errorf("error parsing constraint: %w, constraint string: %s", err, constraint)
+	} else if isValid, _ := c.Validate(version); !isValid {
+		return fmt.Errorf("version %v did not meet constraint requirement %s", version, constraint)
+	}
+	return nil
+}
+
+// checkNumaflowCompatibility checks if the current numaflow version is compatible with the given language's SDK version
+func checkNumaflowCompatibility(numaflowVersion string, minNumaflowVersion string) error {
+	// Check if server info contains MinimumNumaflowVersion
+	if minNumaflowVersion == "" {
+		return fmt.Errorf("server info does not contain minimum numaflow version. Upgrade to newer SDK version")
+	}
+
+	numaflowVersionSemVer, err := semver.NewVersion(numaflowVersion)
+	if err != nil {
+		return fmt.Errorf("error parsing numaflow version: %w", err)
+	}
+
+	numaflowConstraint := fmt.Sprintf(">= %s", minNumaflowVersion)
+	if err = checkConstraint(numaflowVersionSemVer, numaflowConstraint); err != nil {
+		return fmt.Errorf("numaflow version %s must be upgraded to at least %s, in order to work with current SDK version: %w",
+			numaflowVersionSemVer.String(), minNumaflowVersion, err)
+	}
+
+	return nil
+}
+
+// checkSDKCompatibility checks if the current SDK version is compatible with the numaflow version
+func checkSDKCompatibility(sdkVersion string, sdkLanguage info.Language, minSupportedSDKVersions sdkConstraints) error {
+	if sdkRequiredVersion, ok := minSupportedSDKVersions[sdkLanguage]; ok {
+		sdkConstraint := fmt.Sprintf(">= %s", sdkRequiredVersion)
+		if sdkLanguage == info.Python {
+			// Python pre-releases/releases follow PEP440 specification which requires a different library for parsing
+			sdkVersionPEP440, err := pep440.Parse(sdkVersion)
+			if err != nil {
+				return fmt.Errorf("error parsing SDK version: %w", err)
+			}
+
+			c, err := pep440.NewSpecifiers(sdkConstraint)
+			if err != nil {
+				return fmt.Errorf("error parsing SDK constraint: %w", err)
+			}
+
+			if !c.Check(sdkVersionPEP440) {
+				return fmt.Errorf("SDK version %s must be upgraded to at least %s, in order to work with current numaflow version: %w",
+					sdkVersionPEP440.String(), sdkRequiredVersion, err)
+			}
+		} else {
+			sdkVersionSemVer, err := semver.NewVersion(sdkVersion)
+			if err != nil {
+				return fmt.Errorf("error parsing SDK version: %w", err)
+			}
+
+			if err := checkConstraint(sdkVersionSemVer, sdkConstraint); err != nil {
+				return fmt.Errorf("SDK version %s must be upgraded to at least %s, in order to work with current numaflow version: %w",
+					sdkVersionSemVer.String(), sdkRequiredVersion, err)
+			}
+		}
+	}
+
+	return nil
+}
+
 // ToUDFErr converts gRPC error to UDF Error
 func ToUDFErr(name string, err error) error {
 	if err == nil {
@@ -64,7 +134,7 @@ func ToUDFErr(name string, err error) error {
 	}
 }
 
-// WaitForServerInfo waits until the server info is ready. It returns an error if the server info is not ready within the given timeout.
+// WaitForServerInfo waits until the server info is ready. It returns an error if the server info is not ready within the given timeout
 func WaitForServerInfo(timeout time.Duration, filePath string) (*info.ServerInfo, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
@@ -78,6 +148,23 @@ func WaitForServerInfo(timeout time.Duration, filePath string) (*info.ServerInfo
 		return nil, fmt.Errorf("failed to read server info: %w", err)
 	}
 
+	sdkVersion := serverInfo.Version
+	numaflowVersion := numaflow.GetVersion().Version
+
+	// If we are testing locally or in CI, we can skip checking for numaflow compatibility issues
+	// because both return us a version string that the version check libraries can't properly parse. (local: "*latest*" CI: commit SHA)
+	if !strings.Contains(numaflowVersion, "latest") && !strings.Contains(numaflowVersion, numaflow.GetVersion().GitCommit) {
+		if err := checkNumaflowCompatibility(numaflowVersion, serverInfo.MinimumNumaflowVersion); err != nil {
+			return nil, fmt.Errorf("numaflow %s does not satisfy the minimum required by SDK %s: %w",
+				numaflowVersion, sdkVersion, err)
+		}
+	}
+
+	if err := checkSDKCompatibility(sdkVersion, serverInfo.Language, minimumSupportedSDKVersions); err != nil {
+		return nil, fmt.Errorf("SDK %s does not satisfy the minimum required by numaflow %s: %w",
+			sdkVersion, numaflowVersion, err)
+	}
+
 	return serverInfo, nil
 }
 
diff --git a/pkg/shared/util/version_mapping.go b/pkg/shared/util/version_mapping.go
new file mode 100644
index 00000000..7e03fc97
--- /dev/null
+++ b/pkg/shared/util/version_mapping.go
@@ -0,0 +1,29 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package util
+
+import (
+	"github.com/numaproj/numaflow-go/pkg/info"
+)
+
+type sdkConstraints map[info.Language]string
+
+var minimumSupportedSDKVersions = sdkConstraints{
+	info.Go:     "0.7.0-rc2",
+	info.Python: "0.7.0a1",
+	info.Java:   "0.7.2-0",
+}


Test patch:
diff --git a/pkg/shared/util/grpc_utils_test.go b/pkg/shared/util/grpc_utils_test.go
new file mode 100644
index 00000000..2a2fadbe
--- /dev/null
+++ b/pkg/shared/util/grpc_utils_test.go
@@ -0,0 +1,111 @@
+package util
+
+import (
+	"testing"
+
+	"github.com/numaproj/numaflow-go/pkg/info"
+	"github.com/stretchr/testify/assert"
+)
+
+var testMinimumSupportedSDKVersions = sdkConstraints{
+	info.Go:     "0.6.0-0",
+	info.Python: "0.6.0a",
+	info.Java:   "0.6.0-0",
+}
+
+func TestCheckNumaflowCompatibility(t *testing.T) {
+	tests := []struct {
+		name               string
+		numaflowVersion    string
+		minNumaflowVersion string
+		shouldErr          bool
+		errMessage         string
+	}{
+		{
+			name:               "Test with incompatible numaflow version",
+			numaflowVersion:    "v1.1.6",
+			minNumaflowVersion: "1.1.7",
+			shouldErr:          true,
+			errMessage:         "numaflow version 1.1.6 must be upgraded to at least 1.1.7, in order to work with current SDK version",
+		},
+		{
+			name:               "Test with empty MinimumNumaflowVersion field",
+			numaflowVersion:    "1.1.7",
+			minNumaflowVersion: "",
+			shouldErr:          true,
+			errMessage:         "server info does not contain minimum numaflow version. Upgrade to newer SDK version",
+		},
+		{
+			name:               "Test with compatible numaflow version",
+			numaflowVersion:    "1.1.7",
+			minNumaflowVersion: "1.1.6",
+			shouldErr:          false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := checkNumaflowCompatibility(tt.numaflowVersion, tt.minNumaflowVersion)
+			if tt.shouldErr {
+				assert.Error(t, err, "Expected error")
+				assert.Contains(t, err.Error(), tt.errMessage)
+			} else {
+				assert.NoError(t, err, "Expected no error")
+			}
+		})
+	}
+}
+
+func TestCheckSDKCompatibility(t *testing.T) {
+	tests := []struct {
+		name                        string
+		sdkVersion                  string
+		sdkLanguage                 info.Language
+		minimumSupportedSDKVersions sdkConstraints
+		shouldErr                   bool
+		errMessage                  string
+	}{
+		{
+			name:                        "Test with incompatible Python version",
+			sdkVersion:                  "v0.5.3a1",
+			sdkLanguage:                 info.Python,
+			minimumSupportedSDKVersions: testMinimumSupportedSDKVersions,
+			shouldErr:                   true,
+			errMessage:                  "SDK version 0.5.3a1 must be upgraded to at least 0.6.0a, in order to work with current numaflow version",
+		},
+		{
+			name:                        "Test with compatible Python version",
+			sdkVersion:                  "v0.6.0a2",
+			sdkLanguage:                 info.Python,
+			minimumSupportedSDKVersions: testMinimumSupportedSDKVersions,
+			shouldErr:                   false,
+		},
+		{
+			name:                        "Test with incompatible Java version",
+			sdkVersion:                  "v0.4.3",
+			sdkLanguage:                 info.Java,
+			minimumSupportedSDKVersions: testMinimumSupportedSDKVersions,
+			shouldErr:                   true,
+			errMessage:                  "SDK version 0.4.3 must be upgraded to at least 0.6.0-0, in order to work with current numaflow version",
+		},
+		{
+			name:                        "Test with compatible Go version",
+			sdkVersion:                  "v0.6.0-rc2",
+			sdkLanguage:                 info.Go,
+			minimumSupportedSDKVersions: testMinimumSupportedSDKVersions,
+			shouldErr:                   false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := checkSDKCompatibility(tt.sdkVersion, tt.sdkLanguage, tt.minimumSupportedSDKVersions)
+			if tt.shouldErr {
+				assert.Error(t, err, "Expected error")
+				assert.Contains(t, err.Error(), tt.errMessage)
+			} else {
+				assert.NoError(t, err, "Expected no error")
+			}
+		})
+	}
+}
diff --git a/test/sideinputs-e2e/testdata/map-sideinput-pipeline.yaml b/test/sideinputs-e2e/testdata/map-sideinput-pipeline.yaml
index f40dec5b..c60d1721 100644
--- a/test/sideinputs-e2e/testdata/map-sideinput-pipeline.yaml
+++ b/test/sideinputs-e2e/testdata/map-sideinput-pipeline.yaml
@@ -7,8 +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.6.1"
-        imagePullPolicy: Always
+        image: "quay.io/numaio/numaproj-contrib/e2e-map-sideinput-example:latest"
       trigger:
         schedule: "@every 5s"
   vertices:
@@ -19,8 +18,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.6.1"
-          imagePullPolicy: Always
+          image: "quay.io/numaio/numaproj-contrib/e2e-map-sideinput-udf:latest"
       sideInputs:
         - myticker
     - name: sink
@@ -31,9 +29,8 @@ spec:
           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:latest
-            imagePullPolicy: Always
   edges:
     - from: in
       to: si-e2e
     - from: si-e2e
-      to: sink
\ No newline at end of file
+      to: sink
diff --git a/test/sideinputs-e2e/testdata/reduce-sideinput-pipeline.yaml b/test/sideinputs-e2e/testdata/reduce-sideinput-pipeline.yaml
index 689bbf11..54b99889 100644
--- a/test/sideinputs-e2e/testdata/reduce-sideinput-pipeline.yaml
+++ b/test/sideinputs-e2e/testdata/reduce-sideinput-pipeline.yaml
@@ -7,8 +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.6.1"
-        imagePullPolicy: Always
+        image: "quay.io/numaio/numaproj-contrib/e2e-reduce-sideinput-example:latest"
       trigger:
         schedule: "@every 5s"
   vertices:
@@ -27,8 +26,7 @@ spec:
       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:v0.6.1"
-          imagePullPolicy: Always
+          image: "quay.io/numaio/numaproj-contrib/e2e-reduce-sideinput-udf:latest"
         groupBy:
           window:
             fixed:
@@ -54,4 +52,4 @@ spec:
     - from: atoi
       to: si-e2e
     - from: si-e2e
-      to: sink
\ No newline at end of file
+      to: sink
diff --git a/test/sideinputs-e2e/testdata/sideinput_sink.yaml b/test/sideinputs-e2e/testdata/sideinput_sink.yaml
index 83839a8c..8da9f6aa 100644
--- a/test/sideinputs-e2e/testdata/sideinput_sink.yaml
+++ b/test/sideinputs-e2e/testdata/sideinput_sink.yaml
@@ -8,7 +8,6 @@ spec:
       container:
         # see https://github.com/numaproj-contrib/e2e-tests-go/tree/main/sideInput-function
         image: "quay.io/numaio/numaflow-sideinput/sideinput-sink-e2e-test:latest"
-        imagePullPolicy: Always
       trigger:
         schedule: "@every 5s"
         # timezone: America/Los_Angeles
diff --git a/test/sideinputs-e2e/testdata/sideinput_source.yaml b/test/sideinputs-e2e/testdata/sideinput_source.yaml
index 12dc52dc..25e424bc 100644
--- a/test/sideinputs-e2e/testdata/sideinput_source.yaml
+++ b/test/sideinputs-e2e/testdata/sideinput_source.yaml
@@ -8,7 +8,6 @@ spec:
       container:
         # see https://github.com/numaproj-contrib/e2e-tests-go/tree/main/sideInput-function
         image: "quay.io/numaio/numaflow-sideinput/sideinput-sink-e2e-test:latest"
-        imagePullPolicy: Always
       trigger:
         schedule: "@every 5s"
         # timezone: America/Los_Angeles
@@ -34,5 +33,3 @@ spec:
   edges:
     - from: in
       to: redis-uds
-
-