Repository: numaproj/numaflow (go)

Issue:
Add documentations for source data transformer(mapT)
Example: https://github.com/numaproj/numaflow/commit/3c62120725aa940831b901881a2d17192bcb0ee8

Also add built-in mapT function examples.

---
<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->
**Message from the maintainers**:

If you wish to see this enhancement implemented please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.


Pull request:
docs: add docs for source data transformer
Closes #506 

<!--

Before you push your changes:

* Run `make pre-push -B` to fix codegen and lint problems.

Then, you MUST:

* Sign-off your commit (otherwise the DCO check will fail).
* Use [a conventional commit message](https://www.conventionalcommits.org/en/v1.0.0/) (otherwise the commit message check will fail).

If you did not do this, reset all your commit and replace them with a single commit:

```
git reset HEAD~1 ;# change 1 to how many commits you made
git commit --signoff -m 'feat: my feat. Fixes #1234'
```

When creating your PR: 

* Make sure that "Fixes #" or "Closes #" is in both the PR title (for release notes) and description (to automatically link and close the issue).
* Say how you tested your changes. If you changed the UI, attach screenshots.
* Set your PR as a draft initially.
* Your PR needs to pass the required checks before it can be approved. 
* Once required tests have passed, mark your PR "Ready for review".

If changes were requested, once you've made them, you MUST dismiss the review to get it reviewed again.

-->


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2fc88955..49137775 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,41 @@
 # Changelog
 
+## v0.7.2 (2023-03-13)
+
+ * [1196a24](https://github.com/numaproj/numaflow/commit/1196a24b7096273b60c072e803a336c79eef2c5b) Update manifests to v0.7.2
+ * [a6f64f8](https://github.com/numaproj/numaflow/commit/a6f64f8d93f20bfa91b592fd00333068633bc2c2) feat: use randomized shuffle using vertex name as the seed (#601)
+ * [16b20a3](https://github.com/numaproj/numaflow/commit/16b20a357ade0c0bae9edb0461ee01340025eb7f) fix: ack the dropped messages as well (#603)
+ * [4cdef17](https://github.com/numaproj/numaflow/commit/4cdef17494cf72a8d1081945c5bd23c68ffe70cd) feat: enable controller HA (#602)
+ * [d2e0513](https://github.com/numaproj/numaflow/commit/d2e0513c5bb3358649da42b1babff4ee46c9e8a6) feat: expose dnspolicy and dnsconfig to pod template (#598)
+ * [6002c82](https://github.com/numaproj/numaflow/commit/6002c829f4600ef67d72fdda0e4693364dccfc07) Chore: tickgen changes to test reduce pipelines (#587)
+ * [b1aee94](https://github.com/numaproj/numaflow/commit/b1aee94533594cbee6470257d21a3d13c840ce70) feat: bidirectional streaming (#553)
+ * [67fc688](https://github.com/numaproj/numaflow/commit/67fc688ab8be984d48182d3a0f4b9b9b99a8c2c8) feat: use customized binary serde for nats message payload (#585)
+ * [cc10419](https://github.com/numaproj/numaflow/commit/cc1041995d13ed899b3fce8fca955541cebf4588) fix: Idle watermark fix for read batch size > 0 and partial idle outgoing edges (#575)
+ * [df1574d](https://github.com/numaproj/numaflow/commit/df1574dad468cbc68eadb75b7a61c930d6f31578) feat: implement watermark propagation for source data transformer (#557)
+ * [45d5c39](https://github.com/numaproj/numaflow/commit/45d5c396d2f369f0a9724704d91a7e07aa64895f) feat: namespace search (#559)
+ * [b14d470](https://github.com/numaproj/numaflow/commit/b14d470fa2b8bcd746f7e75b4ec20453297d53e6) fix: refine log for buffer validation. Fixes #185 (#573)
+ * [a8e8bb1](https://github.com/numaproj/numaflow/commit/a8e8bb1580dcd80e2a3625a2dc6c0a7c95a823ef) feat: add readiness and liveness check for daemon server. Fixes #543 (#571)
+ * [fd6acb6](https://github.com/numaproj/numaflow/commit/fd6acb6da5b63c743225df3ea3743935f211aaba) feat: marshal/unmarshal binary for read message (#565)
+ * [d1032b4](https://github.com/numaproj/numaflow/commit/d1032b4ce08a509223b5a7dbce4570b54d9e90a5) chore(deps): bump golang.org/x/net from 0.0.0-20220722155237-a158d28d115b to 0.7.0 (#568)
+ * [fd00ebd](https://github.com/numaproj/numaflow/commit/fd00ebdf938933bc2b735e03a7df53bedf1f48d7) chore(deps): bump golang.org/x/text from 0.3.7 to 0.3.8 (#567)
+ * [05ec77f](https://github.com/numaproj/numaflow/commit/05ec77f5d839dd13d6c1ef1cded94178f01d45e6) feat: expose image pull policy to user defined containers (#563)
+ * [ecbe3a0](https://github.com/numaproj/numaflow/commit/ecbe3a0061e02b05077f8a29649b692802a986a5) fix: typos in reduce examples (#556)
+ * [0dc3f5c](https://github.com/numaproj/numaflow/commit/0dc3f5c6d6e99c96937c5ee3a46f3f7e2723363f) feat: edge-watermark (#537)
+ * [77298c8](https://github.com/numaproj/numaflow/commit/77298c853e0184f554bf22b13858cba0e35ed922) feat: enable envFrom for user defined containers (#554)
+
+### Contributors
+
+ * Derek Wang
+ * Dillen Padhiar
+ * Juanlu Yu
+ * Julie Vogelman
+ * Keran Yang
+ * Vedant Gupta
+ * Vigith Maurice
+ * Yashash H L
+ * ashwinidulams
+ * dependabot[bot]
+
 ## v0.7.1 (2023-02-14)
 
  * [92925c1](https://github.com/numaproj/numaflow/commit/92925c15485a802600c5cb54cf603f2c1ceae027) Update manifests to v0.7.1
diff --git a/docs/specifications/edges-and-buffers.md b/docs/specifications/edges-and-buffers.md
index 9a9033d9..b9e2bb08 100644
--- a/docs/specifications/edges-and-buffers.md
+++ b/docs/specifications/edges-and-buffers.md
@@ -48,7 +48,7 @@ spec:
 
 `Buffer` is a concept different from `edge` but has lots of connections. Usually, each `edge` has one or more corresponding buffers, depending on the to vertex type (map or reduce). If the to vertex is a reduce and the edge parallelism > 1, there will be multiple buffers defined for that edge.
 
-`Buffers` are not only defined for `edges`, but also for `Source` and `Sink` vertices - each `Source` and `Sink` vertex has a coresponding `buffer`.
+`Buffers` are not only defined for `edges`, but also for `Source` and `Sink` vertices - each `Source` and `Sink` vertex has a corresponding `buffer`.
 
 In summary, there are 3 types of `buffers` in a pipeline:
 
@@ -56,6 +56,6 @@ In summary, there are 3 types of `buffers` in a pipeline:
 - Source Buffer
 - Sink Buffer
 
-Each buffer has a name, the naming convertion for different type of buffers can be found in the [source code](https://github.com/numaproj/numaflow/blob/main/pkg/apis/numaflow/v1alpha1/vertex_types.go).
+Each buffer has a name, the naming conversions for different type of buffers can be found in the [source code](https://github.com/numaproj/numaflow/blob/main/pkg/apis/numaflow/v1alpha1/vertex_types.go).
 
 `Buffer` is only used internally, it's transparent to the users. Each Inter-Step Buffer implementation should have something physical to map to the `buffers`. For example, In JetStream Inter-Step ISB implementation, a K/V bucket will be created for a `Source Buffer` or a `Sink Buffer`, and a K/V bucket plus a Stream will be created for a `Edge Buffer`. These buffer management operations are done by K8s jobs spawned by the controllers during pipeline creation and deletion.
diff --git a/docs/user-guide/inter-step-buffer-service.md b/docs/user-guide/inter-step-buffer-service.md
index 526c408e..829094c3 100644
--- a/docs/user-guide/inter-step-buffer-service.md
+++ b/docs/user-guide/inter-step-buffer-service.md
@@ -2,7 +2,7 @@
 
 Inter-Step Buffer Service is the service to provide [Inter-Step Buffers](./inter-step-buffer.md).
 
-An Inter-Step Buffer Service is describe by a [Custom Resource](https://kubernetes.io/docs/concepts/extend-kubernetes/api-extension/custom-resources/), it is required to be existing in a namespace before Pipeline objects are created. A sample `InterStepBufferService` with JetStream implementation looks like below.
+An Inter-Step Buffer Service is described by a [Custom Resource](https://kubernetes.io/docs/concepts/extend-kubernetes/api-extension/custom-resources/), it is required to be existing in a namespace before Pipeline objects are created. A sample `InterStepBufferService` with JetStream implementation looks like below.
 
 ```yaml
 apiVersion: numaflow.numaproj.io/v1alpha1
diff --git a/docs/user-guide/sources/transformer/builtin-transformers/README.md b/docs/user-guide/sources/transformer/builtin-transformers/README.md
new file mode 100644
index 00000000..ff4ec8ff
--- /dev/null
+++ b/docs/user-guide/sources/transformer/builtin-transformers/README.md
@@ -0,0 +1,42 @@
+# Built-in Functions
+
+Numaflow provides some built-in source data transformers that can be used directly.
+
+**Filter**
+
+A `filter` built-in transformer filters the message based on expression. `payload` keyword represents message object.
+see documentation for filter expression [here](filter.md#expression)
+
+```yaml
+spec:
+  vertices:
+    - name: in
+      source:
+        http: {}
+        transformer:
+          builtin:
+            name: filter
+            kwargs:
+              expression: int(json(payload).id) < 100
+```
+
+**Event Time Extractor**
+
+A `eventTimeExtractor` built-in transformer extracts event time from the payload of the message, based on expression and user-specified format. `payload` keyword represents message object.
+see documentation for event time extractor expression [here](event-time-extractor.md#expression).
+
+If you want to handle event times in epoch format, you can find helpful resource [here](event-time-extractor.md#epoch-format).
+
+```yaml
+spec:
+  vertices:
+    - name: in
+      source:
+        http: {}
+        transformer:
+          builtin:
+            name: eventTimeExtractor
+            kwargs:
+              expression: json(payload).item[0].time
+              format: 2006-01-02T15:04:05Z07:00
+```
diff --git a/docs/user-guide/sources/transformer/builtin-transformers/event-time-extractor.md b/docs/user-guide/sources/transformer/builtin-transformers/event-time-extractor.md
new file mode 100644
index 00000000..8e0fad96
--- /dev/null
+++ b/docs/user-guide/sources/transformer/builtin-transformers/event-time-extractor.md
@@ -0,0 +1,71 @@
+# Event Time Extractor
+
+A `eventTimeExtractor` built-in transformer extracts event time from the payload of the message, based on a user-provided `expression` and an optional `format` specification.
+
+`expression` is used to compile the payload to a string representation of the event time.
+
+`format` is used to convert the event time in string format to a `time.Time` object.
+
+## Expression (required)
+
+Event Time Extractor expression is implemented with `expr` and `sprig` libraries.
+
+### Data conversion functions
+
+These function can be accessed directly in expression. `payload` keyword represents the message object. It will be the root element to represent the message object in expression.
+
+- `json` - Convert payload in JSON object. e.g: `json(payload)`
+- `int` - Convert element/payload into `int` value. e.g: `int(json(payload).id)`
+- `string` - Convert element/payload into `string` value. e.g: `string(json(payload).amount)`
+
+### Sprig functions
+
+`Sprig` library has 70+ functions. `sprig` prefix need to be added to access the sprig functions.
+
+[sprig functions](http://masterminds.github.io/sprig/)
+
+E.g:
+
+- `sprig.trim(string(json(payload).timestamp))` # Remove spaces from either side of the value of field `timestamp`
+
+## Format (optional)
+
+Depending on whether a `format` is specified, Event Time Extractor uses different approaches to convert the event time string to a `time.Time` object.
+
+### When specified
+When `format` is specified, the native [time.Parse(layout, value string)](https://pkg.go.dev/time#Parse) library is used to make the conversion. In this process, the `format` parameter is passed as the layout input to the time.Parse() function, while the event time string is passed as the value parameter.
+
+### When not specified
+When `format` is not specified, the extractor uses [dateparse](https://github.com/araddon/dateparse) to parse the event time string without knowing the format in advance.
+
+### How to specify format
+Please refer to [golang format library](https://cs.opensource.google/go/go/+/refs/tags/go1.19.5:src/time/format.go).
+
+### Error Scenarios
+When encountering parsing errors, event time extractor skips the extraction and passes on the message without modifying the original input message event time. Errors can occur for a variety of reasons, including:
+
+1. `format` is specified but the event time string can't parse to the specified format.
+1. `format` is not specified but dataparse can't convert the event time string to a `time.Time` object.
+
+### Ambiguous event time strings
+Event time strings can be ambiguous when it comes to date format, such as MM/DD/YYYY versus DD/MM/YYYY. When using such format, you're required to explicitly specify `format`, to avoid confusion.
+If no format is provided, event time extractor treats ambiguous event time strings as an error scenario.
+
+### Epoch format
+If the event time string in your message payload is in epoch format, you can skip specifying a `format`. You can rely on `dateparse` to recognize a wide range of epoch timestamp formats, including Unix seconds, milliseconds, microseconds, and nanoseconds.
+
+## Event Time Extractor Spec
+
+```yaml
+spec:
+  vertices:
+    - name: in
+      source:
+        http: {}
+        transformer:
+          builtin:
+            name: eventTimeExtractor
+            kwargs:
+              expression: sprig.trim(string(json(payload).timestamp))
+              format: 2006-01-02T15:04:05Z07:00
+```
\ No newline at end of file
diff --git a/docs/user-guide/sources/transformer/builtin-transformers/filter.md b/docs/user-guide/sources/transformer/builtin-transformers/filter.md
new file mode 100644
index 00000000..67e4da8a
--- /dev/null
+++ b/docs/user-guide/sources/transformer/builtin-transformers/filter.md
@@ -0,0 +1,46 @@
+# Filter
+
+A `filter` is a special-purpose built-in function. It is used to evaluate on each message in a pipeline and
+is often used to filter the number of messages that are passed to next vertices.
+
+Filter function supports comprehensive expression language which extends flexibility write complex expressions.
+
+`payload` will be root element to represent the message object in expression.
+
+## Expression
+
+Filter expression implemented with `expr` and `sprig` libraries.
+
+### Data conversion functions
+
+These function can be accessed directly in expression.
+
+- `json` - Convert payload in JSON object. e.g: `json(payload)`
+- `int` - Convert element/payload into `int` value. e.g: `int(json(payload).id)`
+- `string` - Convert element/payload into `string` value. e.g: `string(json(payload).amount)`
+
+### Sprig functions
+
+`Sprig` library has 70+ functions. `sprig` prefix need to be added to access the sprig functions.
+
+[sprig functions](http://masterminds.github.io/sprig/)
+
+E.g:
+
+- `sprig.contains('James', json(payload).name)` # `James` is contained in the value of `name`.
+- `int(json(sprig.b64dec(payload)).id) < 100`
+
+## Filter Spec
+
+```yaml
+spec:
+  vertices:
+    - name: in
+      source:
+        http: {}
+        transformer:
+          builtin:
+            name: filter
+            kwargs:
+              expression: int(json(payload).id) < 100
+```
diff --git a/docs/user-guide/sources/transformer/overview.md b/docs/user-guide/sources/transformer/overview.md
new file mode 100644
index 00000000..11b6c27a
--- /dev/null
+++ b/docs/user-guide/sources/transformer/overview.md
@@ -0,0 +1,104 @@
+# Source Data Transformer
+
+The Source Data Transformer is a feature that allows users to execute custom code to transform their data at source.
+
+This functionality offers two primary advantages to users:
+
+1. Event Time Assignment - It enables users to extract the event time from the message payload, providing a more precise and accurate event time than the default mechanisms like LOG_APPEND_TIME of Kafka for Kafka source, custom HTTP header for HTTP source, and others.
+1. Early Data Filtering - It filters out unwanted data at source vertex, saving the cost of creating the filtering UDF vertex and the inter-step buffer between source and the filtering UDF.
+
+Source Data Transformer runs as a sidecar container in a Source Vertex Pod. Data processing in the transformer is supposed to be idempotent.
+The communication between the main container (platform code) and the sidecar container (user code) is through gRPC over Unix Domain Socket.
+
+## Built-in Transformers
+
+There are some [Built-in Transformers](builtin-transformers/README.md) that can be used directly.
+
+## Build Your Own Transformer
+
+You can build your own transformer in multiple languages. A User Defined Transformer could be as simple as the example below in Golang. 
+In the example, the transformer extracts event times from `timestamp` of the JSON payload and assigns them to messages as new event times. It also filters out unwanted messages based on `filterOut` of the payload.
+
+```golang
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	functionsdk "github.com/numaproj/numaflow-go/pkg/function"
+	"github.com/numaproj/numaflow-go/pkg/function/server"
+)
+
+func Handle(_ context.Context, key string, data functionsdk.Datum) functionsdk.MessageTs {
+	/*
+		Input messages are in JSON format. Sample: {"timestamp": "1673239888", "filterOut": "true"}.
+		Field "timestamp" shows the real event time of the message, in format of epoch.
+		Field "filterOut" indicates whether the message should be filtered out, in format of boolean.
+	*/
+	var jsonObject map[string]interface{}
+	json.Unmarshal(data.Value(), &jsonObject)
+
+	// event time assignment
+	eventTime := data.EventTime()
+	// if timestamp field exists, extract event time from payload.
+	if ts, ok := jsonObject["timestamp"]; ok {
+		eventTime = time.Unix(int64(ts.(float64)), 0)
+	}
+	
+	// data filtering
+	var shouldFilter bool
+	if f, ok := jsonObject["filterOut"]; ok {
+		shouldFilter = f.(bool)
+	}
+	if shouldFilter {
+		return functionsdk.MessageTsBuilder().Append(functionsdk.MessageTToDrop())
+	} else {
+		return functionsdk.MessageTsBuilder().Append(functionsdk.MessageTTo(eventTime, key, data.Value()))
+	}
+}
+
+func main() {
+	server.New().RegisterMapperT(functionsdk.MapTFunc(Handle)).Start(context.Background())
+}
+```
+
+Check the links below to see another transformer example in various programming languages, where we apply conditional forwarding based on the input event time.
+
+- [Python](https://github.com/numaproj/numaflow-python/tree/main/examples/function/event_time_filter)
+- [Golang](https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/event_time_filter)
+- [Java](https://github.com/numaproj/numaflow-java/tree/main/examples/src/main/java/io/numaproj/numaflow/examples/function/map/eventtimefilter)
+
+After building a docker image for the written transformer, specify the image as below in the source vertex spec.
+
+```yaml
+spec:
+  vertices:
+    - name: my-vertex
+      source:
+        http: {}
+        transformer:
+          container:
+            image: my-python-transformer-example:latest
+```
+
+### Available Environment Variables
+
+Some environment variables are available in the source vertex Pods, they might be useful in you own source data transformer implementation.
+
+- `NUMAFLOW_NAMESPACE` - Namespace.
+- `NUMAFLOW_POD` - Pod name.
+- `NUMAFLOW_REPLICA` - Replica index.
+- `NUMAFLOW_PIPELINE_NAME` - Name of the pipeline.
+- `NUMAFLOW_VERTEX_NAME` - Name of the vertex.
+
+### Configuration
+
+Configuration data can be provided to the transformer container at runtime multiple ways.
+
+* [`environment variables`](../../environment-variables.md)
+* `args`
+* `command`
+* [`volumes`](../../volumes.md)
+* [`init containers`](../../init-containers.md)
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/mkdocs.yml b/mkdocs.yml
index 201bf384..c134c4bd 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -49,6 +49,12 @@ nav:
           - user-guide/sources/http.md
           - user-guide/sources/kafka.md
           - user-guide/sources/nats.md
+          - Data Transformer:
+              - Overview: "user-guide/sources/transformer/overview.md"
+              - Built-in Transformers:
+                  - Overview: "user-guide/sources/transformer/builtin-transformers/README.md"
+                  - Filter: "user-guide/sources/transformer/builtin-transformers/filter.md"
+                  - Event Time Extractor: "user-guide/sources/transformer/builtin-transformers/event-time-extractor.md"
       - Sinks:
           - user-guide/sinks/kafka.md
           - user-guide/sinks/log.md
diff --git a/pkg/shared/expr/eval_string.go b/pkg/shared/expr/eval_string.go
index be581e82..78a45aa8 100644
--- a/pkg/shared/expr/eval_string.go
+++ b/pkg/shared/expr/eval_string.go
@@ -23,7 +23,7 @@ import (
 )
 
 // EvalStr uses the given input expression to evaluate input message and compile it to a string.
-// See examples in compile_test.go
+// See examples in eval_string_test.go
 func EvalStr(expression string, msg []byte) (string, error) {
 	msgMap := map[string]interface{}{
 		root: string(msg),
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