feat: udsink gRPC stream

[jy4096]

numaproj/numaflow#335
gRPC development reference: #4 (comment)

Test

Client

Use numaflow-go sdk client code as the client

package server

import (
	"context"
	"testing"
	"time"

	sinkpb "github.com/numaproj/numaflow-go/pkg/apis/proto/sink/v1"
	"github.com/numaproj/numaflow-go/pkg/sink/client"
	"github.com/stretchr/testify/assert"
	"google.golang.org/protobuf/types/known/timestamppb"
)

func Test_server_sink(t *testing.T) {
	var ctx = context.Background()
	c, err := client.New(client.WithSockAddr("/tmp/numaflow-test.sock"))
	assert.NoError(t, err)
	defer func() {
		err = c.CloseConn(ctx)
		assert.NoError(t, err)
	}()
	testDatumList := []*sinkpb.Datum{
		{
			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_success`),
			EventTime: &sinkpb.EventTime{EventTime: timestamppb.New(time.Unix(1661169600, 0))},
			Watermark: &sinkpb.Watermark{Watermark: timestamppb.New(time.Time{})},
		},
	}
	responseList, err := c.SinkFn(ctx, testDatumList)
	assert.NoError(t, err)
	expectedResponseList := []*sinkpb.Response{
		{
			Id:      "test_id_0",
			Success: true,
			ErrMsg:  "",
		},
		{
			Id:      "test_id_1",
			Success: true,
			ErrMsg:  "",
		},
	}
	assert.Equal(t, expectedResponseList, responseList)
}

Server

Use numaflow-python sdk sink example log code as the server

from typing import List
from pynumaflow.sink import Datum, Responses, Response, UserDefinedSinkServicer


def udsink_handler(datums: List[Datum]) -> Responses:
    responses = Responses()
    for msg in datums:
        print("User Defined Sink", msg.id, msg.value.decode("utf-8"))
        responses.append(Response.as_success(msg.id))
    return responses


if __name__ == "__main__":
    grpc_server = UserDefinedSinkServicer(udsink_handler, sock_path="/tmp/numaflow-test.sock")
    grpc_server.start()

Result

client test output

=== RUN   Test_server_sink
=== RUN   Test_server_sink/server_sink
--- PASS: Test_server_sink (0.00s)
    --- PASS: Test_server_sink/server_sink (0.00s)
PASS

Process finished with the exit code 0

server udsink output

GRPC Server listening on: unix:///tmp/numaflow-test.sock with max threads: 64
User Defined Sink test_id_0 sink_message_success
User Defined Sink test_id_1 sink_message_success

[codecov]

Codecov Report

Merging #34 (f11a3c4) into main (ad55710) will not change coverage.
The diff coverage is 100.00%.

@@           Coverage Diff           @@
##             main      #34   +/-   ##
=======================================
  Coverage   98.38%   98.38%           
=======================================
  Files           9        9           
  Lines         247      247           
  Branches       14       14           
=======================================
  Hits          243      243           
  Misses          2        2           
  Partials        2        2           

Impacted Files	Coverage Δ
pynumaflow/sink/server.py	95.12% <100.00%> (ø)

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[ab93]

Let's fix the lint/format issues.
Also, this change applies to only udsinks right?

[ab93]

@chromevoid to fix the lint issue, you might have to change the .flake8 file, and make it like this: https://github.com/numaproj/numalogic/blob/main/.flake8

[jy4096]

Let's fix the lint/format issues. Also, this change applies to only udsinks right?

Yes only udsink is updated from unary rpc to streaming rpc.

[ab93]

@vigith I think it might be a better idea to fix the examples once v0.2.7 is released. Otherwise the example codes will fail for the time being. What do you think?