Repository: numaproj/numaflow
Pull request #421 (2022-12-12 19:03:48)

feat: udsink grpc stream
fixes: https://github.com/numaproj/numaflow/issues/335
Update tests based on https://github.com/numaproj/numaflow-go/pull/25


chromevoid: TODO: need to clean up all the numaio images and docs...

Patch:
diff --git a/docs/quick-start.md b/docs/quick-start.md
index 43356f01..201e5b9b 100644
--- a/docs/quick-start.md
+++ b/docs/quick-start.md
@@ -114,7 +114,7 @@ kubectl logs -f even-odd-odd-sink-0-a6p0n
 View the UI for the advanced pipeline at https://localhost:8443/
 ![Numaflow UI](assets/numaflow-ui-advanced-pipeline.png)
 
-The source code of the `even-odd` [User Defined Function](user-guide/user-defined-functions/map/map.md) can be found [here](https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/evenodd). You also can replace the [Log](./user-guide/sinks/log.md) Sink with some other sinks like [Kafka](./user-guide/sinks/kafka.md) to forward the data to Kafka topics.
+The source code of the `even-odd` [User Defined Function](user-guide/user-defined-functions/map/map.md) can be found [here](https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/even_odd). You also can replace the [Log](./user-guide/sinks/log.md) Sink with some other sinks like [Kafka](./user-guide/sinks/kafka.md) to forward the data to Kafka topics.
 
 The pipeline can be deleted by
 
diff --git a/examples/6-simple-reduce-pipeline.yaml b/examples/6-simple-reduce-pipeline.yaml
index b4f54e3d..a536a8d4 100644
--- a/examples/6-simple-reduce-pipeline.yaml
+++ b/examples/6-simple-reduce-pipeline.yaml
@@ -12,13 +12,13 @@ spec:
         min: 1
       udf:
         container:
-          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/evenodd
-          image: quay.io/numaio/go-even-odd-example
+          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/even_odd
+          image: quay.io/numaio/numaflow-go/map-even-odd
     - name: compute-sum
       udf:
         container:
           # compute the sum
-          image: quay.io/numaio/go-integer-sum-example
+          image: quay.io/numaio/numaflow-go/reduce-sum
         groupBy:
           window:
             fixed:
diff --git a/examples/7-reduce-sliding-window.yaml b/examples/7-reduce-sliding-window.yaml
index 1d214bfa..3bbd8503 100644
--- a/examples/7-reduce-sliding-window.yaml
+++ b/examples/7-reduce-sliding-window.yaml
@@ -25,13 +25,13 @@ spec:
         max: 1
       udf:
         container:
-          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/evenodd
-          image: quay.io/numaio/go-even-odd-example
+          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/even_odd
+          image: quay.io/numaio/numaflow-go/map-even-odd
     - name: reduce-sliding
       udf:
         container:
           # compute the sum
-          image: quay.io/numaio/go-integer-sum-example
+          image: quay.io/numaio/numaflow-go/reduce-sum
         groupBy:
           window:
             sliding:
diff --git a/go.mod b/go.mod
index 8f4df94e..ec24c17e 100644
--- a/go.mod
+++ b/go.mod
@@ -25,7 +25,7 @@ require (
 	github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe
 	github.com/nats-io/nats-server/v2 v2.7.5-0.20220415000625-a6b62f61a703
 	github.com/nats-io/nats.go v1.21.0
-	github.com/numaproj/numaflow-go v0.2.4
+	github.com/numaproj/numaflow-go v0.3.0
 	github.com/prometheus/client_golang v1.12.1
 	github.com/prometheus/common v0.32.1
 	github.com/soheilhy/cmux v0.1.5
diff --git a/go.sum b/go.sum
index f4227086..4e040559 100644
--- a/go.sum
+++ b/go.sum
@@ -692,8 +692,8 @@ github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
-github.com/numaproj/numaflow-go v0.2.4 h1:a/O/NJe8sGmRt9UAwYDV2NPvdxnz7VUQ/69rjq/d4Rg=
-github.com/numaproj/numaflow-go v0.2.4/go.mod h1:TOawJdyf1C4V98zKnjjFhbHLBtg/TDyzZM+1MsfZuPo=
+github.com/numaproj/numaflow-go v0.3.0 h1:ry2C67eMcZ6MAgWn8BOPbRjAWIwMg5T5F7mdsvfJAYA=
+github.com/numaproj/numaflow-go v0.3.0/go.mod h1:TOawJdyf1C4V98zKnjjFhbHLBtg/TDyzZM+1MsfZuPo=
 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=


Test patch:
diff --git a/pkg/sinks/udsink/udsink_grpc_test.go b/pkg/sinks/udsink/udsink_grpc_test.go
index 3aeea1e3..f5a7903f 100644
--- a/pkg/sinks/udsink/udsink_grpc_test.go
+++ b/pkg/sinks/udsink/udsink_grpc_test.go
@@ -27,7 +27,6 @@ import (
 	"github.com/numaproj/numaflow-go/pkg/apis/proto/sink/v1/sinkmock"
 	"github.com/numaproj/numaflow-go/pkg/sink/clienttest"
 	"github.com/stretchr/testify/assert"
-	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -57,22 +56,6 @@ func TestGRPCBasedUDF_WaitUntilReadyWithMockClient(t *testing.T) {
 	assert.NoError(t, err)
 }
 
-type rpcMsg struct {
-	msg proto.Message
-}
-
-func (r *rpcMsg) Matches(msg interface{}) bool {
-	m, ok := msg.(proto.Message)
-	if !ok {
-		return false
-	}
-	return proto.Equal(m, r.msg)
-}
-
-func (r *rpcMsg) String() string {
-	return fmt.Sprintf("is %s", r.msg)
-}
-
 func TestGRPCBasedUDF_ApplyWithMockClient(t *testing.T) {
 	t.Run("test success", func(t *testing.T) {
 
@@ -106,14 +89,15 @@ func TestGRPCBasedUDF_ApplyWithMockClient(t *testing.T) {
 			},
 		}
 
-		mockClient := sinkmock.NewMockUserDefinedSinkClient(ctrl)
-
-		mockClient.EXPECT().SinkFn(gomock.Any(), &rpcMsg{msg: &sinkpb.DatumList{
-			Elements: testDatumList,
-		}}).Return(&sinkpb.ResponseList{
+		mockSinkClient := sinkmock.NewMockUserDefinedSink_SinkFnClient(ctrl)
+		mockSinkClient.EXPECT().Send(gomock.Any()).Return(nil).AnyTimes()
+		mockSinkClient.EXPECT().CloseAndRecv().Return(&sinkpb.ResponseList{
 			Responses: testResponseList,
 		}, nil)
 
+		mockClient := sinkmock.NewMockUserDefinedSinkClient(ctrl)
+		mockClient.EXPECT().SinkFn(gomock.Any(), gomock.Any()).Return(mockSinkClient, nil)
+
 		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 		defer cancel()
 		go func() {
@@ -150,13 +134,11 @@ func TestGRPCBasedUDF_ApplyWithMockClient(t *testing.T) {
 			},
 		}
 
-		mockClient := sinkmock.NewMockUserDefinedSinkClient(ctrl)
+		mockSinkErrClient := sinkmock.NewMockUserDefinedSink_SinkFnClient(ctrl)
+		mockSinkErrClient.EXPECT().Send(gomock.Any()).Return(nil).AnyTimes()
 
-		mockClient.EXPECT().SinkFn(gomock.Any(), &rpcMsg{msg: &sinkpb.DatumList{
-			Elements: testDatumList,
-		}}).Return(&sinkpb.ResponseList{
-			Responses: nil,
-		}, fmt.Errorf("mock error"))
+		mockClient := sinkmock.NewMockUserDefinedSinkClient(ctrl)
+		mockClient.EXPECT().SinkFn(gomock.Any(), gomock.Any()).Return(mockSinkErrClient, fmt.Errorf("mock SinkFn error"))
 
 		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 		defer cancel()
@@ -172,7 +154,7 @@ func TestGRPCBasedUDF_ApplyWithMockClient(t *testing.T) {
 		expectedErrList := []error{
 			ApplyUDSinkErr{
 				UserUDSinkErr: false,
-				Message:       "gRPC client.SinkFn failed, failed to execute c.grpcClt.SinkFn(): mock error",
+				Message:       "gRPC client.SinkFn failed, failed to execute c.grpcClt.SinkFn(): mock SinkFn error",
 				InternalErr: InternalErr{
 					Flag:        true,
 					MainCarDown: false,
@@ -180,7 +162,7 @@ func TestGRPCBasedUDF_ApplyWithMockClient(t *testing.T) {
 			},
 			ApplyUDSinkErr{
 				UserUDSinkErr: false,
-				Message:       "gRPC client.SinkFn failed, failed to execute c.grpcClt.SinkFn(): mock error",
+				Message:       "gRPC client.SinkFn failed, failed to execute c.grpcClt.SinkFn(): mock SinkFn error",
 				InternalErr: InternalErr{
 					Flag:        true,
 					MainCarDown: false,
diff --git a/test/e2e/testdata/even-odd.yaml b/test/e2e/testdata/even-odd.yaml
index 01b1a1e4..b06dc2e0 100644
--- a/test/e2e/testdata/even-odd.yaml
+++ b/test/e2e/testdata/even-odd.yaml
@@ -10,8 +10,8 @@ spec:
     - name: even-or-odd
       udf:
         container:
-          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/evenodd
-          image: quay.io/numaio/go-even-odd-example
+          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/even_odd
+          image: quay.io/numaio/numaflow-go/map-even-odd
     - name: even-sink
       sink:
         log: {}
diff --git a/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml b/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml
index ef957825..d40c5a95 100644
--- a/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml
+++ b/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml
@@ -16,13 +16,13 @@ spec:
         min: 1
       udf:
         container:
-          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/evenodd
-          image: quay.io/numaio/go-even-odd-example
+          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/even_odd
+          image: quay.io/numaio/numaflow-go/map-even-odd
     - name: keyed-fixed-sum
       udf:
         container:
           # compute the sum
-          image: quay.io/numaio/go-integer-sum-example
+          image: quay.io/numaio/numaflow-go/reduce-sum
         groupBy:
           window:
             fixed:
@@ -32,7 +32,7 @@ spec:
       udf:
         container:
           # compute the sum
-          image: quay.io/numaio/go-integer-sum-example
+          image: quay.io/numaio/numaflow-go/reduce-sum
         groupBy:
           window:
             fixed:
diff --git a/test/reduce-e2e/testdata/simple-keyed-reduce-pipeline.yaml b/test/reduce-e2e/testdata/simple-keyed-reduce-pipeline.yaml
index 5f1a293c..5b3032be 100644
--- a/test/reduce-e2e/testdata/simple-keyed-reduce-pipeline.yaml
+++ b/test/reduce-e2e/testdata/simple-keyed-reduce-pipeline.yaml
@@ -16,13 +16,13 @@ spec:
         min: 1
       udf:
         container:
-          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/evenodd
-          image: quay.io/numaio/go-even-odd-example
+          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/even_odd
+          image: quay.io/numaio/numaflow-go/map-even-odd
     - name: compute-sum
       udf:
         container:
           # compute the sum
-          image: quay.io/numaio/go-integer-sum-example
+          image: quay.io/numaio/numaflow-go/reduce-sum
         groupBy:
           window:
             fixed:
diff --git a/test/reduce-e2e/testdata/simple-non-keyed-reduce-pipeline.yaml b/test/reduce-e2e/testdata/simple-non-keyed-reduce-pipeline.yaml
index 47924830..c2762eb8 100644
--- a/test/reduce-e2e/testdata/simple-non-keyed-reduce-pipeline.yaml
+++ b/test/reduce-e2e/testdata/simple-non-keyed-reduce-pipeline.yaml
@@ -16,13 +16,13 @@ spec:
         min: 1
       udf:
         container:
-          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/evenodd
-          image: quay.io/numaio/go-even-odd-example
+          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/even_odd
+          image: quay.io/numaio/numaflow-go/map-even-odd
     - name: compute-sum
       udf:
         container:
           # compute the sum
-          image: quay.io/numaio/go-integer-sum-example
+          image: quay.io/numaio/numaflow-go/reduce-sum
         groupBy:
           window:
             fixed:
diff --git a/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml b/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml
index cfdf3b20..f6991086 100644
--- a/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml
+++ b/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml
@@ -14,13 +14,13 @@ spec:
         min: 1
       udf:
         container:
-          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/evenodd
-          image: quay.io/numaio/go-even-odd-example
+          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/even_odd
+          image: quay.io/numaio/numaflow-go/map-even-odd
     - name: compute-sum
       udf:
         container:
           # compute the sum
-          image: quay.io/numaio/go-integer-sum-example
+          image: quay.io/numaio/numaflow-go/reduce-sum
         groupBy:
           window:
             fixed:
diff --git a/test/sdks-e2e/README.md b/test/sdks-e2e/README.md
index 3141cef5..24451e9c 100644
--- a/test/sdks-e2e/README.md
+++ b/test/sdks-e2e/README.md
@@ -2,6 +2,6 @@
 
 Each SDK is supposed to implement:
 
-1. A `flatmap` UDF example, which splits the input message with `,`, and return a list. Build a docker image `quay.io/numaio/flatmap-example:${language}` and push to `quay.io`;
+1. A `flatmap` UDF example, which splits the input message with `,`, and return a list. Build a docker image `quay.io/numaio/numaflow-${language}/map-flatmap:latest` and push to `quay.io`;
    
-2. A `simplesink` UDSink example, which prints out original message in pod logs. Build a docker image `quay.io/numaio/simplesink-example:${language}` and push to `quay.io`.
+2. A `log` UDSink example, which prints out original message in pod logs. Build a docker image `quay.io/numaio/numaflow-${language}/sink-log:latest` and push to `quay.io`.
diff --git a/test/sdks-e2e/testdata/flatmap.yaml b/test/sdks-e2e/testdata/flatmap.yaml
index 416c2879..8b9be264 100644
--- a/test/sdks-e2e/testdata/flatmap.yaml
+++ b/test/sdks-e2e/testdata/flatmap.yaml
@@ -11,13 +11,13 @@ spec:
       udf:
         container:
           # Split input message into an array with comma, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/flatmap
-          image: quay.io/numaio/flatmap-example:go
+          image: quay.io/numaio/numaflow-go/map-flatmap:latest
     - name: go-udsink
       sink:
         udsink:
           container:
-            # https://github.com/numaproj/numaflow-go/tree/main/pkg/sink/examples/udsink_log
-            image: quay.io/numaio/simplesink-example:go
+            # https://github.com/numaproj/numaflow-go/tree/main/pkg/sink/examples/log
+            image: quay.io/numaio/numaflow-go/sink-log:latest
     - name: python-split
       udf:
         container:
@@ -25,7 +25,7 @@ spec:
             - python
             - example.py
           # Split input message into an array with comma, https://github.com/numaproj/numaflow-python/tree/main/examples/function/flatmap
-          image: quay.io/numaio/flatmap-example:python
+          image: quay.io/numaio/numaflow-python/map-flatmap:latest
     - name: python-udsink
       sink:
         udsink:
@@ -33,8 +33,8 @@ spec:
             args:
             - python
             - example.py
-            # https://github.com/numaproj/numaflow-python/tree/main/examples/sink/simplesink
-            image: quay.io/numaio/simplesink-example:python
+            # https://github.com/numaproj/numaflow-python/tree/main/examples/sink/log
+            image: quay.io/numaio/numaflow-python/sink-log:latest
   edges:
     - from: in
       to: go-split