Repository: numaproj/numaflow (go)

Issue:
remove SDK dependency from Numaflow
# Problem statement

Currently, the client code is in `numaflow-go` repo, so every time we have to make a change on the client part of the code we need to change it in` numaflow-go` repo and tag a new version just to update the `go.mod` entry in `numaflow`.

# Proposal

Move the client code to `numaflow` repo, that way we can remove the SDK dependency from `numaflow` and we only keep the server related code in Go SDK repo similar to Java and Python SDKs


Pull request description:
chore: move go client code to numaflow repo
Moves the client code from the go-SDK to the numaflow repository.
Fixes #692 


Patch:
diff --git a/pkg/sdkclient/sink/client/client.go b/pkg/sdkclient/sink/client/client.go
new file mode 100644
index 00000000..bbe637cf
--- /dev/null
+++ b/pkg/sdkclient/sink/client/client.go
@@ -0,0 +1,95 @@
+package client
+
+import (
+	"context"
+	"fmt"
+	"github.com/numaproj/numaflow-go/pkg/sink"
+	"log"
+	"time"
+
+	sinkpb "github.com/numaproj/numaflow-go/pkg/apis/proto/sink/v1"
+	"github.com/numaproj/numaflow-go/pkg/info"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/protobuf/types/known/emptypb"
+)
+
+// client contains the grpc connection and the grpc client.
+type client struct {
+	conn    *grpc.ClientConn
+	grpcClt sinkpb.UserDefinedSinkClient
+}
+
+var _ Client = (*client)(nil)
+
+// New creates a new client object.
+func New(inputOptions ...Option) (*client, error) {
+	var opts = &options{
+		sockAddr:                   sink.Addr,
+		serverInfoFilePath:         info.ServerInfoFilePath,
+		serverInfoReadinessTimeout: 120 * time.Second, // Default timeout is 120 seconds
+	}
+
+	for _, inputOption := range inputOptions {
+		inputOption(opts)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), opts.serverInfoReadinessTimeout)
+	defer cancel()
+
+	if err := info.WaitUntilReady(ctx, info.WithServerInfoFilePath(opts.serverInfoFilePath)); err != nil {
+		return nil, fmt.Errorf("failed to wait until server info is ready: %w", err)
+	}
+
+	serverInfo, err := info.Read(info.WithServerInfoFilePath(opts.serverInfoFilePath))
+	if err != nil {
+		return nil, fmt.Errorf("failed to read server info: %w", err)
+	}
+	// TODO: Use serverInfo to check compatibility.
+	if serverInfo != nil {
+		log.Printf("ServerInfo: %v\n", serverInfo)
+	}
+
+	c := new(client)
+	sockAddr := fmt.Sprintf("%s:%s", sink.Protocol, opts.sockAddr)
+	conn, err := grpc.Dial(sockAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return nil, fmt.Errorf("failed to execute grpc.Dial(%q): %w", sockAddr, err)
+	}
+	c.conn = conn
+	c.grpcClt = sinkpb.NewUserDefinedSinkClient(conn)
+	return c, nil
+}
+
+// CloseConn closes the grpc client connection.
+func (c *client) CloseConn(ctx context.Context) error {
+	return c.conn.Close()
+}
+
+// IsReady returns true if the grpc connection is ready to use.
+func (c *client) IsReady(ctx context.Context, in *emptypb.Empty) (bool, error) {
+	resp, err := c.grpcClt.IsReady(ctx, in)
+	if err != nil {
+		return false, err
+	}
+	return resp.GetReady(), nil
+}
+
+// SinkFn applies a function to a list of datum elements.
+func (c *client) SinkFn(ctx context.Context, datumList []*sinkpb.DatumRequest) ([]*sinkpb.Response, error) {
+	stream, err := c.grpcClt.SinkFn(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to execute c.grpcClt.SinkFn(): %w", err)
+	}
+	for _, datum := range datumList {
+		if err := stream.Send(datum); err != nil {
+			return nil, fmt.Errorf("failed to execute stream.Send(%v): %w", datum, err)
+		}
+	}
+	responseList, err := stream.CloseAndRecv()
+	if err != nil {
+		return nil, fmt.Errorf("failed to execute stream.CloseAndRecv(): %w", err)
+	}
+
+	return responseList.GetResponses(), nil
+}
diff --git a/pkg/sdkclient/sink/client/interface.go b/pkg/sdkclient/sink/client/interface.go
new file mode 100644
index 00000000..96c7b6ef
--- /dev/null
+++ b/pkg/sdkclient/sink/client/interface.go
@@ -0,0 +1,14 @@
+package client
+
+import (
+	"context"
+	sinkpb "github.com/numaproj/numaflow-go/pkg/apis/proto/sink/v1"
+	"google.golang.org/protobuf/types/known/emptypb"
+)
+
+// Client contains methods to call a gRPC client.
+type Client interface {
+	CloseConn(ctx context.Context) error
+	IsReady(ctx context.Context, in *emptypb.Empty) (bool, error)
+	SinkFn(ctx context.Context, datumList []*sinkpb.DatumRequest) ([]*sinkpb.Response, error)
+}
diff --git a/pkg/sdkclient/sink/client/options.go b/pkg/sdkclient/sink/client/options.go
new file mode 100644
index 00000000..6cd3cf29
--- /dev/null
+++ b/pkg/sdkclient/sink/client/options.go
@@ -0,0 +1,33 @@
+package client
+
+import "time"
+
+type options struct {
+	sockAddr                   string
+	serverInfoFilePath         string
+	serverInfoReadinessTimeout time.Duration
+}
+
+// Option is the interface to apply options.
+type Option func(*options)
+
+// WithSockAddr start the client with the given sock addr. This is mainly used for testing purpose.
+func WithSockAddr(addr string) Option {
+	return func(opts *options) {
+		opts.sockAddr = addr
+	}
+}
+
+// WithServerInfoFilePath start the client with the given server info file path. This is mainly used for testing purpose.
+func WithServerInfoFilePath(f string) Option {
+	return func(o *options) {
+		o.serverInfoFilePath = f
+	}
+}
+
+// WithServerInfoReadinessTimeout sets the server info readiness timeout to the given timeout.
+func WithServerInfoReadinessTimeout(t time.Duration) Option {
+	return func(o *options) {
+		o.serverInfoReadinessTimeout = t
+	}
+}
diff --git a/pkg/sdkclient/udf/client/client.go b/pkg/sdkclient/udf/client/client.go
new file mode 100644
index 00000000..ab860bf6
--- /dev/null
+++ b/pkg/sdkclient/udf/client/client.go
@@ -0,0 +1,255 @@
+package client
+
+import (
+	"context"
+	"fmt"
+	"github.com/numaproj/numaflow/pkg/udferr"
+	"io"
+	"log"
+	"strconv"
+	"time"
+
+	"golang.org/x/sync/errgroup"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/resolver"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/emptypb"
+
+	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
+	"github.com/numaproj/numaflow-go/pkg/function"
+	"github.com/numaproj/numaflow-go/pkg/info"
+)
+
+// client contains the grpc connection and the grpc client.
+type client struct {
+	conn    *grpc.ClientConn
+	grpcClt functionpb.UserDefinedFunctionClient
+}
+
+// New creates a new client object.
+func New(inputOptions ...Option) (*client, error) {
+	var opts = &options{
+		maxMessageSize:             function.DefaultMaxMessageSize,
+		serverInfoFilePath:         info.ServerInfoFilePath,
+		tcpSockAddr:                function.TCP_ADDR,
+		udsSockAddr:                function.UDS_ADDR,
+		serverInfoReadinessTimeout: 120 * time.Second, // Default timeout is 120 seconds
+	}
+
+	for _, inputOption := range inputOptions {
+		inputOption(opts)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), opts.serverInfoReadinessTimeout)
+	defer cancel()
+
+	if err := info.WaitUntilReady(ctx, info.WithServerInfoFilePath(opts.serverInfoFilePath)); err != nil {
+		return nil, fmt.Errorf("failed to wait until server info is ready: %w", err)
+	}
+
+	serverInfo, err := info.Read(info.WithServerInfoFilePath(opts.serverInfoFilePath))
+	if err != nil {
+		return nil, fmt.Errorf("failed to read server info: %w", err)
+	}
+	// TODO: Use serverInfo to check compatibility.
+	if serverInfo != nil {
+		log.Printf("ServerInfo: %v\n", serverInfo)
+	}
+
+	c := new(client)
+	var conn *grpc.ClientConn
+	var sockAddr string
+	// Make a TCP connection client for multiprocessing grpc server
+	if serverInfo.Protocol == info.TCP {
+		// Populate connection variables for client connection
+		// based on multiprocessing enabled/disabled
+		if err := regMultiProcResolver(serverInfo); err != nil {
+			return nil, fmt.Errorf("failed to start Multiproc Client: %w", err)
+		}
+
+		sockAddr = fmt.Sprintf("%s%s", connAddr, opts.tcpSockAddr)
+		log.Println("Multiprocessing TCP Client:", sockAddr)
+		conn, err = grpc.Dial(
+			fmt.Sprintf("%s:///%s", custScheme, custServiceName),
+			// This sets the initial load balancing policy as Round Robin
+			grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"round_robin":{}}]}`),
+			grpc.WithTransportCredentials(insecure.NewCredentials()),
+			grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(opts.maxMessageSize), grpc.MaxCallSendMsgSize(opts.maxMessageSize)),
+		)
+	} else {
+		sockAddr = fmt.Sprintf("%s:%s", function.UDS, opts.udsSockAddr)
+		log.Println("UDS Client:", sockAddr)
+		conn, err = grpc.Dial(sockAddr, grpc.WithTransportCredentials(insecure.NewCredentials()),
+			grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(opts.maxMessageSize), grpc.MaxCallSendMsgSize(opts.maxMessageSize)))
+	}
+	if err != nil {
+		return nil, fmt.Errorf("failed to execute grpc.Dial(%q): %w", sockAddr, err)
+	}
+	c.conn = conn
+	c.grpcClt = functionpb.NewUserDefinedFunctionClient(conn)
+	return c, nil
+}
+
+// CloseConn closes the grpc client connection.
+func (c *client) CloseConn(ctx context.Context) error {
+	return c.conn.Close()
+}
+
+// IsReady returns true if the grpc connection is ready to use.
+func (c *client) IsReady(ctx context.Context, in *emptypb.Empty) (bool, error) {
+	resp, err := c.grpcClt.IsReady(ctx, in)
+	if err != nil {
+		return false, err
+	}
+	return resp.GetReady(), nil
+}
+
+// MapFn applies a function to each datum element.
+func (c *client) MapFn(ctx context.Context, datum *functionpb.DatumRequest) ([]*functionpb.DatumResponse, error) {
+	mappedDatumList, err := c.grpcClt.MapFn(ctx, datum)
+	err = toUDFErr("c.grpcClt.MapFn", err)
+	if err != nil {
+		return nil, err
+	}
+	return mappedDatumList.GetElements(), nil
+}
+
+// MapStreamFn applies a function to each datum element and returns a stream.
+func (c *client) MapStreamFn(ctx context.Context, datum *functionpb.DatumRequest, datumCh chan<- *functionpb.DatumResponse) error {
+	defer close(datumCh)
+	stream, err := c.grpcClt.MapStreamFn(ctx, datum)
+	if err != nil {
+		return fmt.Errorf("failed to execute c.grpcClt.MapStreamFn(): %w", err)
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+			var resp *functionpb.DatumResponse
+			resp, err = stream.Recv()
+			if err == io.EOF {
+				return nil
+			}
+			if err != nil {
+				return err
+			}
+			datumCh <- resp
+		}
+	}
+}
+
+// MapTFn applies a function to each datum element.
+// In addition to map function, MapTFn also supports assigning a new event time to datum.
+// MapTFn can be used only at source vertex by source data transformer.
+func (c *client) MapTFn(ctx context.Context, datum *functionpb.DatumRequest) ([]*functionpb.DatumResponse, error) {
+	mappedDatumList, err := c.grpcClt.MapTFn(ctx, datum)
+	err = toUDFErr("c.grpcClt.MapTFn", err)
+	if err != nil {
+		return nil, err
+	}
+	return mappedDatumList.GetElements(), nil
+}
+
+// ReduceFn applies a reduce function to a datum stream.
+func (c *client) ReduceFn(ctx context.Context, datumStreamCh <-chan *functionpb.DatumRequest) ([]*functionpb.DatumResponse, error) {
+	var g errgroup.Group
+	datumList := make([]*functionpb.DatumResponse, 0)
+
+	stream, err := c.grpcClt.ReduceFn(ctx)
+	err = toUDFErr("c.grpcClt.ReduceFn", err)
+	if err != nil {
+		return nil, err
+	}
+	// stream the messages to server
+	g.Go(func() error {
+		var sendErr error
+		for datum := range datumStreamCh {
+			select {
+			case <-ctx.Done():
+				return status.FromContextError(ctx.Err()).Err()
+			default:
+				if sendErr = stream.Send(datum); sendErr != nil {
+					// we don't need to invoke close on the stream
+					// if there is an error gRPC will close the stream.
+					return sendErr
+				}
+			}
+		}
+		return stream.CloseSend()
+	})
+
+	// read the response from the server stream
+outputLoop:
+	for {
+		select {
+		case <-ctx.Done():
+			return nil, toUDFErr("ReduceFn OutputLoop", status.FromContextError(ctx.Err()).Err())
+		default:
+			var resp *functionpb.DatumResponseList
+			resp, err = stream.Recv()
+			if err == io.EOF {
+				break outputLoop
+			}
+			err = toUDFErr("ReduceFn stream.Recv()", err)
+			if err != nil {
+				return nil, err
+			}
+			datumList = append(datumList, resp.Elements...)
+		}
+	}
+
+	err = g.Wait()
+	err = toUDFErr("ReduceFn errorGroup", err)
+	if err != nil {
+		return nil, err
+	}
+
+	return datumList, nil
+}
+
+func toUDFErr(name string, err error) error {
+	if err == nil {
+		return nil
+	}
+	statusCode, ok := status.FromError(err)
+	// default udfError
+	udfError := udferr.New(udferr.NonRetryable, statusCode.Message())
+	// check if it's a standard status code
+	if !ok {
+		// if not, the status code will be unknown which we consider as non retryable
+		// return default udfError
+		log.Printf("failed %s: %s", name, udfError.Error())
+		return udfError
+	}
+	switch statusCode.Code() {
+	case codes.OK:
+		return nil
+	case codes.DeadlineExceeded, codes.Unavailable, codes.Unknown:
+		// update to retryable err
+		udfError = udferr.New(udferr.Retryable, statusCode.Message())
+		log.Printf("failed %s: %s", name, udfError.Error())
+		return udfError
+	default:
+		log.Printf("failed %s: %s", name, udfError.Error())
+		return udfError
+	}
+}
+
+// setConn function is used to populate the connection properties based
+// on multiprocessing TCP or UDS connection
+func regMultiProcResolver(svrInfo *info.ServerInfo) error {
+	numCpu, err := strconv.Atoi(svrInfo.Metadata["CPU_LIMIT"])
+	if err != nil {
+		return err
+	}
+	log.Println("Num CPU:", numCpu)
+	conn := buildConnAddrs(numCpu)
+	res := &multiProcResolverBuilder{addrsList: conn}
+	resolver.Register(res)
+	log.Println("TCP client list:", res.addrsList)
+	return nil
+}
diff --git a/pkg/sdkclient/udf/client/client_resolver.go b/pkg/sdkclient/udf/client/client_resolver.go
new file mode 100644
index 00000000..acc48e2a
--- /dev/null
+++ b/pkg/sdkclient/udf/client/client_resolver.go
@@ -0,0 +1,67 @@
+package client
+
+import (
+	"strconv"
+	"strings"
+
+	"github.com/numaproj/numaflow-go/pkg/function"
+	"google.golang.org/grpc/resolver"
+)
+
+const (
+	custScheme      = "udf"
+	custServiceName = "numaflow.numaproj.io"
+	connAddr        = "0.0.0.0"
+)
+
+type multiProcResolverBuilder struct {
+	addrsList []string
+}
+
+func (m *multiProcResolverBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
+	r := &multiProcResolver{
+		target: target,
+		cc:     cc,
+		addrsStore: map[string][]string{
+			custServiceName: m.addrsList,
+		},
+	}
+	r.start()
+	return r, nil
+}
+func (*multiProcResolverBuilder) Scheme() string { return custScheme }
+
+type multiProcResolver struct {
+	target     resolver.Target
+	cc         resolver.ClientConn
+	addrsStore map[string][]string
+}
+
+func (r *multiProcResolver) start() {
+	addrStrs := r.addrsStore[r.target.Endpoint()]
+	addrs := make([]resolver.Address, len(addrStrs))
+	//attrs := make(*attributes.Attributes, len(addrStrs))
+	for i, s := range addrStrs {
+		adr := strings.Split(s, ",")
+		addr := adr[0]
+		serv := adr[1]
+		addrs[i] = resolver.Address{Addr: addr, ServerName: serv}
+	}
+	err := r.cc.UpdateState(resolver.State{Addresses: addrs})
+	if err != nil {
+		return
+	}
+}
+func (*multiProcResolver) ResolveNow(o resolver.ResolveNowOptions) {}
+func (*multiProcResolver) Close()                                  {}
+func (*multiProcResolver) Resolve(target resolver.Target)          {}
+
+// Populate the connection list for the clients
+// Format (serverAddr, serverIdx) : (0.0.0.0:5551, 1)
+func buildConnAddrs(numCpu int) []string {
+	var conn = make([]string, numCpu)
+	for i := 0; i < numCpu; i++ {
+		conn[i] = connAddr + function.TCP_ADDR + "," + strconv.Itoa(i+1)
+	}
+	return conn
+}
diff --git a/pkg/sdkclient/udf/client/interface.go b/pkg/sdkclient/udf/client/interface.go
new file mode 100644
index 00000000..9083baef
--- /dev/null
+++ b/pkg/sdkclient/udf/client/interface.go
@@ -0,0 +1,17 @@
+package client
+
+import (
+	"context"
+	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
+	"google.golang.org/protobuf/types/known/emptypb"
+)
+
+// Client contains methods to call a gRPC client.
+type Client interface {
+	CloseConn(ctx context.Context) error
+	IsReady(ctx context.Context, in *emptypb.Empty) (bool, error)
+	MapFn(ctx context.Context, datum *functionpb.DatumRequest) ([]*functionpb.DatumResponse, error)
+	MapStreamFn(ctx context.Context, datum *functionpb.DatumRequest, datumCh chan<- *functionpb.DatumResponse) error
+	MapTFn(ctx context.Context, datum *functionpb.DatumRequest) ([]*functionpb.DatumResponse, error)
+	ReduceFn(ctx context.Context, datumStreamCh <-chan *functionpb.DatumRequest) ([]*functionpb.DatumResponse, error)
+}
diff --git a/pkg/sdkclient/udf/client/options.go b/pkg/sdkclient/udf/client/options.go
new file mode 100644
index 00000000..e3a21e6d
--- /dev/null
+++ b/pkg/sdkclient/udf/client/options.go
@@ -0,0 +1,49 @@
+package client
+
+import "time"
+
+type options struct {
+	tcpSockAddr                string
+	udsSockAddr                string
+	maxMessageSize             int
+	serverInfoFilePath         string
+	serverInfoReadinessTimeout time.Duration
+}
+
+// Option is the interface to apply options.
+type Option func(*options)
+
+// WithUdsSockAddr start the client with the given UDS sock addr. This is mainly used for testing purpose.
+func WithUdsSockAddr(addr string) Option {
+	return func(opts *options) {
+		opts.udsSockAddr = addr
+	}
+}
+
+// WithTcpSockAddr start the client with the given TCP sock addr. This is mainly used for testing purpose.
+func WithTcpSockAddr(addr string) Option {
+	return func(opts *options) {
+		opts.tcpSockAddr = addr
+	}
+}
+
+// WithMaxMessageSize sets the server max receive message size and the server max send message size to the given size.
+func WithMaxMessageSize(size int) Option {
+	return func(opts *options) {
+		opts.maxMessageSize = size
+	}
+}
+
+// WithServerInfoFilePath sets the server info file path to the given path.
+func WithServerInfoFilePath(f string) Option {
+	return func(o *options) {
+		o.serverInfoFilePath = f
+	}
+}
+
+// WithServerInfoReadinessTimeout sets the server info readiness timeout to the given timeout.
+func WithServerInfoReadinessTimeout(t time.Duration) Option {
+	return func(o *options) {
+		o.serverInfoReadinessTimeout = t
+	}
+}
diff --git a/pkg/sinks/udsink/udsink_grpc.go b/pkg/sinks/udsink/udsink_grpc.go
index 27aab81d..2940b967 100644
--- a/pkg/sinks/udsink/udsink_grpc.go
+++ b/pkg/sinks/udsink/udsink_grpc.go
@@ -19,22 +19,21 @@ package udsink
 import (
 	"context"
 	"fmt"
+	sinkclient "github.com/numaproj/numaflow/pkg/sdkclient/sink/client"
 	"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"
 )
 
 // UDSgRPCBasedUDSink applies user defined sink over gRPC (over Unix Domain Socket) client/server where server is the UDSink.
 type UDSgRPCBasedUDSink struct {
-	client sinksdk.Client
+	client sinkclient.Client
 }
 
 // NewUDSgRPCBasedUDSink returns UDSgRPCBasedUDSink
 func NewUDSgRPCBasedUDSink() (*UDSgRPCBasedUDSink, error) {
-	c, err := client.New()
+	c, err := sinkclient.New()
 	if err != nil {
 		return nil, fmt.Errorf("failed to create a new gRPC client: %w", err)
 	}
diff --git a/pkg/sources/transformer/grpc_transformer.go b/pkg/sources/transformer/grpc_transformer.go
index fdd89988..1a01be8c 100644
--- a/pkg/sources/transformer/grpc_transformer.go
+++ b/pkg/sources/transformer/grpc_transformer.go
@@ -19,17 +19,15 @@ package transformer
 import (
 	"context"
 	"fmt"
+	"github.com/numaproj/numaflow/pkg/sdkclient/udf/client"
 	"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-go/pkg/function/udferr"
-	"k8s.io/apimachinery/pkg/util/wait"
-
 	"github.com/numaproj/numaflow/pkg/forward/applier"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/udf/function"
+	"github.com/numaproj/numaflow/pkg/udferr"
+	"k8s.io/apimachinery/pkg/util/wait"
 
 	"google.golang.org/protobuf/types/known/emptypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -37,7 +35,7 @@ import (
 
 // gRPCBasedTransformer applies user defined transformer over gRPC (over Unix Domain Socket) client/server where server is the transformer.
 type gRPCBasedTransformer struct {
-	client functionsdk.Client
+	client client.Client
 }
 
 var _ applier.MapApplier = (*gRPCBasedTransformer)(nil)
@@ -52,7 +50,7 @@ func NewGRPCBasedTransformer() (*gRPCBasedTransformer, error) {
 }
 
 // NewGRPCBasedTransformerWithClient need this for testing
-func NewGRPCBasedTransformerWithClient(client functionsdk.Client) *gRPCBasedTransformer {
+func NewGRPCBasedTransformerWithClient(client client.Client) *gRPCBasedTransformer {
 	return &gRPCBasedTransformer{client: client}
 }
 
diff --git a/pkg/udf/function/uds_grpc.go b/pkg/udf/function/uds_grpc.go
index 3f968424..465d198f 100644
--- a/pkg/udf/function/uds_grpc.go
+++ b/pkg/udf/function/uds_grpc.go
@@ -19,13 +19,14 @@ package function
 import (
 	"context"
 	"fmt"
+	clientsdk "github.com/numaproj/numaflow/pkg/sdkclient/udf/client"
 	"strconv"
 	"sync"
 	"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/udferr"
+	"github.com/numaproj/numaflow/pkg/udferr"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/types/known/emptypb"
@@ -40,19 +41,19 @@ import (
 
 // UDSgRPCBasedUDF applies user defined function over gRPC (over Unix Domain Socket) client/server where server is the UDF.
 type UDSgRPCBasedUDF struct {
-	client functionsdk.Client
+	client clientsdk.Client
 }
 
 var _ map_applier.MapApplier = (*UDSgRPCBasedUDF)(nil)
 var _ reduce_applier.ReduceApplier = (*UDSgRPCBasedUDF)(nil)
 
 // NewUDSgRPCBasedUDF returns a new UDSgRPCBasedUDF object.
-func NewUDSgRPCBasedUDF(c functionsdk.Client) (*UDSgRPCBasedUDF, error) {
+func NewUDSgRPCBasedUDF(c clientsdk.Client) (*UDSgRPCBasedUDF, error) {
 	return &UDSgRPCBasedUDF{c}, nil
 }
 
 // NewUDSgRPCBasedUDFWithClient need this for testing
-func NewUDSgRPCBasedUDFWithClient(client functionsdk.Client) *UDSgRPCBasedUDF {
+func NewUDSgRPCBasedUDFWithClient(client clientsdk.Client) *UDSgRPCBasedUDF {
 	return &UDSgRPCBasedUDF{client: client}
 }
 
diff --git a/pkg/udf/map_udf.go b/pkg/udf/map_udf.go
index 7363994d..138ef896 100644
--- a/pkg/udf/map_udf.go
+++ b/pkg/udf/map_udf.go
@@ -19,9 +19,9 @@ package udf
 import (
 	"context"
 	"fmt"
+	clientsdk "github.com/numaproj/numaflow/pkg/sdkclient/udf/client"
 	"sync"
 
-	"github.com/numaproj/numaflow-go/pkg/function/client"
 	"go.uber.org/zap"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
@@ -125,7 +125,7 @@ func (u *MapUDFProcessor) Start(ctx context.Context) error {
 
 	log = log.With("protocol", "uds-grpc-map-udf")
 	maxMessageSize := sharedutil.LookupEnvIntOr(dfv1.EnvGRPCMaxMessageSize, dfv1.DefaultGRPCMaxMessageSize)
-	c, err := client.New(client.WithMaxMessageSize(maxMessageSize))
+	c, err := clientsdk.New(clientsdk.WithMaxMessageSize(maxMessageSize))
 	if err != nil {
 		return fmt.Errorf("failed to create a new gRPC client: %w", err)
 	}
diff --git a/pkg/udf/reduce_udf.go b/pkg/udf/reduce_udf.go
index 8d6cadde..38a008f0 100644
--- a/pkg/udf/reduce_udf.go
+++ b/pkg/udf/reduce_udf.go
@@ -19,13 +19,12 @@ package udf
 import (
 	"context"
 	"fmt"
+	clientsdk "github.com/numaproj/numaflow/pkg/sdkclient/udf/client"
 	"strings"
 	"sync"
 
 	"go.uber.org/zap"
 
-	"github.com/numaproj/numaflow-go/pkg/function/client"
-
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/forward"
 	"github.com/numaproj/numaflow/pkg/isb"
@@ -160,7 +159,7 @@ func (u *ReduceUDFProcessor) Start(ctx context.Context) error {
 	log = log.With("protocol", "uds-grpc-reduce-udf")
 
 	maxMessageSize := sharedutil.LookupEnvIntOr(dfv1.EnvGRPCMaxMessageSize, dfv1.DefaultGRPCMaxMessageSize)
-	c, err := client.New(client.WithMaxMessageSize(maxMessageSize))
+	c, err := clientsdk.New(clientsdk.WithMaxMessageSize(maxMessageSize))
 	if err != nil {
 		return fmt.Errorf("failed to create a new gRPC client: %w", err)
 	}
diff --git a/pkg/udferr/error.go b/pkg/udferr/error.go
new file mode 100644
index 00000000..b435db99
--- /dev/null
+++ b/pkg/udferr/error.go
@@ -0,0 +1,66 @@
+package udferr
+
+import (
+	"fmt"
+)
+
+// ErrKind represents if the error is retryable
+type ErrKind int16
+
+const (
+	Retryable    ErrKind = iota // The error is retryable
+	NonRetryable                // The error is non-retryable
+	Unknown                     // Unknown err kind
+)
+
+func (ek ErrKind) String() string {
+	switch ek {
+	case Retryable:
+		return "Retryable"
+	case NonRetryable:
+		return "NonRetryable"
+	case Unknown:
+		return "Unknown"
+	default:
+		return "Unknown"
+	}
+}
+
+// UDFError is returned to the main numaflow indicates the status of the error
+type UDFError struct {
+	errKind    ErrKind
+	errMessage string
+}
+
+func New(kind ErrKind, msg string) *UDFError {
+	return &UDFError{
+		errKind:    kind,
+		errMessage: msg,
+	}
+}
+
+func (e *UDFError) Error() string {
+	return fmt.Sprintf("%s: %s", e.errKind, e.errMessage)
+}
+
+func (e *UDFError) ErrorKind() ErrKind {
+	return e.errKind
+}
+
+func (e *UDFError) ErrorMessage() string {
+	return e.errMessage
+}
+
+// FromError gets error information from the UDFError
+func FromError(err error) (udfErr *UDFError, ok bool) {
+	if err == nil {
+		return nil, true
+	}
+	if se, ok := err.(interface {
+		ErrorKind() ErrKind
+		ErrorMessage() string
+	}); ok {
+		return &UDFError{se.ErrorKind(), se.ErrorMessage()}, true
+	}
+	return &UDFError{Unknown, err.Error()}, false
+}


Test patch:
diff --git a/pkg/reduce/pnf/processandforward_test.go b/pkg/reduce/pnf/processandforward_test.go
index db18859d..69671921 100644
--- a/pkg/reduce/pnf/processandforward_test.go
+++ b/pkg/reduce/pnf/processandforward_test.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/numaproj/numaflow/pkg/sdkclient/udf/clienttest"
 	"io"
 	"strings"
 	"testing"
@@ -43,7 +44,6 @@ import (
 	"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/stretchr/testify/assert"
 
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
diff --git a/pkg/sdkclient/sink/clienttest/client_test.go b/pkg/sdkclient/sink/clienttest/client_test.go
new file mode 100644
index 00000000..ebf3f4c2
--- /dev/null
+++ b/pkg/sdkclient/sink/clienttest/client_test.go
@@ -0,0 +1,137 @@
+package clienttest
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"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/stretchr/testify/assert"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/emptypb"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+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 TestIsReady(t *testing.T) {
+	var ctx = context.Background()
+	LintCleanCall()
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockClient := sinkmock.NewMockUserDefinedSinkClient(ctrl)
+	mockClient.EXPECT().IsReady(gomock.Any(), gomock.Any()).Return(&sinkpb.ReadyResponse{Ready: true}, nil)
+	mockClient.EXPECT().IsReady(gomock.Any(), gomock.Any()).Return(&sinkpb.ReadyResponse{Ready: false}, fmt.Errorf("mock connection refused"))
+
+	testClient, err := New(mockClient)
+	assert.NoError(t, err)
+	reflect.DeepEqual(testClient, &client{
+		grpcClt: mockClient,
+	})
+
+	ready, err := testClient.IsReady(ctx, &emptypb.Empty{})
+	assert.True(t, ready)
+	assert.NoError(t, err)
+
+	ready, err = testClient.IsReady(ctx, &emptypb.Empty{})
+	assert.False(t, ready)
+	assert.EqualError(t, err, "mock connection refused")
+}
+
+func TestSinkFn(t *testing.T) {
+	var ctx = context.Background()
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockSinkClient := sinkmock.NewMockUserDefinedSink_SinkFnClient(ctrl)
+	mockSinkClient.EXPECT().Send(gomock.Any()).Return(nil).AnyTimes()
+	mockSinkClient.EXPECT().CloseAndRecv().Return(&sinkpb.ResponseList{
+		Responses: []*sinkpb.Response{
+			{
+				Id:      "test_id_0",
+				Success: true,
+				ErrMsg:  "",
+			},
+			{
+				Id:      "test_id_1",
+				Success: false,
+				ErrMsg:  "mock sink message error",
+			},
+		},
+	}, nil)
+
+	mockSinkErrClient := sinkmock.NewMockUserDefinedSink_SinkFnClient(ctrl)
+	mockSinkErrClient.EXPECT().Send(gomock.Any()).Return(nil).AnyTimes()
+
+	mockClient := sinkmock.NewMockUserDefinedSinkClient(ctrl)
+	testDatumList := []*sinkpb.DatumRequest{
+		{
+			Id:        "test_id_0",
+			Value:     []byte(`sink_message_success`),
+			EventTime: &sinkpb.EventTime{EventTime: timestamppb.New(time.Unix(1661169600, 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(1661169600, 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.EXPECT().SinkFn(gomock.Any(), gomock.Any()).Return(mockSinkClient, nil)
+	mockClient.EXPECT().SinkFn(gomock.Any(), gomock.Any()).Return(mockSinkErrClient, fmt.Errorf("mock SinkFn error"))
+
+	testClient, err := New(mockClient)
+	assert.NoError(t, err)
+	reflect.DeepEqual(testClient, &client{
+		grpcClt: mockClient,
+	})
+
+	got, err := testClient.SinkFn(ctx, testDatumList)
+	reflect.DeepEqual(got, testResponseList)
+	assert.NoError(t, err)
+
+	got, err = testClient.SinkFn(ctx, testDatumList)
+	assert.Nil(t, got)
+	assert.EqualError(t, err, "failed to execute c.grpcClt.SinkFn(): mock SinkFn error")
+}
+
+// Check if there is a better way to resolve
+func LintCleanCall() {
+	var m = rpcMsg{}
+	fmt.Println(m.Matches(m))
+	fmt.Println(m)
+}
diff --git a/pkg/sdkclient/sink/clienttest/clienttest.go b/pkg/sdkclient/sink/clienttest/clienttest.go
new file mode 100644
index 00000000..20e353f1
--- /dev/null
+++ b/pkg/sdkclient/sink/clienttest/clienttest.go
@@ -0,0 +1,56 @@
+package clienttest
+
+import (
+	"context"
+	"fmt"
+	sinksdk "github.com/numaproj/numaflow/pkg/sdkclient/sink/client"
+
+	sinkpb "github.com/numaproj/numaflow-go/pkg/apis/proto/sink/v1"
+	"github.com/numaproj/numaflow-go/pkg/apis/proto/sink/v1/sinkmock"
+	"google.golang.org/protobuf/types/known/emptypb"
+)
+
+// client contains the grpc client for testing.
+type client struct {
+	grpcClt sinkpb.UserDefinedSinkClient
+}
+
+var _ sinksdk.Client = (*client)(nil)
+
+// New creates a new mock client object.
+func New(c *sinkmock.MockUserDefinedSinkClient) (*client, error) {
+	return &client{c}, nil
+}
+
+// CloseConn closes the grpc client connection.
+func (c *client) CloseConn(ctx context.Context) error {
+	return nil
+}
+
+// IsReady returns true if the grpc connection is ready to use.
+func (c *client) IsReady(ctx context.Context, in *emptypb.Empty) (bool, error) {
+	resp, err := c.grpcClt.IsReady(ctx, in)
+	if err != nil {
+		return false, err
+	}
+	return resp.GetReady(), nil
+}
+
+// SinkFn applies a function to a list of datum elements.
+func (c *client) SinkFn(ctx context.Context, datumList []*sinkpb.DatumRequest) ([]*sinkpb.Response, error) {
+	stream, err := c.grpcClt.SinkFn(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to execute c.grpcClt.SinkFn(): %w", err)
+	}
+	for _, datum := range datumList {
+		if err := stream.Send(datum); err != nil {
+			return nil, fmt.Errorf("failed to execute stream.Send(%v): %w", datum, err)
+		}
+	}
+	responseList, err := stream.CloseAndRecv()
+	if err != nil {
+		return nil, fmt.Errorf("failed to execute stream.CloseAndRecv(): %w", err)
+	}
+
+	return responseList.GetResponses(), nil
+}
diff --git a/pkg/sdkclient/udf/clienttest/client_test.go b/pkg/sdkclient/udf/clienttest/client_test.go
new file mode 100644
index 00000000..87bc8818
--- /dev/null
+++ b/pkg/sdkclient/udf/clienttest/client_test.go
@@ -0,0 +1,311 @@
+package clienttest
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"reflect"
+	"testing"
+	"time"
+
+	"golang.org/x/sync/errgroup"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/emptypb"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
+	"github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1/funcmock"
+)
+
+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 TestIsReady(t *testing.T) {
+	LintCleanCall()
+	var ctx = context.Background()
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockClient := funcmock.NewMockUserDefinedFunctionClient(ctrl)
+	mockClient.EXPECT().IsReady(gomock.Any(), gomock.Any()).Return(&functionpb.ReadyResponse{Ready: true}, nil)
+	mockClient.EXPECT().IsReady(gomock.Any(), gomock.Any()).Return(&functionpb.ReadyResponse{Ready: false}, fmt.Errorf("mock connection refused"))
+
+	testClient, err := New(mockClient)
+	assert.NoError(t, err)
+	reflect.DeepEqual(testClient, &client{
+		grpcClt: mockClient,
+	})
+
+	ready, err := testClient.IsReady(ctx, &emptypb.Empty{})
+	assert.True(t, ready)
+	assert.NoError(t, err)
+
+	ready, err = testClient.IsReady(ctx, &emptypb.Empty{})
+	assert.False(t, ready)
+	assert.EqualError(t, err, "mock connection refused")
+}
+
+func TestMapFn(t *testing.T) {
+	var ctx = context.Background()
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockClient := funcmock.NewMockUserDefinedFunctionClient(ctrl)
+	testRequestDatum := &functionpb.DatumRequest{
+		Keys:      []string{"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{})},
+	}
+	testResponseDatum := &functionpb.DatumResponse{
+		Keys:      []string{"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: testRequestDatum}).Return(&functionpb.DatumResponseList{
+		Elements: []*functionpb.DatumResponse{
+			testResponseDatum,
+		},
+	}, nil)
+	mockClient.EXPECT().MapFn(gomock.Any(), &rpcMsg{msg: testRequestDatum}).Return(&functionpb.DatumResponseList{
+		Elements: []*functionpb.DatumResponse{
+			nil,
+		},
+	}, fmt.Errorf("mock MapFn error"))
+	testClient, err := New(mockClient)
+	assert.NoError(t, err)
+	reflect.DeepEqual(testClient, &client{
+		grpcClt: mockClient,
+	})
+	got, err := testClient.MapFn(ctx, testRequestDatum)
+	reflect.DeepEqual(got, testRequestDatum)
+	assert.NoError(t, err)
+
+	got, err = testClient.MapFn(ctx, testRequestDatum)
+	assert.Nil(t, got)
+	assert.EqualError(t, err, "NonRetryable: mock MapFn error")
+}
+
+func TestMapStreamFn(t *testing.T) {
+	var ctx = context.Background()
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockClient := funcmock.NewMockUserDefinedFunctionClient(ctrl)
+	mockMapStreamClient := funcmock.NewMockUserDefinedFunction_MapStreamFnClient(ctrl)
+
+	requestDatum := &functionpb.DatumRequest{
+		Keys:      []string{"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{})},
+	}
+	expectedDatum := &functionpb.DatumResponse{
+		Keys:      []string{"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{})},
+	}
+
+	mockMapStreamClient.EXPECT().Recv().Return(expectedDatum, nil).Times(1)
+	mockMapStreamClient.EXPECT().Recv().Return(nil, io.EOF).Times(1)
+
+	mockClient.EXPECT().MapStreamFn(gomock.Any(), &rpcMsg{msg: requestDatum}).Return(mockMapStreamClient, nil)
+
+	testClient, err := New(mockClient)
+	assert.NoError(t, err)
+	reflect.DeepEqual(testClient, &client{
+		grpcClt: mockClient,
+	})
+
+	datumCh := make(chan *functionpb.DatumResponse)
+	datumResponses := make([]*functionpb.DatumResponse, 0)
+
+	go func() {
+		err := testClient.MapStreamFn(ctx, requestDatum, datumCh)
+		assert.NoError(t, err)
+	}()
+
+	for msg := range datumCh {
+		datumResponses = append(datumResponses, msg)
+	}
+	assert.True(t, reflect.DeepEqual(datumResponses, []*functionpb.DatumResponse{expectedDatum}))
+}
+
+func TestMapStreamFnErr(t *testing.T) {
+	var ctx = context.Background()
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockClient := funcmock.NewMockUserDefinedFunctionClient(ctrl)
+	mockMapStreamClient := funcmock.NewMockUserDefinedFunction_MapStreamFnClient(ctrl)
+
+	requestDatum := &functionpb.DatumRequest{
+		Keys:      []string{"test_error_key"},
+		Value:     []byte(`forward_message`),
+		EventTime: &functionpb.EventTime{EventTime: timestamppb.New(time.Unix(1661169600, 0))},
+		Watermark: &functionpb.Watermark{Watermark: timestamppb.New(time.Time{})},
+	}
+	expectedDatum := &functionpb.DatumResponse{
+		Keys:      []string{"test_error_key"},
+		Value:     []byte(`forward_message`),
+		EventTime: &functionpb.EventTime{EventTime: timestamppb.New(time.Unix(1661169600, 0))},
+		Watermark: &functionpb.Watermark{Watermark: timestamppb.New(time.Time{})},
+	}
+
+	mockMapStreamClient.EXPECT().Recv().Return(expectedDatum, nil).Times(1)
+	mockMapStreamClient.EXPECT().Recv().Return(
+		nil, errors.New("mock error for map")).AnyTimes()
+
+	mockClient.EXPECT().MapStreamFn(gomock.Any(), &rpcMsg{msg: requestDatum}).Return(mockMapStreamClient, nil)
+
+	testClient, err := New(mockClient)
+	assert.NoError(t, err)
+	reflect.DeepEqual(testClient, &client{
+		grpcClt: mockClient,
+	})
+
+	datumCh := make(chan *functionpb.DatumResponse)
+	errs, ctx := errgroup.WithContext(ctx)
+	errs.Go(func() error {
+		return testClient.MapStreamFn(ctx, requestDatum, datumCh)
+	})
+
+	datumResponses := make([]*functionpb.DatumResponse, 0)
+	for msg := range datumCh {
+		datumResponses = append(datumResponses, msg)
+	}
+	assert.True(t, reflect.DeepEqual(datumResponses, []*functionpb.DatumResponse{expectedDatum}))
+	err = errs.Wait()
+	assert.Error(t, err)
+}
+
+func TestMapTFn(t *testing.T) {
+	var ctx = context.Background()
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockClient := funcmock.NewMockUserDefinedFunctionClient(ctrl)
+	testDatumRequest := &functionpb.DatumRequest{
+		Keys:      []string{"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{})},
+	}
+	testDatumResponse := &functionpb.DatumResponse{
+		Keys:      []string{"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().MapTFn(gomock.Any(), &rpcMsg{msg: testDatumRequest}).Return(&functionpb.DatumResponseList{
+		Elements: []*functionpb.DatumResponse{
+			testDatumResponse,
+		},
+	}, nil)
+	mockClient.EXPECT().MapTFn(gomock.Any(), &rpcMsg{msg: testDatumRequest}).Return(&functionpb.DatumResponseList{
+		Elements: []*functionpb.DatumResponse{
+			nil,
+		},
+	}, fmt.Errorf("mock MapTFn error"))
+	testClient, err := New(mockClient)
+	assert.NoError(t, err)
+	reflect.DeepEqual(testClient, &client{
+		grpcClt: mockClient,
+	})
+	got, err := testClient.MapTFn(ctx, testDatumRequest)
+	reflect.DeepEqual(got, testDatumRequest)
+	assert.NoError(t, err)
+
+	got, err = testClient.MapTFn(ctx, testDatumRequest)
+	assert.Nil(t, got)
+	assert.EqualError(t, err, "NonRetryable: mock MapTFn error")
+}
+
+func TestReduceFn(t *testing.T) {
+	var ctx = context.Background()
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockClient := funcmock.NewMockUserDefinedFunctionClient(ctrl)
+	mockReduceClient := funcmock.NewMockUserDefinedFunction_ReduceFnClient(ctrl)
+
+	mockReduceClient.EXPECT().Send(gomock.Any()).Return(nil).AnyTimes()
+	mockReduceClient.EXPECT().CloseSend().Return(nil).AnyTimes()
+
+	testDatumList := &functionpb.DatumResponseList{
+		Elements: []*functionpb.DatumResponse{
+			{
+				Keys:  []string{"reduced_result_key"},
+				Value: []byte(`forward_message`),
+			},
+		},
+	}
+
+	mockReduceClient.EXPECT().Recv().Return(testDatumList, nil).Times(3)
+
+	mockReduceClient.EXPECT().Recv().Return(nil, io.EOF).Times(1)
+
+	datumCh := make(chan *functionpb.DatumRequest)
+
+	mockClient.EXPECT().ReduceFn(gomock.Any(), gomock.Any()).Return(mockReduceClient, nil)
+
+	testClient, err := New(mockClient)
+	assert.NoError(t, err)
+	reflect.DeepEqual(testClient, &client{
+		grpcClt: mockClient,
+	})
+
+	expectedDatumList := &functionpb.DatumResponseList{
+		Elements: []*functionpb.DatumResponse{
+			{
+				Keys:  []string{"reduced_result_key"},
+				Value: []byte(`forward_message`),
+			},
+			{
+				Keys:  []string{"reduced_result_key"},
+				Value: []byte(`forward_message`),
+			},
+			{
+				Keys:  []string{"reduced_result_key"},
+				Value: []byte(`forward_message`),
+			},
+		},
+	}
+	close(datumCh)
+	got, err := testClient.ReduceFn(ctx, datumCh)
+	assert.True(t, reflect.DeepEqual(got, expectedDatumList.Elements))
+	assert.NoError(t, err)
+}
+
+// Check if there is a better way to resolve
+func LintCleanCall() {
+	var m = rpcMsg{}
+	fmt.Println(m.Matches(m))
+	fmt.Println(m)
+
+}
diff --git a/pkg/sdkclient/udf/clienttest/clienttest.go b/pkg/sdkclient/udf/clienttest/clienttest.go
new file mode 100644
index 00000000..acf0f09e
--- /dev/null
+++ b/pkg/sdkclient/udf/clienttest/clienttest.go
@@ -0,0 +1,174 @@
+package clienttest
+
+import (
+	"context"
+	"fmt"
+	"github.com/numaproj/numaflow/pkg/udferr"
+	"io"
+	"log"
+
+	"golang.org/x/sync/errgroup"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/emptypb"
+
+	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
+	"github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1/funcmock"
+)
+
+// client contains the grpc client for testing.
+type client struct {
+	grpcClt functionpb.UserDefinedFunctionClient
+}
+
+// New creates a new mock client object.
+func New(c *funcmock.MockUserDefinedFunctionClient) (*client, error) {
+	return &client{c}, nil
+}
+
+// CloseConn closes the grpc client connection.
+func (c *client) CloseConn(ctx context.Context) error {
+	return nil
+}
+
+// IsReady returns true if the grpc connection is ready to use.
+func (c *client) IsReady(ctx context.Context, in *emptypb.Empty) (bool, error) {
+	resp, err := c.grpcClt.IsReady(ctx, in)
+	if err != nil {
+		return false, err
+	}
+	return resp.GetReady(), nil
+}
+
+// MapFn applies a function to each datum element.
+func (c *client) MapFn(ctx context.Context, datum *functionpb.DatumRequest) ([]*functionpb.DatumResponse, error) {
+	mappedDatumList, err := c.grpcClt.MapFn(ctx, datum)
+	err = toUDFErr("c.grpcClt.MapFn", err)
+	if err != nil {
+		return nil, err
+	}
+	return mappedDatumList.GetElements(), nil
+}
+
+// MapStreamFn applies a function to each datum element and returns a stream.
+func (c *client) MapStreamFn(ctx context.Context, datum *functionpb.DatumRequest, datumCh chan<- *functionpb.DatumResponse) error {
+	defer close(datumCh)
+	stream, err := c.grpcClt.MapStreamFn(ctx, datum)
+	if err != nil {
+		return fmt.Errorf("failed to execute c.grpcClt.MapStreamFn(): %w", err)
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+			var resp *functionpb.DatumResponse
+			resp, err = stream.Recv()
+			if err == io.EOF {
+				return nil
+			}
+			if err != nil {
+				return err
+			}
+			datumCh <- resp
+		}
+	}
+}
+
+// MapTFn applies a function to each datum element.
+// In addition to map function, MapTFn also supports assigning a new event time to datum.
+// MapTFn can be used only at source vertex by source data transformer.
+func (c *client) MapTFn(ctx context.Context, datum *functionpb.DatumRequest) ([]*functionpb.DatumResponse, error) {
+	mappedDatumList, err := c.grpcClt.MapTFn(ctx, datum)
+	err = toUDFErr("c.grpcClt.MapTFn", err)
+	if err != nil {
+		return nil, err
+	}
+	return mappedDatumList.GetElements(), nil
+}
+
+// ReduceFn applies a reduce function to a datum stream.
+func (c *client) ReduceFn(ctx context.Context, datumStreamCh <-chan *functionpb.DatumRequest) ([]*functionpb.DatumResponse, error) {
+	var g errgroup.Group
+	datumList := make([]*functionpb.DatumResponse, 0)
+
+	stream, err := c.grpcClt.ReduceFn(ctx)
+	err = toUDFErr("c.grpcClt.ReduceFn", err)
+	if err != nil {
+		return nil, err
+	}
+	// stream the messages to server
+	g.Go(func() error {
+		var sendErr error
+		for datum := range datumStreamCh {
+			select {
+			case <-ctx.Done():
+				return status.FromContextError(ctx.Err()).Err()
+			default:
+				if sendErr = stream.Send(datum); sendErr != nil {
+					// we don't need to invoke close on the stream
+					// if there is an error gRPC will close the stream.
+					return sendErr
+				}
+			}
+		}
+		return stream.CloseSend()
+	})
+
+	// read the response from the server stream
+outputLoop:
+	for {
+		select {
+		case <-ctx.Done():
+			return nil, toUDFErr("ReduceFn OutputLoop", status.FromContextError(ctx.Err()).Err())
+		default:
+			var resp *functionpb.DatumResponseList
+			resp, err = stream.Recv()
+			if err == io.EOF {
+				break outputLoop
+			}
+			err = toUDFErr("ReduceFn stream.Recv()", err)
+			if err != nil {
+				return nil, err
+			}
+			datumList = append(datumList, resp.Elements...)
+		}
+	}
+
+	err = g.Wait()
+	err = toUDFErr("ReduceFn errorGroup", err)
+	if err != nil {
+		return nil, err
+	}
+
+	return datumList, nil
+}
+
+func toUDFErr(name string, err error) error {
+	if err == nil {
+		return nil
+	}
+	statusCode, ok := status.FromError(err)
+	// default udfError
+	udfError := udferr.New(udferr.NonRetryable, statusCode.Message())
+	// check if it's a standard status code
+	if !ok {
+		// if not, the status code will be unknown which we consider as non retryable
+		// return default udfError
+		log.Printf("failed %s: %s", name, udfError.Error())
+		return udfError
+	}
+	switch statusCode.Code() {
+	case codes.OK:
+		return nil
+	case codes.DeadlineExceeded, codes.Unavailable, codes.Unknown:
+		// update to retryable err
+		udfError = udferr.New(udferr.Retryable, statusCode.Message())
+		log.Printf("failed %s: %s", name, udfError.Error())
+		return udfError
+	default:
+		log.Printf("failed %s: %s", name, udfError.Error())
+		return udfError
+	}
+}
diff --git a/pkg/sinks/udsink/udsink_grpc_test.go b/pkg/sinks/udsink/udsink_grpc_test.go
index e1e87846..da75f92a 100644
--- a/pkg/sinks/udsink/udsink_grpc_test.go
+++ b/pkg/sinks/udsink/udsink_grpc_test.go
@@ -19,13 +19,13 @@ package udsink
 import (
 	"context"
 	"fmt"
+	"github.com/numaproj/numaflow/pkg/sdkclient/sink/clienttest"
 	"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/types/known/timestamppb"
 )
diff --git a/pkg/sources/transformer/grpc_transformer_test.go b/pkg/sources/transformer/grpc_transformer_test.go
index bb50b771..a542f5a2 100644
--- a/pkg/sources/transformer/grpc_transformer_test.go
+++ b/pkg/sources/transformer/grpc_transformer_test.go
@@ -20,13 +20,13 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/numaproj/numaflow/pkg/sdkclient/udf/clienttest"
 	"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/stretchr/testify/assert"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
diff --git a/pkg/udf/function/uds_grpc_test.go b/pkg/udf/function/uds_grpc_test.go
index 613876cb..2d1e118a 100644
--- a/pkg/udf/function/uds_grpc_test.go
+++ b/pkg/udf/function/uds_grpc_test.go
@@ -21,13 +21,13 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/numaproj/numaflow/pkg/sdkclient/udf/clienttest"
 	"io"
 	"testing"
 	"time"
 
 	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"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
diff --git a/pkg/udferr/error_test.go b/pkg/udferr/error_test.go
new file mode 100644
index 00000000..34eca429
--- /dev/null
+++ b/pkg/udferr/error_test.go
@@ -0,0 +1,204 @@
+package udferr
+
+import (
+	"fmt"
+	"reflect"
+	"testing"
+)
+
+func TestErrKind_String(t *testing.T) {
+	tests := []struct {
+		name string
+		ek   ErrKind
+		want string
+	}{
+		{
+			name: "retry",
+			ek:   0,
+			want: "Retryable",
+		},
+		{
+			name: "non-retry",
+			ek:   1,
+			want: "NonRetryable",
+		},
+		{
+			name: "unknown_1",
+			ek:   2,
+			want: "Unknown",
+		},
+		{
+			name: "unknown_2",
+			ek:   3,
+			want: "Unknown",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.ek.String(); got != tt.want {
+				t.Errorf("String() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestFromError(t *testing.T) {
+	type args struct {
+		err error
+	}
+	tests := []struct {
+		name   string
+		args   args
+		want   *UDFError
+		wantOk bool
+	}{
+		{
+			name: "good_retry",
+			args: args{
+				err: &UDFError{
+					errKind:    0,
+					errMessage: "Retryable Error",
+				},
+			},
+			want:   &UDFError{Retryable, "Retryable Error"},
+			wantOk: true,
+		},
+		{
+			name: "good_non_retry",
+			args: args{
+				err: &UDFError{
+					errKind:    1,
+					errMessage: "NonRetryable Error",
+				},
+			},
+			want:   &UDFError{NonRetryable, "NonRetryable Error"},
+			wantOk: true,
+		},
+		{
+			name: "good_unknown_1",
+			args: args{
+				err: &UDFError{
+					errKind:    2,
+					errMessage: "Unknown Error",
+				},
+			},
+			want:   &UDFError{Unknown, "Unknown Error"},
+			wantOk: true,
+		},
+		{
+			name: "good_not_standard",
+			args: args{
+				err: fmt.Errorf("not a standard error"),
+			},
+			want:   &UDFError{Unknown, "not a standard error"},
+			wantOk: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, gotOk := FromError(tt.args.err)
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("FromError() gotK = %v, want %v", got, tt.want)
+			}
+			if gotOk != tt.wantOk {
+				t.Errorf("FromError() gotOk = %v, want %v", gotOk, tt.wantOk)
+			}
+		})
+	}
+}
+
+func TestUDFError_Error(t *testing.T) {
+	type fields struct {
+		ErrKind    ErrKind
+		ErrMessage string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		want   string
+	}{
+		{
+			name: "retryable",
+			fields: fields{
+				ErrKind:    0,
+				ErrMessage: "deadline exceeds",
+			},
+			want: "Retryable: deadline exceeds",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			e := UDFError{
+				errKind:    tt.fields.ErrKind,
+				errMessage: tt.fields.ErrMessage,
+			}
+			if got := e.Error(); got != tt.want {
+				t.Errorf("Error() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestUDFError_ErrorKind(t *testing.T) {
+	type fields struct {
+		ErrKind    ErrKind
+		ErrMessage string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		want   ErrKind
+	}{
+		{
+			name: "retryable",
+			fields: fields{
+				ErrKind:    0,
+				ErrMessage: "deadline exceeds",
+			},
+			want: Retryable,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			e := UDFError{
+				errKind:    tt.fields.ErrKind,
+				errMessage: tt.fields.ErrMessage,
+			}
+			if got := e.ErrorKind(); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("ErrorKind() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestUDFError_ErrorMessage(t *testing.T) {
+	type fields struct {
+		ErrKind    ErrKind
+		ErrMessage string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		want   string
+	}{
+		{
+			name: "retryable",
+			fields: fields{
+				ErrKind:    0,
+				ErrMessage: "deadline exceeds",
+			},
+			want: "deadline exceeds",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			e := UDFError{
+				errKind:    tt.fields.ErrKind,
+				errMessage: tt.fields.ErrMessage,
+			}
+			if got := e.ErrorMessage(); got != tt.want {
+				t.Errorf("ErrorMessage() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}