Repository: numaproj/numaflow
Pull request #1709 (2024-04-30 18:56:14)

chore: bypass transformer call if UDTransformer is not defined
related to: #1612 

<!--

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.

-->


KeranYang: @ayildirim21, let's also think about how we can unit-test this change.
KeranYang: @ayildirim21 awesome job on collecting the perf data!
vigith: @yhl25, can you look into why idle-source for kafka is failing?

Patch:
diff --git a/pkg/sources/forward/applier/sourcetransformer.go b/pkg/sources/forward/applier/sourcetransformer.go
index bd814361..795cd4c5 100644
--- a/pkg/sources/forward/applier/sourcetransformer.go
+++ b/pkg/sources/forward/applier/sourcetransformer.go
@@ -35,12 +35,3 @@ type ApplySourceTransformFunc func(ctx context.Context, message *isb.ReadMessage
 func (f ApplySourceTransformFunc) ApplyTransform(ctx context.Context, message *isb.ReadMessage) ([]*isb.WriteMessage, error) {
 	return f(ctx, message)
 }
-
-var (
-	// Terminal Applier do not make any change to the message
-	Terminal = ApplySourceTransformFunc(func(ctx context.Context, msg *isb.ReadMessage) ([]*isb.WriteMessage, error) {
-		return []*isb.WriteMessage{{
-			Message: msg.Message,
-		}}, nil
-	})
-)
diff --git a/pkg/sources/forward/data_forward.go b/pkg/sources/forward/data_forward.go
index f3e3f5c7..328941db 100644
--- a/pkg/sources/forward/data_forward.go
+++ b/pkg/sources/forward/data_forward.go
@@ -32,7 +32,6 @@ import (
 	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/shared/idlehandler"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
-	"github.com/numaproj/numaflow/pkg/sources/forward/applier"
 	"github.com/numaproj/numaflow/pkg/sources/sourcer"
 	"github.com/numaproj/numaflow/pkg/watermark/entity"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
@@ -51,7 +50,6 @@ type DataForward struct {
 	reader               sourcer.SourceReader          // reader reads data from source.
 	toBuffers            map[string][]isb.BufferWriter // toBuffers store the toVertex name to its owned buffers mapping.
 	toWhichStepDecider   forwarder.ToWhichStepDecider
-	transformer          applier.SourceTransformApplier
 	wmFetcher            fetch.SourceFetcher
 	toVertexWMStores     map[string]store.WatermarkStore
 	toVertexWMPublishers map[string]map[int32]publish.Publisher // toVertexWMPublishers stores the toVertex to publisher mapping.
@@ -72,7 +70,6 @@ func NewDataForward(
 	reader sourcer.SourceReader,
 	toSteps map[string][]isb.BufferWriter,
 	toWhichStepDecider forwarder.ToWhichStepDecider,
-	transformer applier.SourceTransformApplier,
 	fetchWatermark fetch.SourceFetcher,
 	srcWMPublisher publish.SourcePublisher,
 	toVertexWmStores map[string]store.WatermarkStore,
@@ -101,7 +98,6 @@ func NewDataForward(
 		reader:               reader,
 		toBuffers:            toSteps,
 		toWhichStepDecider:   toWhichStepDecider,
-		transformer:          transformer,
 		wmFetcher:            fetchWatermark,
 		toVertexWMStores:     toVertexWmStores,
 		toVertexWMPublishers: toVertexWMPublishers,
@@ -164,7 +160,10 @@ func (df *DataForward) Start() <-chan struct{} {
 		for _, buffer := range df.toBuffers {
 			for _, partition := range buffer {
 				if err := partition.Close(); err != nil {
-					log.Errorw("Failed to close partition writer, shutdown anyways...", zap.Error(err), zap.String("bufferTo", partition.GetName()))
+					log.Errorw("Failed to close partition writer, shutdown anyways...",
+						zap.Error(err),
+						zap.String("bufferTo", partition.GetName()),
+					)
 				} else {
 					log.Infow("Closed partition writer", zap.String("bufferTo", partition.GetName()))
 				}
@@ -204,7 +203,13 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 	readMessages, err := df.reader.Read(ctx, df.opts.readBatchSize)
 	if err != nil {
 		df.opts.logger.Warnw("failed to read from source", zap.Error(err))
-		metrics.ReadMessagesError.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexType: string(dfv1.VertexTypeSource), metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)), metrics.LabelPartitionName: df.reader.GetName()}).Inc()
+		metrics.ReadMessagesError.With(map[string]string{
+			metrics.LabelVertex:             df.vertexName,
+			metrics.LabelPipeline:           df.pipelineName,
+			metrics.LabelVertexType:         string(dfv1.VertexTypeSource),
+			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+			metrics.LabelPartitionName:      df.reader.GetName(),
+		}).Inc()
 	}
 
 	// if there are no read messages, we return early.
@@ -236,7 +241,8 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 							publisher = df.createToVertexWatermarkPublisher(toVertexName, sp)
 							vertexPublishers[sp] = publisher
 						}
-						idlehandler.PublishIdleWatermark(ctx, wmb.PARTITION_0, df.toBuffers[toVertexName][index], publisher, df.idleManager, df.opts.logger, df.vertexName, df.pipelineName, dfv1.VertexTypeSource, df.vertexReplica, fetchedWm)
+						idlehandler.PublishIdleWatermark(ctx, wmb.PARTITION_0, df.toBuffers[toVertexName][index], publisher,
+							df.idleManager, df.opts.logger, df.vertexName, df.pipelineName, dfv1.VertexTypeSource, df.vertexReplica, fetchedWm)
 					}
 				}
 			}
@@ -248,9 +254,21 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 
 	// reset the idle handler because we have read messages
 	df.srcIdleHandler.Reset()
-
-	metrics.ReadDataMessagesCount.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexType: string(dfv1.VertexTypeSource), metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)), metrics.LabelPartitionName: df.reader.GetName()}).Add(float64(len(readMessages)))
-	metrics.ReadMessagesCount.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexType: string(dfv1.VertexTypeSource), metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)), metrics.LabelPartitionName: df.reader.GetName()}).Add(float64(len(readMessages)))
+	metrics.ReadDataMessagesCount.With(map[string]string{
+		metrics.LabelVertex:             df.vertexName,
+		metrics.LabelPipeline:           df.pipelineName,
+		metrics.LabelVertexType:         string(dfv1.VertexTypeSource),
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+		metrics.LabelPartitionName:      df.reader.GetName()},
+	).Add(float64(len(readMessages)))
+
+	metrics.ReadMessagesCount.With(map[string]string{
+		metrics.LabelVertex:             df.vertexName,
+		metrics.LabelPipeline:           df.pipelineName,
+		metrics.LabelVertexType:         string(dfv1.VertexTypeSource),
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+		metrics.LabelPartitionName:      df.reader.GetName(),
+	}).Add(float64(len(readMessages)))
 
 	// store the offsets of the messages we read from source
 	var readOffsets = make([]isb.Offset, len(readMessages))
@@ -270,51 +288,80 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 		messageToStep[toVertex] = make([][]isb.Message, len(df.toBuffers[toVertex]))
 	}
 
-	// FIXME: when the transformer is not defined, we should avoid doing this.
-	// user-defined transformer concurrent processing request channel
-	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(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
-	var wg sync.WaitGroup
-	for i := 0; i < df.opts.transformerConcurrency; i++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			df.concurrentApplyTransformer(ctx, transformerCh)
-		}()
-	}
-	concurrentTransformerProcessingStart := time.Now()
-
-	// send to transformer only the data messages
-	for idx, m := range readMessages {
-		// emit message size metric
-		metrics.ReadBytesCount.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexType: string(dfv1.VertexTypeSource), metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)), metrics.LabelPartitionName: df.reader.GetName()}).Add(float64(len(m.Payload)))
-		// assign watermark to the message
-		m.Watermark = time.Time(processorWM)
-		// send transformer processing work to the channel
-		transformerResults[idx].readMessage = m
-		transformerCh <- &transformerResults[idx]
+	readWriteMessagePairs := make([]readWriteMessagePair, len(readMessages))
+
+	// If a user-defined transformer exists, apply it
+	if df.opts.transformer != nil {
+		// user-defined transformer concurrent processing request channel
+		transformerCh := make(chan *readWriteMessagePair)
+
+		// create a pool of Transformer Processors
+		var wg sync.WaitGroup
+		for i := 0; i < df.opts.transformerConcurrency; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				df.concurrentApplyTransformer(ctx, transformerCh)
+			}()
+		}
+		concurrentTransformerProcessingStart := time.Now()
+		for idx, m := range readMessages {
+			metrics.ReadBytesCount.With(map[string]string{
+				metrics.LabelVertex:             df.vertexName,
+				metrics.LabelPipeline:           df.pipelineName,
+				metrics.LabelVertexType:         string(dfv1.VertexTypeSource),
+				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+				metrics.LabelPartitionName:      df.reader.GetName(),
+			}).Add(float64(len(m.Payload)))
+
+			// assign watermark to the message
+			m.Watermark = time.Time(processorWM)
+			readWriteMessagePairs[idx].readMessage = m
+			// send transformer processing work to the channel. Thus, the results of the transformer
+			// application on a read message will be stored as the corresponding writeMessage in readWriteMessagePairs
+			transformerCh <- &readWriteMessagePairs[idx]
+		}
+		// let the go routines know that there is no more work
+		close(transformerCh)
+		// wait till the processing is done. this will not be an infinite wait because the transformer processing will exit if
+		// context.Done() is closed.
+		wg.Wait()
+		df.opts.logger.Debugw("concurrent applyTransformer completed",
+			zap.Int("concurrency", df.opts.transformerConcurrency),
+			zap.Duration("took", time.Since(concurrentTransformerProcessingStart)),
+		)
+
+		metrics.SourceTransformerConcurrentProcessingTime.With(map[string]string{
+			metrics.LabelVertex:             df.vertexName,
+			metrics.LabelPipeline:           df.pipelineName,
+			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+			metrics.LabelPartitionName:      df.reader.GetName(),
+		}).Observe(float64(time.Since(concurrentTransformerProcessingStart).Microseconds()))
+	} else {
+		for idx, m := range readMessages {
+			metrics.ReadBytesCount.With(map[string]string{
+				metrics.LabelVertex:             df.vertexName,
+				metrics.LabelPipeline:           df.pipelineName,
+				metrics.LabelVertexType:         string(dfv1.VertexTypeSource),
+				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+				metrics.LabelPartitionName:      df.reader.GetName(),
+			}).Add(float64(len(m.Payload)))
+
+			// assign watermark to the message
+			m.Watermark = time.Time(processorWM)
+			readWriteMessagePairs[idx].readMessage = m
+			// if no user-defined transformer exists, then the messages to write will be identical to the message read from source
+			// thus, the unmodified read message will be stored as the corresponding writeMessage in readWriteMessagePairs
+			readWriteMessagePairs[idx].writeMessages = []*isb.WriteMessage{{Message: m.Message}}
+		}
 	}
-	// let the go routines know that there is no more work
-	close(transformerCh)
-	// wait till the processing is done. this will not be an infinite wait because the transformer processing will exit if
-	// context.Done() is closed.
-	wg.Wait()
-	df.opts.logger.Debugw("concurrent applyTransformer completed", zap.Int("concurrency", df.opts.transformerConcurrency), zap.Duration("took", time.Since(concurrentTransformerProcessingStart)))
-	metrics.SourceTransformerConcurrentProcessingTime.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)), metrics.LabelPartitionName: df.reader.GetName()}).Observe(float64(time.Since(concurrentTransformerProcessingStart).Microseconds()))
-	// transformer processing is done.
 
 	// publish source watermark and assign IsLate attribute based on new event time.
 	var writeMessages []*isb.WriteMessage
 	var transformedReadMessages []*isb.ReadMessage
 	latestEtMap := make(map[int32]int64)
 
-	for _, m := range transformerResults {
+	for _, m := range readWriteMessagePairs {
 		writeMessages = append(writeMessages, m.writeMessages...)
 		for _, message := range m.writeMessages {
 			message.Headers = m.readMessage.Headers
@@ -345,11 +392,17 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 	var sourcePartitionsIndices = make(map[int32]bool)
 	// let's figure out which vertex to send the results to.
 	// update the toBuffer(s) with writeMessages.
-	for _, m := range transformerResults {
+	for _, m := range readWriteMessagePairs {
 		// Look for errors in transformer processing if we see even 1 error we return.
 		// Handling partial retrying is not worth ATM.
 		if m.transformerError != nil {
-			metrics.SourceTransformerError.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)), metrics.LabelPartitionName: df.reader.GetName()}).Inc()
+			metrics.SourceTransformerError.With(map[string]string{
+				metrics.LabelVertex:             df.vertexName,
+				metrics.LabelPipeline:           df.pipelineName,
+				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+				metrics.LabelPartitionName:      df.reader.GetName(),
+			}).Inc()
+
 			df.opts.logger.Errorw("failed to apply source transformer", zap.Error(m.transformerError))
 			return
 		}
@@ -418,7 +471,8 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 							publisher = df.createToVertexWatermarkPublisher(toVertexName, sp)
 							vertexPublishers[sp] = publisher
 						}
-						idlehandler.PublishIdleWatermark(ctx, wmb.PARTITION_0, df.toBuffers[toVertexName][index], publisher, df.idleManager, df.opts.logger, df.vertexName, df.pipelineName, dfv1.VertexTypeSource, df.vertexReplica, processorWM)
+						idlehandler.PublishIdleWatermark(ctx, wmb.PARTITION_0, df.toBuffers[toVertexName][index], publisher,
+							df.idleManager, df.opts.logger, df.vertexName, df.pipelineName, dfv1.VertexTypeSource, df.vertexReplica, processorWM)
 					}
 				}
 			}
@@ -431,13 +485,31 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 	// implicit return for posterity :-)
 	if err != nil {
 		df.opts.logger.Errorw("failed to ack from source", zap.Error(err))
-		metrics.AckMessageError.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexType: string(dfv1.VertexTypeSource), metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)), metrics.LabelPartitionName: df.reader.GetName()}).Add(float64(len(readOffsets)))
+		metrics.AckMessageError.With(map[string]string{
+			metrics.LabelVertex:             df.vertexName,
+			metrics.LabelPipeline:           df.pipelineName,
+			metrics.LabelVertexType:         string(dfv1.VertexTypeSource),
+			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+			metrics.LabelPartitionName:      df.reader.GetName(),
+		}).Add(float64(len(readOffsets)))
+
 		return
 	}
-	metrics.AckMessagesCount.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexType: string(dfv1.VertexTypeSource), metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)), metrics.LabelPartitionName: df.reader.GetName()}).Add(float64(len(readOffsets)))
+	metrics.AckMessagesCount.With(map[string]string{
+		metrics.LabelVertex:             df.vertexName,
+		metrics.LabelPipeline:           df.pipelineName,
+		metrics.LabelVertexType:         string(dfv1.VertexTypeSource),
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+		metrics.LabelPartitionName:      df.reader.GetName(),
+	}).Add(float64(len(readOffsets)))
 
 	// ProcessingTimes of the entire forwardAChunk
-	metrics.ForwardAChunkProcessingTime.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexType: string(dfv1.VertexTypeSource), metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica))}).Observe(float64(time.Since(start).Microseconds()))
+	metrics.ForwardAChunkProcessingTime.With(map[string]string{
+		metrics.LabelVertex:             df.vertexName,
+		metrics.LabelPipeline:           df.pipelineName,
+		metrics.LabelVertexType:         string(dfv1.VertexTypeSource),
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+	}).Observe(float64(time.Since(start).Microseconds()))
 }
 
 func (df *DataForward) ackFromSource(ctx context.Context, offsets []isb.Offset) error {
@@ -509,15 +581,32 @@ func (df *DataForward) writeToBuffer(ctx context.Context, toBufferPartition isb.
 						metrics.LabelReason:             err.Error(),
 					}).Add(float64(len(msg.Payload)))
 
-					df.opts.logger.Infow("Dropped message", zap.String("reason", err.Error()), zap.String("partition", toBufferPartition.GetName()), zap.String("vertex", df.vertexName), zap.String("pipeline", df.pipelineName))
+					df.opts.logger.Infow("Dropped message",
+						zap.String("reason", err.Error()),
+						zap.String("partition", toBufferPartition.GetName()),
+						zap.String("vertex", df.vertexName), zap.String("pipeline", df.pipelineName),
+					)
 				} else {
 					needRetry = true
 					// we retry only failed messages
 					failedMessages = append(failedMessages, msg)
-					metrics.WriteMessagesError.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexType: string(dfv1.VertexTypeSource), metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)), metrics.LabelPartitionName: toBufferPartition.GetName()}).Inc()
+					metrics.WriteMessagesError.With(map[string]string{
+						metrics.LabelVertex:             df.vertexName,
+						metrics.LabelPipeline:           df.pipelineName,
+						metrics.LabelVertexType:         string(dfv1.VertexTypeSource),
+						metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+						metrics.LabelPartitionName:      toBufferPartition.GetName(),
+					}).Inc()
+
 					// a shutdown can break the blocking loop caused due to InternalErr
 					if ok, _ := df.IsShuttingDown(); ok {
-						metrics.PlatformError.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexType: string(dfv1.VertexTypeSource), metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica))}).Inc()
+						metrics.PlatformError.With(map[string]string{
+							metrics.LabelVertex:             df.vertexName,
+							metrics.LabelPipeline:           df.pipelineName,
+							metrics.LabelVertexType:         string(dfv1.VertexTypeSource),
+							metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+						}).Inc()
+
 						return writeOffsets, fmt.Errorf("writeToBuffer failed, Stop called while stuck on an internal error with failed messages:%d, %v", len(failedMessages), errs)
 					}
 				}
@@ -547,8 +636,22 @@ func (df *DataForward) writeToBuffer(ctx context.Context, toBufferPartition isb.
 		}
 	}
 
-	metrics.WriteMessagesCount.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexType: string(dfv1.VertexTypeSource), metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)), metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(float64(writeCount))
-	metrics.WriteBytesCount.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexType: string(dfv1.VertexTypeSource), metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)), metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(writeBytes)
+	metrics.WriteMessagesCount.With(map[string]string{
+		metrics.LabelVertex:             df.vertexName,
+		metrics.LabelPipeline:           df.pipelineName,
+		metrics.LabelVertexType:         string(dfv1.VertexTypeSource),
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+		metrics.LabelPartitionName:      toBufferPartition.GetName(),
+	}).Add(float64(writeCount))
+
+	metrics.WriteBytesCount.With(map[string]string{
+		metrics.LabelVertex:             df.vertexName,
+		metrics.LabelPipeline:           df.pipelineName,
+		metrics.LabelVertexType:         string(dfv1.VertexTypeSource),
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+		metrics.LabelPartitionName:      toBufferPartition.GetName(),
+	}).Add(writeBytes)
+
 	return writeOffsets, nil
 }
 
@@ -556,12 +659,29 @@ func (df *DataForward) writeToBuffer(ctx context.Context, toBufferPartition isb.
 func (df *DataForward) concurrentApplyTransformer(ctx context.Context, readMessagePair <-chan *readWriteMessagePair) {
 	for message := range readMessagePair {
 		start := time.Now()
-		metrics.SourceTransformerReadMessagesCount.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)), metrics.LabelPartitionName: df.reader.GetName()}).Inc()
+		metrics.SourceTransformerReadMessagesCount.With(map[string]string{
+			metrics.LabelVertex:             df.vertexName,
+			metrics.LabelPipeline:           df.pipelineName,
+			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+			metrics.LabelPartitionName:      df.reader.GetName(),
+		}).Inc()
+
 		writeMessages, err := df.applyTransformer(ctx, message.readMessage)
-		metrics.SourceTransformerWriteMessagesCount.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)), metrics.LabelPartitionName: df.reader.GetName()}).Add(float64(len(writeMessages)))
+		metrics.SourceTransformerWriteMessagesCount.With(map[string]string{
+			metrics.LabelVertex:             df.vertexName,
+			metrics.LabelPipeline:           df.pipelineName,
+			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+			metrics.LabelPartitionName:      df.reader.GetName(),
+		}).Add(float64(len(writeMessages)))
+
 		message.writeMessages = append(message.writeMessages, writeMessages...)
 		message.transformerError = err
-		metrics.SourceTransformerProcessingTime.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)), metrics.LabelPartitionName: df.reader.GetName()}).Observe(float64(time.Since(start).Microseconds()))
+		metrics.SourceTransformerProcessingTime.With(map[string]string{
+			metrics.LabelVertex:             df.vertexName,
+			metrics.LabelPipeline:           df.pipelineName,
+			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+			metrics.LabelPartitionName:      df.reader.GetName(),
+		}).Observe(float64(time.Since(start).Microseconds()))
 	}
 }
 
@@ -570,7 +690,7 @@ func (df *DataForward) concurrentApplyTransformer(ctx context.Context, readMessa
 // The UserError retry will be done on the applyTransformer.
 func (df *DataForward) applyTransformer(ctx context.Context, readMessage *isb.ReadMessage) ([]*isb.WriteMessage, error) {
 	for {
-		writeMessages, err := df.transformer.ApplyTransform(ctx, readMessage)
+		writeMessages, err := df.opts.transformer.ApplyTransform(ctx, readMessage)
 		if err != nil {
 			df.opts.logger.Errorw("Transformer.Apply error", zap.Error(err))
 			// TODO: implement retry with backoff etc.
@@ -580,7 +700,13 @@ func (df *DataForward) applyTransformer(ctx context.Context, readMessage *isb.Re
 			// this does not mean we should prohibit this from a shutdown.
 			if ok, _ := df.IsShuttingDown(); ok {
 				df.opts.logger.Errorw("Transformer.Apply, Stop called while stuck on an internal error", zap.Error(err))
-				metrics.PlatformError.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexType: string(dfv1.VertexTypeSource), metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica))}).Inc()
+				metrics.PlatformError.With(map[string]string{
+					metrics.LabelVertex:             df.vertexName,
+					metrics.LabelPipeline:           df.pipelineName,
+					metrics.LabelVertexType:         string(dfv1.VertexTypeSource),
+					metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+				}).Inc()
+
 				return nil, err
 			}
 			continue
@@ -598,11 +724,23 @@ func (df *DataForward) whereToStep(writeMessage *isb.WriteMessage, messageToStep
 	// call WhereTo and drop it on errors
 	to, err := df.toWhichStepDecider.WhereTo(writeMessage.Keys, writeMessage.Tags, writeMessage.ID)
 	if err != nil {
-		df.opts.logger.Errorw("failed in whereToStep", zap.Error(isb.MessageWriteErr{Name: df.reader.GetName(), Header: readMessage.Header, Body: readMessage.Body, Message: fmt.Sprintf("WhereTo failed, %s", err)}))
+		df.opts.logger.Errorw("failed in whereToStep", zap.Error(isb.MessageWriteErr{
+			Name:    df.reader.GetName(),
+			Header:  readMessage.Header,
+			Body:    readMessage.Body,
+			Message: fmt.Sprintf("WhereTo failed, %s", err),
+		}))
+
 		// a shutdown can break the blocking loop caused due to InternalErr
 		if ok, _ := df.IsShuttingDown(); ok {
 			err := fmt.Errorf("whereToStep, Stop called while stuck on an internal error, %v", err)
-			metrics.PlatformError.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelVertexType: string(dfv1.VertexTypeSource), metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica))}).Inc()
+			metrics.PlatformError.With(map[string]string{
+				metrics.LabelVertex:             df.vertexName,
+				metrics.LabelPipeline:           df.pipelineName,
+				metrics.LabelVertexType:         string(dfv1.VertexTypeSource),
+				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+			}).Inc()
+
 			return err
 		}
 		return err
@@ -610,7 +748,12 @@ func (df *DataForward) whereToStep(writeMessage *isb.WriteMessage, messageToStep
 
 	for _, t := range to {
 		if _, ok := messageToStep[t.ToVertexName]; !ok {
-			df.opts.logger.Errorw("failed in whereToStep", zap.Error(isb.MessageWriteErr{Name: df.reader.GetName(), Header: readMessage.Header, Body: readMessage.Body, Message: fmt.Sprintf("no such destination (%s)", t.ToVertexName)}))
+			df.opts.logger.Errorw("failed in whereToStep", zap.Error(isb.MessageWriteErr{
+				Name:    df.reader.GetName(),
+				Header:  readMessage.Header,
+				Body:    readMessage.Body,
+				Message: fmt.Sprintf("no such destination (%s)", t.ToVertexName),
+			}))
 		}
 		messageToStep[t.ToVertexName][t.ToVertexPartitionIdx] = append(messageToStep[t.ToVertexName][t.ToVertexPartitionIdx], writeMessage.Message)
 	}
diff --git a/pkg/sources/forward/options.go b/pkg/sources/forward/options.go
index 5360b677..1195ff4c 100644
--- a/pkg/sources/forward/options.go
+++ b/pkg/sources/forward/options.go
@@ -23,12 +23,15 @@ import (
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
+	"github.com/numaproj/numaflow/pkg/sources/forward/applier"
 )
 
 // options for forwarding the message
 type options struct {
 	// readBatchSize is the default batch size
 	readBatchSize int64
+	// transformer defines the way to transform data at source
+	transformer applier.SourceTransformApplier
 	// transformerConcurrency sets the concurrency for concurrent transformer processing
 	transformerConcurrency int
 	// retryInterval is the time.Duration to sleep before retrying
@@ -42,6 +45,7 @@ type Option func(*options) error
 func defaultOptions() *options {
 	return &options{
 		readBatchSize:          dfv1.DefaultReadBatchSize,
+		transformer:            nil,
 		transformerConcurrency: dfv1.DefaultReadBatchSize,
 		retryInterval:          time.Millisecond,
 		logger:                 logging.NewLogger(),
@@ -79,3 +83,11 @@ func WithLogger(l *zap.SugaredLogger) Option {
 		return nil
 	}
 }
+
+// WithTransformer sets the transformer to be applied
+func WithTransformer(f applier.SourceTransformApplier) Option {
+	return func(o *options) error {
+		o.transformer = f
+		return nil
+	}
+}
diff --git a/pkg/sources/source.go b/pkg/sources/source.go
index 682a7e3f..26fba228 100644
--- a/pkg/sources/source.go
+++ b/pkg/sources/source.go
@@ -41,7 +41,6 @@ import (
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"github.com/numaproj/numaflow/pkg/shuffle"
 	sourceforward "github.com/numaproj/numaflow/pkg/sources/forward"
-	"github.com/numaproj/numaflow/pkg/sources/forward/applier"
 	"github.com/numaproj/numaflow/pkg/sources/generator"
 	"github.com/numaproj/numaflow/pkg/sources/http"
 	"github.com/numaproj/numaflow/pkg/sources/kafka"
@@ -147,7 +146,7 @@ func (sp *SourceProcessor) Start(ctx context.Context) error {
 		}
 
 		// created watermark related components only if watermark is enabled
-		// otherwise no op will used
+		// otherwise no op will be used
 		if !sp.VertexInstance.Vertex.Spec.Watermark.Disabled {
 			// build watermark stores for from vertex
 			sourceWmStores, err = jetstream.BuildFromVertexWatermarkStores(ctx, sp.VertexInstance, natsClientPool.NextAvailableClient())
@@ -222,6 +221,14 @@ func (sp *SourceProcessor) Start(ctx context.Context) error {
 		healthCheckers = append(healthCheckers, udsGRPCClient)
 	}
 
+	var forwardOpts []sourceforward.Option
+
+	if x := sp.VertexInstance.Vertex.Spec.Limits; x != nil {
+		if x.ReadBatchSize != nil {
+			forwardOpts = append(forwardOpts, sourceforward.WithReadBatchSize(int64(*x.ReadBatchSize)))
+		}
+	}
+
 	if sp.VertexInstance.Vertex.HasUDTransformer() {
 		// Wait for server info to be ready
 		serverInfo, err := sdkserverinfo.SDKServerInfo(sdkserverinfo.WithServerInfoFilePath(sdkclient.SourceTransformerServerInfoFile))
@@ -250,6 +257,7 @@ func (sp *SourceProcessor) Start(ctx context.Context) error {
 		}
 
 		healthCheckers = append(healthCheckers, srcTransformerGRPCClient)
+		forwardOpts = append(forwardOpts, sourceforward.WithTransformer(srcTransformerGRPCClient))
 	}
 
 	sourceReader, err := sp.createSourceReader(ctx, udsGRPCClient)
@@ -259,19 +267,13 @@ func (sp *SourceProcessor) Start(ctx context.Context) error {
 
 	// create a source watermark publisher
 	sourceWmPublisher := publish.NewSourcePublish(ctx, pipelineName, vertexName, sourcePublisherStores, publish.WithDelay(sp.VertexInstance.Vertex.Spec.Watermark.GetMaxDelay()))
-	var forwardOpts []sourceforward.Option
-	if x := sp.VertexInstance.Vertex.Spec.Limits; x != nil {
-		if x.ReadBatchSize != nil {
-			forwardOpts = append(forwardOpts, sourceforward.WithReadBatchSize(int64(*x.ReadBatchSize)))
-		}
-	}
 
 	// create source data forwarder
 	var sourceForwarder *sourceforward.DataForward
 	if sp.VertexInstance.Vertex.HasUDTransformer() {
-		sourceForwarder, err = sourceforward.NewDataForward(sp.VertexInstance, sourceReader, writersMap, sp.getTransformerGoWhereDecider(shuffleFuncMap), srcTransformerGRPCClient, fetchWatermark, sourceWmPublisher, toVertexWatermarkStores, idleManager, forwardOpts...)
+		sourceForwarder, err = sourceforward.NewDataForward(sp.VertexInstance, sourceReader, writersMap, sp.getTransformerGoWhereDecider(shuffleFuncMap), fetchWatermark, sourceWmPublisher, toVertexWatermarkStores, idleManager, forwardOpts...)
 	} else {
-		sourceForwarder, err = sourceforward.NewDataForward(sp.VertexInstance, sourceReader, writersMap, sp.getSourceGoWhereDecider(shuffleFuncMap), applier.Terminal, fetchWatermark, sourceWmPublisher, toVertexWatermarkStores, idleManager, forwardOpts...)
+		sourceForwarder, err = sourceforward.NewDataForward(sp.VertexInstance, sourceReader, writersMap, sp.getSourceGoWhereDecider(shuffleFuncMap), fetchWatermark, sourceWmPublisher, toVertexWatermarkStores, idleManager, forwardOpts...)
 	}
 	if err != nil {
 		return fmt.Errorf("failed to create source forwarder, error: %w", err)


Test patch:
diff --git a/pkg/sources/forward/data_forward_test.go b/pkg/sources/forward/data_forward_test.go
index 55254a78..ad4ad596 100644
--- a/pkg/sources/forward/data_forward_test.go
+++ b/pkg/sources/forward/data_forward_test.go
@@ -165,7 +165,7 @@ func TestNewDataForward(t *testing.T) {
 			fetchWatermark, _ := generic.BuildNoOpSourceWatermarkProgressorsFromBufferMap(toSteps)
 			noOpStores := buildNoOpToVertexStores(toSteps)
 			idleManager, _ := wmb.NewIdleManager(1, len(toSteps))
-			f, err := NewDataForward(vertexInstance, fromStep, toSteps, &mySourceForwardTestRoundRobin{}, myForwardTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, noOpStores, idleManager, WithReadBatchSize(batchSize))
+			f, err := NewDataForward(vertexInstance, fromStep, toSteps, &mySourceForwardTestRoundRobin{}, fetchWatermark, TestSourceWatermarkPublisher{}, noOpStores, idleManager, WithReadBatchSize(batchSize), WithTransformer(myForwardTest{}))
 
 			assert.NoError(t, err)
 			assert.False(t, to11.IsFull())
@@ -265,7 +265,7 @@ func TestNewDataForward(t *testing.T) {
 			toVertexStores := buildToVertexWatermarkStores(toSteps)
 
 			idleManager, _ := wmb.NewIdleManager(1, len(toSteps))
-			f, err := NewDataForward(vertexInstance, fromStep, toSteps, &myForwardToAllTest{}, &myForwardToAllTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(batchSize))
+			f, err := NewDataForward(vertexInstance, fromStep, toSteps, &myForwardToAllTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(batchSize), WithTransformer(&myForwardToAllTest{}))
 
 			assert.NoError(t, err)
 			assert.False(t, to11.IsFull())
@@ -418,7 +418,7 @@ func TestNewDataForward(t *testing.T) {
 			toVertexStores := buildToVertexWatermarkStores(toSteps)
 
 			idleManager, _ := wmb.NewIdleManager(1, len(toSteps))
-			f, err := NewDataForward(vertexInstance, fromStep, toSteps, myForwardDropTest{}, myForwardDropTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(batchSize))
+			f, err := NewDataForward(vertexInstance, fromStep, toSteps, myForwardDropTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(batchSize), WithTransformer(myForwardDropTest{}))
 
 			assert.NoError(t, err)
 			assert.False(t, to11.IsFull())
@@ -583,7 +583,7 @@ func TestNewDataForward(t *testing.T) {
 			toVertexStores := buildToVertexWatermarkStores(toSteps)
 
 			idleManager, _ := wmb.NewIdleManager(1, len(toSteps))
-			f, err := NewDataForward(vertexInstance, fromStep, toSteps, &mySourceForwardTestRoundRobin{}, myForwardTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(batchSize))
+			f, err := NewDataForward(vertexInstance, fromStep, toSteps, &mySourceForwardTestRoundRobin{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(batchSize), WithTransformer(myForwardTest{}))
 
 			assert.NoError(t, err)
 			assert.False(t, to11.IsFull())
@@ -720,7 +720,7 @@ func TestNewDataForward(t *testing.T) {
 			toVertexStores := buildNoOpToVertexStores(toSteps)
 
 			idleManager, _ := wmb.NewIdleManager(1, len(toSteps))
-			f, err := NewDataForward(vertexInstance, fromStep, toSteps, myForwardApplyTransformerErrTest{}, myForwardApplyTransformerErrTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(batchSize))
+			f, err := NewDataForward(vertexInstance, fromStep, toSteps, myForwardApplyTransformerErrTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(batchSize), WithTransformer(myForwardApplyTransformerErrTest{}))
 
 			assert.NoError(t, err)
 			assert.False(t, to1.IsFull())
@@ -766,7 +766,7 @@ func TestNewDataForward(t *testing.T) {
 			toVertexStores := buildNoOpToVertexStores(toSteps)
 
 			idleManager, _ := wmb.NewIdleManager(1, len(toSteps))
-			f, err := NewDataForward(vertexInstance, fromStep, toSteps, myForwardApplyWhereToErrTest{}, myForwardApplyWhereToErrTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(batchSize))
+			f, err := NewDataForward(vertexInstance, fromStep, toSteps, myForwardApplyWhereToErrTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(batchSize), WithTransformer(myForwardApplyWhereToErrTest{}))
 
 			assert.NoError(t, err)
 			assert.True(t, to1.IsEmpty())
@@ -810,7 +810,7 @@ func TestNewDataForward(t *testing.T) {
 			toVertexStores := buildNoOpToVertexStores(toSteps)
 
 			idleManager, _ := wmb.NewIdleManager(1, len(toSteps))
-			f, err := NewDataForward(vertexInstance, fromStep, toSteps, myForwardInternalErrTest{}, myForwardInternalErrTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(batchSize))
+			f, err := NewDataForward(vertexInstance, fromStep, toSteps, myForwardInternalErrTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(batchSize), WithTransformer(myForwardInternalErrTest{}))
 
 			assert.NoError(t, err)
 			assert.False(t, to1.IsFull())
@@ -921,7 +921,7 @@ func TestDataForwardSinglePartition(t *testing.T) {
 	toVertexStores := buildNoOpToVertexStores(toSteps)
 
 	idleManager, _ := wmb.NewIdleManager(1, len(toSteps))
-	f, err := NewDataForward(vertexInstance, fromStep, toSteps, mySourceForwardTest{}, mySourceForwardTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(5))
+	f, err := NewDataForward(vertexInstance, fromStep, toSteps, mySourceForwardTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(5), WithTransformer(mySourceForwardTest{}))
 	assert.NoError(t, err)
 	assert.False(t, to1.IsFull())
 	assert.True(t, to1.IsEmpty())
@@ -976,7 +976,7 @@ func TestDataForwardMultiplePartition(t *testing.T) {
 	toVertexStores := buildNoOpToVertexStores(toSteps)
 
 	idleManager, _ := wmb.NewIdleManager(1, len(toSteps))
-	f, err := NewDataForward(vertexInstance, fromStep, toSteps, &mySourceForwardTestRoundRobin{}, mySourceForwardTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(5))
+	f, err := NewDataForward(vertexInstance, fromStep, toSteps, &mySourceForwardTestRoundRobin{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(5), WithTransformer(mySourceForwardTest{}))
 	assert.NoError(t, err)
 	assert.False(t, to11.IsFull())
 	assert.False(t, to12.IsFull())
@@ -1085,7 +1085,7 @@ func TestWriteToBuffer(t *testing.T) {
 			toVertexStores := buildNoOpToVertexStores(toSteps)
 
 			idleManager, _ := wmb.NewIdleManager(1, len(toSteps))
-			f, err := NewDataForward(vertexInstance, fromStep, toSteps, myForwardTest{}, myForwardTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(value.batchSize))
+			f, err := NewDataForward(vertexInstance, fromStep, toSteps, myForwardTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexStores, idleManager, WithReadBatchSize(value.batchSize), WithTransformer(myForwardTest{}))
 			assert.NoError(t, err)
 			assert.False(t, buffer.IsFull())
 			assert.True(t, buffer.IsEmpty())
diff --git a/pkg/sources/forward/shutdown_test.go b/pkg/sources/forward/shutdown_test.go
index 54210b4f..0f9bbbf3 100644
--- a/pkg/sources/forward/shutdown_test.go
+++ b/pkg/sources/forward/shutdown_test.go
@@ -89,7 +89,7 @@ func TestInterStepDataForward(t *testing.T) {
 			fetchWatermark, _ := generic.BuildNoOpSourceWatermarkProgressorsFromBufferMap(toSteps)
 			toVertexWmStores := buildNoOpToVertexStores(toSteps)
 			idleManager, _ := wmb.NewIdleManager(1, len(toSteps))
-			f, err := NewDataForward(vertexInstance, fromStep, toSteps, myShutdownTest{}, myShutdownTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexWmStores, idleManager, WithReadBatchSize(batchSize))
+			f, err := NewDataForward(vertexInstance, fromStep, toSteps, myShutdownTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexWmStores, idleManager, WithReadBatchSize(batchSize), WithTransformer(myShutdownTest{}))
 			assert.NoError(t, err)
 			stopped := f.Start()
 			// write some data but buffer is not full even though we are not reading
@@ -129,7 +129,7 @@ func TestInterStepDataForward(t *testing.T) {
 			toVertexWmStores := buildNoOpToVertexStores(toSteps)
 
 			idleManager, _ := wmb.NewIdleManager(1, len(toSteps))
-			f, err := NewDataForward(vertexInstance, fromStep, toSteps, myShutdownTest{}, myShutdownTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexWmStores, idleManager, WithReadBatchSize(batchSize))
+			f, err := NewDataForward(vertexInstance, fromStep, toSteps, myShutdownTest{}, fetchWatermark, TestSourceWatermarkPublisher{}, toVertexWmStores, idleManager, WithReadBatchSize(batchSize), WithTransformer(myShutdownTest{}))
 			assert.NoError(t, err)
 			stopped := f.Start()
 			// write some data such that the reader can be empty, that is toBuffer gets full
diff --git a/pkg/sources/http/http_test.go b/pkg/sources/http/http_test.go
index c812cf9a..9e5b1717 100644
--- a/pkg/sources/http/http_test.go
+++ b/pkg/sources/http/http_test.go
@@ -61,6 +61,7 @@ func Test_NewHTTP(t *testing.T) {
 		Hostname: "test-host",
 		Replica:  0,
 	}
+
 	h, err := NewHttpSource(ctx, vi)
 	assert.NoError(t, err)
 	assert.NotNil(t, h.(*httpSource).shutdown)