Merge branch 'main' into aug-21-continue-1

docs/user-guide/reference/join-vertex.md

@@ -1,11 +1,13 @@
 # Joins and Cycles
 
 As of Numaflow v0.10, Pipeline Edges can be defined such that multiple Vertices send to a single vertex. This includes:
+
 - UDF Map Vertices
 - UDF Reduce Vertices
 - Sink Vertices
 
 Please see the following examples:
+
 - [Join on Map Vertex](https://github.com/numaproj/numaflow/blob/main/examples/11-join-on-map.yaml)
 - [Join on Reduce Vertex](https://github.com/numaproj/numaflow/blob/main/examples/11-join-on-reduce.yaml)
 - [Join on Sink Vertex](https://github.com/numaproj/numaflow/blob/main/examples/11-join-on-sink.yaml)
@@ -17,5 +19,6 @@ A special case of a "Join" is a **Cycle** (a Vertex which can send either to its
 Cycles are permitted, except in the case that there's a Reduce Vertex at or downstream of the cycle. (This is because a cycle inevitably produces late data, which would get dropped by the Reduce Vertex. For this reason, cycles should be used sparingly.)
 
 The following examples are of Cycles:
+
 - [Cycle to Self](https://github.com/numaproj/numaflow/blob/main/examples/10-cycle-to-self.yaml)
 - [Cycle to Previous](https://github.com/numaproj/numaflow/blob/main/examples/10-cycle-to-prev.yaml)

pkg/sinks/udsink/sink.go

@@ -108,9 +108,9 @@ func (s *UserDefinedSink) Write(ctx context.Context, messages []isb.Message) ([]
 	msgs := make([]*sinkpb.SinkRequest, len(messages))
 	for i, m := range messages {
 		msgs[i] = &sinkpb.SinkRequest{
-			// NOTE: key is not used anywhere ATM
 			Id:        m.ID,
 			Value:     m.Payload,
+			Keys:      m.Keys,
 			EventTime: timestamppb.New(m.EventTime),
 			// Watermark is only available in readmessage....
 			Watermark: timestamppb.New(time.Time{}), // TODO: insert the correct watermark