Repository: numaproj/numaflow
Pull request #173 (2022-09-16 07:06:34)

chore: enable jetstream 2.9.0
Signed-off-by: Derek Wang <whynowy@gmail.com>

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/README.md b/README.md
index 8a663849..8da5e62f 100644
--- a/README.md
+++ b/README.md
@@ -8,19 +8,24 @@
 
 ## Summary
 
-Numaflow is a Kubernetes-native platform for running massive parallel data processing and streaming jobs.
+Numaflow is a Kubernetes-native tool for running massively parallel stream processing. A Numaflow Pipeline is implemented as a Kubernetes custom resource and consists of one or more source, data processing, and sink vertices.
 
-A Numaflow Pipeline is implemented as a Kubernetes custom resource, and consists of one or more source, data processing, and sink vertices.
 
-Numaflow installs in a few minutes and is easier and cheaper to use for simple data processing applications than full-featured stream processing platforms.
+Numaflow installs in a few minutes and is easier and cheaper to use for simple data processing applications than a full-featured stream processing platforms.
 
 ## Key Features
 
-- Kubernetes-native: If you know Kubernetes, you already know 90% of what you need to use Numaflow.
+- Kubernetes-native: If you know Kubernetes, you already know how to use Numaflow.
 - Language agnostic: Use your favorite programming language.
 - Exactly-Once semantics: No input element is duplicated or lost even as pods are rescheduled or restarted.
 - Auto-scaling with back-pressure: Each vertex automatically scales from zero to whatever is needed.
 
+
+## Data Integrity Guarantees:
+- Minimally provide at-least-once semantics
+- Provide exactly-once semantics for unbounded and near real-time data sources 
+- Preserving order is not required
+
 ## Roadmap
 
 - Data aggregation (e.g. group-by)
diff --git a/config/advanced-install/namespaced-controller-wo-crds.yaml b/config/advanced-install/namespaced-controller-wo-crds.yaml
index e19a2ef7..cbfe5fce 100644
--- a/config/advanced-install/namespaced-controller-wo-crds.yaml
+++ b/config/advanced-install/namespaced-controller-wo-crds.yaml
@@ -161,7 +161,7 @@ data:
             replicas: 3
         versions:
         - version: latest
-          natsImage: nats:2.8.3
+          natsImage: nats:2.9.0
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: /nats-server
@@ -185,6 +185,16 @@ data:
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: nats-server
+        - version: 2.9.0
+          natsImage: nats:2.9.0
+          metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
+          configReloaderImage: natsio/nats-server-config-reloader:0.7.0
+          startCommand: /nats-server
+        - version: 2.9.0-alpine
+          natsImage: nats:2.9.0-alpine
+          metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
+          configReloaderImage: natsio/nats-server-config-reloader:0.7.0
+          startCommand: nats-server
 kind: ConfigMap
 metadata:
   name: numaflow-controller-config
diff --git a/config/base/controller-manager/numaflow-controller-config.yaml b/config/base/controller-manager/numaflow-controller-config.yaml
index 54d6a233..ea8de432 100644
--- a/config/base/controller-manager/numaflow-controller-config.yaml
+++ b/config/base/controller-manager/numaflow-controller-config.yaml
@@ -80,7 +80,7 @@ data:
             replicas: 3
         versions:
         - version: latest
-          natsImage: nats:2.8.3
+          natsImage: nats:2.9.0
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: /nats-server
@@ -104,3 +104,13 @@ data:
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: nats-server
+        - version: 2.9.0
+          natsImage: nats:2.9.0
+          metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
+          configReloaderImage: natsio/nats-server-config-reloader:0.7.0
+          startCommand: /nats-server
+        - version: 2.9.0-alpine
+          natsImage: nats:2.9.0-alpine
+          metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
+          configReloaderImage: natsio/nats-server-config-reloader:0.7.0
+          startCommand: nats-server
diff --git a/config/install.yaml b/config/install.yaml
index 888bf183..f02d4c74 100644
--- a/config/install.yaml
+++ b/config/install.yaml
@@ -12870,7 +12870,7 @@ data:
             replicas: 3
         versions:
         - version: latest
-          natsImage: nats:2.8.3
+          natsImage: nats:2.9.0
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: /nats-server
@@ -12894,6 +12894,16 @@ data:
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: nats-server
+        - version: 2.9.0
+          natsImage: nats:2.9.0
+          metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
+          configReloaderImage: natsio/nats-server-config-reloader:0.7.0
+          startCommand: /nats-server
+        - version: 2.9.0-alpine
+          natsImage: nats:2.9.0-alpine
+          metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
+          configReloaderImage: natsio/nats-server-config-reloader:0.7.0
+          startCommand: nats-server
 kind: ConfigMap
 metadata:
   name: numaflow-controller-config
diff --git a/config/namespace-install.yaml b/config/namespace-install.yaml
index 266eb38f..76f4e273 100644
--- a/config/namespace-install.yaml
+++ b/config/namespace-install.yaml
@@ -12789,7 +12789,7 @@ data:
             replicas: 3
         versions:
         - version: latest
-          natsImage: nats:2.8.3
+          natsImage: nats:2.9.0
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: /nats-server
@@ -12813,6 +12813,16 @@ data:
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: nats-server
+        - version: 2.9.0
+          natsImage: nats:2.9.0
+          metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
+          configReloaderImage: natsio/nats-server-config-reloader:0.7.0
+          startCommand: /nats-server
+        - version: 2.9.0-alpine
+          natsImage: nats:2.9.0-alpine
+          metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
+          configReloaderImage: natsio/nats-server-config-reloader:0.7.0
+          startCommand: nats-server
 kind: ConfigMap
 metadata:
   name: numaflow-controller-config
diff --git a/docs/quick-start.md b/docs/quick-start.md
index 0f356131..9008c3ed 100644
--- a/docs/quick-start.md
+++ b/docs/quick-start.md
@@ -5,11 +5,13 @@ Install Numaflow and run a couple of example pipelines.
 ## Prerequisites
 
 A Kubernetes cluster is needed to try out Numaflow. A simple way to create a local cluster is using Docker Desktop.
-* https://www.docker.com/
-* https://www.docker.com/blog/how-kubernetes-works-under-the-hood-with-docker-desktop/
+
+- [Docker](https://www.docker.com)
+- [Docker Desktop](https://www.docker.com/blog/how-kubernetes-works-under-the-hood-with-docker-desktop)
 
 You will also need `kubectl` to manage the cluster.
-* [`kubectl`](https://kubernetes.io/docs/tasks/tools/install-kubectl/)
+
+- [kubectl](https://kubernetes.io/docs/tasks/tools/install-kubectl/)
 
 ## Installation
 
@@ -107,7 +109,7 @@ kubectl logs -f even-odd-odd-sink-0-a6p0n main
 View the UI for the advanced pipleline at https://localhost:8443/
 ![Numaflow UI](assets/numaflow-ui-advanced-pipeline.png)
 
-The source code of the `even-odd` [User Defined Function](./user-defined-functions.md) can be found [here](https://github.com/numaproj/numaflow-go/tree/main/examples/function/evenodd). You also can replace the [Log](./sinks/log.md) Sink with some other sinks like [Kafka](./sinks/kafka.md) to forward the data to Kafka topics.
+The source code of the `even-odd` [User Defined Function](./user-defined-functions.md) can be found [here](https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/evenodd). You also can replace the [Log](./sinks/log.md) Sink with some other sinks like [Kafka](./sinks/kafka.md) to forward the data to Kafka topics.
 
 The pipeline can be deleted by
 ```shell
diff --git a/docs/user-defined-functions.md b/docs/user-defined-functions.md
index 5f2fae47..26c396af 100644
--- a/docs/user-defined-functions.md
+++ b/docs/user-defined-functions.md
@@ -20,23 +20,24 @@ package main
 import (
 	"context"
 
-	funcsdk "github.com/numaproj/numaflow-go/function"
+	functionsdk "github.com/numaproj/numaflow-go/pkg/function"
+	"github.com/numaproj/numaflow-go/pkg/function/server"
 )
 
-// Simply return the same msg
-func handle(ctx context.Context, key, msg []byte) (funcsdk.Messages, error) {
-	return funcsdk.MessagesBuilder().Append(funcsdk.MessageToAll(msg)), nil
+func mapHandle(_ context.Context, key string, d functionsdk.Datum) functionsdk.Messages {
+	// directly forward the input to the output
+	return functionsdk.MessagesBuilder().Append(functionsdk.MessageToAll(d.Value()))
 }
 
 func main() {
-	funcsdk.Start(context.Background(), handle)
+	server.New().RegisterMapper(functionsdk.MapFunc(mapHandle)).Start(context.Background())
 }
 ```
 
 Check the links below to see the UDF examples for different languages.
 
 - [Python](https://github.com/numaproj/numaflow-python/tree/main/examples/function)
-- [Golang](https://github.com/numaproj/numaflow-go/tree/main/examples/function)
+- [Golang](https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples)
 
 After building a docker image for the written UDF, specify the image as below in the vertex spec.
 
diff --git a/go.mod b/go.mod
index 177efbb7..3a7d76c0 100644
--- a/go.mod
+++ b/go.mod
@@ -24,14 +24,13 @@ require (
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe
 	github.com/nats-io/nats.go v1.16.0
-	github.com/numaproj/numaflow-go v0.2.1
+	github.com/numaproj/numaflow-go v0.2.3
 	github.com/prometheus/client_golang v1.12.1
 	github.com/prometheus/common v0.32.1
 	github.com/soheilhy/cmux v0.1.5
 	github.com/spf13/cobra v1.2.1
 	github.com/spf13/viper v1.9.0
 	github.com/stretchr/testify v1.7.2
-	github.com/vmihailenco/msgpack/v5 v5.3.5
 	go.uber.org/atomic v1.9.0
 	go.uber.org/zap v1.19.1
 	google.golang.org/genproto v0.0.0-20220118154757-00ab72f36ad5
@@ -153,6 +152,7 @@ require (
 	github.com/ugorji/go/codec v1.2.7 // indirect
 	github.com/valyala/bytebufferpool v1.0.0 // indirect
 	github.com/valyala/fasthttp v1.37.0 // indirect
+	github.com/vmihailenco/msgpack/v5 v5.3.5 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f // indirect
 	github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect
@@ -162,7 +162,7 @@ require (
 	github.com/yudai/golcs v0.0.0-20170316035057-ecda9a501e82 // indirect
 	go.mongodb.org/mongo-driver v1.7.3 // indirect
 	go.uber.org/multierr v1.7.0 // indirect
-	golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d // indirect
+	golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 // indirect
 	golang.org/x/mod v0.5.1 // indirect
 	golang.org/x/net v0.0.0-20220225172249-27dd8689420f // indirect
 	golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8 // indirect
diff --git a/go.sum b/go.sum
index 2cbe9766..0b73fcb0 100644
--- a/go.sum
+++ b/go.sum
@@ -690,8 +690,8 @@ github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV
 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.2.1 h1:yxTwMZF/11Q+q+qbG+v5QETxPeksvIVnWKLzZFhlSJU=
-github.com/numaproj/numaflow-go v0.2.1/go.mod h1:q4POBeeCxEZ3sA6WMrZkZRgK28YkoDULEnAhGRJVO3k=
+github.com/numaproj/numaflow-go v0.2.3 h1:DDyLK6SehUmXUq/RwQemawPd3TfXqJLepA1fwYUrN0U=
+github.com/numaproj/numaflow-go v0.2.3/go.mod h1:jwhmgurhkWs3YMzpPPl4qXf3OohdE18jc0YKM7Ux2iQ=
 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=
@@ -968,8 +968,8 @@ golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5y
 golang.org/x/crypto v0.0.0-20210817164053-32db794688a5/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20210920023735-84f357641f63/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20220214200702-86341886e292/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d h1:sK3txAijHtOK88l68nt020reeT1ZdKLIYetKl95FzVY=
-golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 h1:Y/gsMcFOcR+6S6f3YeMKl5g+dZMEWqcz5Czj/GWYbkM=
+golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
diff --git a/pkg/apis/numaflow/v1alpha1/vertex_types.go b/pkg/apis/numaflow/v1alpha1/vertex_types.go
index 4fff97c8..51404c32 100644
--- a/pkg/apis/numaflow/v1alpha1/vertex_types.go
+++ b/pkg/apis/numaflow/v1alpha1/vertex_types.go
@@ -469,7 +469,7 @@ func (s Scale) GetMinReplicas() int32 {
 }
 
 func (s Scale) GetMaxReplicas() int32 {
-	if x := s.Max; x == nil || *x > DefaultMaxReplicas {
+	if x := s.Max; x == nil {
 		return DefaultMaxReplicas
 	} else {
 		return *x
diff --git a/pkg/sinks/udsink/sink.go b/pkg/sinks/udsink/sink.go
index c3bfd2fd..61eff987 100644
--- a/pkg/sinks/udsink/sink.go
+++ b/pkg/sinks/udsink/sink.go
@@ -2,18 +2,19 @@ package udsink
 
 import (
 	"context"
+	"fmt"
 	"time"
 
-	sinksdk "github.com/numaproj/numaflow-go/sink"
+	sinkpb "github.com/numaproj/numaflow-go/pkg/apis/proto/sink/v1"
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/forward"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
-	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	"go.uber.org/zap"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 type userDefinedSink struct {
@@ -21,7 +22,7 @@ type userDefinedSink struct {
 	pipelineName string
 	isdf         *forward.InterStepDataForward
 	logger       *zap.SugaredLogger
-	udsink       *udsHTTPBasedUDSink
+	udsink       *udsGRPCBasedUDSink
 }
 
 type Option func(*userDefinedSink) error
@@ -54,8 +55,11 @@ func NewUserDefinedSink(vertex *dfv1.Vertex, fromBuffer isb.BufferReader, fetchW
 			forwardOpts = append(forwardOpts, forward.WithReadBatchSize(int64(*x.ReadBatchSize)))
 		}
 	}
-	contentType := sharedutil.LookupEnvStringOr(dfv1.EnvUDSinkContentType, string(dfv1.MsgPackType))
-	s.udsink = NewUDSHTTPBasedUDSink(dfv1.PathVarRun+"/udsink.sock", withTimeout(20*time.Second), withContentType(dfv1.ContentType(contentType)))
+	udsink, err := NewUDSGRPCBasedUDSink()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create gRPC client, %w", err)
+	}
+	s.udsink = udsink
 	isdf, err := forward.NewInterStepDataForward(vertex, fromBuffer, map[string]isb.BufferWriter{vertex.GetToBuffers()[0].Name: s}, forward.All, applier.Terminal, fetchWatermark, publishWatermark, forwardOpts...)
 	if err != nil {
 		return nil, err
@@ -74,15 +78,22 @@ func (s *userDefinedSink) IsFull() bool {
 
 // Write writes to the UDSink container.
 func (s *userDefinedSink) Write(ctx context.Context, messages []isb.Message) ([]isb.Offset, []error) {
-	msgs := make([]sinksdk.Message, len(messages))
+	msgs := make([]*sinkpb.Datum, len(messages))
 	for i, m := range messages {
-		msgs[i] = sinksdk.Message{ID: m.ID, Payload: m.Payload}
+		msgs[i] = &sinkpb.Datum{
+			// NOTE: key is not used anywhere ATM
+			Id:        m.ID,
+			Value:     m.Payload,
+			EventTime: &sinkpb.EventTime{EventTime: timestamppb.New(m.EventTime)},
+			Watermark: &sinkpb.Watermark{Watermark: timestamppb.New(time.Time{})}, // TODO: insert the correct watermark
+		}
 	}
 	return nil, s.udsink.Apply(ctx, msgs)
 }
 
 func (br *userDefinedSink) Close() error {
-	return nil
+	// TODO: context.Background()
+	return br.udsink.CloseConn(context.Background())
 }
 
 func (s *userDefinedSink) Start() <-chan struct{} {
diff --git a/pkg/sinks/udsink/udsink_grpc.go b/pkg/sinks/udsink/udsink_grpc.go
new file mode 100644
index 00000000..f2ef87f3
--- /dev/null
+++ b/pkg/sinks/udsink/udsink_grpc.go
@@ -0,0 +1,82 @@
+package udsink
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	sinkpb "github.com/numaproj/numaflow-go/pkg/apis/proto/sink/v1"
+	sinksdk "github.com/numaproj/numaflow-go/pkg/sink"
+	"github.com/numaproj/numaflow-go/pkg/sink/client"
+	"google.golang.org/protobuf/types/known/emptypb"
+)
+
+type udsGRPCBasedUDSink struct {
+	client sinksdk.Client
+}
+
+func NewUDSGRPCBasedUDSink() (*udsGRPCBasedUDSink, error) {
+	c, err := client.New()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create a new gRPC client: %w", err)
+	}
+	return &udsGRPCBasedUDSink{c}, nil
+}
+
+// CloseConn closes the gRPC client connection.
+func (u *udsGRPCBasedUDSink) CloseConn(ctx context.Context) error {
+	return u.client.CloseConn(ctx)
+}
+
+// WaitUntilReady waits until the client is connected.
+func (u *udsGRPCBasedUDSink) WaitUntilReady(ctx context.Context) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("failed on readiness check: %w", ctx.Err())
+		default:
+			if _, err := u.client.IsReady(ctx, &emptypb.Empty{}); err == nil {
+				return nil
+			}
+			time.Sleep(1 * time.Second)
+		}
+	}
+}
+
+func (u *udsGRPCBasedUDSink) Apply(ctx context.Context, dList []*sinkpb.Datum) []error {
+	errs := make([]error, len(dList))
+
+	responseList, err := u.client.SinkFn(ctx, dList)
+	if err != nil {
+		for i := range dList {
+			errs[i] = ApplyUDSinkErr{
+				UserUDSinkErr: false,
+				Message:       fmt.Sprintf("gRPC client.SinkFn failed, %s", err),
+				InternalErr: InternalErr{
+					Flag:        true,
+					MainCarDown: false,
+				},
+			}
+		}
+		return errs
+	}
+	// Use ID to map the response messages, so that there's no strict requirement for the user defined sink to return the responseList in order.
+	resMap := make(map[string]*sinkpb.Response)
+	for _, res := range responseList {
+		resMap[res.GetId()] = res
+	}
+	for i, m := range dList {
+		if r, existing := resMap[m.GetId()]; !existing {
+			errs[i] = fmt.Errorf("not found in responseList")
+		} else {
+			if !r.Success {
+				if r.GetErrMsg() != "" {
+					errs[i] = fmt.Errorf(r.GetErrMsg())
+				} else {
+					errs[i] = fmt.Errorf("unsuccessful due to unknown reason")
+				}
+			}
+		}
+	}
+	return errs
+}
diff --git a/pkg/sinks/udsink/udsink_http.go b/pkg/sinks/udsink/udsink_http.go
deleted file mode 100644
index 28dc9e78..00000000
--- a/pkg/sinks/udsink/udsink_http.go
+++ /dev/null
@@ -1,236 +0,0 @@
-package udsink
-
-import (
-	"bytes"
-	"context"
-	"encoding/json"
-	"fmt"
-	"io"
-	"net"
-	"net/http"
-	"time"
-
-	sinksdk "github.com/numaproj/numaflow-go/sink"
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/shared/logging"
-	"github.com/vmihailenco/msgpack/v5"
-)
-
-type udsHTTPBasedUDSink struct {
-	client      *http.Client
-	contentType dfv1.ContentType
-}
-
-// options for HTTPBasedUDF
-type options struct {
-	httpClientTimeout time.Duration
-	contentType       dfv1.ContentType
-}
-
-type udsSinkOption func(o *options)
-
-func withTimeout(t time.Duration) udsSinkOption {
-	return func(o *options) {
-		o.httpClientTimeout = t
-	}
-}
-
-func withContentType(t dfv1.ContentType) udsSinkOption {
-	return func(o *options) {
-		o.contentType = t
-	}
-}
-
-func NewUDSHTTPBasedUDSink(socketPath string, opts ...udsSinkOption) *udsHTTPBasedUDSink {
-	options := &options{
-		httpClientTimeout: time.Second * 10,
-		contentType:       dfv1.MsgPackType,
-	}
-	for _, o := range opts {
-		o(options)
-	}
-	var httpClient *http.Client
-	httpTransport := http.DefaultTransport.(*http.Transport).Clone()
-	httpTransport.MaxIdleConns = 100
-	httpTransport.MaxConnsPerHost = 100
-	httpTransport.MaxIdleConnsPerHost = 100
-	httpTransport.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) {
-		return net.Dial("unix", socketPath)
-	}
-	httpClient = &http.Client{
-		Transport: httpTransport,
-		Timeout:   options.httpClientTimeout,
-	}
-	return &udsHTTPBasedUDSink{
-		client:      httpClient,
-		contentType: options.contentType,
-	}
-}
-
-func (u *udsHTTPBasedUDSink) Apply(ctx context.Context, msgs []sinksdk.Message) []error {
-	errs := make([]error, len(msgs))
-	payload, err := marshalMessages(u.contentType, msgs)
-	if err != nil {
-		for i := range msgs {
-			errs[i] = ApplyUDSinkErr{
-				UserUDSinkErr: false,
-				Message:       fmt.Sprintf("marshal messages failed, %s", err),
-				InternalErr: InternalErr{
-					Flag:        true,
-					MainCarDown: false,
-				},
-			}
-		}
-		return errs
-	}
-	req, err := http.NewRequestWithContext(ctx, "POST", "http://unix/messages", bytes.NewBuffer(payload))
-	if err != nil {
-		for i := range msgs {
-			errs[i] = ApplyUDSinkErr{
-				UserUDSinkErr: false,
-				Message:       fmt.Sprintf("http.NewRequestWithContext failed, %s", err),
-				InternalErr: InternalErr{
-					Flag:        true,
-					MainCarDown: false,
-				},
-			}
-		}
-		return errs
-	}
-	req.Header.Set("Content-Type", string(u.contentType))
-
-	// hold results (from response)
-	var data []byte
-	var responseContentType dfv1.ContentType
-
-	i := 0
-	retryCount := 3
-	var failed = true
-retry:
-	for i < retryCount && failed {
-		i++
-		failed = false
-		if resp, err := u.client.Do(req); err != nil {
-			for i := range msgs {
-				errs[i] = ApplyUDSinkErr{
-					UserUDSinkErr: false,
-					Message:       fmt.Sprintf("client.Do failed, %s", err),
-					InternalErr: InternalErr{
-						Flag:        true,
-						MainCarDown: false,
-					},
-				}
-			}
-			return errs
-		} else {
-			data, err = io.ReadAll(resp.Body)
-			if err != nil {
-				failed = true
-				logging.FromContext(ctx).Warnf("io.ReadAll failed (%d/%d), %w", i, retryCount, err)
-				goto retry
-			}
-			_ = resp.Body.Close()
-			if resp.StatusCode >= 300 {
-				failed = true
-				logging.FromContext(ctx).Warnf("maincar returned statusCode=%d (%d/%d) %s", resp.StatusCode, i, retryCount, data)
-				goto retry
-			}
-			responseContentType = dfv1.ContentType(resp.Header.Get("Content-Type"))
-		}
-	}
-	// ran out of retry limit
-	if failed {
-		for i := range msgs {
-			errs[i] = ApplyUDSinkErr{
-				UserUDSinkErr: true,
-				Message:       "ran out of retry limit",
-				InternalErr: InternalErr{
-					Flag:        false,
-					MainCarDown: false,
-				},
-			}
-		}
-		return errs
-	}
-
-	responses, err := unmarshalMessageResponses(responseContentType, data)
-	if err != nil {
-		for i := range msgs {
-			errs[i] = ApplyUDSinkErr{
-				UserUDSinkErr: false,
-				Message:       fmt.Sprintf("failed to unmarshal message responses, %s", err),
-				InternalErr: InternalErr{
-					Flag:        true,
-					MainCarDown: false,
-				},
-			}
-		}
-		return errs
-	}
-	// Use ID to map the response messages, so that there's no strict requirement for the user defined sink to return the responses in order.
-	resMap := make(map[string]sinksdk.Response)
-	for _, res := range responses.Items() {
-		resMap[res.ID] = res
-	}
-	for i, m := range msgs {
-		if r, existing := resMap[m.ID]; !existing {
-			errs[i] = fmt.Errorf("not found in responses")
-		} else {
-			if !r.Success {
-				if r.Err != "" {
-					errs[i] = fmt.Errorf(r.Err)
-				} else {
-					errs[i] = fmt.Errorf("unsuccessful due to unknown reason")
-				}
-			}
-		}
-	}
-	return errs
-}
-
-// WaitUntilReady waits till the readyURL is available
-func (u *udsHTTPBasedUDSink) WaitUntilReady(ctx context.Context) error {
-	for {
-		select {
-		case <-ctx.Done():
-			return fmt.Errorf("failed to wait for ready: %w", ctx.Err())
-		default:
-			if resp, err := u.client.Get("http://unix/ready"); err == nil {
-				_, _ = io.Copy(io.Discard, resp.Body)
-				_ = resp.Body.Close()
-				if resp.StatusCode < 300 {
-					return nil
-				}
-			}
-			time.Sleep(1 * time.Second)
-		}
-	}
-}
-
-func marshalMessages(contentType dfv1.ContentType, msgs []sinksdk.Message) ([]byte, error) {
-	switch contentType {
-	case dfv1.JsonType:
-		return json.Marshal(msgs)
-	case dfv1.MsgPackType:
-		return msgpack.Marshal(msgs)
-	default:
-		return nil, fmt.Errorf("unsupported UDSink Content-Type %q", string(contentType))
-	}
-}
-
-func unmarshalMessageResponses(contentType dfv1.ContentType, data []byte) (sinksdk.Responses, error) {
-	responses := sinksdk.Responses{}
-	switch contentType {
-	case dfv1.JsonType:
-		if err := json.Unmarshal(data, &responses); err != nil {
-			return nil, fmt.Errorf("failed to unmarshal data from the response with json, %w", err)
-		}
-	case dfv1.MsgPackType:
-		if err := msgpack.Unmarshal(data, &responses); err != nil {
-			return nil, fmt.Errorf("failed to unmarshal data from the response with msgpack, %w", err)
-		}
-	default:
-		return nil, fmt.Errorf("unsupported UDSink Content-Type %q", string(contentType))
-	}
-	return responses, nil
-}
diff --git a/pkg/udf/applier/uds_grpc.go b/pkg/udf/applier/uds_grpc.go
index 8cec61aa..d509506b 100644
--- a/pkg/udf/applier/uds_grpc.go
+++ b/pkg/udf/applier/uds_grpc.go
@@ -14,29 +14,29 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
-// UDSGRPCBasedUDF applies user defined function over gRPC (over Unix Domain Socket) client/server where server is the UDF.
-type UDSGRPCBasedUDF struct {
+// udsGRPCBasedUDF applies user defined function over gRPC (over Unix Domain Socket) client/server where server is the UDF.
+type udsGRPCBasedUDF struct {
 	client functionsdk.Client
 }
 
-var _ Applier = (*UDSGRPCBasedUDF)(nil)
+var _ Applier = (*udsGRPCBasedUDF)(nil)
 
-// NewUDSGRPCBasedUDF returns a new UDSGRPCBasedUDF object.
-func NewUDSGRPCBasedUDF() (*UDSGRPCBasedUDF, error) {
+// NewUDSGRPCBasedUDF returns a new udsGRPCBasedUDF object.
+func NewUDSGRPCBasedUDF() (*udsGRPCBasedUDF, error) {
 	c, err := client.New()
 	if err != nil {
 		return nil, fmt.Errorf("failed to create a new gRPC client: %w", err)
 	}
-	return &UDSGRPCBasedUDF{c}, nil
+	return &udsGRPCBasedUDF{c}, nil
 }
 
 // CloseConn closes the gRPC client connection.
-func (u *UDSGRPCBasedUDF) CloseConn(ctx context.Context) error {
+func (u *udsGRPCBasedUDF) CloseConn(ctx context.Context) error {
 	return u.client.CloseConn(ctx)
 }
 
 // WaitUntilReady waits until the client is connected.
-func (u *UDSGRPCBasedUDF) WaitUntilReady(ctx context.Context) error {
+func (u *udsGRPCBasedUDF) WaitUntilReady(ctx context.Context) error {
 	for {
 		select {
 		case <-ctx.Done():
@@ -50,7 +50,7 @@ func (u *UDSGRPCBasedUDF) WaitUntilReady(ctx context.Context) error {
 	}
 }
 
-func (u *UDSGRPCBasedUDF) Apply(ctx context.Context, readMessage *isb.ReadMessage) ([]*isb.Message, error) {
+func (u *udsGRPCBasedUDF) Apply(ctx context.Context, readMessage *isb.ReadMessage) ([]*isb.Message, error) {
 	key := readMessage.Key
 	payload := readMessage.Body.Payload
 	offset := readMessage.ReadOffset
diff --git a/pkg/udf/applier/uds_http.go b/pkg/udf/applier/uds_http.go
deleted file mode 100644
index 2a3b7e7b..00000000
--- a/pkg/udf/applier/uds_http.go
+++ /dev/null
@@ -1,214 +0,0 @@
-/*
-Package udf calls the user defined function. It supports HTTP based invocation and
-can be extended to stdin/stdout FIFO etc.
-*/
-
-package applier
-
-import (
-	"bytes"
-	"context"
-	"encoding/json"
-	"fmt"
-	"io"
-	"net"
-	"net/http"
-	"time"
-
-	funcsdk "github.com/numaproj/numaflow-go/function"
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/shared/logging"
-	"github.com/vmihailenco/msgpack/v5"
-)
-
-// UDSHTTPBasedUDF applies user defined function over HTTP (over Unix Domain Socket) client/server where server is the UDF.
-type UDSHTTPBasedUDF struct {
-	client *http.Client
-}
-
-var _ Applier = (*UDSHTTPBasedUDF)(nil)
-
-// options for HTTPBasedUDF
-type options struct {
-	httpClientTimeout time.Duration
-}
-
-// Option to apply different options
-type Option interface {
-	apply(*options)
-}
-
-type httpClientTimeout time.Duration
-
-func (h httpClientTimeout) apply(opts *options) {
-	opts.httpClientTimeout = time.Duration(h)
-}
-
-// WithHTTPClientTimeout sets timeout for the HTTP Client
-func WithHTTPClientTimeout(t time.Duration) Option {
-	return httpClientTimeout(t)
-}
-
-// NewUDSHTTPBasedUDF returns UDSHTTPBasedUDF.
-// Parameter - socketPath, Unix Domain Socket path
-func NewUDSHTTPBasedUDF(socketPath string, opts ...Option) *UDSHTTPBasedUDF {
-	options := options{
-		// default REST standard is 10 seconds
-		httpClientTimeout: time.Second * 10,
-	}
-	for _, o := range opts {
-		o.apply(&options)
-	}
-
-	var httpClient *http.Client
-	// https://www.loginradius.com/blog/async/tune-the-go-http-client-for-high-performance/
-	httpTransport := http.DefaultTransport.(*http.Transport).Clone()
-	// all our connects are loop back
-	httpTransport.MaxIdleConns = 100
-	httpTransport.MaxConnsPerHost = 100
-	httpTransport.MaxIdleConnsPerHost = 100
-	httpTransport.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) {
-		return net.Dial("unix", socketPath)
-	}
-	httpClient = &http.Client{
-		Transport: httpTransport,
-		Timeout:   options.httpClientTimeout,
-	}
-
-	return &UDSHTTPBasedUDF{
-		client: httpClient,
-	}
-}
-
-// Apply applies the user defined function.
-func (u *UDSHTTPBasedUDF) Apply(ctx context.Context, readMessage *isb.ReadMessage) ([]*isb.Message, error) {
-	payload := readMessage.Body.Payload
-	offset := readMessage.ReadOffset
-	parentPaneInfo := readMessage.PaneInfo
-
-	req, err := http.NewRequestWithContext(ctx, "POST", "http://unix/messages", bytes.NewBuffer(payload))
-	// looking at the code, err returned by NewRequestWithContext are InternalErrs which cannot be resolved by retrying.
-	// Let's return InternalErr and pause the pipeline.
-	if err != nil {
-		return nil, ApplyUDFErr{
-			UserUDFErr: false,
-			Message:    fmt.Sprintf("http.NewRequestWithContext failed, %s", err),
-			InternalErr: InternalErr{
-				Flag:        true,
-				MainCarDown: false,
-			},
-		}
-	}
-	req.Header.Set(dfv1.UDFApplierMessageKey, string(readMessage.Key))
-
-	// hold results (from response)
-	var data []byte
-	var contentType dfv1.ContentType
-
-	// retry if we think it is a transient error. maybe we should make this retry more elegant :-)
-	i := 0
-	retryCount := 3
-	var failed = true
-retry:
-	for i < retryCount && failed {
-		i++
-		failed = false
-		if resp, err := u.client.Do(req); err != nil {
-			return nil, ApplyUDFErr{
-				UserUDFErr: false,
-				Message:    fmt.Sprintf("client.Do failed, %s", err),
-				InternalErr: InternalErr{
-					Flag:        true,
-					MainCarDown: false,
-				},
-			}
-		} else {
-			data, err = io.ReadAll(resp.Body)
-			if err != nil {
-				failed = true
-				logging.FromContext(ctx).Warnf("io.ReadAll failed (%d/%d), %s", i, retryCount, err)
-				goto retry
-			}
-			_ = resp.Body.Close()
-			// TODO: differentiate between 502, 504 from others etc
-			if resp.StatusCode >= 300 {
-				failed = true
-				logging.FromContext(ctx).Warnf("maincar returned statusCode=%d (%d/%d) %s", resp.StatusCode, i, retryCount, data)
-				goto retry
-			}
-			contentType = dfv1.ContentType(resp.Header.Get("Content-Type"))
-		}
-	}
-	// ran out of retry limit
-	if failed {
-		return nil, ApplyUDFErr{
-			UserUDFErr: true,
-			Message:    "ran out of retry limit",
-			InternalErr: InternalErr{
-				Flag:        false,
-				MainCarDown: false,
-			},
-		}
-	}
-	messages, err := unmarshalMessages(contentType, data)
-	if err != nil {
-		return nil, err
-	}
-	writeMessages := []*isb.Message{}
-	for i, m := range messages.Items() {
-		key := m.Key
-		if key == nil {
-			key = []byte{}
-		}
-		writeMessage := &isb.Message{
-			Header: isb.Header{
-				PaneInfo: parentPaneInfo,
-				ID:       fmt.Sprintf("%s-%d", offset.String(), i),
-				// TODO: change to use string directly after SDK upgrade
-				Key: string(key),
-			},
-			Body: isb.Body{
-				Payload: m.Value,
-			},
-		}
-		writeMessages = append(writeMessages, writeMessage)
-	}
-	return writeMessages, nil
-}
-
-// WaitUntilReady waits till the readyURL is available
-func (u *UDSHTTPBasedUDF) WaitUntilReady(ctx context.Context) error {
-	for {
-		select {
-		case <-ctx.Done():
-			return fmt.Errorf("failed to wait for ready: %w", ctx.Err())
-		default:
-			if resp, err := u.client.Get("http://unix/ready"); err == nil {
-				_, _ = io.Copy(io.Discard, resp.Body)
-				_ = resp.Body.Close()
-				if resp.StatusCode < 300 {
-					return nil
-				}
-			}
-			time.Sleep(1 * time.Second)
-		}
-	}
-}
-
-func unmarshalMessages(contentType dfv1.ContentType, data []byte) (*funcsdk.Messages, error) {
-	messages := &funcsdk.Messages{}
-	switch contentType {
-	case dfv1.JsonType:
-		if err := json.Unmarshal(data, &messages); err != nil {
-			return nil, fmt.Errorf("failed to unmarshal data from the response with json, %w", err)
-		}
-	case dfv1.MsgPackType:
-		if err := msgpack.Unmarshal(data, &messages); err != nil {
-			return nil, fmt.Errorf("failed to unmarshal data from the response with msgpack, %w", err)
-		}
-	default:
-		return nil, fmt.Errorf("unsupported UDF Content-Type %q", string(contentType))
-	}
-	return messages, nil
-}


Test patch:
diff --git a/pkg/apis/numaflow/v1alpha1/vertex_types_test.go b/pkg/apis/numaflow/v1alpha1/vertex_types_test.go
index b9ce79a3..3f02e6de 100644
--- a/pkg/apis/numaflow/v1alpha1/vertex_types_test.go
+++ b/pkg/apis/numaflow/v1alpha1/vertex_types_test.go
@@ -407,4 +407,6 @@ func Test_Scale_Parameters(t *testing.T) {
 	assert.Equal(t, int(tbu), s.GetTargetBufferUsage())
 	assert.Equal(t, int(tps), s.GetTargetProcessingSeconds())
 	assert.Equal(t, int(zrss), s.GetZeroReplicaSleepSeconds())
+	s.Max = pointer.Int32(500)
+	assert.Equal(t, int32(500), s.GetMaxReplicas())
 }
diff --git a/pkg/sinks/udsink/udsink_grpc_test.go b/pkg/sinks/udsink/udsink_grpc_test.go
new file mode 100644
index 00000000..1d0a4b32
--- /dev/null
+++ b/pkg/sinks/udsink/udsink_grpc_test.go
@@ -0,0 +1,177 @@
+package udsink
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/golang/mock/gomock"
+	sinkpb "github.com/numaproj/numaflow-go/pkg/apis/proto/sink/v1"
+	"github.com/numaproj/numaflow-go/pkg/apis/proto/sink/v1/sinkmock"
+	"github.com/numaproj/numaflow-go/pkg/sink/clienttest"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+func NewMockUDSGRPCBasedUDF(mockClient *sinkmock.MockUserDefinedSinkClient) *udsGRPCBasedUDSink {
+	c, _ := clienttest.New(mockClient)
+	return &udsGRPCBasedUDSink{c}
+}
+
+func TestGRPCBasedUDF_WaitUntilReadyWithMockClient(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockClient := sinkmock.NewMockUserDefinedSinkClient(ctrl)
+	mockClient.EXPECT().IsReady(gomock.Any(), gomock.Any()).Return(&sinkpb.ReadyResponse{Ready: true}, nil)
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	go func() {
+		<-ctx.Done()
+		if ctx.Err() == context.DeadlineExceeded {
+			t.Log(t.Name(), "test timeout")
+		}
+	}()
+
+	u := NewMockUDSGRPCBasedUDF(mockClient)
+	err := u.WaitUntilReady(ctx)
+	assert.NoError(t, err)
+}
+
+type rpcMsg struct {
+	msg proto.Message
+}
+
+func (r *rpcMsg) Matches(msg interface{}) bool {
+	m, ok := msg.(proto.Message)
+	if !ok {
+		return false
+	}
+	return proto.Equal(m, r.msg)
+}
+
+func (r *rpcMsg) String() string {
+	return fmt.Sprintf("is %s", r.msg)
+}
+
+func TestGRPCBasedUDF_ApplyWithMockClient(t *testing.T) {
+	t.Run("test success", func(t *testing.T) {
+
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+
+		testDatumList := []*sinkpb.Datum{
+			{
+				Id:        "test_id_0",
+				Value:     []byte(`sink_message_success`),
+				EventTime: &sinkpb.EventTime{EventTime: timestamppb.New(time.Unix(1661169660, 0))},
+				Watermark: &sinkpb.Watermark{Watermark: timestamppb.New(time.Time{})},
+			},
+			{
+				Id:        "test_id_1",
+				Value:     []byte(`sink_message_err`),
+				EventTime: &sinkpb.EventTime{EventTime: timestamppb.New(time.Unix(1661169660, 0))},
+				Watermark: &sinkpb.Watermark{Watermark: timestamppb.New(time.Time{})},
+			},
+		}
+		testResponseList := []*sinkpb.Response{
+			{
+				Id:      "test_id_0",
+				Success: true,
+				ErrMsg:  "",
+			},
+			{
+				Id:      "test_id_1",
+				Success: false,
+				ErrMsg:  "mock sink message error",
+			},
+		}
+
+		mockClient := sinkmock.NewMockUserDefinedSinkClient(ctrl)
+
+		mockClient.EXPECT().SinkFn(gomock.Any(), &rpcMsg{msg: &sinkpb.DatumList{
+			Elements: testDatumList,
+		}}).Return(&sinkpb.ResponseList{
+			Responses: testResponseList,
+		}, nil)
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		go func() {
+			<-ctx.Done()
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Log(t.Name(), "test timeout")
+			}
+		}()
+
+		u := NewMockUDSGRPCBasedUDF(mockClient)
+		gotErrList := u.Apply(ctx, testDatumList)
+		assert.Equal(t, 2, len(gotErrList))
+		assert.Equal(t, nil, gotErrList[0])
+		assert.Equal(t, fmt.Errorf("mock sink message error"), gotErrList[1])
+	})
+
+	t.Run("test err", func(t *testing.T) {
+
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+
+		testDatumList := []*sinkpb.Datum{
+			{
+				Id:        "test_id_0",
+				Value:     []byte(`sink_message_grpc_err`),
+				EventTime: &sinkpb.EventTime{EventTime: timestamppb.New(time.Unix(1661169660, 0))},
+				Watermark: &sinkpb.Watermark{Watermark: timestamppb.New(time.Time{})},
+			},
+			{
+				Id:        "test_id_1",
+				Value:     []byte(`sink_message_grpc_err`),
+				EventTime: &sinkpb.EventTime{EventTime: timestamppb.New(time.Unix(1661169660, 0))},
+				Watermark: &sinkpb.Watermark{Watermark: timestamppb.New(time.Time{})},
+			},
+		}
+
+		mockClient := sinkmock.NewMockUserDefinedSinkClient(ctrl)
+
+		mockClient.EXPECT().SinkFn(gomock.Any(), &rpcMsg{msg: &sinkpb.DatumList{
+			Elements: testDatumList,
+		}}).Return(&sinkpb.ResponseList{
+			Responses: nil,
+		}, fmt.Errorf("mock error"))
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		go func() {
+			<-ctx.Done()
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Log(t.Name(), "test timeout")
+			}
+		}()
+
+		u := NewMockUDSGRPCBasedUDF(mockClient)
+		gotErrList := u.Apply(ctx, testDatumList)
+		expectedErrList := []error{
+			ApplyUDSinkErr{
+				UserUDSinkErr: false,
+				Message:       "gRPC client.SinkFn failed, failed to execute c.grpcClt.SinkFn(): mock error",
+				InternalErr: InternalErr{
+					Flag:        true,
+					MainCarDown: false,
+				},
+			},
+			ApplyUDSinkErr{
+				UserUDSinkErr: false,
+				Message:       "gRPC client.SinkFn failed, failed to execute c.grpcClt.SinkFn(): mock error",
+				InternalErr: InternalErr{
+					Flag:        true,
+					MainCarDown: false,
+				},
+			},
+		}
+		assert.Equal(t, 2, len(gotErrList))
+		assert.Equal(t, expectedErrList, gotErrList)
+	})
+}
diff --git a/pkg/sinks/udsink/udssink_http_test.go b/pkg/sinks/udsink/udssink_http_test.go
deleted file mode 100644
index 3f72016e..00000000
--- a/pkg/sinks/udsink/udssink_http_test.go
+++ /dev/null
@@ -1,99 +0,0 @@
-package udsink
-
-import (
-	"context"
-	"fmt"
-	"net"
-	"net/http"
-	"net/http/httptest"
-	"os"
-	"testing"
-	"time"
-
-	sinksdk "github.com/numaproj/numaflow-go/sink"
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/stretchr/testify/assert"
-	"github.com/vmihailenco/msgpack/v5"
-)
-
-const (
-	testSocketPath = "/tmp/test-sink.sock"
-)
-
-var (
-	testTransport = &http.Transport{
-		DialContext: func(_ context.Context, _, _ string) (net.Conn, error) {
-			return net.Dial("unix", testSocketPath)
-		},
-	}
-
-	testClient = &http.Client{
-		Transport: testTransport,
-	}
-)
-
-func TestHTTPBasedUDSink_WaitUntilReady(t *testing.T) {
-	ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		w.WriteHeader(http.StatusOK)
-		_, _ = fmt.Fprintln(w, "Hello, client")
-	}))
-	_ = os.Remove(testSocketPath)
-	listener, _ := net.Listen("unix", testSocketPath)
-	defer func() { _ = listener.Close() }()
-	ts.Listener = listener
-	ts.Start()
-	defer ts.Close()
-
-	u := NewUDSHTTPBasedUDSink(testSocketPath)
-	u.client = testClient
-	ctx := context.Background()
-	err := u.WaitUntilReady(ctx)
-	assert.NoError(t, err)
-}
-
-func TestHTTPBasedUDSink_BasicApply(t *testing.T) {
-	t.Run("test 200", func(t *testing.T) {
-		u := NewUDSHTTPBasedUDSink(testSocketPath)
-		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-		defer cancel()
-
-		// 200
-		ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-			w.Header().Add("Content-Type", string(dfv1.MsgPackType))
-			w.WriteHeader(http.StatusOK)
-			responses := sinksdk.ResponsesBuilder().Append(sinksdk.ResponseOK("abc"))
-			b, _ := msgpack.Marshal(responses)
-			_, _ = w.Write(b)
-		}))
-		_ = os.Remove(testSocketPath)
-		listener, _ := net.Listen("unix", testSocketPath)
-		defer func() { _ = listener.Close() }()
-		ts.Listener = listener
-		ts.Start()
-		u.client = testClient
-		errs := u.Apply(ctx, []sinksdk.Message{{ID: "abc", Payload: []byte("hello")}})
-		assert.Equal(t, 1, len(errs))
-		assert.Nil(t, errs[0])
-		ts.Close()
-	})
-
-	t.Run("test 503", func(t *testing.T) {
-		u := NewUDSHTTPBasedUDSink(testSocketPath)
-		ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-			w.WriteHeader(http.StatusInternalServerError)
-			_, _ = fmt.Fprintln(w, "test")
-		}))
-		_ = os.Remove(testSocketPath)
-		listener, _ := net.Listen("unix", testSocketPath)
-		defer func() { _ = listener.Close() }()
-		ts.Listener = listener
-		ts.Start()
-		u.client = testClient
-		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-		defer cancel()
-		errs := u.Apply(ctx, []sinksdk.Message{{ID: "abc", Payload: []byte("hello")}})
-		assert.Equal(t, 1, len(errs))
-		assert.NotNil(t, errs[0])
-		ts.Close()
-	})
-}
diff --git a/pkg/udf/applier/uds_grpc_test.go b/pkg/udf/applier/uds_grpc_test.go
index b8ea6f7a..1eef624b 100644
--- a/pkg/udf/applier/uds_grpc_test.go
+++ b/pkg/udf/applier/uds_grpc_test.go
@@ -19,9 +19,9 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
-func NewMockUDSGRPCBasedUDF(mockClient *funcmock.MockUserDefinedFunctionClient) *UDSGRPCBasedUDF {
+func NewMockUDSGRPCBasedUDF(mockClient *funcmock.MockUserDefinedFunctionClient) *udsGRPCBasedUDF {
 	c, _ := clienttest.New(mockClient)
-	return &UDSGRPCBasedUDF{c}
+	return &udsGRPCBasedUDF{c}
 }
 
 func TestGRPCBasedUDF_WaitUntilReadyWithMockClient(t *testing.T) {
diff --git a/pkg/udf/applier/uds_http_test.go b/pkg/udf/applier/uds_http_test.go
deleted file mode 100644
index 14bcbe6e..00000000
--- a/pkg/udf/applier/uds_http_test.go
+++ /dev/null
@@ -1,170 +0,0 @@
-package applier
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"io"
-	"net"
-	"net/http"
-	"net/http/httptest"
-	"os"
-	"testing"
-	"time"
-
-	funcsdk "github.com/numaproj/numaflow-go/function"
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/isb/testutils"
-	"github.com/stretchr/testify/assert"
-	"github.com/vmihailenco/msgpack/v5"
-)
-
-const (
-	testSocketPath = "/tmp/test.sock"
-)
-
-var (
-	testTransport = &http.Transport{
-		DialContext: func(_ context.Context, _, _ string) (net.Conn, error) {
-			return net.Dial("unix", testSocketPath)
-		},
-	}
-
-	testClient = &http.Client{
-		Transport: testTransport,
-	}
-)
-
-func TestHTTPBasedUDF_WaitUntilReady(t *testing.T) {
-	ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		w.WriteHeader(http.StatusOK)
-		_, _ = fmt.Fprintln(w, "Hello, client")
-	}))
-	_ = os.Remove(testSocketPath)
-	listener, _ := net.Listen("unix", testSocketPath)
-	defer func() { _ = listener.Close() }()
-	ts.Listener = listener
-	ts.Start()
-	defer ts.Close()
-
-	u := NewUDSHTTPBasedUDF(testSocketPath)
-	u.client = testClient
-	ctx := context.Background()
-	err := u.WaitUntilReady(ctx)
-	assert.NoError(t, err)
-}
-
-func TestHTTPBasedUDF_BasicApply(t *testing.T) {
-	t.Run("test 200", func(t *testing.T) {
-		u := NewUDSHTTPBasedUDF(testSocketPath)
-		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-		defer cancel()
-
-		// 200
-		ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-			w.Header().Add("Content-Type", string(dfv1.MsgPackType))
-			w.WriteHeader(http.StatusOK)
-			msgs := funcsdk.MessagesBuilder().Append(funcsdk.Message{Key: []byte{}, Value: []byte("test")})
-			b, _ := msgpack.Marshal(msgs)
-			_, _ = w.Write(b)
-		}))
-		_ = os.Remove(testSocketPath)
-		listener, _ := net.Listen("unix", testSocketPath)
-		defer func() { _ = listener.Close() }()
-		ts.Listener = listener
-		ts.Start()
-		u.client = testClient
-		_, err := u.Apply(ctx, &isb.ReadMessage{ReadOffset: isb.SimpleOffset(func() string { return "" })})
-		assert.NoError(t, err)
-		ts.Close()
-	})
-
-	t.Run("test 503", func(t *testing.T) {
-		u := NewUDSHTTPBasedUDF(testSocketPath)
-		ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-			w.WriteHeader(http.StatusInternalServerError)
-			_, _ = fmt.Fprintln(w, "test")
-		}))
-		_ = os.Remove(testSocketPath)
-		listener, _ := net.Listen("unix", testSocketPath)
-		defer func() { _ = listener.Close() }()
-		ts.Listener = listener
-		ts.Start()
-		u.client = testClient
-		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-		defer cancel()
-		_, err := u.Apply(ctx, &isb.ReadMessage{})
-		assert.Error(t, err)
-		ts.Close()
-	})
-}
-
-func TestHTTPBasedUDF_Apply(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-	defer cancel()
-
-	multiplyBy2 := func(body []byte) interface{} {
-		var result testutils.PayloadForTest
-		_ = json.Unmarshal(body, &result)
-		result.Value = result.Value * 2
-		return result
-	}
-
-	s := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		body, _ := io.ReadAll(r.Body)
-
-		// set header as even or odd based on the original request
-		var orignalReq testutils.PayloadForTest
-		_ = json.Unmarshal(body, &orignalReq)
-		result, _ := json.Marshal(multiplyBy2(body).(testutils.PayloadForTest))
-		m := &funcsdk.Message{Value: result}
-		if orignalReq.Value%2 == 0 {
-			m.Key = []byte("even")
-		} else {
-			m.Key = []byte("odd")
-		}
-		b, _ := msgpack.Marshal(funcsdk.MessagesBuilder().Append(*m))
-		w.Header().Add("Content-Type", string(dfv1.MsgPackType))
-		w.WriteHeader(http.StatusOK)
-		_, _ = w.Write(b)
-
-		_, _ = w.Write(result)
-	}))
-	_ = os.Remove(testSocketPath)
-	listener, _ := net.Listen("unix", testSocketPath)
-	defer func() { _ = listener.Close() }()
-	s.Listener = listener
-	s.Start()
-	defer s.Close()
-
-	u := NewUDSHTTPBasedUDF(testSocketPath)
-	u.client = testClient
-	var count = int64(10)
-	readMessages := testutils.BuildTestReadMessages(count, time.Unix(1636470000, 0))
-	var expectedResults = make([][]byte, count)
-	var expectedKeys = make([]string, count)
-	for idx, readMessage := range readMessages {
-		var readMessagePayload testutils.PayloadForTest
-		_ = json.Unmarshal(readMessage.Payload, &readMessagePayload)
-		if readMessagePayload.Value%2 == 0 {
-			expectedKeys[idx] = "even"
-		} else {
-			expectedKeys[idx] = "odd"
-		}
-		marshal, _ := json.Marshal(multiplyBy2(readMessage.Payload))
-		expectedResults[idx] = marshal
-	}
-
-	var results = make([][]byte, len(readMessages))
-	var resultKeys = make([]string, len(readMessages))
-	for idx, readMessage := range readMessages {
-		apply, err := u.Apply(ctx, &readMessage)
-		assert.NoError(t, err)
-		results[idx] = apply[0].Payload
-		resultKeys[idx] = string(apply[0].Header.Key)
-	}
-
-	assert.Equal(t, expectedResults, results)
-	assert.Equal(t, expectedKeys, resultKeys)
-}