Repository: numaproj/numaflow
Pull request #146 (2022-08-24 23:40:28)

feat: Add udf grpc support Fixes #145 
Signed-off-by: Juanlu Yu <juanlu_yu@intuit.com>

`make pre-push -B`
failed with
```
# marker file, based on it's modification time, we know how long ago this target was run
touch dist/pre-push
touch: dist/pre-push: No such file or directory
make: *** [pre-push] Error 1
```

Replace HTTP with gRPC to communicate between the main container and the UDF container.
Use mock clients from function sdk to test the uds_grpc.go.

DRAFT PR, DO NOT MERGE



chromevoid: tested gRPC map function using simple pipeline example

seeing `starting the gRPC server with unix domain socket` from the new sdk
```shell
➜  numaflow git:(udf-grpc-client-init-ver) kubectl logs -f simple-pipeline-cat-0-2xx5c udf
{"level":"info","ts":1662099248.8747523,"logger":"numaflow.builtin-udf","caller":"builtin/builtin.go:23","msg":"Start a builtin function","name":"cat","args":[],"kwargs":{}}
2022/09/02 06:14:08 starting the gRPC server with unix domain socket...
```
seeing `uds-grpc-udf` tagged to the logger
```shell
➜  numaflow git:(udf-grpc-client-init-ver) kubectl logs -f simple-pipeline-cat-0-2xx5c main
...
{"level":"info","ts":1662099249.8015754,"logger":"numaflow.udf-processor","caller":"udf/udf.go:138","msg":"Start processing udf messages","vertex":"simple-pipeline-cat","protocol":"uds-grpc-udf","isbs":"jetstream","from":"default-simple-pipeline-in-cat","to":[{"Name":"default-simple-pipeline-cat-out","Type":"ed"}]}
...
```
seeing data flowing
```shell
2022/09/02 06:15:36 (out) {"Data":"Jj1XYYr2EBc=","Createdts":1662099336698740006}
2022/09/02 06:15:37 (out) {"Data":"5Tz4nIr2EBc=","Createdts":1662099337699146981}
2022/09/02 06:15:37 (out) {"Data":"vof4nIr2EBc=","Createdts":1662099337699166142}
2022/09/02 06:15:37 (out) {"Data":"HIv4nIr2EBc=","Createdts":1662099337699167004}
2022/09/02 06:15:37 (out) {"Data":"EI34nIr2EBc=","Createdts":1662099337699167504}
2022/09/02 06:15:37 (out) {"Data":"Po/4nIr2EBc=","Createdts":1662099337699168062}
2022/09/02 06:15:38 (out) {"Data":"DBaY2Ir2EBc=","Createdts":1662099338699478540}
2022/09/02 06:15:38 (out) {"Data":"cV6Y2Ir2EBc=","Createdts":1662099338699497073}
2022/09/02 06:15:38 (out) {"Data":"GmKY2Ir2EBc=","Createdts":1662099338699498010}
2022/09/02 06:15:38 (out) {"Data":"MWSY2Ir2EBc=","Createdts":1662099338699498545}
2022/09/02 06:15:38 (out) {"Data":"F2aY2Ir2EBc=","Createdts":1662099338699499031}
2022/09/02 06:15:39 (out) {"Data":"mzkoFIv2EBc=","Createdts":1662099339698780571}
2022/09/02 06:15:39 (out) {"Data":"44koFIv2EBc=","Createdts":1662099339698801123}
2022/09/02 06:15:39 (out) {"Data":"RY0oFIv2EBc=","Createdts":1662099339698801989}
2022/09/02 06:15:39 (out) {"Data":"d48oFIv2EBc=","Createdts":1662099339698802551}
2022/09/02 06:15:39 (out) {"Data":"m5IoFIv2EBc=","Createdts":1662099339698803355}
```
chromevoid: As discussed the reduce applier will be done later

Patch:
diff --git a/go.mod b/go.mod
index f2af46f5..345744fa 100644
--- a/go.mod
+++ b/go.mod
@@ -15,6 +15,7 @@ require (
 	github.com/go-swagger/go-swagger v0.28.0
 	github.com/goccy/go-json v0.9.7
 	github.com/gogo/protobuf v1.3.2
+	github.com/golang/mock v1.6.0
 	github.com/golang/protobuf v1.5.2
 	github.com/google/uuid v1.2.0
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
@@ -23,7 +24,7 @@ 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.0.1
+	github.com/numaproj/numaflow-go v0.2.1
 	github.com/prometheus/client_golang v1.12.1
 	github.com/prometheus/common v0.32.1
 	github.com/soheilhy/cmux v0.1.5
@@ -34,8 +35,8 @@ require (
 	go.uber.org/atomic v1.9.0
 	go.uber.org/zap v1.19.1
 	google.golang.org/genproto v0.0.0-20220118154757-00ab72f36ad5
-	google.golang.org/grpc v1.43.0
-	google.golang.org/protobuf v1.28.0
+	google.golang.org/grpc v1.48.0
+	google.golang.org/protobuf v1.28.1
 	k8s.io/api v0.23.3
 	k8s.io/apimachinery v0.23.3
 	k8s.io/client-go v0.23.3
diff --git a/go.sum b/go.sum
index 3aa08240..2cbe9766 100644
--- a/go.sum
+++ b/go.sum
@@ -130,8 +130,8 @@ github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnht
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4/go.mod h1:6pvJx4me5XPnfI9Z40ddWsdw2W/uZgQLFXToKeRcDiI=
 github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
-github.com/cncf/xds/go v0.0.0-20210805033703-aa0b78936158/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20210922020428-25de7278fc84/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
+github.com/cncf/xds/go v0.0.0-20211001041855-01bcc9b48dfe/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cockroachdb/datadriven v0.0.0-20200714090401-bf6692d28da5/go.mod h1:h6jFvWxBdQXxjopDMZyH2UVceIRfR84bdzbkoKrsWNo=
 github.com/cockroachdb/errors v1.2.4/go.mod h1:rQD95gz6FARkaKkQXUksEje/d9a6wBJoCr5oaCLELYA=
@@ -177,7 +177,7 @@ github.com/envoyproxy/go-control-plane v0.9.7/go.mod h1:cwu0lG7PUMfa9snN8LXBig5y
 github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.mod h1:hliV/p42l8fGbc6Y9bQ70uLwIvmJyVE5k4iMKlh8wCQ=
-github.com/envoyproxy/go-control-plane v0.9.10-0.20210907150352-cf90f659a021/go.mod h1:AFq3mo9L8Lqqiid3OhADV3RfLJnjiw63cSpi+fDTRC0=
+github.com/envoyproxy/go-control-plane v0.10.2-0.20220325020618-49ff273808a1/go.mod h1:KJwIaB5Mv44NWtYuAOFCVOjcI94vtpEz2JU/D2v6IjE=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/evanphx/json-patch v0.5.2/go.mod h1:ZWS5hhDbVDyob71nXKNL0+PWn6ToqBHMikGIFbs31qQ=
 github.com/evanphx/json-patch v4.12.0+incompatible h1:4onqiflcdA9EOZ4RxV643DvftH5pOlLGNtQ5lPWQu84=
@@ -404,6 +404,7 @@ github.com/golang/mock v1.4.1/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt
 github.com/golang/mock v1.4.3/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
 github.com/golang/mock v1.4.4/go.mod h1:l3mdAwkq5BuhzHwde/uurv3sEJeZMXNpwsxVWU71h+4=
 github.com/golang/mock v1.5.0/go.mod h1:CWnOUgYIOo4TcNZ0wHX3YZCqsaM1I1Jvs6v3mP3KVu8=
+github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
 github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
@@ -689,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.0.1 h1:ZUo+axC0cYK/50Wgxt/sK8AlRYogESkO6sJnG9xijSk=
-github.com/numaproj/numaflow-go v0.0.1/go.mod h1:df8MZ5SavR1dCz/pw7eYBVrA0v0GtAgxnsAfbAJovDo=
+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/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=
@@ -1400,8 +1401,8 @@ google.golang.org/grpc v1.38.0/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQ
 google.golang.org/grpc v1.39.0/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnDzfrE=
 google.golang.org/grpc v1.39.1/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnDzfrE=
 google.golang.org/grpc v1.40.0/go.mod h1:ogyxbiOoUXAkP+4+xa6PZSE9DZgIHtSpzjDTB9KAK34=
-google.golang.org/grpc v1.43.0 h1:Eeu7bZtDZ2DpRCsLhUlcrLnvYaMK1Gz86a+hMVvELmM=
-google.golang.org/grpc v1.43.0/go.mod h1:k+4IHHFw41K8+bbowsex27ge2rCb65oeWqe4jJ590SU=
+google.golang.org/grpc v1.48.0 h1:rQOsyJ/8+ufEDJd/Gdsz7HG220Mh9HAhFHRGnIjda0w=
+google.golang.org/grpc v1.48.0/go.mod h1:vN9eftEi1UMyUsIF80+uQXhHjbXYbm0uXoFCACuMGWk=
 google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0/go.mod h1:6Kw0yEErY5E/yWrBtf03jp27GLLJujG4z/JK95pnjjw=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
@@ -1416,8 +1417,9 @@ google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlba
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-google.golang.org/protobuf v1.28.0 h1:w43yiav+6bVFTBQFZX0r7ipe9JQ1QsbMgHwbBziscLw=
 google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
+google.golang.org/protobuf v1.28.1 h1:d0NfwRgPtno5B1Wa6L2DAG+KivqkdutMf1UhdNx175w=
+google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
diff --git a/pkg/udf/applier/uds_grpc.go b/pkg/udf/applier/uds_grpc.go
new file mode 100644
index 00000000..96657569
--- /dev/null
+++ b/pkg/udf/applier/uds_grpc.go
@@ -0,0 +1,97 @@
+package applier
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
+	functionsdk "github.com/numaproj/numaflow-go/pkg/function"
+	"github.com/numaproj/numaflow-go/pkg/function/client"
+	"github.com/numaproj/numaflow/pkg/isb"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/protobuf/types/known/emptypb"
+	"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 {
+	client functionsdk.Client
+}
+
+var _ Applier = (*UDSGRPCBasedUDF)(nil)
+
+// 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
+}
+
+// CloseConn closes the gRPC client connection.
+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 {
+	var err error
+	for {
+		select {
+		case <-ctx.Done():
+			// using %v for ctx.Err() because only one %w can exist in the fmt.Errorf
+			return fmt.Errorf("failed to wait for ready: %v, %w", ctx.Err(), err)
+		default:
+			if _, err = u.client.IsReady(ctx, &emptypb.Empty{}); err == nil {
+				return nil
+			}
+			time.Sleep(1 * time.Second)
+		}
+	}
+}
+
+func (u *UDSGRPCBasedUDF) Apply(ctx context.Context, readMessage *isb.ReadMessage) ([]*isb.Message, error) {
+	key := readMessage.Key
+	payload := readMessage.Body.Payload
+	offset := readMessage.ReadOffset
+	parentPaneInfo := readMessage.PaneInfo
+
+	var d = &functionpb.Datum{
+		Key:       key,
+		Value:     payload,
+		EventTime: &functionpb.EventTime{EventTime: timestamppb.New(parentPaneInfo.EventTime)},
+		Watermark: &functionpb.Watermark{Watermark: timestamppb.New(time.Time{})}, // TODO: insert the correct watermark
+	}
+
+	ctx = metadata.NewOutgoingContext(ctx, metadata.New(map[string]string{functionsdk.DatumKey: key}))
+	datumList, err := u.client.MapFn(ctx, d)
+	if err != nil {
+		return nil, ApplyUDFErr{
+			UserUDFErr: false,
+			Message:    fmt.Sprintf("gRPC client.MapFn failed, %s", err),
+			InternalErr: InternalErr{
+				Flag:        true,
+				MainCarDown: false,
+			},
+		}
+	}
+
+	writeMessages := make([]*isb.Message, 0)
+	for i, datum := range datumList {
+		key := datum.Key
+		writeMessage := &isb.Message{
+			Header: isb.Header{
+				PaneInfo: parentPaneInfo,
+				ID:       fmt.Sprintf("%s-%d", offset.String(), i),
+				Key:      key,
+			},
+			Body: isb.Body{
+				Payload: datum.Value,
+			},
+		}
+		writeMessages = append(writeMessages, writeMessage)
+	}
+	return writeMessages, nil
+}
diff --git a/pkg/udf/builtin/builtin.go b/pkg/udf/builtin/builtin.go
index 3fe7d579..33c1e274 100644
--- a/pkg/udf/builtin/builtin.go
+++ b/pkg/udf/builtin/builtin.go
@@ -4,7 +4,8 @@ import (
 	"context"
 	"fmt"
 
-	funcsdk "github.com/numaproj/numaflow-go/function"
+	functionsdk "github.com/numaproj/numaflow-go/pkg/function"
+	"github.com/numaproj/numaflow-go/pkg/function/server"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"github.com/numaproj/numaflow/pkg/udf/builtin/cat"
 	"github.com/numaproj/numaflow/pkg/udf/builtin/filter"
@@ -20,22 +21,22 @@ type Builtin struct {
 func (b *Builtin) Start(ctx context.Context) error {
 	log := logging.FromContext(ctx)
 	log.Infow("Start a builtin function", zap.Any("name", b.Name), zap.Strings("args", b.Args), zap.Any("kwargs", b.KWArgs))
-	excutor, err := b.excutor()
+
+	executor, err := b.executor()
 	if err != nil {
 		return err
 	}
-	funcsdk.Start(ctx, excutor)
+	server.New().RegisterMapper(executor).Start(ctx)
 	return nil
 }
 
-func (b *Builtin) excutor() (funcsdk.Handle, error) {
+func (b *Builtin) executor() (functionsdk.MapFunc, error) {
 	// TODO: deal with args later
 	switch b.Name {
 	case "cat":
 		return cat.New(), nil
 	case "filter":
 		return filter.New(b.KWArgs)
-
 	default:
 		return nil, fmt.Errorf("unrecognized function %q", b.Name)
 	}
diff --git a/pkg/udf/builtin/cat/cat.go b/pkg/udf/builtin/cat/cat.go
index f1396d5f..c9412077 100644
--- a/pkg/udf/builtin/cat/cat.go
+++ b/pkg/udf/builtin/cat/cat.go
@@ -3,11 +3,11 @@ package cat
 import (
 	"context"
 
-	funcsdk "github.com/numaproj/numaflow-go/function"
+	functionsdk "github.com/numaproj/numaflow-go/pkg/function"
 )
 
-func New() funcsdk.Handle {
-	return func(ctx context.Context, key, msg []byte) (funcsdk.Messages, error) {
-		return funcsdk.MessagesBuilder().Append(funcsdk.MessageToAll(msg)), nil
+func New() functionsdk.MapFunc {
+	return func(ctx context.Context, key string, datum functionsdk.Datum) functionsdk.Messages {
+		return functionsdk.MessagesBuilder().Append(functionsdk.MessageToAll(datum.Value()))
 	}
 }
diff --git a/pkg/udf/builtin/filter/filter.go b/pkg/udf/builtin/filter/filter.go
index 6dc763b7..55303b8b 100644
--- a/pkg/udf/builtin/filter/filter.go
+++ b/pkg/udf/builtin/filter/filter.go
@@ -4,15 +4,16 @@ import (
 	"context"
 	"fmt"
 
-	funcsdk "github.com/numaproj/numaflow-go/function"
+	functionsdk "github.com/numaproj/numaflow-go/pkg/function"
 	"github.com/numaproj/numaflow/pkg/shared/expr"
+	"github.com/numaproj/numaflow/pkg/shared/logging"
 )
 
 type filter struct {
 	expression string
 }
 
-func New(args map[string]string) (funcsdk.Handle, error) {
+func New(args map[string]string) (functionsdk.MapFunc, error) {
 	expr, existing := args["expression"]
 	if !existing {
 		return nil, fmt.Errorf("missing \"expression\"")
@@ -20,20 +21,25 @@ func New(args map[string]string) (funcsdk.Handle, error) {
 	f := filter{
 		expression: expr,
 	}
-	return func(ctx context.Context, key, msg []byte) (funcsdk.Messages, error) {
-		resultMsg, err := f.apply(msg)
-		return funcsdk.MessagesBuilder().Append(resultMsg), err
+
+	return func(ctx context.Context, key string, datum functionsdk.Datum) functionsdk.Messages {
+		log := logging.FromContext(ctx)
+		resultMsg, err := f.apply(datum.Value())
+		if err != nil {
+			log.Errorf("filter map function apply got an error: %v", err)
+		}
+		return functionsdk.MessagesBuilder().Append(resultMsg)
 	}, nil
 }
 
-func (f filter) apply(msg []byte) (funcsdk.Message, error) {
+func (f filter) apply(msg []byte) (functionsdk.Message, error) {
 
 	result, err := expr.EvalBool(f.expression, msg)
 	if err != nil {
-		return funcsdk.MessageToDrop(), err
+		return functionsdk.MessageToDrop(), err
 	}
 	if result {
-		return funcsdk.MessageToAll(msg), nil
+		return functionsdk.MessageToAll(msg), nil
 	}
-	return funcsdk.MessageToDrop(), nil
+	return functionsdk.MessageToDrop(), nil
 }
diff --git a/pkg/udf/udf.go b/pkg/udf/udf.go
index 0df8c824..2f1592a8 100644
--- a/pkg/udf/udf.go
+++ b/pkg/udf/udf.go
@@ -4,10 +4,8 @@ import (
 	"context"
 	"fmt"
 	"sync"
-	"time"
 
 	"github.com/numaproj/numaflow/pkg/watermark/generic/jetstream"
-
 	"go.uber.org/zap"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
@@ -124,12 +122,23 @@ func (u *UDFProcessor) Start(ctx context.Context) error {
 		return result, nil
 	})
 
-	udfHandler := applier.NewUDSHTTPBasedUDF(dfv1.PathVarRun+"/udf.sock", applier.WithHTTPClientTimeout(120*time.Second))
+	log = log.With("protocol", "uds-grpc-udf")
+	udfHandler, err := applier.NewUDSGRPCBasedUDF()
+	if err != nil {
+		return fmt.Errorf("failed to create gRPC client, %w", err)
+	}
 	// Readiness check
 	if err := udfHandler.WaitUntilReady(ctx); err != nil {
 		return fmt.Errorf("failed on UDF readiness check, %w", err)
 	}
+	defer func() {
+		err = udfHandler.CloseConn(ctx)
+		if err != nil {
+			log.Warnw("Failed to close gRPC client conn", zap.Error(err))
+		}
+	}()
 	log.Infow("Start processing udf messages", zap.String("isbs", string(u.ISBSvcType)), zap.String("from", fromBufferName), zap.Any("to", toBuffers))
+
 	opts := []forward.Option{forward.WithLogger(log)}
 	if x := u.VertexInstance.Vertex.Spec.Limits; x != nil {
 		if x.ReadBatchSize != nil {


Test patch:
diff --git a/pkg/udf/applier/uds_grpc_test.go b/pkg/udf/applier/uds_grpc_test.go
new file mode 100644
index 00000000..b8ea6f7a
--- /dev/null
+++ b/pkg/udf/applier/uds_grpc_test.go
@@ -0,0 +1,242 @@
+package applier
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/golang/mock/gomock"
+	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
+	"github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1/funcmock"
+	"github.com/numaproj/numaflow-go/pkg/function/clienttest"
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/isb/testutils"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+func NewMockUDSGRPCBasedUDF(mockClient *funcmock.MockUserDefinedFunctionClient) *UDSGRPCBasedUDF {
+	c, _ := clienttest.New(mockClient)
+	return &UDSGRPCBasedUDF{c}
+}
+
+func TestGRPCBasedUDF_WaitUntilReadyWithMockClient(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockClient := funcmock.NewMockUserDefinedFunctionClient(ctrl)
+	mockClient.EXPECT().IsReady(gomock.Any(), gomock.Any()).Return(&functionpb.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_BasicApplyWithMockClient(t *testing.T) {
+	t.Run("test success", func(t *testing.T) {
+
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+
+		mockClient := funcmock.NewMockUserDefinedFunctionClient(ctrl)
+		req := &functionpb.Datum{
+			Key:       "test_success_key",
+			Value:     []byte(`forward_message`),
+			EventTime: &functionpb.EventTime{EventTime: timestamppb.New(time.Unix(1661169600, 0))},
+			Watermark: &functionpb.Watermark{Watermark: timestamppb.New(time.Time{})},
+		}
+		mockClient.EXPECT().MapFn(gomock.Any(), &rpcMsg{msg: req}).Return(&functionpb.DatumList{
+			Elements: []*functionpb.Datum{
+				{
+					Key:   "test_success_key",
+					Value: []byte(`forward_message`),
+				},
+			},
+		}, 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)
+		got, err := u.Apply(ctx, &isb.ReadMessage{
+			Message: isb.Message{
+				Header: isb.Header{
+					PaneInfo: isb.PaneInfo{
+						EventTime: time.Unix(1661169600, 0),
+					},
+					ID:  "test_id",
+					Key: `test_success_key`,
+				},
+				Body: isb.Body{
+					Payload: []byte(`forward_message`),
+				},
+			},
+			ReadOffset: isb.SimpleOffset(func() string { return "0" }),
+		},
+		)
+		assert.NoError(t, err)
+		assert.Equal(t, req.Key, string(got[0].Key))
+		assert.Equal(t, req.Value, got[0].Payload)
+	})
+
+	t.Run("test error", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+
+		mockClient := funcmock.NewMockUserDefinedFunctionClient(ctrl)
+		req := &functionpb.Datum{
+			Key:       "test_error_key",
+			Value:     []byte(`forward_message`),
+			EventTime: &functionpb.EventTime{EventTime: timestamppb.New(time.Unix(1661169660, 0))},
+			Watermark: &functionpb.Watermark{Watermark: timestamppb.New(time.Time{})},
+		}
+		mockClient.EXPECT().MapFn(gomock.Any(), &rpcMsg{msg: req}).Return(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)
+		_, err := u.Apply(ctx, &isb.ReadMessage{
+			Message: isb.Message{
+				Header: isb.Header{
+					PaneInfo: isb.PaneInfo{
+						EventTime: time.Unix(1661169660, 0),
+					},
+					ID:  "test_id",
+					Key: `test_error_key`,
+				},
+				Body: isb.Body{
+					Payload: []byte(`forward_message`),
+				},
+			},
+			ReadOffset: isb.SimpleOffset(func() string { return "0" }),
+		},
+		)
+		assert.ErrorIs(t, err, ApplyUDFErr{
+			UserUDFErr: false,
+			Message:    fmt.Sprintf("%s", err),
+			InternalErr: InternalErr{
+				Flag:        true,
+				MainCarDown: false,
+			},
+		})
+	})
+}
+
+func TestHGRPCBasedUDF_ApplyWithMockClient(t *testing.T) {
+	multiplyBy2 := func(body []byte) interface{} {
+		var result testutils.PayloadForTest
+		_ = json.Unmarshal(body, &result)
+		result.Value = result.Value * 2
+		return result
+	}
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockClient := funcmock.NewMockUserDefinedFunctionClient(ctrl)
+	mockClient.EXPECT().MapFn(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(_ context.Context, datum *functionpb.Datum, opts ...grpc.CallOption) (*functionpb.DatumList, error) {
+			var originalValue testutils.PayloadForTest
+			_ = json.Unmarshal(datum.GetValue(), &originalValue)
+			doubledValue, _ := json.Marshal(multiplyBy2(datum.GetValue()).(testutils.PayloadForTest))
+			var elements []*functionpb.Datum
+			if originalValue.Value%2 == 0 {
+				elements = append(elements, &functionpb.Datum{
+					Key:   "even",
+					Value: doubledValue,
+				})
+			} else {
+				elements = append(elements, &functionpb.Datum{
+					Key:   "odd",
+					Value: doubledValue,
+				})
+			}
+			datumList := &functionpb.DatumList{
+				Elements: elements,
+			}
+			return datumList, nil
+		},
+	).AnyTimes()
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+	go func() {
+		<-ctx.Done()
+		if ctx.Err() == context.DeadlineExceeded {
+			t.Log(t.Name(), "test timeout")
+		}
+	}()
+
+	u := NewMockUDSGRPCBasedUDF(mockClient)
+
+	var count = int64(10)
+	readMessages := testutils.BuildTestReadMessages(count, time.Unix(1661169600, 0))
+
+	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)
+	}
+
+	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
+	}
+
+	assert.Equal(t, expectedResults, results)
+	assert.Equal(t, expectedKeys, resultKeys)
+}
diff --git a/pkg/udf/builtin/builtin_test.go b/pkg/udf/builtin/builtin_test.go
index 415a5671..6332e7e9 100644
--- a/pkg/udf/builtin/builtin_test.go
+++ b/pkg/udf/builtin/builtin_test.go
@@ -18,7 +18,7 @@ func TestGetExecutors(t *testing.T) {
 			},
 		}
 		for _, b := range builtins {
-			e, err := b.excutor()
+			e, err := b.executor()
 			assert.NoError(t, err)
 			assert.NotNil(t, e)
 		}
@@ -28,7 +28,7 @@ func TestGetExecutors(t *testing.T) {
 		b := &Builtin{
 			Name: "catt",
 		}
-		_, err := b.excutor()
+		_, err := b.executor()
 		assert.Error(t, err)
 		assert.Contains(t, err.Error(), "unrecognized function")
 	})
diff --git a/pkg/udf/builtin/cat/cat_test.go b/pkg/udf/builtin/cat/cat_test.go
index 91b52e79..9676329c 100644
--- a/pkg/udf/builtin/cat/cat_test.go
+++ b/pkg/udf/builtin/cat/cat_test.go
@@ -3,18 +3,40 @@ package cat
 import (
 	"context"
 	"testing"
+	"time"
 
 	funcsdk "github.com/numaproj/numaflow-go/function"
 	"github.com/stretchr/testify/assert"
 )
 
+type testDatum struct {
+	value     []byte
+	eventTime time.Time
+	watermark time.Time
+}
+
+func (h *testDatum) Value() []byte {
+	return h.value
+}
+
+func (h *testDatum) EventTime() time.Time {
+	return h.eventTime
+}
+
+func (h *testDatum) Watermark() time.Time {
+	return h.watermark
+}
+
 func TestNew(t *testing.T) {
 	ctx := context.Background()
 	p := New()
 	req := []byte{0}
-	messages, err := p(ctx, []byte(""), req)
-	assert.NoError(t, err)
+	messages := p(ctx, "", &testDatum{
+		value:     req,
+		eventTime: time.Time{},
+		watermark: time.Time{},
+	})
 	assert.Equal(t, 1, len(messages))
-	assert.Equal(t, funcsdk.ALL, string(messages[0].Key))
+	assert.Equal(t, funcsdk.ALL, messages[0].Key)
 	assert.Equal(t, req, messages[0].Value)
 }
diff --git a/pkg/udf/builtin/filter/filter_test.go b/pkg/udf/builtin/filter/filter_test.go
index c4192ae4..666807a6 100644
--- a/pkg/udf/builtin/filter/filter_test.go
+++ b/pkg/udf/builtin/filter/filter_test.go
@@ -4,15 +4,34 @@ import (
 	"context"
 	"encoding/base64"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 )
 
-var _key = []byte("")
+var _key = ""
 var jsonMsg = `{"test": 21, "item": [{"id": 1, "name": "bala"},{"id": 2, "name": "bala"}]}`
 var strMsg = `welcome to numaflow`
 var base64Msg = base64.StdEncoding.EncodeToString([]byte(strMsg))
 
+type testDatum struct {
+	value     []byte
+	eventTime time.Time
+	watermark time.Time
+}
+
+func (h *testDatum) Value() []byte {
+	return h.value
+}
+
+func (h *testDatum) EventTime() time.Time {
+	return h.eventTime
+}
+
+func (h *testDatum) Watermark() time.Time {
+	return h.watermark
+}
+
 func TestExpression(t *testing.T) {
 	t.Run("missing expression", func(t *testing.T) {
 		_, err := New(map[string]string{})
@@ -26,8 +45,11 @@ func TestExpression(t *testing.T) {
 		handle, err := New(args)
 		assert.NoError(t, err)
 
-		result, err := handle(context.Background(), _key, []byte(jsonMsg))
-		assert.NoError(t, err)
+		result := handle(context.Background(), _key, &testDatum{
+			value:     []byte(jsonMsg),
+			eventTime: time.Time{},
+			watermark: time.Time{},
+		})
 		assert.Equal(t, jsonMsg, string(result.Items()[0].Value))
 	})
 
@@ -37,8 +59,11 @@ func TestExpression(t *testing.T) {
 		handle, err := New(args)
 		assert.NoError(t, err)
 
-		result, err := handle(context.Background(), _key, []byte(jsonMsg))
-		assert.Error(t, err)
+		result := handle(context.Background(), _key, &testDatum{
+			value:     []byte(jsonMsg),
+			eventTime: time.Time{},
+			watermark: time.Time{},
+		})
 		assert.Equal(t, "", string(result.Items()[0].Value))
 	})
 
@@ -48,8 +73,11 @@ func TestExpression(t *testing.T) {
 		handle, err := New(args)
 		assert.NoError(t, err)
 
-		result, err := handle(context.Background(), _key, []byte(jsonMsg))
-		assert.NoError(t, err)
+		result := handle(context.Background(), _key, &testDatum{
+			value:     []byte(jsonMsg),
+			eventTime: time.Time{},
+			watermark: time.Time{},
+		})
 		assert.Equal(t, "", string(result.Items()[0].Value))
 	})
 
@@ -59,8 +87,11 @@ func TestExpression(t *testing.T) {
 		handle, err := New(args)
 		assert.NoError(t, err)
 
-		result, err := handle(context.Background(), _key, []byte(strMsg))
-		assert.NoError(t, err)
+		result := handle(context.Background(), _key, &testDatum{
+			value:     []byte(jsonMsg),
+			eventTime: time.Time{},
+			watermark: time.Time{},
+		})
 		assert.Equal(t, "", string(result.Items()[0].Value))
 	})
 
@@ -70,8 +101,11 @@ func TestExpression(t *testing.T) {
 		handle, err := New(args)
 		assert.NoError(t, err)
 
-		result, err := handle(context.Background(), _key, []byte(base64Msg))
-		assert.NoError(t, err)
+		result := handle(context.Background(), _key, &testDatum{
+			value:     []byte(base64Msg),
+			eventTime: time.Time{},
+			watermark: time.Time{},
+		})
 		assert.Equal(t, base64Msg, string(result.Items()[0].Value))
 	})