Repository: numaproj/numaflow
Pull request #880 (2023-07-19 06:41:06)

chore: some initial code for side inputs manager
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/go.mod b/go.mod
index 66e27bdf..40df5519 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/redis/go-redis/v9 v9.0.3
+	github.com/robfig/cron/v3 v3.0.1
 	github.com/soheilhy/cmux v0.1.5
 	github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72
 	github.com/spf13/cobra v1.2.1
diff --git a/go.sum b/go.sum
index 964590fa..cf0f3ef1 100644
--- a/go.sum
+++ b/go.sum
@@ -789,6 +789,8 @@ github.com/redis/go-redis/v9 v9.0.3 h1:+7mmR26M0IvyLxGZUHxu4GiBkJkVDid0Un+j4ScYu
 github.com/redis/go-redis/v9 v9.0.3/go.mod h1:WqMKv5vnQbRuZstUwxQI195wHy+t4PuXDOjzMvcuQHk=
 github.com/rivo/tview v0.0.0-20200219210816-cd38d7432498/go.mod h1:6lkG1x+13OShEf0EaOCaTQYyB7d5nSbb181KtjlS+84=
 github.com/rivo/uniseg v0.1.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/rogpeppe/go-internal v1.1.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
diff --git a/pkg/forward/doc.go b/pkg/forward/doc.go
new file mode 100644
index 00000000..30acdc3f
--- /dev/null
+++ b/pkg/forward/doc.go
@@ -0,0 +1,15 @@
+/*
+Copyright 2022 The Numaproj Authors.
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package forward is used for creating a data forwarder for the UDF and sink vertex.
+package forward
diff --git a/pkg/forward/forward.go b/pkg/forward/forward.go
index 5aa0fda9..a86027c3 100644
--- a/pkg/forward/forward.go
+++ b/pkg/forward/forward.go
@@ -69,7 +69,8 @@ type InterStepDataForward struct {
 }
 
 // NewInterStepDataForward creates an inter-step forwarder.
-func NewInterStepDataForward(vertex *dfv1.Vertex,
+func NewInterStepDataForward(
+	vertex *dfv1.Vertex,
 	fromStep isb.BufferReader,
 	toSteps map[string][]isb.BufferWriter,
 	fsd ToWhichStepDecider,
@@ -110,19 +111,14 @@ func NewInterStepDataForward(vertex *dfv1.Vertex,
 	// Add logger from parent ctx to child context.
 	isdf.ctx = logging.WithLogger(ctx, options.logger)
 
-	if isdf.opts.vertexType == dfv1.VertexTypeSource {
-		if isdf.opts.srcWatermarkPublisher == nil {
-			return nil, fmt.Errorf("failed to assign a non-nil source watermark publisher for source vertex data forwarder")
-		}
-		if isdf.opts.enableMapUdfStream {
-			return nil, fmt.Errorf("stream is not supported for source data transformer")
-		}
-	}
-
 	if isdf.opts.enableMapUdfStream && isdf.opts.readBatchSize != 1 {
 		return nil, fmt.Errorf("batch size is not 1 with UDF streaming")
 	}
 
+	if isdf.opts.vertexType == dfv1.VertexTypeSource {
+		return nil, fmt.Errorf("source vertex is not supported by inter-step forwarder, please use source forwarder instead")
+	}
+
 	return &isdf, nil
 }
 
@@ -258,20 +254,12 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		}
 	}
 
-	var processorWM wmb.Watermark
-	if isdf.opts.vertexType == dfv1.VertexTypeSource {
-		// for source vertex, the udf is the source data transformer.
-		// in this case, we assign time.UnixMilli(-1) to processorWM.
-		// source data transformer applies filtering and assigns event time to source data, which doesn't require watermarks.
-		processorWM = wmb.Watermark(time.UnixMilli(-1))
-	} else {
-		// fetch watermark if available
-		// TODO: make it async (concurrent and wait later)
-		// let's track only the first element's watermark. This is important because we reassign the watermark we fetch
-		// to all the elements in the batch. If we were to assign last element's watermark, we will wrongly mark on-time data as late.
-		// we fetch the watermark for the partition from which we read the message.
-		processorWM = isdf.wmFetcher.GetWatermark(readMessages[0].ReadOffset, isdf.fromBufferPartition.GetPartitionIdx())
-	}
+	// fetch watermark if available
+	// TODO: make it async (concurrent and wait later)
+	// let's track only the first element's watermark. This is important because we reassign the watermark we fetch
+	// to all the elements in the batch. If we were to assign last element's watermark, we will wrongly mark on-time data as late.
+	// we fetch the watermark for the partition from which we read the message.
+	processorWM := isdf.wmFetcher.GetWatermark(readMessages[0].ReadOffset, isdf.fromBufferPartition.GetPartitionIdx())
 
 	var writeOffsets map[string][][]isb.Offset
 	if !isdf.opts.enableMapUdfStream {
@@ -304,7 +292,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		for idx, m := range dataMessages {
 			// emit message size metric
 			ReadBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(m.Payload)))
-			// assign watermark to the message. assign time.UnixMilli(-1) as watermark when we are at source vertex.
+			// assign watermark to the message
 			m.Watermark = time.Time(processorWM)
 			// send UDF processing work to the channel
 			udfResults[idx].readMessage = m
@@ -319,34 +307,6 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		ConcurrentUDFProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Observe(float64(time.Since(concurrentUDFProcessingStart).Microseconds()))
 		// UDF processing is done.
 
-		// if vertex type is source, it means we have finished the source data transformation.
-		// let's publish source watermark and assign IsLate attribute based on new event time.
-		if isdf.opts.vertexType == dfv1.VertexTypeSource {
-			var writeMessages []*isb.WriteMessage
-			var transformedReadMessages []*isb.ReadMessage
-			for _, m := range udfResults {
-				writeMessages = append(writeMessages, m.writeMessages...)
-				for _, message := range m.writeMessages {
-					// we convert each writeMessage to isb.ReadMessage by providing its parent ReadMessage's ReadOffset.
-					// since we use message event time instead of the watermark to determine and publish source watermarks,
-					// time.UnixMilli(-1) is assigned to the message watermark. transformedReadMessages are immediately
-					// used below for publishing source watermarks.
-					transformedReadMessages = append(transformedReadMessages, message.ToReadMessage(m.readMessage.ReadOffset, time.UnixMilli(-1)))
-				}
-			}
-			// publish source watermark
-			isdf.opts.srcWatermarkPublisher.PublishSourceWatermarks(transformedReadMessages)
-			// fetch the source watermark again, we might not get the latest watermark because of publishing delay,
-			// but ideally we should use the latest to determine the IsLate attribute.
-			processorWM = isdf.wmFetcher.GetWatermark(readMessages[0].ReadOffset, isdf.fromBufferPartition.GetPartitionIdx())
-			// assign isLate
-			for _, m := range writeMessages {
-				if processorWM.After(m.EventTime) { // Set late data at source level
-					m.IsLate = true
-				}
-			}
-		}
-
 		// let's figure out which vertex to send the results to.
 		// update the toBuffer(s) with writeMessages.
 		for _, m := range udfResults {
@@ -397,8 +357,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		activeWatermarkBuffers[toVertexName] = make([]bool, len(toVertexBufferOffsets))
 		if publisher, ok := isdf.wmPublishers[toVertexName]; ok {
 			for index, offsets := range toVertexBufferOffsets {
-				if isdf.opts.vertexType == dfv1.VertexTypeSource || isdf.opts.vertexType == dfv1.VertexTypeMapUDF ||
-					isdf.opts.vertexType == dfv1.VertexTypeReduceUDF {
+				if isdf.opts.vertexType == dfv1.VertexTypeMapUDF || isdf.opts.vertexType == dfv1.VertexTypeReduceUDF {
 					if len(offsets) > 0 {
 						publisher.PublishWatermark(processorWM, offsets[len(offsets)-1], int32(index))
 						activeWatermarkBuffers[toVertexName][index] = true
@@ -480,7 +439,7 @@ func (isdf *InterStepDataForward) streamMessage(
 		// emit message size metric
 		ReadBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).
 			Add(float64(len(dataMessages[0].Payload)))
-		// assign watermark to the message. assign time.UnixMilli(-1) as watermark when we are at source vertex.
+		// assign watermark to the message
 		dataMessages[0].Watermark = time.Time(processorWM)
 
 		// process the UDF and get the result
diff --git a/pkg/forward/options.go b/pkg/forward/options.go
index 96d6b26c..28e2453e 100644
--- a/pkg/forward/options.go
+++ b/pkg/forward/options.go
@@ -22,7 +22,6 @@ import (
 	"go.uber.org/zap"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 )
 
@@ -36,8 +35,6 @@ type options struct {
 	retryInterval time.Duration
 	// vertexType indicates the type of the vertex
 	vertexType dfv1.VertexType
-	// srcWatermarkPublisher is used to publish source watermark
-	srcWatermarkPublisher isb.SourceWatermarkPublisher
 	// logger is used to pass the logger variable
 	logger *zap.SugaredLogger
 	// enableMapUdfStream indicates whether the message streaming is enabled or not for UDF processing
@@ -96,14 +93,6 @@ func WithVertexType(t dfv1.VertexType) Option {
 	}
 }
 
-// WithSourceWatermarkPublisher sets the source watermark publisher
-func WithSourceWatermarkPublisher(p isb.SourceWatermarkPublisher) Option {
-	return func(o *options) error {
-		o.srcWatermarkPublisher = p
-		return nil
-	}
-}
-
 // WithUDFStreaming sets streaming for UDF processing
 func WithUDFStreaming(f bool) Option {
 	return func(o *options) error {
diff --git a/pkg/sideinputs/manager/manager.go b/pkg/sideinputs/manager/manager.go
index 3db3341c..f835a914 100644
--- a/pkg/sideinputs/manager/manager.go
+++ b/pkg/sideinputs/manager/manager.go
@@ -19,7 +19,9 @@ package manager
 import (
 	"context"
 	"fmt"
+	"time"
 
+	cronlib "github.com/robfig/cron/v3"
 	"go.uber.org/zap"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
@@ -65,11 +67,56 @@ func (sim *sideInputsManager) Start(ctx context.Context) error {
 		return fmt.Errorf("unrecognized isbsvc type %q", sim.isbSvcType)
 	}
 
-	// TODO(SI): do something
-	// Periodically call the ud container and write data to the store.
+	// TODO(SI): remove it.
 	fmt.Printf("ISB Svc Client nil: %v\n", isbSvcClient == nil)
-	fmt.Printf("Pipeline: %s, SideInput: %s\n", sim.pipelineName, sim.sideInput.Name)
+
+	f := func() {
+		if err := sim.execute(ctx); err != nil {
+			log.Errorw("Failed to execute the call to fetch Side Inputs.", zap.Error(err))
+		}
+	}
+
+	cron, err := resolveCron(sim.sideInput.Trigger, f)
+	if err != nil {
+		return fmt.Errorf("failed to resolve cron: %w", err)
+	}
+
+	// Run the first time immediately.
+	f()
+
+	cron.Start()
+	defer func() { _ = cron.Stop() }()
 
 	<-ctx.Done()
 	return nil
 }
+
+func (sim *sideInputsManager) execute(ctx context.Context) error {
+	log := logging.FromContext(ctx)
+	// TODO(SI): call ud container to fetch data and write to store.
+	log.Info("Executing ...")
+	return nil
+}
+
+func resolveCron(trigger *dfv1.SideInputTrigger, cmd func()) (*cronlib.Cron, error) {
+	var opts []cronlib.Option
+	if trigger.Timezone != nil {
+		location, err := time.LoadLocation(*trigger.Timezone)
+		if err != nil {
+			return nil, fmt.Errorf("failed to load location %q, error: %w", *trigger.Timezone, err)
+		}
+		opts = append(opts, cronlib.WithLocation(location))
+	}
+	cron := cronlib.New(opts...)
+	if trigger.Schedule != nil {
+		if _, err := cron.AddFunc(*trigger.Schedule, cmd); err != nil {
+			return nil, fmt.Errorf("failed to add cron schedule %q, error: %w", *trigger.Schedule, err)
+		}
+	} else {
+		duration := trigger.Interval.Duration.String()
+		if _, err := cron.AddFunc(fmt.Sprintf("@every %s", duration), cmd); err != nil {
+			return nil, fmt.Errorf("failed to add interval based cron job %q, error: %w", duration, err)
+		}
+	}
+	return cron, nil
+}
diff --git a/pkg/sources/forward/data_forward.go b/pkg/sources/forward/data_forward.go
index bfb9b872..2f842cf4 100644
--- a/pkg/sources/forward/data_forward.go
+++ b/pkg/sources/forward/data_forward.go
@@ -62,8 +62,6 @@ type DataForward struct {
 	pipelineName   string
 	// idleManager manages the idle watermark status.
 	idleManager *wmb.IdleManager
-	// wmbChecker checks if the idle watermark is valid.
-	wmbChecker wmb.WMBChecker
 	Shutdown
 }
 
@@ -99,7 +97,6 @@ func NewDataForward(
 		vertexName:           vertex.Spec.Name,
 		pipelineName:         vertex.Spec.PipelineName,
 		idleManager:          wmb.NewIdleManager(len(toSteps)),
-		wmbChecker:           wmb.NewWMBChecker(2), // TODO: make configurable
 		Shutdown: Shutdown{
 			rwlock: new(sync.RWMutex),
 		},
@@ -199,44 +196,13 @@ func (isdf *DataForward) forwardAChunk(ctx context.Context) {
 	// process only if we have any read messages. There is a natural looping here if there is an internal error while
 	// reading, and we are not able to proceed.
 	if len(readMessages) == 0 {
-		// When the read length is zero, the write length is definitely zero too
-		// meaning there's no data to be published to the next vertex, and we consider this
-		// situation as idling.
-		// In order to continue propagating watermark, we will set watermark idle=true and publish it.
-		// We also publish a control message if this is the first time we get this idle situation.
-		// we use the HeadWMB as the watermark for the idle
-		// we get the HeadWMB for the partition from which we read the messages
-		var processorWMB = isdf.wmFetcher.GetHeadWMB(isdf.reader.GetPartitionIdx())
-		if !isdf.wmbChecker.ValidateHeadWMB(processorWMB) {
-			// validation failed, skip publishing
-			isdf.opts.logger.Debugw("skip publishing idle watermark",
-				zap.Int("counter", isdf.wmbChecker.GetCounter()),
-				zap.Int64("offset", processorWMB.Offset),
-				zap.Int64("watermark", processorWMB.Watermark),
-				zap.Bool("Idle", processorWMB.Idle))
-			return
-		}
-
-		// if the validation passed, we will publish the watermark to all the toBuffer partitions.
-		for toVertexName, toVertexBuffer := range isdf.toBuffers {
-			for _, partition := range toVertexBuffer {
-				if p, ok := isdf.toVertexWMPublishers[toVertexName]; ok {
-					idlehandler.PublishIdleWatermark(ctx, partition, p, isdf.idleManager, isdf.opts.logger, dfv1.VertexTypeSource, wmb.Watermark(time.UnixMilli(processorWMB.Watermark)))
-				}
-			}
-		}
 		return
 	}
 
-	var dataMessages = make([]*isb.ReadMessage, 0, len(readMessages))
 	// store the offsets of the messages we read from source
 	var readOffsets = make([]isb.Offset, len(readMessages))
 	for idx, m := range readMessages {
 		readOffsets[idx] = m.ReadOffset
-		// for source forwarder, below if statement should always hold true, since we never publish idle watermark to source reader buffer
-		if m.Kind == isb.Data {
-			dataMessages = append(dataMessages, m)
-		}
 	}
 
 	// source data transformer applies filtering and assigns event time to source data, which doesn't require watermarks.
@@ -255,7 +221,7 @@ func (isdf *DataForward) forwardAChunk(ctx context.Context) {
 	transformerCh := make(chan *readWriteMessagePair)
 	// transformerResults stores the results after user defined transformer processing for all read messages. It indexes
 	// a read message to the corresponding write message
-	transformerResults := make([]readWriteMessagePair, len(dataMessages))
+	transformerResults := make([]readWriteMessagePair, len(readMessages))
 	// applyTransformer, if there is an Internal error it is a blocking call and will return only if shutdown has been initiated.
 
 	// create a pool of Transformer Processors
@@ -270,7 +236,7 @@ func (isdf *DataForward) forwardAChunk(ctx context.Context) {
 	concurrentTransformerProcessingStart := time.Now()
 
 	// send to transformer only the data messages
-	for idx, m := range dataMessages {
+	for idx, m := range readMessages {
 		// emit message size metric
 		forward.ReadBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Add(float64(len(m.Payload)))
 		// assign watermark to the message
@@ -364,24 +330,16 @@ func (isdf *DataForward) forwardAChunk(ctx context.Context) {
 			}
 		}
 	}
-	// - condition1 "len(dataMessages) > 0" :
-	//   Meaning, we do have some data messages, but we may not have written to all out buffers or its partitions.
-	//   It could be all data messages are dropped, or conditional forwarding to part of the out buffers.
-	//   If we don't have this condition check, when dataMessages is zero but ctrlMessages > 0, we will
-	//   wrongly publish an idle watermark without the ctrl message and the ctrl message tracking map.
-	// - condition 2 "len(activeWatermarkBuffers) < len(isdf.toVertexWMPublishers)" :
-	//   send idle watermark only if we have idle out buffers
-	// Note: When the len(dataMessages) is 0, meaning all the readMessages are control messages, we choose not to do extra steps
-	// This is because, if the idle continues, we will eventually handle the idle watermark when we read the next batch where the len(readMessages) will be zero
-	if len(dataMessages) > 0 {
-		for bufferName := range isdf.toVertexWMPublishers {
-			for index, activePartition := range activeWatermarkBuffers[bufferName] {
-				if !activePartition {
-					// use the watermark of the current read batch for the idle watermark
-					// same as read len==0 because there's no event published to the buffer
-					if p, ok := isdf.toVertexWMPublishers[bufferName]; ok {
-						idlehandler.PublishIdleWatermark(ctx, isdf.toBuffers[bufferName][index], p, isdf.idleManager, isdf.opts.logger, dfv1.VertexTypeSource, processorWM)
-					}
+
+	// condition "len(activeWatermarkBuffers) < len(isdf.toVertexWMPublishers)" :
+	// send idle watermark only if we have idle out buffers
+	for bufferName := range isdf.toVertexWMPublishers {
+		for index, activePartition := range activeWatermarkBuffers[bufferName] {
+			if !activePartition {
+				// use the watermark of the current read batch for the idle watermark
+				// same as read len==0 because there's no event published to the buffer
+				if p, ok := isdf.toVertexWMPublishers[bufferName]; ok {
+					idlehandler.PublishIdleWatermark(ctx, isdf.toBuffers[bufferName][index], p, isdf.idleManager, isdf.opts.logger, dfv1.VertexTypeSource, processorWM)
 				}
 			}
 		}
diff --git a/ui/yarn.lock b/ui/yarn.lock
index ec91606b..5085b04e 100644
--- a/ui/yarn.lock
+++ b/ui/yarn.lock
@@ -11022,9 +11022,9 @@ which@^2.0.1:
     isexe "^2.0.0"
 
 word-wrap@^1.2.3, word-wrap@~1.2.3:
-  version "1.2.3"
-  resolved "https://registry.yarnpkg.com/word-wrap/-/word-wrap-1.2.3.tgz#610636f6b1f703891bd34771ccb17fb93b47079c"
-  integrity sha512-Hz/mrNwitNRh/HUAtM/VT/5VH+ygD6DV7mYKZAtHOrbs8U7lvPS6xf7EJKMF0uW1KJCl0H701g3ZGus+muE5vQ==
+  version "1.2.4"
+  resolved "https://registry.yarnpkg.com/word-wrap/-/word-wrap-1.2.4.tgz#cb4b50ec9aca570abd1f52f33cd45b6c61739a9f"
+  integrity sha512-2V81OA4ugVo5pRo46hAoD2ivUJx8jXmWXfUkY4KFNw0hEptvN0QfH3K4nHiwzGeKl5rFKedV48QVoqYavy4YpA==
 
 workbox-background-sync@6.5.4:
   version "6.5.4"


Test patch:
diff --git a/pkg/forward/forward_test.go b/pkg/forward/forward_test.go
index 0f0890fb..47fd7a0d 100644
--- a/pkg/forward/forward_test.go
+++ b/pkg/forward/forward_test.go
@@ -1140,10 +1140,6 @@ func (f *mySourceForwardTestRoundRobin) WhereTo(_ []string, _ []string) ([]Verte
 	return output, nil
 }
 
-// for source forward test, in transformer, we assign to the message a new event time that is before test source watermark,
-// such that we can verify message IsLate attribute gets set to true.
-var testSourceNewEventTime = testSourceWatermark.Add(time.Duration(-1) * time.Minute)
-
 func (f mySourceForwardTest) ApplyMap(ctx context.Context, message *isb.ReadMessage) ([]*isb.WriteMessage, error) {
 	return func(ctx context.Context, readMessage *isb.ReadMessage) ([]*isb.WriteMessage, error) {
 		_ = ctx
@@ -1155,8 +1151,6 @@ func (f mySourceForwardTest) ApplyMap(ctx context.Context, message *isb.ReadMess
 		_ = payload
 		// copy the payload
 		result := payload
-		// assign new event time
-		parentPaneInfo.EventTime = testSourceNewEventTime
 		var key []string
 
 		writeMessage := isb.Message{
@@ -1186,8 +1180,6 @@ func (f mySourceForwardTest) ApplyMapStream(ctx context.Context, message *isb.Re
 		_ = payload
 		// copy the payload
 		result := payload
-		// assign new event time
-		parentPaneInfo.EventTime = testSourceNewEventTime
 		var key []string
 
 		writeMessage := isb.Message{
@@ -1214,7 +1206,7 @@ func (p TestSourceWatermarkPublisher) PublishSourceWatermarks([]*isb.ReadMessage
 	// PublishSourceWatermarks is not tested in forwarder_test.go
 }
 
-func TestSourceInterStepDataForwardSinglePartition(t *testing.T) {
+func TestInterStepDataForwardSinglePartition(t *testing.T) {
 	fromStep := simplebuffer.NewInMemoryBuffer("from", 25, 0)
 	to1 := simplebuffer.NewInMemoryBuffer("to1", 10, 0, simplebuffer.WithReadTimeOut(time.Second*10))
 	toSteps := map[string][]isb.BufferWriter{
@@ -1233,14 +1225,8 @@ func TestSourceInterStepDataForwardSinglePartition(t *testing.T) {
 	fetchWatermark := &testForwardFetcher{}
 	_, publishWatermark := generic.BuildNoOpWatermarkProgressorsFromBufferMap(toSteps)
 
-	// verify if source watermark publisher is not set, NewInterStepDataForward throws.
-	failedForwarder, err := NewInterStepDataForward(vertex, fromStep, toSteps, mySourceForwardTest{}, mySourceForwardTest{}, fetchWatermark, publishWatermark, WithReadBatchSize(5), WithVertexType(dfv1.VertexTypeSource))
-	assert.True(t, failedForwarder == nil)
-	assert.Error(t, err)
-	assert.Equal(t, fmt.Errorf("failed to assign a non-nil source watermark publisher for source vertex data forwarder"), err)
-
-	// create a valid source forwarder
-	f, err := NewInterStepDataForward(vertex, fromStep, toSteps, mySourceForwardTest{}, mySourceForwardTest{}, fetchWatermark, publishWatermark, WithReadBatchSize(5), WithVertexType(dfv1.VertexTypeSource), WithSourceWatermarkPublisher(TestSourceWatermarkPublisher{}))
+	// create a forwarder
+	f, err := NewInterStepDataForward(vertex, fromStep, toSteps, mySourceForwardTest{}, mySourceForwardTest{}, fetchWatermark, publishWatermark, WithReadBatchSize(5), WithVertexType(dfv1.VertexTypeMapUDF))
 	assert.NoError(t, err)
 	assert.False(t, to1.IsFull())
 	assert.True(t, to1.IsEmpty())
@@ -1257,12 +1243,7 @@ func TestSourceInterStepDataForwardSinglePartition(t *testing.T) {
 	assert.Len(t, readMessages, int(count))
 	assert.Equal(t, []interface{}{writeMessages[0].Header.Keys, writeMessages[1].Header.Keys}, []interface{}{readMessages[0].Header.Keys, readMessages[1].Header.Keys})
 	assert.Equal(t, []interface{}{writeMessages[0].Header.ID + "-0-0", writeMessages[1].Header.ID + "-0-0"}, []interface{}{readMessages[0].Header.ID, readMessages[1].Header.ID})
-	for _, m := range readMessages {
-		// verify new event time gets assigned to messages.
-		assert.Equal(t, testSourceNewEventTime, m.EventTime)
-		// verify messages are marked as late because of event time update.
-		assert.Equal(t, true, m.IsLate)
-	}
+
 	f.Stop()
 	time.Sleep(1 * time.Millisecond)
 	// only for shutdown will work as from buffer is not empty
@@ -1270,7 +1251,7 @@ func TestSourceInterStepDataForwardSinglePartition(t *testing.T) {
 	<-stopped
 }
 
-func TestSourceInterStepDataForwardMultiplePartition(t *testing.T) {
+func TestInterStepDataForwardMultiplePartition(t *testing.T) {
 	fromStep := simplebuffer.NewInMemoryBuffer("from", 25, 0)
 	to11 := simplebuffer.NewInMemoryBuffer("to1-0", 10, 0, simplebuffer.WithReadTimeOut(time.Second*10))
 	to12 := simplebuffer.NewInMemoryBuffer("to1-1", 10, 1, simplebuffer.WithReadTimeOut(time.Second*10))
@@ -1290,14 +1271,8 @@ func TestSourceInterStepDataForwardMultiplePartition(t *testing.T) {
 	fetchWatermark := &testForwardFetcher{}
 	_, publishWatermark := generic.BuildNoOpWatermarkProgressorsFromBufferMap(toSteps)
 
-	// verify if source watermark publisher is not set, NewInterStepDataForward throws.
-	failedForwarder, err := NewInterStepDataForward(vertex, fromStep, toSteps, mySourceForwardTest{}, mySourceForwardTest{}, fetchWatermark, publishWatermark, WithReadBatchSize(5), WithVertexType(dfv1.VertexTypeSource))
-	assert.True(t, failedForwarder == nil)
-	assert.Error(t, err)
-	assert.Equal(t, fmt.Errorf("failed to assign a non-nil source watermark publisher for source vertex data forwarder"), err)
-
-	// create a valid source forwarder
-	f, err := NewInterStepDataForward(vertex, fromStep, toSteps, &mySourceForwardTestRoundRobin{}, mySourceForwardTest{}, fetchWatermark, publishWatermark, WithReadBatchSize(5), WithVertexType(dfv1.VertexTypeSource), WithSourceWatermarkPublisher(TestSourceWatermarkPublisher{}))
+	// create a forwarder
+	f, err := NewInterStepDataForward(vertex, fromStep, toSteps, &mySourceForwardTestRoundRobin{}, mySourceForwardTest{}, fetchWatermark, publishWatermark, WithReadBatchSize(5), WithVertexType(dfv1.VertexTypeMapUDF))
 	assert.NoError(t, err)
 	assert.False(t, to11.IsFull())
 	assert.False(t, to12.IsFull())
@@ -1318,12 +1293,6 @@ func TestSourceInterStepDataForwardMultiplePartition(t *testing.T) {
 	assert.Len(t, readMessages, 2)
 	assert.Equal(t, []interface{}{writeMessages[0].Header.Keys, writeMessages[2].Header.Keys}, []interface{}{readMessages[0].Header.Keys, readMessages[1].Header.Keys})
 	assert.Equal(t, []interface{}{writeMessages[0].Header.ID + "-0-0", writeMessages[2].Header.ID + "-0-0"}, []interface{}{readMessages[0].Header.ID, readMessages[1].Header.ID})
-	for _, m := range readMessages {
-		// verify new event time gets assigned to messages.
-		assert.Equal(t, testSourceNewEventTime, m.EventTime)
-		// verify messages are marked as late because of event time update.
-		assert.Equal(t, true, m.IsLate)
-	}
 
 	time.Sleep(time.Second)
 
@@ -1332,12 +1301,7 @@ func TestSourceInterStepDataForwardMultiplePartition(t *testing.T) {
 	assert.Len(t, readMessages, 2)
 	assert.Equal(t, []interface{}{writeMessages[1].Header.Keys, writeMessages[3].Header.Keys}, []interface{}{readMessages[0].Header.Keys, readMessages[1].Header.Keys})
 	assert.Equal(t, []interface{}{writeMessages[1].Header.ID + "-0-0", writeMessages[3].Header.ID + "-0-0"}, []interface{}{readMessages[0].Header.ID, readMessages[1].Header.ID})
-	for _, m := range readMessages {
-		// verify new event time gets assigned to messages.
-		assert.Equal(t, testSourceNewEventTime, m.EventTime)
-		// verify messages are marked as late because of event time update.
-		assert.Equal(t, true, m.IsLate)
-	}
+
 	f.Stop()
 	time.Sleep(1 * time.Millisecond)
 	// only for shutdown will work as from buffer is not empty
diff --git a/pkg/sideinputs/manager/manager_test.go b/pkg/sideinputs/manager/manager_test.go
new file mode 100644
index 00000000..d024c2a9
--- /dev/null
+++ b/pkg/sideinputs/manager/manager_test.go
@@ -0,0 +1,62 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package manager
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+)
+
+func Test_resolveCron(t *testing.T) {
+	f := func() { fmt.Println("hello") }
+
+	t.Run("valid cron schedule", func(t *testing.T) {
+		trigger := &dfv1.SideInputTrigger{
+			Schedule: pointer.String("*/5 * * * *"),
+		}
+		c, err := resolveCron(trigger, f)
+		assert.NoError(t, err)
+		assert.NotNil(t, c)
+		assert.Len(t, c.Entries(), 1)
+	})
+
+	t.Run("invalid cron schedule", func(t *testing.T) {
+		trigger := &dfv1.SideInputTrigger{
+			Schedule: pointer.String("*/ab * * * *"),
+		}
+		c, err := resolveCron(trigger, f)
+		assert.Error(t, err)
+		assert.Nil(t, c)
+	})
+
+	t.Run("valid interval", func(t *testing.T) {
+		trigger := &dfv1.SideInputTrigger{
+			Interval: &metav1.Duration{Duration: time.Second * 20},
+		}
+		c, err := resolveCron(trigger, f)
+		assert.NoError(t, err)
+		assert.NotNil(t, c)
+		assert.Len(t, c.Entries(), 1)
+	})
+}
diff --git a/pkg/sources/forward/data_forward_test.go b/pkg/sources/forward/data_forward_test.go
index 79217864..5ee1f5e6 100644
--- a/pkg/sources/forward/data_forward_test.go
+++ b/pkg/sources/forward/data_forward_test.go
@@ -54,7 +54,6 @@ const (
 var (
 	testStartTime       = time.Unix(1636470000, 0).UTC()
 	testSourceWatermark = time.Unix(1636460000, 0).UTC()
-	testWMBWatermark    = time.Unix(1636470000, 0).UTC()
 )
 
 type testForwardFetcher struct {
@@ -746,370 +745,6 @@ func TestNewDataForward(t *testing.T) {
 	}
 }
 
-type testWMBFetcher struct {
-	// for forward_test.go only
-	WMBTestSameHeadWMB bool // for testing same head wmb, if set true then WMBTestDiffHeadWMB must be false
-	sameCounter        int
-	sameLock           sync.RWMutex
-	WMBTestDiffHeadWMB bool // for testing different head wmb, if set true then WMBTestSameHeadWMB must be false
-	diffCounter        int
-	diffLock           sync.RWMutex
-}
-
-// RevertBoolValue set WMBTestSameHeadWMB and WMBTestDiffHeadWMB to opposite value
-func (t *testWMBFetcher) RevertBoolValue() {
-	t.sameLock.Lock()
-	defer t.sameLock.Unlock()
-	t.diffLock.Lock()
-	defer t.diffLock.Unlock()
-	t.WMBTestSameHeadWMB = !t.WMBTestSameHeadWMB
-	t.WMBTestDiffHeadWMB = !t.WMBTestDiffHeadWMB
-}
-
-func (t *testWMBFetcher) Close() error {
-	// won't be used
-	return nil
-}
-
-// GetWatermark uses current time as the watermark because we want to make sure
-// the test publisher is publishing watermark
-func (t *testWMBFetcher) GetWatermark(offset isb.Offset, partition int32) wmb.Watermark {
-	return wmb.Watermark(testWMBWatermark)
-}
-
-func (t *testWMBFetcher) GetHeadWatermark(int32) wmb.Watermark {
-	// won't be used
-	return wmb.Watermark{}
-}
-
-func (t *testWMBFetcher) GetHeadWMB(int32) wmb.WMB {
-	t.sameLock.RLock()
-	defer t.sameLock.RUnlock()
-	t.diffLock.RLock()
-	defer t.diffLock.RUnlock()
-	if t.WMBTestSameHeadWMB {
-		t.sameCounter++
-		if t.sameCounter == 1 || t.sameCounter == 2 {
-			return wmb.WMB{
-				Idle:      true,
-				Offset:    100,
-				Watermark: 1636440000000,
-			}
-		}
-		if t.sameCounter == 3 || t.sameCounter == 4 {
-			return wmb.WMB{
-				Idle:      true,
-				Offset:    102,
-				Watermark: 1636480000000,
-			}
-		}
-	}
-	if t.WMBTestDiffHeadWMB {
-		t.diffCounter++
-		if t.diffCounter == 1 {
-			return wmb.WMB{
-				Idle:      true,
-				Offset:    100,
-				Watermark: 1636440000000,
-			}
-		}
-		if t.diffCounter == 2 {
-			return wmb.WMB{
-				Idle:      false,
-				Offset:    101,
-				Watermark: 1636450000000,
-			}
-		}
-	}
-	// won't be used
-	return wmb.WMB{}
-}
-
-func TestNewDataForwardIdleWatermark(t *testing.T) {
-	fromStep := simplebuffer.NewInMemoryBuffer("from", 25, 0, simplebuffer.WithReadTimeOut(time.Second)) // default read timeout is 1s
-	to1 := simplebuffer.NewInMemoryBuffer("to1", 10, 0)
-	toSteps := map[string][]isb.BufferWriter{
-		"to1": {to1},
-	}
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
-	defer cancel()
-
-	vertex := &dfv1.Vertex{Spec: dfv1.VertexSpec{
-		PipelineName: "testPipeline",
-		AbstractVertex: dfv1.AbstractVertex{
-			Name: "testVertex",
-		},
-	}}
-
-	ctrlMessage := []isb.Message{{Header: isb.Header{Kind: isb.WMB}}}
-	writeMessages := testutils.BuildTestWriteMessages(int64(20), testStartTime)
-
-	fetchWatermark := &testWMBFetcher{WMBTestSameHeadWMB: true}
-	publishWatermark, otStores := buildPublisherMapAndOTStore(toSteps)
-	f, err := NewDataForward(vertex, fromStep, toSteps, myForwardTest{}, myForwardTest{}, fetchWatermark, publishWatermark, TestSourceWatermarkPublisher{}, WithReadBatchSize(2))
-	assert.NoError(t, err)
-	assert.False(t, to1.IsFull())
-	assert.True(t, to1.IsEmpty())
-
-	stopped := f.Start()
-	assert.True(t, fromStep.IsEmpty())
-	// first batch: read message size is 1 with one ctrl message
-	// the ctrl message should be acked
-	// no message published to the next vertex
-	// so the timeline should be empty
-	_, errs := fromStep.Write(ctx, ctrlMessage)
-	assert.Equal(t, make([]error, 1), errs)
-	// waiting for the ctrl message to be acked
-	for !fromStep.IsEmpty() {
-		select {
-		case <-ctx.Done():
-			if ctx.Err() == context.DeadlineExceeded {
-				t.Fatal("expected the buffer to be empty", ctx.Err())
-			}
-		default:
-			time.Sleep(1 * time.Millisecond)
-		}
-	}
-	// it should not publish any wm because
-	// 1. readLength != 0
-	// 2. we only have one ctrl message
-	// 3. meaning dataMessage=0
-	otNil, _ := otStores["to1"].GetAllKeys(ctx)
-	assert.Nil(t, otNil)
-
-	// 2nd and 3rd batches: read message size is 0
-	// should send idle watermark
-	var wg sync.WaitGroup
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		otKeys1, _ := otStores["to1"].GetAllKeys(ctx)
-		for otKeys1 == nil {
-			otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
-			time.Sleep(time.Millisecond * 10)
-		}
-	}()
-	wg.Wait()
-
-	otKeys1, _ := otStores["to1"].GetAllKeys(ctx)
-	otValue1, _ := otStores["to1"].GetValue(ctx, otKeys1[0])
-	otDecode1, _ := wmb.DecodeToWMB(otValue1)
-	for otDecode1.Offset != 0 { // the first ctrl message written to isb. can't use idle because default idle=false
-		select {
-		case <-ctx.Done():
-			if ctx.Err() == context.DeadlineExceeded {
-				t.Fatal("expected to have idle watermark in to1 timeline", ctx.Err())
-			}
-		default:
-			time.Sleep(1 * time.Millisecond)
-			otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
-			otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
-			otDecode1, _ = wmb.DecodeToWMB(otValue1)
-		}
-	}
-	assert.Equal(t, wmb.WMB{
-		Idle:      true,
-		Offset:    0, // the first ctrl message written to isb
-		Watermark: 1636440000000,
-	}, otDecode1)
-	assert.Equal(t, isb.WMB, to1.GetMessages(1)[0].Kind)
-
-	// 4th batch: read message size = 1
-	// a new active watermark should be inserted
-	_, errs = fromStep.Write(ctx, writeMessages[:2])
-	assert.Equal(t, make([]error, 2), errs)
-
-	otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
-	otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
-	otDecode1, _ = wmb.DecodeToWMB(otValue1)
-	for otDecode1.Idle {
-		select {
-		case <-ctx.Done():
-			if ctx.Err() == context.DeadlineExceeded {
-				t.Fatal("expected to have active watermark in to1 timeline", ctx.Err())
-			}
-		default:
-			time.Sleep(1 * time.Millisecond)
-			otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
-			otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
-			otDecode1, _ = wmb.DecodeToWMB(otValue1)
-		}
-	}
-	assert.Equal(t, wmb.WMB{
-		Idle:      false,
-		Offset:    2, // the second message written to isb, read batch size is 2 so the offset is 0+2=2
-		Watermark: testWMBWatermark.UnixMilli(),
-	}, otDecode1)
-
-	// 5th & 6th batch: again idling but got diff head WMB
-	// so the head is still the same active watermark
-	// and no new ctrl message to the next vertex
-	f.wmFetcher.(*testWMBFetcher).RevertBoolValue()
-	time.Sleep(2 * time.Second) // default read timeout is 1s
-	otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
-	otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
-	otDecode1, _ = wmb.DecodeToWMB(otValue1)
-	assert.Equal(t, wmb.WMB{
-		Idle:      false,
-		Offset:    2, // the second message written to isb, read batch size is 2 so the offset is 0+2=2
-		Watermark: testWMBWatermark.UnixMilli(),
-	}, otDecode1)
-
-	var wantKind = []isb.MessageKind{
-		isb.WMB,
-		isb.Data,
-		isb.Data,
-		isb.Data, // this is data because it's the default value
-	}
-	var wantBody = []bool{
-		false,
-		true,
-		true,
-		false,
-	}
-	for idx, msg := range to1.GetMessages(4) {
-		assert.Equal(t, wantKind[idx], msg.Kind)
-		assert.Equal(t, wantBody[idx], len(msg.Body.Payload) > 0)
-	}
-
-	// stop will cancel the contexts and therefore the forwarder stops without waiting
-	f.Stop()
-
-	<-stopped
-}
-
-func TestNewDataForwardIdleWatermark_Reset(t *testing.T) {
-	fromStep := simplebuffer.NewInMemoryBuffer("from", 25, 0, simplebuffer.WithReadTimeOut(time.Second)) // default read timeout is 1s
-	to1 := simplebuffer.NewInMemoryBuffer("to1", 10, 0)
-	toSteps := map[string][]isb.BufferWriter{
-		"to1": {to1},
-	}
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
-	defer cancel()
-
-	vertex := &dfv1.Vertex{Spec: dfv1.VertexSpec{
-		PipelineName: "testPipeline",
-		AbstractVertex: dfv1.AbstractVertex{
-			Name: "testVertex",
-		},
-	}}
-
-	writeMessages := testutils.BuildTestWriteMessages(int64(20), testStartTime)
-
-	fetchWatermark := &testWMBFetcher{WMBTestSameHeadWMB: true}
-	publishWatermark, otStores := buildPublisherMapAndOTStore(toSteps)
-	f, err := NewDataForward(vertex, fromStep, toSteps, myForwardTest{}, myForwardTest{}, fetchWatermark, publishWatermark, TestSourceWatermarkPublisher{}, WithReadBatchSize(2))
-	assert.NoError(t, err)
-	assert.False(t, to1.IsFull())
-	assert.True(t, to1.IsEmpty())
-
-	stopped := f.Start()
-	var wg sync.WaitGroup
-
-	// 1st and 2nd batches: read message size is 0
-	// should send idle watermark
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		otKeys1, _ := otStores["to1"].GetAllKeys(ctx)
-		for otKeys1 == nil {
-			otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
-			time.Sleep(time.Millisecond * 10)
-		}
-	}()
-	wg.Wait()
-
-	otKeys1, _ := otStores["to1"].GetAllKeys(ctx)
-	otValue1, _ := otStores["to1"].GetValue(ctx, otKeys1[0])
-	otDecode1, _ := wmb.DecodeToWMB(otValue1)
-	for otDecode1.Offset != 0 { // the first ctrl message written to isb. can't use idle because default idle=false
-		select {
-		case <-ctx.Done():
-			if ctx.Err() == context.DeadlineExceeded {
-				t.Fatal("expected to have idle watermark in to1 timeline", ctx.Err())
-			}
-		default:
-			time.Sleep(1 * time.Millisecond)
-			otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
-			otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
-			otDecode1, _ = wmb.DecodeToWMB(otValue1)
-		}
-	}
-	assert.Equal(t, wmb.WMB{
-		Idle:      true,
-		Offset:    0, // the first ctrl message written to isb
-		Watermark: 1636440000000,
-	}, otDecode1)
-	assert.Equal(t, isb.WMB, to1.GetMessages(1)[0].Kind)
-
-	// 3rd batch: read message size = 1
-	// a new active watermark should be inserted
-	_, errs := fromStep.Write(ctx, writeMessages[:2])
-	assert.Equal(t, make([]error, 2), errs)
-
-	otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
-	otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
-	otDecode1, _ = wmb.DecodeToWMB(otValue1)
-	for otDecode1.Idle {
-		select {
-		case <-ctx.Done():
-			if ctx.Err() == context.DeadlineExceeded {
-				t.Fatal("expected to have active watermark in to1 timeline", ctx.Err())
-			}
-		default:
-			time.Sleep(1 * time.Millisecond)
-			otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
-			otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
-			otDecode1, _ = wmb.DecodeToWMB(otValue1)
-		}
-	}
-	assert.Equal(t, wmb.WMB{
-		Idle:      false,
-		Offset:    2, // the second message written to isb, read batch size is 2 so the offset is 0+2=2
-		Watermark: testWMBWatermark.UnixMilli(),
-	}, otDecode1)
-
-	// 5th & 6th batch: again idling should send a new ctrl message
-	otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
-	otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
-	otDecode1, _ = wmb.DecodeToWMB(otValue1)
-	for otDecode1.Offset != 3 { // the second ctrl message written to isb. can't use idle because default idle=false
-		select {
-		case <-ctx.Done():
-			if ctx.Err() == context.DeadlineExceeded {
-				t.Fatal("expected to have idle watermark in to1 timeline", ctx.Err())
-			}
-		default:
-			time.Sleep(1 * time.Millisecond)
-			otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
-			otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
-			otDecode1, _ = wmb.DecodeToWMB(otValue1)
-		}
-	}
-
-	var wantKind = []isb.MessageKind{
-		isb.WMB,
-		isb.Data,
-		isb.Data,
-		isb.WMB,
-	}
-	var wantBody = []bool{
-		false,
-		true,
-		true,
-		false,
-	}
-	for idx, msg := range to1.GetMessages(4) {
-		assert.Equal(t, wantKind[idx], msg.Kind)
-		assert.Equal(t, wantBody[idx], len(msg.Body.Payload) > 0)
-	}
-
-	// stop will cancel the contexts and therefore the forwarder stops without waiting
-	f.Stop()
-
-	<-stopped
-}
-
 // mySourceForwardTest tests source data transformer by updating message event time, and then verifying new event time and IsLate assignments.
 type mySourceForwardTest struct {
 }