Repository: numaproj/numaflow (go)

Issue:
Shuffling algorithm might be biased and make a partition of the reduce edge idle.
Shuffling before the Vnth vertex creates a key to edge-buffer-index affinity, which will not change from Vn to Vn+1 Reduce vertices if there is no re-keying between these vertices.

Pull request description:
feat: use randomized shuffle using vertex name as the seed
fixes #600 

Move from `fnv` to `mumur3` for randomized keying when there is no rekeying using the vertex name as the seed.

Patch:
diff --git a/go.mod b/go.mod
index 8b2dcc1f..ce46d48e 100644
--- a/go.mod
+++ b/go.mod
@@ -31,6 +31,7 @@ require (
 	github.com/prometheus/client_golang v1.12.1
 	github.com/prometheus/common v0.32.1
 	github.com/soheilhy/cmux v0.1.5
+	github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72
 	github.com/spf13/cobra v1.2.1
 	github.com/spf13/viper v1.9.0
 	github.com/stretchr/testify v1.7.2
diff --git a/go.sum b/go.sum
index 3bc3a892..7f7a3afd 100644
--- a/go.sum
+++ b/go.sum
@@ -810,6 +810,7 @@ github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=
 github.com/soheilhy/cmux v0.1.5 h1:jjzc5WVemNEDTLwv9tlmemhC73tI08BNOIGwBOo10Js=
 github.com/soheilhy/cmux v0.1.5/go.mod h1:T7TcVDs9LWfQgPlPsdngu6I6QIoyIFZDDC6sNE1GqG0=
+github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72 h1:qLC7fQah7D6K1B0ujays3HV9gkFtllcxhzImRR7ArPQ=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
 github.com/spf13/afero v1.2.2/go.mod h1:9ZxEEn6pIJ8Rxe320qSDBk6AsU0r9pR7Q4OcevTdifk=
diff --git a/pkg/reduce/readloop/readloop.go b/pkg/reduce/readloop/readloop.go
index 893c6eef..b1c1d40b 100644
--- a/pkg/reduce/readloop/readloop.go
+++ b/pkg/reduce/readloop/readloop.go
@@ -181,6 +181,9 @@ messagesLoop:
 				metrics.LabelPipeline:           rl.pipelineName,
 				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(rl.vertexReplica)),
 				LabelReason:                     "late"}).Inc()
+
+			// mark it as a successfully written message as the message will be acked to avoid subsequent retries
+			writtenMessages = append(writtenMessages, message)
 			continue
 		}
 
diff --git a/pkg/shuffle/shuffle.go b/pkg/shuffle/shuffle.go
index 8d09f0b8..5b2183e2 100644
--- a/pkg/shuffle/shuffle.go
+++ b/pkg/shuffle/shuffle.go
@@ -18,25 +18,38 @@ package shuffle
 
 import (
 	"hash"
-	"hash/fnv"
 
 	"github.com/numaproj/numaflow/pkg/isb"
+
+	"github.com/spaolacci/murmur3"
 )
 
 // Shuffle shuffles messages among ISB
 type Shuffle struct {
+	vertexName        string
 	bufferIdentifiers []string
 	buffersCount      int
 	hash              hash.Hash64
 }
 
 // NewShuffle accepts list of buffer identifiers(unique identifier of isb)
-// and returns new shuffle instance
-func NewShuffle(bufferIdentifiers []string) *Shuffle {
+// and returns new shuffle instance. It uses vertex-name as seed, without a seed, we will end with the problem where
+// Shuffling before the Vnth vertex creates a key to edge-buffer-index affinity,
+// which will not change from Vn to Vn+1 Reduce vertices if there is no re-keying between these vertices causing
+// idle partitions.
+func NewShuffle(vertexName string, bufferIdentifiers []string) *Shuffle {
+	// We use vertex name as seed.
+	vertexHash := murmur3.New64()
+	_, _ = vertexHash.Write([]byte(vertexName))
+
 	return &Shuffle{
+		vertexName:        vertexName,
 		bufferIdentifiers: bufferIdentifiers,
 		buffersCount:      len(bufferIdentifiers),
-		hash:              fnv.New64(),
+		// we use murmur3, we are open for suggestions. fnv did not work for us because of lack of re-keying in
+		// some cases causing idle partitions in reduce edges. We need to revisit the below link
+		// https://softwareengineering.stackexchange.com/questions/49550/which-hashing-algorithm-is-best-for-uniqueness-and-speed
+		hash: murmur3.New64WithSeed(uint32(vertexHash.Sum64())),
 	}
 }
 
diff --git a/pkg/sources/source.go b/pkg/sources/source.go
index d3fce3ae..bbed6869 100644
--- a/pkg/sources/source.go
+++ b/pkg/sources/source.go
@@ -221,7 +221,7 @@ func (sp *SourceProcessor) getTransformerGoWhereDecider() forward.GoWhere {
 	shuffleFuncMap := make(map[string]*shuffle.Shuffle)
 	for _, edge := range sp.VertexInstance.Vertex.Spec.ToEdges {
 		if edge.Parallelism != nil && *edge.Parallelism > 1 {
-			s := shuffle.NewShuffle(dfv1.GenerateEdgeBufferNames(sp.VertexInstance.Vertex.Namespace, sp.VertexInstance.Vertex.Spec.PipelineName, edge))
+			s := shuffle.NewShuffle(sp.VertexInstance.Vertex.GetName(), dfv1.GenerateEdgeBufferNames(sp.VertexInstance.Vertex.Namespace, sp.VertexInstance.Vertex.Spec.PipelineName, edge))
 			shuffleFuncMap[fmt.Sprintf("%s:%s", edge.From, edge.To)] = s
 		}
 	}
diff --git a/pkg/udf/map_udf.go b/pkg/udf/map_udf.go
index f3803624..d9d45062 100644
--- a/pkg/udf/map_udf.go
+++ b/pkg/udf/map_udf.go
@@ -77,7 +77,7 @@ func (u *MapUDFProcessor) Start(ctx context.Context) error {
 	shuffleFuncMap := make(map[string]*shuffle.Shuffle)
 	for _, edge := range u.VertexInstance.Vertex.Spec.ToEdges {
 		if edge.Parallelism != nil && *edge.Parallelism > 1 {
-			s := shuffle.NewShuffle(dfv1.GenerateEdgeBufferNames(u.VertexInstance.Vertex.Namespace, u.VertexInstance.Vertex.Spec.PipelineName, edge))
+			s := shuffle.NewShuffle(u.VertexInstance.Vertex.GetName(), dfv1.GenerateEdgeBufferNames(u.VertexInstance.Vertex.Namespace, u.VertexInstance.Vertex.Spec.PipelineName, edge))
 			shuffleFuncMap[fmt.Sprintf("%s:%s", edge.From, edge.To)] = s
 		}
 	}
diff --git a/pkg/udf/reduce_udf.go b/pkg/udf/reduce_udf.go
index 8855c3f5..e2590837 100644
--- a/pkg/udf/reduce_udf.go
+++ b/pkg/udf/reduce_udf.go
@@ -106,7 +106,7 @@ func (u *ReduceUDFProcessor) Start(ctx context.Context) error {
 	shuffleFuncMap := make(map[string]*shuffle.Shuffle)
 	for _, edge := range u.VertexInstance.Vertex.Spec.ToEdges {
 		if edge.Parallelism != nil && *edge.Parallelism > 1 {
-			s := shuffle.NewShuffle(dfv1.GenerateEdgeBufferNames(u.VertexInstance.Vertex.Namespace, u.VertexInstance.Vertex.Spec.PipelineName, edge))
+			s := shuffle.NewShuffle(u.VertexInstance.Vertex.GetName(), dfv1.GenerateEdgeBufferNames(u.VertexInstance.Vertex.Namespace, u.VertexInstance.Vertex.Spec.PipelineName, edge))
 			shuffleFuncMap[fmt.Sprintf("%s:%s", edge.From, edge.To)] = s
 		}
 	}


Test patch:
diff --git a/pkg/shuffle/shuffle_test.go b/pkg/shuffle/shuffle_test.go
index e58f4031..0171a4e1 100644
--- a/pkg/shuffle/shuffle_test.go
+++ b/pkg/shuffle/shuffle_test.go
@@ -18,11 +18,13 @@ package shuffle
 
 import (
 	"fmt"
-	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/isb/testutils"
-	"github.com/stretchr/testify/assert"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/isb/testutils"
 )
 
 func TestShuffle_ShuffleMessages(t *testing.T) {
@@ -81,7 +83,7 @@ func TestShuffle_ShuffleMessages(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			// create shuffle with buffer id list
-			shuffler := NewShuffle(test.buffersIdentifiers)
+			shuffler := NewShuffle(test.name, test.buffersIdentifiers)
 
 			bufferIdMessageMap := shuffler.ShuffleMessages(test.messages)
 			sum := 0
diff --git a/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml b/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml
index d21d1f20..126d5f48 100644
--- a/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml
+++ b/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml
@@ -54,7 +54,8 @@ spec:
       to: atoi
     - from: atoi
       to: keyed-fixed-sum
-      parallelism: 2
+      # FIXME: use parallelism of 2 after we implement reduce watermark
+      parallelism: 1
     - from: keyed-fixed-sum
       to: non-keyed-fixed-sum
     - from: non-keyed-fixed-sum
diff --git a/test/reduce-e2e/testdata/complex-sliding-window-pipeline.yaml b/test/reduce-e2e/testdata/complex-sliding-window-pipeline.yaml
index 9dcf0a03..5932eea1 100644
--- a/test/reduce-e2e/testdata/complex-sliding-window-pipeline.yaml
+++ b/test/reduce-e2e/testdata/complex-sliding-window-pipeline.yaml
@@ -67,7 +67,8 @@ spec:
       to: atoi
     - from: atoi
       to: keyed-fixed-sum
-      parallelism: 2
+      # FIXME: use parallelism of 2 after we implement reduce watermark
+      parallelism: 1
     - from: keyed-fixed-sum
       to: non-keyed-fixed-sum
     - from: non-keyed-fixed-sum