Repository: numaproj/numaflow
Pull request #814 (2023-06-29 05:36:13)

 fix(docs): fixed some incorrect docs and renamed a timeExtractionFilter arg
Explain what this PR does.

<!--

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/docs/specifications/edges-buffers-buckets.md b/docs/specifications/edges-buffers-buckets.md
index ba33aa3a..ac87013b 100644
--- a/docs/specifications/edges-buffers-buckets.md
+++ b/docs/specifications/edges-buffers-buckets.md
@@ -27,6 +27,7 @@ spec:
         container:
           image: quay.io/numaio/numaflow-go/map-even-odd
     - name: compute-sum
+      partitions: 2
       udf:
         container:
           image: quay.io/numaio/numaflow-go/reduce-sum
@@ -45,7 +46,6 @@ spec:
       to: atoi
     - from: atoi
       to: compute-sum
-      parallelism: 2
     - from: compute-sum
       to: out
 ```
diff --git a/docs/user-guide/sources/transformer/builtin-transformers/README.md b/docs/user-guide/sources/transformer/builtin-transformers/README.md
index ff4ec8ff..a3174f63 100644
--- a/docs/user-guide/sources/transformer/builtin-transformers/README.md
+++ b/docs/user-guide/sources/transformer/builtin-transformers/README.md
@@ -40,3 +40,22 @@ spec:
               expression: json(payload).item[0].time
               format: 2006-01-02T15:04:05Z07:00
 ```
+
+**Time Extraction Filter**
+
+A `timeExtractionFilter` implements both the `eventTimeExtractor` and `filter` built-in functions. It evaluates a message on a pipeline and if valid, extracts event time from the payload of the messsage.
+
+```yaml
+spec:
+  vertices:
+    - name: in
+      source:
+        http: {}
+        transformer:
+          builtin:
+            name: timeExtractionFilter
+            kwargs:
+              filterExpr: int(json(payload).id) < 100
+              eventTimeExpr: json(payload).item[1].time
+              eventTimeFormat: 2006-01-02T15:04:05Z07:00
+```
diff --git a/docs/user-guide/sources/transformer/builtin-transformers/time-extraction-filter.md b/docs/user-guide/sources/transformer/builtin-transformers/time-extraction-filter.md
index 9a034284..e6a6a027 100644
--- a/docs/user-guide/sources/transformer/builtin-transformers/time-extraction-filter.md
+++ b/docs/user-guide/sources/transformer/builtin-transformers/time-extraction-filter.md
@@ -48,5 +48,5 @@ spec:
             kwargs:
               filterExpr: int(json(payload).id) < 100
               eventTimeExpr: json(payload).item[1].time
-              format: 2006-01-02T15:04:05Z07:00
-```
\ No newline at end of file
+              eventTimeFormat: 2006-01-02T15:04:05Z07:00
+```
diff --git a/docs/user-guide/sources/transformer/overview.md b/docs/user-guide/sources/transformer/overview.md
index d5db6224..c150c59d 100644
--- a/docs/user-guide/sources/transformer/overview.md
+++ b/docs/user-guide/sources/transformer/overview.md
@@ -5,7 +5,7 @@ The Source Data Transformer is a feature that allows users to execute custom cod
 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.
+2. Early data processing - It pre-processes the data, or filters out unwanted data at source vertex, saving the cost of creating another UDF vertex and an inter-step buffer.
 
 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.
@@ -16,7 +16,7 @@ There are some [Built-in Transformers](builtin-transformers/README.md) that can
 
 ## 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. 
+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
@@ -46,7 +46,7 @@ func Handle(_ context.Context, keys []string, data functionsdk.Datum) functionsd
 	if ts, ok := jsonObject["timestamp"]; ok {
 		eventTime = time.Unix(int64(ts.(float64)), 0)
 	}
-	
+
 	// data filtering
 	var filterOut bool
 	if f, ok := jsonObject["filterOut"]; ok {
@@ -97,8 +97,8 @@ Some environment variables are available in the source vertex Pods, they might b
 
 Configuration data can be provided to the transformer container at runtime multiple ways.
 
-* [`environment variables`](../../reference/configuration/environment-variables.md)
-* `args`
-* `command`
-* [`volumes`](../../reference/configuration/volumes.md)
-* [`init containers`](../../reference/configuration/init-containers.md)
+- [`environment variables`](../../reference/configuration/environment-variables.md)
+- `args`
+- `command`
+- [`volumes`](../../reference/configuration/volumes.md)
+- [`init containers`](../../reference/configuration/init-containers.md)
diff --git a/docs/user-guide/user-defined-functions/reduce/examples.md b/docs/user-guide/user-defined-functions/reduce/examples.md
index 7e63a0b5..0030faba 100644
--- a/docs/user-guide/user-defined-functions/reduce/examples.md
+++ b/docs/user-guide/user-defined-functions/reduce/examples.md
@@ -10,8 +10,8 @@ Install the ISB
 kubectl apply -f https://raw.githubusercontent.com/numaproj/numaflow/stable/examples/0-isbsvc-jetstream.yaml
 ```
 
-Source used in the examples is an HTTP source producing messages with value 5 and 10 with event time 
-starting from 60000. Please refer the doc [http source](../../sources/http.md) on how to use an HTTP 
+Source used in the examples is an HTTP source producing messages with value 5 and 10 with event time
+starting from 60000. Please refer the doc [http source](../../sources/http.md) on how to use an HTTP
 source.
 An example will be as follows,
 
@@ -21,6 +21,7 @@ curl -kq -X POST -H "x-numaflow-event-time: 60000" -d "10" ${http-source-url}
 ```
 
 ## sum pipeline using fixed window
+
 This is a simple reduce pipeline that just does summation (sum of numbers) but uses fixed window.
 The snippet for the reduce vertex is as follows.
 
@@ -42,13 +43,12 @@ The snippet for the reduce vertex is as follows.
 [6-reduce-fixed-window.yaml](https://github.com/numaproj/numaflow/blob/main/examples/6-reduce-fixed-window.yaml)
 has the complete pipeline definition.
 
-In this example we use a `parallelism` of `2`. We are setting a parallelism > 1 because it is a
+In this example we use a `partitions` of `2`. We are setting a partitions > 1 because it is a
 keyed window.
 
 ```yaml
-- from: atoi
-  to: compute-sum
-  parallelism: 2
+- name: compute-sum
+  partitions: 2
 ```
 
 ```shell
@@ -56,6 +56,7 @@ kubectl apply -f https://raw.githubusercontent.com/numaproj/numaflow/main/exampl
 ```
 
 Output :
+
 ```text
 2023/01/05 11:54:41 (sink)  Payload -  300  Key -  odd  Start -  60000  End -  120000
 2023/01/05 11:54:41 (sink)  Payload -  600  Key -  even  Start -  60000  End -  120000
@@ -69,12 +70,13 @@ In our example, input is an HTTP source producing 2 messages each second with va
 and the event time starts from 60000. Since we have considered a fixed window of length 60s,
 and also we are producing two messages with different keys "even" and "odd", Numaflow will create
 two different windows with a start time of 60000 and an end time of 120000. So the output will be
-300(5 * 60) and 600(10 * 60).
+300(5 \* 60) and 600(10 \* 60).
 
 If we had used a non keyed window (`keyed: false`), we would have seen one single output with value
 of 900(300 of odd + 600 of even) for each window.
 
 ## sum pipeline using sliding window
+
 This is a simple reduce pipeline that just does summation (sum of numbers) but uses sliding window.
 The snippet for the reduce vertex is as follows.
 
@@ -100,7 +102,9 @@ has the complete pipeline definition
 ```shell
 kubectl apply -f https://raw.githubusercontent.com/numaproj/numaflow/main/examples/7-reduce-sliding-window.yaml
 ```
+
 Output:
+
 ```text
 2023/01/05 15:13:16 (sink)  Payload -  300  Key -  odd  Start -  60000  End -  120000
 2023/01/05 15:13:16 (sink)  Payload -  600  Key -  even  Start -  60000  End -  120000
@@ -115,7 +119,7 @@ and the event time starts from 60000. Since we have considered a sliding window
 and slide 10s, and also we are producing two messages with different keys "even" and "odd".
 Numaflow will create two different windows with a start time of 60000 and an end time of 120000,
 and because the slide duration is 10s, a next set of windows will be created with start time of
-70000 and an end time of 130000. Since it's a sum operation the output will be 300(5 * 60) and 600(10 * 60).
+70000 and an end time of 130000. Since it's a sum operation the output will be 300(5 \* 60) and 600(10 \* 60).
 
 `Payload -  50  Key -  odd  Start -  10000  End -  70000`, we see 50 here for odd because the
 first window has only 10 elements
@@ -123,9 +127,10 @@ first window has only 10 elements
 ## complex reduce pipeline
 
 In the complex reduce example, we will
-* chain of reduce functions
-* use both fixed and sliding windows
-* use keyed and non-keyed windowing
+
+- chain of reduce functions
+- use both fixed and sliding windows
+- use keyed and non-keyed windowing
 
 ![plot](../../../assets/complex-reduce.png)
 
@@ -137,6 +142,7 @@ kubectl apply -f https://raw.githubusercontent.com/numaproj/numaflow/main/exampl
 ```
 
 Output:
+
 ```text
 2023/01/05 15:33:55 (sink)  Payload -  900  Key -  NON_KEYED_STREAM  Start -  80000  End -  140000
 2023/01/05 15:33:55 (sink)  Payload -  900  Key -  NON_KEYED_STREAM  Start -  90000  End -  150000
@@ -147,8 +153,8 @@ Output:
 ```
 
 In our example, first we have the reduce vertex with a fixed window of duration 5s. Since the input is 5
-and 10, the output from the first reduce vertex will be 25 (5 * 5) and 50 (5 * 10). This will be passed
+and 10, the output from the first reduce vertex will be 25 (5 \* 5) and 50 (5 \* 10). This will be passed
 to the next non-keyed reduce vertex with the fixed window duration of 10s. This being a non-keyed, it will
-combine the inputs and produce the output of 150(25 * 2 + 50 * 2), which will be passed to the reduce
+combine the inputs and produce the output of 150(25 \* 2 + 50 \* 2), which will be passed to the reduce
 vertex with a sliding window of duration 60s and with the slide duration of 10s. Hence the final output
-will be 900(150 * 6).
+will be 900(150 \* 6).
diff --git a/docs/user-guide/user-defined-functions/reduce/reduce.md b/docs/user-guide/user-defined-functions/reduce/reduce.md
index e19b37fc..6d2271b8 100644
--- a/docs/user-guide/user-defined-functions/reduce/reduce.md
+++ b/docs/user-guide/user-defined-functions/reduce/reduce.md
@@ -1,17 +1,18 @@
 # Reduce UDF
 
-## Overview 
+## Overview
 
-Reduce is one of the most commonly used abstractions in a stream processing pipeline to define 
-aggregation functions on a stream of data. It is the reduce feature that helps us solve problems like 
-"performs a summary operation(such as counting the number of occurrence of a key, yielding user login 
+Reduce is one of the most commonly used abstractions in a stream processing pipeline to define
+aggregation functions on a stream of data. It is the reduce feature that helps us solve problems like
+"performs a summary operation(such as counting the number of occurrence of a key, yielding user login
 frequencies), etc."Since the input an unbounded stream (with infinite entries), we need an additional
 parameter to convert the unbounded problem to a bounded problem and provide results on that. That
-bounding condition is "time", eg, "number of users logged in per minute". So while processing an 
+bounding condition is "time", eg, "number of users logged in per minute". So while processing an
 unbounded stream of data, we need a way to group elements into finite chunks using time. To build these
 chunks the reduce function is applied to the set of records produced using the concept of [windowing](./windowing/windowing.md).
 
 ## Reduce Pseudo code
+
 Unlike in _map_ vertex where only an element is given to user-defined function, in _reduce_ since
 there is a group of elements, an iterator is passed to the reduce function. The following is a generic
 outlook of a reduce function. I have written the pseudo-code using the accumulator to show that very
@@ -25,7 +26,7 @@ def reduceFn(keys: List[str], datums: Iterator[Datum], md: Metadata) -> Messages
     accumulator = initialize_accumalor()
     # we are iterating on the input set of elements
     for d in datums:
-        # accumulator.add_input() can be any function. 
+        # accumulator.add_input() can be any function.
         # e.g., it could be as simple as accumulator += 1
         accumulator.add_input(d)
     # once we are done with iterating on the elements, we return the result
@@ -36,17 +37,16 @@ def reduceFn(keys: List[str], datums: Iterator[Datum], md: Metadata) -> Messages
 ## Specification
 
 The structure for defining a reduce vertex is as follows.
+
 ```yaml
 - name: my-reduce-udf
   udf:
     container:
       image: my-reduce-udf:latest
     groupBy:
-      window:
-        ...
+      window: ...
       keyed: ...
-      storage:
-        ...
+      storage: ...
 ```
 
 The reduce spec adds a new section called `groupBy` and this how we differentiate a _map_ vertex
@@ -54,41 +54,39 @@ from _reduce_ vertex. There are two important fields, the [_window_](./windowing
 and [_keyed_](./windowing/windowing.md#non-keyed-vs-keyed-windows). These two fields play an
 important role in grouping the data together and pass it to the user-defined reduce code.
 
-The reduce supports a parallelism value while defining the edge. This is because auto-scaling is 
-not supported in reduce vertex. If `parallelism` is not defined default of one will be used.
+The reduce supports parallelism processing by defining a `partitions` in the vertex. This is because auto-scaling is not supported in reduce vertex. If `partitions` is not defined default of one will be used.
 
 ```yaml
-- from: edge1
-  to: my-reduce-reduce
-  parallelism: integer
+- name: my-reduce-udf
+  partitions: integer
 ```
 
-It is wrong to give a `parallelism` > `1` if it is a _non-keyed_ vertex (`keyed: false`).
+It is wrong to give a `partitions` > `1` if it is a _non-keyed_ vertex (`keyed: false`).
 
 There are a couple of [examples](examples.md) that demonstrates Fixed windows, Sliding windows,
 chaining of windows, keyed streams, etc.
 
 ## Time Characteristics
 
-All windowing operations generate new records as an output of reduce operations. Event-time and Watermark 
-are two main primitives that determine how the time propagates in a streaming application. so for all new 
-records generated in a reduce operation, event time is set to the end time of the window. 
+All windowing operations generate new records as an output of reduce operations. Event-time and Watermark
+are two main primitives that determine how the time propagates in a streaming application. so for all new
+records generated in a reduce operation, event time is set to the end time of the window.
 
-For example, for a reduce operation over a keyed/non-keyed window with a start and end defined by 
-`[2031-09-29T18:47:00Z, 2031-09-29T18:48:00Z)`, event time for all the records generated will be set to 
-`2031-09-29T18:47:59.999Z` since millisecond is the smallest granularity (as of now) event time is set to 
-the last timestamp that belongs to a window. 
+For example, for a reduce operation over a keyed/non-keyed window with a start and end defined by
+`[2031-09-29T18:47:00Z, 2031-09-29T18:48:00Z)`, event time for all the records generated will be set to
+`2031-09-29T18:47:59.999Z` since millisecond is the smallest granularity (as of now) event time is set to
+the last timestamp that belongs to a window.
 
-Watermark is treated similarly, the watermark is set to the last timestamp for a given window. 
+Watermark is treated similarly, the watermark is set to the last timestamp for a given window.
 So for the example above, the value of the watermark will be set to the last timestamp, i.e., `2031-09-29T18:47:59.999Z`.
 
-This applies to all the window types regardless of whether they are keyed or non-keyed windows. 
+This applies to all the window types regardless of whether they are keyed or non-keyed windows.
 
 ## Allowed Lateness
 
-`allowedLateness` flag on the Reduce vertex will allow late data to be 
+`allowedLateness` flag on the Reduce vertex will allow late data to be
 processed by slowing the down the close-of-book operation of the Reduce vertex. Late data will be included for
-the Reduce operation as long as the late data is not later than `(CurrentWatermark - AllowedLateness)`. 
+the Reduce operation as long as the late data is not later than `(CurrentWatermark - AllowedLateness)`.
 Without `allowedLateness`, late data will be rejected and dropped. Each Reduce vertex can have its own `allowedLateness`.
 
 ```yaml
@@ -101,7 +99,7 @@ vertices:
 
 ## Storage
 
-Reduce unlike map requires persistence. To support persistence user has to define the 
+Reduce unlike map requires persistence. To support persistence user has to define the
 `storage` configuration. We replay the data stored in this storage on pod startup if there has
 been a restart of the reduce pod caused due to pod migrations, etc.
 
@@ -110,20 +108,20 @@ vertices:
   - name: my-udf
     udf:
       groupBy:
-        storage:
-           ....
+        storage: ....
 ```
 
 ### Persistent Volume Claim (PVC)
 
 `persistentVolumeClaim` supports the following fields, `volumeSize`, `storageClassName`, and`accessMode`.
-As name suggests,`volumeSize` specifies the size of the volume. `accessMode` can be of many types, but for 
+As name suggests,`volumeSize` specifies the size of the volume. `accessMode` can be of many types, but for
 reduce use case we need only `ReadWriteOnce`. `storageClassName` can also be provided, more info on storage class
 can be found [here](https://kubernetes.io/docs/concepts/storage/persistent-volumes#class-1). The default
-value of `storageClassName` is `default` which is default StorageClass may be deployed to a Kubernetes 
+value of `storageClassName` is `default` which is default StorageClass may be deployed to a Kubernetes
 cluster by addon manager during installation.
 
 #### Example
+
 ```yaml
 vertices:
   - name: my-udf
@@ -135,13 +133,13 @@ vertices:
             accessMode: ReadWriteOnce
 ```
 
-### EmptyDir 
+### EmptyDir
 
 We also support `emptyDir` for quick experimentation. We do not recommend this in production
-setup. If we use `emptyDir`, we will end up in data loss if there are pod migrations. `emptyDir` 
+setup. If we use `emptyDir`, we will end up in data loss if there are pod migrations. `emptyDir`
 also takes an optional `sizeLimit`. `medium` field controls where emptyDir volumes are stored.
-By default emptyDir volumes are stored on whatever medium that backs the node such as disk, SSD, 
-or network storage, depending on your environment. If you set the `medium` field to `"Memory"`, 
+By default emptyDir volumes are stored on whatever medium that backs the node such as disk, SSD,
+or network storage, depending on your environment. If you set the `medium` field to `"Memory"`,
 Kubernetes mounts a tmpfs (RAM-backed filesystem) for you instead.
 
 #### Example
@@ -153,4 +151,4 @@ vertices:
       groupBy:
         storage:
           emptyDir: {}
-```
\ No newline at end of file
+```
diff --git a/docs/user-guide/user-defined-functions/reduce/windowing/windowing.md b/docs/user-guide/user-defined-functions/reduce/windowing/windowing.md
index 83eaafed..d9fe734a 100644
--- a/docs/user-guide/user-defined-functions/reduce/windowing/windowing.md
+++ b/docs/user-guide/user-defined-functions/reduce/windowing/windowing.md
@@ -61,8 +61,8 @@ to a set of keyed problems and apply a non-keyed function at the end. This will
 help solve the original problem in a scalable manner without affecting the
 result's completeness and/or accuracy.
 
-When a `keyed` window is used, an optional `parallelism` can be specified in the
-`edge` leading to the vertex for parallel processing.
+When a `keyed` window is used, an optional `partitions` can be specified in the
+vertex for parallel processing.
 
 ### Usage
 
@@ -73,12 +73,9 @@ are not horizontally scalable as mentioned above.
 ```yaml
 vertices:
   - name: my-reduce
+    partitions: 5 # Optional, defaults to 1
     udf:
       groupBy:
         window: ...
         keyed: true # Optional, defaults to false
-edges:
-  - from: prev-udf
-    to: my-reduce
-    parallelism: 5 # Optional, defaults to 1
 ```
diff --git a/pkg/sources/transformer/builtin/time_extraction_filter/time_extraction_filter.go b/pkg/sources/transformer/builtin/time_extraction_filter/time_extraction_filter.go
index 02c514ac..738def97 100644
--- a/pkg/sources/transformer/builtin/time_extraction_filter/time_extraction_filter.go
+++ b/pkg/sources/transformer/builtin/time_extraction_filter/time_extraction_filter.go
@@ -28,9 +28,9 @@ import (
 )
 
 type expressions struct {
-	filterExpr    string
-	eventTimeExpr string
-	format        string
+	filterExpr      string
+	eventTimeExpr   string
+	eventTimeFormat string
 }
 
 func New(args map[string]string) (functionsdk.MapTFunc, error) {
@@ -45,15 +45,15 @@ func New(args map[string]string) (functionsdk.MapTFunc, error) {
 		return nil, fmt.Errorf(`missing "eventTimeExpr"`)
 	}
 
-	var format string
-	if format, existing = args["format"]; !existing {
-		format = ""
+	var eventTimeFormat string
+	if eventTimeFormat, existing = args["eventTimeFormat"]; !existing {
+		eventTimeFormat = ""
 	}
 
 	e := expressions{
-		filterExpr:    filterExpr,
-		eventTimeExpr: eventTimeExpr,
-		format:        format,
+		filterExpr:      filterExpr,
+		eventTimeExpr:   eventTimeExpr,
+		eventTimeFormat: eventTimeFormat,
 	}
 
 	return func(ctx context.Context, keys []string, datum functionsdk.Datum) functionsdk.MessageTs {
@@ -79,8 +79,8 @@ func (e expressions) apply(et time.Time, payload []byte) (functionsdk.MessageT,
 		}
 		var newEventTime time.Time
 		time.Local, _ = time.LoadLocation("UTC")
-		if e.format != "" {
-			newEventTime, err = time.Parse(e.format, timeStr)
+		if e.eventTimeFormat != "" {
+			newEventTime, err = time.Parse(e.eventTimeFormat, timeStr)
 		} else {
 			newEventTime, err = dateparse.ParseStrict(timeStr)
 		}


Test patch:
diff --git a/pkg/sources/transformer/builtin/time_extraction_filter/time_extraction_filter_test.go b/pkg/sources/transformer/builtin/time_extraction_filter/time_extraction_filter_test.go
index 0d43ccf7..20833fab 100644
--- a/pkg/sources/transformer/builtin/time_extraction_filter/time_extraction_filter_test.go
+++ b/pkg/sources/transformer/builtin/time_extraction_filter/time_extraction_filter_test.go
@@ -86,7 +86,7 @@ func TestFilterEventTime(t *testing.T) {
 	})
 
 	t.Run("Valid JSON expression for filter and eventTimeExtractor", func(t *testing.T) {
-		handle, err := New(map[string]string{"filterExpr": "int(json(payload).item[1].id) == 2", "eventTimeExpr": "json(payload).item[1].time", "format": time.RFC3339})
+		handle, err := New(map[string]string{"filterExpr": "int(json(payload).item[1].id) == 2", "eventTimeExpr": "json(payload).item[1].time", "eventTimeFormat": time.RFC3339})
 		assert.NoError(t, err)
 
 		result := handle(context.Background(), []string{"test-key"}, &testDatum{
@@ -105,7 +105,7 @@ func TestFilterEventTime(t *testing.T) {
 	})
 
 	t.Run("Invalid JSON expression for filter", func(t *testing.T) {
-		handle, err := New(map[string]string{"filterExpr": "int(json(payload).item[1].id) == 3", "eventTimeExpr": "json(payload).item[1].time", "format": time.RFC3339})
+		handle, err := New(map[string]string{"filterExpr": "int(json(payload).item[1].id) == 3", "eventTimeExpr": "json(payload).item[1].time", "eventTimeFormat": time.RFC3339})
 		assert.NoError(t, err)
 
 		result := handle(context.Background(), []string{"test-key"}, &testDatum{
@@ -118,7 +118,7 @@ func TestFilterEventTime(t *testing.T) {
 	})
 
 	t.Run("Valid JSON expression for filter, incorrect format to eventTime", func(t *testing.T) {
-		handle, err := New(map[string]string{"filterExpr": "int(json(payload).item[1].id) == 2", "eventTimeExpr": "json(payload).item[1].time", "format": time.ANSIC})
+		handle, err := New(map[string]string{"filterExpr": "int(json(payload).item[1].id) == 2", "eventTimeExpr": "json(payload).item[1].time", "eventTimeFormat": time.ANSIC})
 		assert.NoError(t, err)
 
 		testInputEventTime := time.Date(2022, 1, 4, 2, 3, 4, 5, time.UTC)
diff --git a/ui/src/utils/fetcherHooks/pipelineViewFetch.test.ts b/ui/src/utils/fetcherHooks/pipelineViewFetch.test.ts
index 0559acb5..428b4a22 100644
--- a/ui/src/utils/fetcherHooks/pipelineViewFetch.test.ts
+++ b/ui/src/utils/fetcherHooks/pipelineViewFetch.test.ts
@@ -54,13 +54,11 @@ describe("Custom Pipeline hook", () => {
               from: "in",
               to: "cat",
               conditions: null,
-              parallelism: null,
             },
             {
               from: "cat",
               to: "out",
               conditions: null,
-              parallelism: null,
             },
           ],
           watermark: {