revert: metadata for udsink (#46)

Signed-off-by: Hao Hao <[email protected]>
numaproj · Apr 10, 2023 · 2ecaa6c · 2ecaa6c
1 parent d1c512a
commit 2ecaa6c
Show file tree

Hide file tree

Showing 9 changed files with 58 additions and 198 deletions.
diff --git a/go.mod b/go.mod
@@ -23,3 +23,5 @@ require (
 	google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
+
+retract v0.4.1
diff --git a/pkg/apis/proto/sink/v1/udsink.pb.go b/pkg/apis/proto/sink/v1/udsink.pb.go
diff --git a/pkg/apis/proto/sink/v1/udsink.proto b/pkg/apis/proto/sink/v1/udsink.proto
@@ -34,15 +34,7 @@ message Datum {
   bytes value = 2;
   EventTime event_time = 3;
   Watermark watermark = 4;
-  Metadata metadata = 5;
-}
-
-/**
- * Metadata of a datum element.
- */
-message Metadata {
-  string id = 1;
-  uint64 num_delivered = 2;
+  string id = 5;
 }
 
 /**

diff --git a/pkg/sink/clienttest/client_test.go b/pkg/sink/clienttest/client_test.go
@@ -86,22 +86,16 @@ func TestSinkFn(t *testing.T) {
 	mockClient := sinkmock.NewMockUserDefinedSinkClient(ctrl)
 	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{})},
-			Metadata: &sinkpb.Metadata{
-				Id:           "test_id_0",
-				NumDelivered: 1,
-			},
 		},
 		{
+			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{})},
-			Metadata: &sinkpb.Metadata{
-				Id:           "test_id_1",
-				NumDelivered: 1,
-			},
 		},
 	}
 	testResponseList := []*sinkpb.Response{

diff --git a/pkg/sink/examples/log/main.go b/pkg/sink/examples/log/main.go
@@ -14,7 +14,7 @@ func handle(_ context.Context, datumStreamCh <-chan sinksdk.Datum) sinksdk.Respo
 		_ = d.EventTime()
 		_ = d.Watermark()
 		fmt.Println("User Defined Sink:", string(d.Value()))
-		id := d.Metadata().ID()
+		id := d.ID()
 		result = result.Append(sinksdk.ResponseOK(id))
 	}
 	return result

diff --git a/pkg/sink/interface.go b/pkg/sink/interface.go
@@ -13,15 +13,7 @@ type Datum interface {
 	Value() []byte
 	EventTime() time.Time
 	Watermark() time.Time
-	Metadata() DatumMetadata
-}
-
-// DatumMetadata contains metadata of a datum.
-type DatumMetadata interface {
-	// ID returns the ID of the datum.
 	ID() string
-	// NumDelivered returns the number of times the datum has been delivered.
-	NumDelivered() uint64
 }
 
 // Client contains methods to call a gRPC client.

diff --git a/pkg/sink/server/server_test.go b/pkg/sink/server/server_test.go
@@ -35,7 +35,7 @@ func Test_server_sink(t *testing.T) {
 				sinkHandler: sinksdk.SinkFunc(func(ctx context.Context, datumStreamCh <-chan sinksdk.Datum) sinksdk.Responses {
 					result := sinksdk.ResponsesBuilder()
 					for d := range datumStreamCh {
-						id := d.Metadata().ID()
+						id := d.ID()
 						if strings.Contains(string(d.Value()), "err") {
 							result = result.Append(sinksdk.ResponseFailure(id, "mock sink message error"))
 						} else {
@@ -63,22 +63,16 @@ func Test_server_sink(t *testing.T) {
 			}()
 			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{})},
-					Metadata: &sinkpb.Metadata{
-						Id:           "test_id_0",
-						NumDelivered: 1,
-					},
 				},
 				{
+					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{})},
-					Metadata: &sinkpb.Metadata{
-						Id:           "test_id_1",
-						NumDelivered: 1,
-					},
 				},
 			}
 			responseList, err := c.SinkFn(ctx, testDatumList)