pass eot request

Signed-off-by: Yashash H L <[email protected]>
numaproj · Sep 26, 2024 · 516ac07 · 516ac07
1 parent a799310
commit 516ac07
Showing 1 changed file with 11 additions and 0 deletions.
diff --git a/pkg/sdkclient/sinker/client.go b/pkg/sdkclient/sinker/client.go
@@ -154,6 +154,17 @@ func (c *client) SinkFn(ctx context.Context, requests []*sinkpb.SinkRequest) ([]
 		}
 	}
 
+	// send eot request
+	eotRequest := &sinkpb.SinkRequest{
+		Status: &sinkpb.SinkRequest_Status{
+			Eot: true,
+		},
+	}
+
+	if err := c.sinkStream.Send(eotRequest); err != nil {
+		return nil, fmt.Errorf("failed to send eot request: %v", err)
+	}
+
 	// Wait for the corresponding responses
 	var responses []*sinkpb.SinkResponse
 	for i := 0; i < len(requests); i++ {