Repository: numaproj/numaflow
Pull request #535 (2023-02-08 19:32:54)

chore: expose UDSgRPC type because it is required in healthcheck
Signed-off-by: Vigith Maurice <vigith@gmail.com>

Fixes: #533 

whynowy: What is the change in health check?
chromevoid: > What is the change in health check?

Will do the health check refactor in another PR.

Patch:
diff --git a/pkg/sinks/udsink/sink.go b/pkg/sinks/udsink/sink.go
index 1b00ca7d..dbc6bce1 100644
--- a/pkg/sinks/udsink/sink.go
+++ b/pkg/sinks/udsink/sink.go
@@ -40,7 +40,7 @@ type UserDefinedSink struct {
 	pipelineName string
 	isdf         *forward.InterStepDataForward
 	logger       *zap.SugaredLogger
-	udsink       *udsGRPCBasedUDSink
+	udsink       *UDSgRPCBasedUDSink
 }
 
 type Option func(*UserDefinedSink) error
@@ -73,7 +73,7 @@ func NewUserDefinedSink(vertex *dfv1.Vertex, fromBuffer isb.BufferReader, fetchW
 			forwardOpts = append(forwardOpts, forward.WithReadBatchSize(int64(*x.ReadBatchSize)))
 		}
 	}
-	udsink, err := NewUDSGRPCBasedUDSink()
+	udsink, err := NewUDSgRPCBasedUDSink()
 	if err != nil {
 		return nil, fmt.Errorf("failed to create gRPC client, %w", err)
 	}
diff --git a/pkg/sinks/udsink/udsink_grpc.go b/pkg/sinks/udsink/udsink_grpc.go
index df6be998..19feed72 100644
--- a/pkg/sinks/udsink/udsink_grpc.go
+++ b/pkg/sinks/udsink/udsink_grpc.go
@@ -27,25 +27,27 @@ import (
 	"google.golang.org/protobuf/types/known/emptypb"
 )
 
-type udsGRPCBasedUDSink struct {
+// UDSgRPCBasedUDSink applies user defined sink over gRPC (over Unix Domain Socket) client/server where server is the UDSink.
+type UDSgRPCBasedUDSink struct {
 	client sinksdk.Client
 }
 
-func NewUDSGRPCBasedUDSink() (*udsGRPCBasedUDSink, error) {
+// NewUDSgRPCBasedUDSink returns UDSgRPCBasedUDSink
+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
+	return &UDSgRPCBasedUDSink{c}, nil
 }
 
 // CloseConn closes the gRPC client connection.
-func (u *udsGRPCBasedUDSink) CloseConn(ctx context.Context) error {
+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 {
+func (u *UDSgRPCBasedUDSink) WaitUntilReady(ctx context.Context) error {
 	for {
 		select {
 		case <-ctx.Done():
@@ -59,7 +61,7 @@ func (u *udsGRPCBasedUDSink) WaitUntilReady(ctx context.Context) error {
 	}
 }
 
-func (u *udsGRPCBasedUDSink) Apply(ctx context.Context, dList []*sinkpb.Datum) []error {
+func (u *UDSgRPCBasedUDSink) Apply(ctx context.Context, dList []*sinkpb.Datum) []error {
 	errs := make([]error, len(dList))
 
 	responseList, err := u.client.SinkFn(ctx, dList)
diff --git a/pkg/udf/function/uds_grpc.go b/pkg/udf/function/uds_grpc.go
index 76436ddf..18839298 100644
--- a/pkg/udf/function/uds_grpc.go
+++ b/pkg/udf/function/uds_grpc.go
@@ -35,31 +35,31 @@ import (
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
 )
 
-// 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 _ map_applier.MapApplier = (*udsGRPCBasedUDF)(nil)
-var _ reduce_applier.ReduceApplier = (*udsGRPCBasedUDF)(nil)
+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) {
-	return &udsGRPCBasedUDF{c}, nil
+// NewUDSgRPCBasedUDF returns a new UDSgRPCBasedUDF object.
+func NewUDSgRPCBasedUDF(c functionsdk.Client) (*UDSgRPCBasedUDF, error) {
+	return &UDSgRPCBasedUDF{c}, nil
 }
 
-// NewUdsGRPCBasedUDFWithClient need this for testing
-func NewUdsGRPCBasedUDFWithClient(client functionsdk.Client) *udsGRPCBasedUDF {
-	return &udsGRPCBasedUDF{client: client}
+// NewUDSgRPCBasedUDFWithClient need this for testing
+func NewUDSgRPCBasedUDFWithClient(client functionsdk.Client) *UDSgRPCBasedUDF {
+	return &UDSgRPCBasedUDF{client: client}
 }
 
 // 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():
@@ -73,7 +73,7 @@ func (u *udsGRPCBasedUDF) WaitUntilReady(ctx context.Context) error {
 	}
 }
 
-func (u *udsGRPCBasedUDF) ApplyMap(ctx context.Context, readMessage *isb.ReadMessage) ([]*isb.Message, error) {
+func (u *UDSgRPCBasedUDF) ApplyMap(ctx context.Context, readMessage *isb.ReadMessage) ([]*isb.Message, error) {
 	key := readMessage.Key
 	payload := readMessage.Body.Payload
 	offset := readMessage.ReadOffset
@@ -119,7 +119,7 @@ func (u *udsGRPCBasedUDF) ApplyMap(ctx context.Context, readMessage *isb.ReadMes
 // should we pass metadata information ?
 
 // ApplyReduce accepts a channel of isbMessages and returns the aggregated result
-func (u *udsGRPCBasedUDF) ApplyReduce(ctx context.Context, partitionID *partition.ID, messageStream <-chan *isb.ReadMessage) ([]*isb.Message, error) {
+func (u *UDSgRPCBasedUDF) ApplyReduce(ctx context.Context, partitionID *partition.ID, messageStream <-chan *isb.ReadMessage) ([]*isb.Message, error) {
 	datumCh := make(chan *functionpb.Datum)
 	var wg sync.WaitGroup
 	var result []*functionpb.Datum
diff --git a/pkg/udf/map_udf.go b/pkg/udf/map_udf.go
index a7b7972d..51449fdd 100644
--- a/pkg/udf/map_udf.go
+++ b/pkg/udf/map_udf.go
@@ -108,7 +108,7 @@ func (u *MapUDFProcessor) Start(ctx context.Context) error {
 		return fmt.Errorf("failed to create a new gRPC client: %w", err)
 	}
 
-	udfHandler, err := function.NewUDSGRPCBasedUDF(c)
+	udfHandler, err := function.NewUDSgRPCBasedUDF(c)
 	if err != nil {
 		return fmt.Errorf("failed to create gRPC client, %w", err)
 	}
diff --git a/pkg/udf/reduce_udf.go b/pkg/udf/reduce_udf.go
index d0a225f3..7d0dd14d 100644
--- a/pkg/udf/reduce_udf.go
+++ b/pkg/udf/reduce_udf.go
@@ -138,7 +138,7 @@ func (u *ReduceUDFProcessor) Start(ctx context.Context) error {
 		return fmt.Errorf("failed to create a new gRPC client: %w", err)
 	}
 
-	udfHandler, err := function.NewUDSGRPCBasedUDF(c)
+	udfHandler, err := function.NewUDSgRPCBasedUDF(c)
 	if err != nil {
 		return fmt.Errorf("failed to create gRPC client, %w", err)
 	}


Test patch:
diff --git a/pkg/reduce/pnf/processandforward_test.go b/pkg/reduce/pnf/processandforward_test.go
index 94eeacfc..f2246d86 100644
--- a/pkg/reduce/pnf/processandforward_test.go
+++ b/pkg/reduce/pnf/processandforward_test.go
@@ -139,7 +139,7 @@ func TestProcessAndForward_Process(t *testing.T) {
 	mockClient.EXPECT().ReduceFn(gomock.Any(), gomock.Any()).Return(mockReduceClient, nil)
 
 	c, _ := clienttest.New(mockClient)
-	client := udfcall.NewUdsGRPCBasedUDFWithClient(c)
+	client := udfcall.NewUDSgRPCBasedUDFWithClient(c)
 
 	assert.NoError(t, err)
 	_, publishWatermark := generic.BuildNoOpWatermarkProgressorsFromBufferMap(make(map[string]isb.BufferWriter))
diff --git a/pkg/sinks/udsink/udsink_grpc_test.go b/pkg/sinks/udsink/udsink_grpc_test.go
index f5a7903f..77b30946 100644
--- a/pkg/sinks/udsink/udsink_grpc_test.go
+++ b/pkg/sinks/udsink/udsink_grpc_test.go
@@ -30,12 +30,12 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
-func NewMockUDSGRPCBasedUDF(mockClient *sinkmock.MockUserDefinedSinkClient) *udsGRPCBasedUDSink {
+func NewMockUDSgRPCBasedUDSink(mockClient *sinkmock.MockUserDefinedSinkClient) *UDSgRPCBasedUDSink {
 	c, _ := clienttest.New(mockClient)
-	return &udsGRPCBasedUDSink{c}
+	return &UDSgRPCBasedUDSink{c}
 }
 
-func TestGRPCBasedUDF_WaitUntilReadyWithMockClient(t *testing.T) {
+func Test_gRPCBasedUDSink_WaitUntilReadyWithMockClient(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 
@@ -51,12 +51,12 @@ func TestGRPCBasedUDF_WaitUntilReadyWithMockClient(t *testing.T) {
 		}
 	}()
 
-	u := NewMockUDSGRPCBasedUDF(mockClient)
+	u := NewMockUDSgRPCBasedUDSink(mockClient)
 	err := u.WaitUntilReady(ctx)
 	assert.NoError(t, err)
 }
 
-func TestGRPCBasedUDF_ApplyWithMockClient(t *testing.T) {
+func Test_gRPCBasedUDSink_ApplyWithMockClient(t *testing.T) {
 	t.Run("test success", func(t *testing.T) {
 
 		ctrl := gomock.NewController(t)
@@ -107,7 +107,7 @@ func TestGRPCBasedUDF_ApplyWithMockClient(t *testing.T) {
 			}
 		}()
 
-		u := NewMockUDSGRPCBasedUDF(mockClient)
+		u := NewMockUDSgRPCBasedUDSink(mockClient)
 		gotErrList := u.Apply(ctx, testDatumList)
 		assert.Equal(t, 2, len(gotErrList))
 		assert.Equal(t, nil, gotErrList[0])
@@ -149,7 +149,7 @@ func TestGRPCBasedUDF_ApplyWithMockClient(t *testing.T) {
 			}
 		}()
 
-		u := NewMockUDSGRPCBasedUDF(mockClient)
+		u := NewMockUDSgRPCBasedUDSink(mockClient)
 		gotErrList := u.Apply(ctx, testDatumList)
 		expectedErrList := []error{
 			ApplyUDSinkErr{
diff --git a/pkg/udf/function/uds_grpc_test.go b/pkg/udf/function/uds_grpc_test.go
index 15d4f370..a286f125 100644
--- a/pkg/udf/function/uds_grpc_test.go
+++ b/pkg/udf/function/uds_grpc_test.go
@@ -39,9 +39,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) {