Repository: numaproj/numaflow
Pull request #1662 (2024-04-13 07:18:02)

chore: Refactor Sinks to Avoid Implementing Forwarder Interface
Now that we have a separate forwarder for the sink, our sinks don't need to implement the forwarder interface.




Patch:
diff --git a/pkg/sinks/blackhole/blackhole.go b/pkg/sinks/blackhole/blackhole.go
index e72c9e05..3ef03321 100644
--- a/pkg/sinks/blackhole/blackhole.go
+++ b/pkg/sinks/blackhole/blackhole.go
@@ -25,65 +25,22 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
-	sinkforward "github.com/numaproj/numaflow/pkg/sinks/forward"
-	"github.com/numaproj/numaflow/pkg/watermark/fetch"
-	"github.com/numaproj/numaflow/pkg/watermark/publish"
-	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 // Blackhole is a sink to emulate /dev/null
 type Blackhole struct {
 	name         string
 	pipelineName string
-	isdf         *sinkforward.DataForward
 	logger       *zap.SugaredLogger
 }
 
-type Option func(*Blackhole) error
-
-func WithLogger(log *zap.SugaredLogger) Option {
-	return func(bl *Blackhole) error {
-		bl.logger = log
-		return nil
-	}
-}
-
-// NewBlackhole returns Blackhole type.
-func NewBlackhole(vertexInstance *dfv1.VertexInstance,
-	fromBuffer isb.BufferReader,
-	fetchWatermark fetch.Fetcher,
-	publishWatermark publish.Publisher,
-	idleManager wmb.IdleManager,
-	opts ...Option) (*Blackhole, error) {
-
-	bh := new(Blackhole)
-	name := vertexInstance.Vertex.Spec.Name
-	bh.name = name
-	bh.pipelineName = vertexInstance.Vertex.Spec.PipelineName
-
-	for _, o := range opts {
-		if err := o(bh); err != nil {
-			return nil, err
-		}
-	}
-	if bh.logger == nil {
-		bh.logger = logging.NewLogger()
-	}
-
-	forwardOpts := []sinkforward.Option{sinkforward.WithLogger(bh.logger)}
-	if x := vertexInstance.Vertex.Spec.Limits; x != nil {
-		if x.ReadBatchSize != nil {
-			forwardOpts = append(forwardOpts, sinkforward.WithReadBatchSize(int64(*x.ReadBatchSize)))
-		}
-	}
-
-	isdf, err := sinkforward.NewDataForward(vertexInstance, fromBuffer, bh, fetchWatermark, publishWatermark, idleManager, forwardOpts...)
-	if err != nil {
-		return nil, err
-	}
-	bh.isdf = isdf
-
-	return bh, nil
+// NewBlackhole returns a new Blackhole sink.
+func NewBlackhole(ctx context.Context, vertexInstance *dfv1.VertexInstance) (*Blackhole, error) {
+	return &Blackhole{
+		name:         vertexInstance.Vertex.Spec.Name,
+		pipelineName: vertexInstance.Vertex.Spec.PipelineName,
+		logger:       logging.FromContext(ctx),
+	}, nil
 }
 
 // GetName returns the name.
@@ -113,18 +70,3 @@ func (b *Blackhole) Write(_ context.Context, messages []isb.Message) ([]isb.Offs
 func (b *Blackhole) Close() error {
 	return nil
 }
-
-// Start starts the sink.
-func (b *Blackhole) Start() <-chan struct{} {
-	return b.isdf.Start()
-}
-
-// Stop stops sinking
-func (b *Blackhole) Stop() {
-	b.isdf.Stop()
-}
-
-// ForceStop stops sinking
-func (b *Blackhole) ForceStop() {
-	b.isdf.ForceStop()
-}
diff --git a/pkg/sinks/kafka/kafka.go b/pkg/sinks/kafka/kafka.go
index ac3bb06a..cfbd2d27 100644
--- a/pkg/sinks/kafka/kafka.go
+++ b/pkg/sinks/kafka/kafka.go
@@ -29,10 +29,6 @@ import (
 	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"github.com/numaproj/numaflow/pkg/shared/util"
-	sinkforward "github.com/numaproj/numaflow/pkg/sinks/forward"
-	"github.com/numaproj/numaflow/pkg/watermark/fetch"
-	"github.com/numaproj/numaflow/pkg/watermark/publish"
-	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 // ToKafka produce the output to a kafka sinks.
@@ -42,65 +38,29 @@ type ToKafka struct {
 	producer     sarama.AsyncProducer
 	connected    bool
 	topic        string
-	isdf         *sinkforward.DataForward
 	kafkaSink    *dfv1.KafkaSink
 	log          *zap.SugaredLogger
 }
 
-type Option func(*ToKafka) error
-
-func WithLogger(log *zap.SugaredLogger) Option {
-	return func(t *ToKafka) error {
-		t.log = log
-		return nil
-	}
-}
-
 // NewToKafka returns ToKafka type.
-func NewToKafka(vertexInstance *dfv1.VertexInstance,
-	fromBuffer isb.BufferReader,
-	fetchWatermark fetch.Fetcher,
-	publishWatermark publish.Publisher,
-	idleManager wmb.IdleManager,
-	opts ...Option) (*ToKafka, error) {
+func NewToKafka(ctx context.Context, vertexInstance *dfv1.VertexInstance) (*ToKafka, error) {
 
 	kafkaSink := vertexInstance.Vertex.Spec.Sink.Kafka
 	toKafka := new(ToKafka)
-	// apply options for kafka sink
-	for _, o := range opts {
-		if err := o(toKafka); err != nil {
-			return nil, err
-		}
-	}
 
-	// set default logger
-	if toKafka.log == nil {
-		toKafka.log = logging.NewLogger()
-	}
-	toKafka.log = toKafka.log.With("sinkType", "kafka").With("topic", kafkaSink.Topic)
 	toKafka.name = vertexInstance.Vertex.Spec.Name
 	toKafka.pipelineName = vertexInstance.Vertex.Spec.PipelineName
 	toKafka.topic = kafkaSink.Topic
 	toKafka.kafkaSink = kafkaSink
 
-	forwardOpts := []sinkforward.Option{sinkforward.WithLogger(toKafka.log)}
-	if x := vertexInstance.Vertex.Spec.Limits; x != nil {
-		if x.ReadBatchSize != nil {
-			forwardOpts = append(forwardOpts, sinkforward.WithReadBatchSize(int64(*x.ReadBatchSize)))
-		}
-	}
-
-	f, err := sinkforward.NewDataForward(vertexInstance, fromBuffer, toKafka, fetchWatermark, publishWatermark, idleManager, forwardOpts...)
-	if err != nil {
-		return nil, err
-	}
-	toKafka.isdf = f
 	producer, err := connect(kafkaSink)
 	if err != nil {
 		return nil, err
 	}
 	toKafka.producer = producer
 	toKafka.connected = true
+
+	toKafka.log = logging.FromContext(ctx).With("sinkType", "kafka").With("topic", kafkaSink.Topic)
 	return toKafka, nil
 }
 
@@ -213,20 +173,3 @@ func (tk *ToKafka) Close() error {
 	tk.log.Info("Closing kafka producer...")
 	return tk.producer.Close()
 }
-
-// Start starts sinking to kafka.
-func (tk *ToKafka) Start() <-chan struct{} {
-	return tk.isdf.Start()
-}
-
-// Stop stops sinking
-func (tk *ToKafka) Stop() {
-	tk.isdf.Stop()
-	tk.log.Info("forwarder stopped successfully")
-}
-
-// ForceStop stops sinking
-func (tk *ToKafka) ForceStop() {
-	tk.isdf.ForceStop()
-	tk.log.Info("forwarder force stopped successfully")
-}
diff --git a/pkg/sinks/logger/log.go b/pkg/sinks/logger/log.go
index ce6d787b..3d7e31f9 100644
--- a/pkg/sinks/logger/log.go
+++ b/pkg/sinks/logger/log.go
@@ -28,65 +28,22 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
-	sinkforward "github.com/numaproj/numaflow/pkg/sinks/forward"
-	"github.com/numaproj/numaflow/pkg/watermark/fetch"
-	"github.com/numaproj/numaflow/pkg/watermark/publish"
-	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 // ToLog prints the output to a log sinks.
 type ToLog struct {
 	name         string
 	pipelineName string
-	isdf         *sinkforward.DataForward
 	logger       *zap.SugaredLogger
 }
 
-type Option func(*ToLog) error
-
-func WithLogger(log *zap.SugaredLogger) Option {
-	return func(t *ToLog) error {
-		t.logger = log
-		return nil
-	}
-}
-
 // NewToLog returns ToLog type.
-func NewToLog(vertexInstance *dfv1.VertexInstance,
-	fromBuffer isb.BufferReader,
-	fetchWatermark fetch.Fetcher,
-	publishWatermark publish.Publisher,
-	idleManager wmb.IdleManager,
-	opts ...Option) (*ToLog, error) {
-
-	toLog := new(ToLog)
-	name := vertexInstance.Vertex.Spec.Name
-	toLog.name = name
-	toLog.pipelineName = vertexInstance.Vertex.Spec.PipelineName
-	// use opts in future for specifying logger format etc
-	for _, o := range opts {
-		if err := o(toLog); err != nil {
-			return nil, err
-		}
-	}
-	if toLog.logger == nil {
-		toLog.logger = logging.NewLogger()
-	}
-
-	forwardOpts := []sinkforward.Option{sinkforward.WithLogger(toLog.logger)}
-	if x := vertexInstance.Vertex.Spec.Limits; x != nil {
-		if x.ReadBatchSize != nil {
-			forwardOpts = append(forwardOpts, sinkforward.WithReadBatchSize(int64(*x.ReadBatchSize)))
-		}
-	}
-
-	isdf, err := sinkforward.NewDataForward(vertexInstance, fromBuffer, toLog, fetchWatermark, publishWatermark, idleManager, forwardOpts...)
-	if err != nil {
-		return nil, err
-	}
-	toLog.isdf = isdf
-
-	return toLog, nil
+func NewToLog(ctx context.Context, vertexInstance *dfv1.VertexInstance) (*ToLog, error) {
+	return &ToLog{
+		name:         vertexInstance.Vertex.Spec.Name,
+		pipelineName: vertexInstance.Vertex.Spec.PipelineName,
+		logger:       logging.FromContext(ctx),
+	}, nil
 }
 
 // GetName returns the name.
@@ -123,18 +80,3 @@ func (t *ToLog) Write(_ context.Context, messages []isb.Message) ([]isb.Offset,
 func (t *ToLog) Close() error {
 	return nil
 }
-
-// Start starts sinking to Log.
-func (t *ToLog) Start() <-chan struct{} {
-	return t.isdf.Start()
-}
-
-// Stop stops sinking
-func (t *ToLog) Stop() {
-	t.isdf.Stop()
-}
-
-// ForceStop stops sinking
-func (t *ToLog) ForceStop() {
-	t.isdf.ForceStop()
-}
diff --git a/pkg/sinks/sink.go b/pkg/sinks/sink.go
index 8c7b1490..4438d136 100644
--- a/pkg/sinks/sink.go
+++ b/pkg/sinks/sink.go
@@ -23,9 +23,11 @@ import (
 
 	"go.uber.org/zap"
 
+	"github.com/numaproj/numaflow/pkg/forwarder"
 	"github.com/numaproj/numaflow/pkg/sdkclient"
 	"github.com/numaproj/numaflow/pkg/sdkserverinfo"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
+	sinkforward "github.com/numaproj/numaflow/pkg/sinks/forward"
 	"github.com/numaproj/numaflow/pkg/watermark/generic/jetstream"
 	"github.com/numaproj/numaflow/pkg/watermark/store"
 	"github.com/numaproj/numaflow/pkg/watermark/wmb"
@@ -46,7 +48,6 @@ import (
 	"github.com/numaproj/numaflow/pkg/sinks/udsink"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
 	"github.com/numaproj/numaflow/pkg/watermark/generic"
-	"github.com/numaproj/numaflow/pkg/watermark/publish"
 )
 
 type SinkProcessor struct {
@@ -174,31 +175,45 @@ func (u *SinkProcessor) Start(ctx context.Context) error {
 	var finalWg sync.WaitGroup
 	for index := range u.VertexInstance.Vertex.OwnedBuffers() {
 		finalWg.Add(1)
-		sinker, err := u.getSinker(readers[index], log, fetchWatermark, publishWatermark[u.VertexInstance.Vertex.Spec.Name], idleManager, sinkHandler)
+		sinkWriter, err := u.createSinkWriter(ctx, sinkHandler)
 		if err != nil {
-			return fmt.Errorf("failed to find a sink, errpr: %w", err)
+			return fmt.Errorf("failed to find a sink, error: %w", err)
 		}
-		// start sinker using a goroutine for each partition.
-		go func(sinker Sinker, fromBufferPartitionName string) {
+
+		forwardOpts := []sinkforward.Option{sinkforward.WithLogger(log)}
+		if x := u.VertexInstance.Vertex.Spec.Limits; x != nil {
+			if x.ReadBatchSize != nil {
+				forwardOpts = append(forwardOpts, sinkforward.WithReadBatchSize(int64(*x.ReadBatchSize)))
+			}
+		}
+
+		// NOTE: forwarder can make multiple sink writers when we introduce fallback sinks
+		df, err := sinkforward.NewDataForward(u.VertexInstance, readers[index], sinkWriter, fetchWatermark, publishWatermark[u.VertexInstance.Vertex.Spec.Name], idleManager, forwardOpts...)
+		if err != nil {
+			return fmt.Errorf("failed to create data forward, error: %w", err)
+		}
+
+		// start a forwarder using a goroutine for each partition.
+		go func(sinkForwarder forwarder.StarterStopper, fromBufferPartitionName string) {
 			defer finalWg.Done()
 			log.Infow("Start processing sink messages ", zap.String("isbsvc", string(u.ISBSvcType)), zap.String("fromPartition ", fromBufferPartitionName))
-			stopped := sinker.Start()
+			stopped := sinkForwarder.Start()
 			wg := &sync.WaitGroup{}
 			wg.Add(1)
 			go func() {
 				defer wg.Done()
 				for {
 					<-stopped
-					log.Info("Sinker stopped, exiting sink processor...")
+					log.Info("Sink forwarder stopped, exiting sink processor...")
 					return
 				}
 			}()
 			<-ctx.Done()
 			log.Infow("SIGTERM exiting inside partition...", zap.String("fromPartition", fromBufferPartitionName))
-			sinker.Stop()
+			sinkForwarder.Stop()
 			wg.Wait()
 			log.Infow("Exited for partition...", zap.String("fromPartition", fromBufferPartitionName))
-		}(sinker, readers[index].GetName())
+		}(df, readers[index].GetName())
 	}
 
 	// create lag readers from buffer readers
@@ -215,7 +230,7 @@ func (u *SinkProcessor) Start(ctx context.Context) error {
 		defer func() { _ = shutdown(context.Background()) }()
 	}
 
-	// wait for all the sinkers to exit
+	// wait for all the forwarders to exit
 	finalWg.Wait()
 
 	// closing the publisher will only delete the keys from the store, but not the store itself
@@ -244,18 +259,18 @@ func (u *SinkProcessor) Start(ctx context.Context) error {
 	return nil
 }
 
-// getSinker takes in the logger from the parent context
-func (u *SinkProcessor) getSinker(reader isb.BufferReader, logger *zap.SugaredLogger, fetchWM fetch.Fetcher, publishWM publish.Publisher, idleManager wmb.IdleManager, sinkHandler udsink.SinkApplier) (Sinker, error) {
+// createSinkWriter creates a sink writer based on the sink spec
+func (u *SinkProcessor) createSinkWriter(ctx context.Context, sinkHandler udsink.SinkApplier) (isb.BufferWriter, error) {
 	sink := u.VertexInstance.Vertex.Spec.Sink
 	if x := sink.Log; x != nil {
-		return logsink.NewToLog(u.VertexInstance, reader, fetchWM, publishWM, idleManager, logsink.WithLogger(logger))
+		return logsink.NewToLog(ctx, u.VertexInstance)
 	} else if x := sink.Kafka; x != nil {
-		return kafkasink.NewToKafka(u.VertexInstance, reader, fetchWM, publishWM, idleManager, kafkasink.WithLogger(logger))
+		return kafkasink.NewToKafka(ctx, u.VertexInstance)
 	} else if x := sink.Blackhole; x != nil {
-		return blackhole.NewBlackhole(u.VertexInstance, reader, fetchWM, publishWM, idleManager, blackhole.WithLogger(logger))
+		return blackhole.NewBlackhole(ctx, u.VertexInstance)
 	} else if x := sink.UDSink; x != nil {
 		// if the sink is a user defined sink, then we need to pass the sinkHandler to it which will be used to invoke the user defined sink
-		return udsink.NewUserDefinedSink(u.VertexInstance, reader, fetchWM, publishWM, idleManager, sinkHandler, udsink.WithLogger(logger))
+		return udsink.NewUserDefinedSink(ctx, u.VertexInstance, sinkHandler)
 	}
 	return nil, fmt.Errorf("invalid sink spec")
 }
diff --git a/pkg/sinks/sinker.go b/pkg/sinks/sinker.go
index e2045a67..5b974e32 100644
--- a/pkg/sinks/sinker.go
+++ b/pkg/sinks/sinker.go
@@ -17,12 +17,10 @@ limitations under the License.
 package sinks
 
 import (
-	"github.com/numaproj/numaflow/pkg/forwarder"
 	"github.com/numaproj/numaflow/pkg/isb"
 )
 
 // Sinker interface defines what a Sink should implement.
 type Sinker interface {
 	isb.BufferWriter
-	forwarder.StarterStopper
 }
diff --git a/pkg/sinks/udsink/sink.go b/pkg/sinks/udsink/sink.go
index e230ff4c..c301e137 100644
--- a/pkg/sinks/udsink/sink.go
+++ b/pkg/sinks/udsink/sink.go
@@ -27,64 +27,24 @@ import (
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
-	sinkforward "github.com/numaproj/numaflow/pkg/sinks/forward"
-	"github.com/numaproj/numaflow/pkg/watermark/fetch"
-	"github.com/numaproj/numaflow/pkg/watermark/publish"
-	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 type UserDefinedSink struct {
 	name         string
 	pipelineName string
-	isdf         *sinkforward.DataForward
 	logger       *zap.SugaredLogger
 	udsink       SinkApplier
 }
 
-type Option func(*UserDefinedSink) error
-
-func WithLogger(log *zap.SugaredLogger) Option {
-	return func(t *UserDefinedSink) error {
-		t.logger = log
-		return nil
-	}
-}
-
 // NewUserDefinedSink returns genericSink type.
-func NewUserDefinedSink(vertexInstance *dfv1.VertexInstance,
-	fromBuffer isb.BufferReader,
-	fetchWatermark fetch.Fetcher,
-	publishWatermark publish.Publisher,
-	idleManager wmb.IdleManager,
-	udsink SinkApplier,
-	opts ...Option) (*UserDefinedSink, error) {
-
-	s := new(UserDefinedSink)
-	name := vertexInstance.Vertex.Spec.Name
-	s.name = name
-	s.pipelineName = vertexInstance.Vertex.Spec.PipelineName
-	for _, o := range opts {
-		if err := o(s); err != nil {
-			return nil, err
-		}
-	}
-	if s.logger == nil {
-		s.logger = logging.NewLogger()
-	}
-
-	forwardOpts := []sinkforward.Option{sinkforward.WithLogger(s.logger)}
-	if x := vertexInstance.Vertex.Spec.Limits; x != nil {
-		if x.ReadBatchSize != nil {
-			forwardOpts = append(forwardOpts, sinkforward.WithReadBatchSize(int64(*x.ReadBatchSize)))
-		}
-	}
-	s.udsink = udsink
-	isdf, err := sinkforward.NewDataForward(vertexInstance, fromBuffer, s, fetchWatermark, publishWatermark, idleManager, forwardOpts...)
-	if err != nil {
-		return nil, err
-	}
-	s.isdf = isdf
-	return s, nil
+func NewUserDefinedSink(ctx context.Context, vertexInstance *dfv1.VertexInstance, udsink SinkApplier) (*UserDefinedSink, error) {
+
+	return &UserDefinedSink{
+		name:         vertexInstance.Vertex.Spec.Name,
+		pipelineName: vertexInstance.Vertex.Spec.PipelineName,
+		logger:       logging.FromContext(ctx),
+		udsink:       udsink,
+	}, nil
 }
 
 func (s *UserDefinedSink) GetName() string {
@@ -121,15 +81,3 @@ func (s *UserDefinedSink) Write(ctx context.Context, messages []isb.Message) ([]
 func (s *UserDefinedSink) Close() error {
 	return nil
 }
-
-func (s *UserDefinedSink) Start() <-chan struct{} {
-	return s.isdf.Start()
-}
-
-func (s *UserDefinedSink) Stop() {
-	s.isdf.Stop()
-}
-
-func (s *UserDefinedSink) ForceStop() {
-	s.isdf.ForceStop()
-}


Test patch:
diff --git a/pkg/sinks/blackhole/blackhole_test.go b/pkg/sinks/blackhole/blackhole_test.go
index 60e9736c..38fee753 100644
--- a/pkg/sinks/blackhole/blackhole_test.go
+++ b/pkg/sinks/blackhole/blackhole_test.go
@@ -24,14 +24,10 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
-	"github.com/numaproj/numaflow/pkg/watermark/generic"
-	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 func TestBlackhole_Start(t *testing.T) {
-	fromStep := simplebuffer.NewInMemoryBuffer("from", 25, 0)
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
 	defer cancel()
 
@@ -50,21 +46,13 @@ func TestBlackhole_Start(t *testing.T) {
 		Vertex:  vertex,
 		Replica: 0,
 	}
-	fetchWatermark, publishWatermark := generic.BuildNoOpWatermarkProgressorsFromBufferList([]string{vertex.Spec.Name})
-	idleManager, _ := wmb.NewIdleManager(1, 1)
-	s, err := NewBlackhole(vertexInstance, fromStep, fetchWatermark, publishWatermark[vertex.Spec.Name], idleManager)
+	s, err := NewBlackhole(ctx, vertexInstance)
 	assert.NoError(t, err)
 
-	stopped := s.Start()
-	// write some data
-	_, errs := fromStep.Write(ctx, writeMessages[0:5])
+	_, errs := s.Write(ctx, writeMessages[0:5])
 	assert.Equal(t, make([]error, 5), errs)
 
 	// write some data
-	_, errs = fromStep.Write(ctx, writeMessages[5:20])
+	_, errs = s.Write(ctx, writeMessages[5:20])
 	assert.Equal(t, make([]error, 15), errs)
-
-	s.Stop()
-
-	<-stopped
 }
diff --git a/pkg/sinks/kafka/kafka_test.go b/pkg/sinks/kafka/kafka_test.go
index 6575e1f6..9a1a5f73 100644
--- a/pkg/sinks/kafka/kafka_test.go
+++ b/pkg/sinks/kafka/kafka_test.go
@@ -26,16 +26,11 @@ import (
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
-	sinkforward "github.com/numaproj/numaflow/pkg/sinks/forward"
-	"github.com/numaproj/numaflow/pkg/watermark/generic"
-	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 func TestWriteSuccessToKafka(t *testing.T) {
 	var err error
-	fromStep := simplebuffer.NewInMemoryBuffer("toKafka", 25, 0)
 	toKafka := new(ToKafka)
 	vertex := &dfv1.Vertex{Spec: dfv1.VertexSpec{
 		PipelineName: "testPipeline",
@@ -46,13 +41,6 @@ func TestWriteSuccessToKafka(t *testing.T) {
 			},
 		},
 	}}
-	vertexInstance := &dfv1.VertexInstance{
-		Vertex:  vertex,
-		Replica: 0,
-	}
-	fetchWatermark, publishWatermark := generic.BuildNoOpWatermarkProgressorsFromBufferList([]string{vertex.Spec.Name})
-	idleManager, _ := wmb.NewIdleManager(1, 1)
-	toKafka.isdf, err = sinkforward.NewDataForward(vertexInstance, fromStep, toKafka, fetchWatermark, publishWatermark["testVertex"], idleManager)
 	assert.NoError(t, err)
 	toKafka.kafkaSink = vertex.Spec.Sink.Kafka
 	toKafka.name = "Test"
@@ -66,7 +54,6 @@ func TestWriteSuccessToKafka(t *testing.T) {
 	producer.ExpectInputAndSucceed()
 	toKafka.producer = producer
 	toKafka.connected = true
-	toKafka.Start()
 	msgs := []isb.Message{
 		{
 			Header: isb.Header{
@@ -87,30 +74,11 @@ func TestWriteSuccessToKafka(t *testing.T) {
 	for _, err := range errs {
 		assert.Nil(t, err)
 	}
-	toKafka.Stop()
 }
 
 func TestWriteFailureToKafka(t *testing.T) {
 	var err error
-	fromStep := simplebuffer.NewInMemoryBuffer("toKafka", 25, 0)
 	toKafka := new(ToKafka)
-	vertex := &dfv1.Vertex{Spec: dfv1.VertexSpec{
-		PipelineName: "testPipeline",
-		AbstractVertex: dfv1.AbstractVertex{
-			Name: "testVertex",
-			Sink: &dfv1.Sink{
-				Kafka: &dfv1.KafkaSink{},
-			},
-		},
-	}}
-	vertexInstance := &dfv1.VertexInstance{
-		Vertex:  vertex,
-		Replica: 0,
-	}
-	toSteps := map[string][]isb.BufferWriter{vertex.Spec.Name: {toKafka}}
-	fetchWatermark, publishWatermark := generic.BuildNoOpWatermarkProgressorsFromBufferMap(toSteps)
-	idleManager, _ := wmb.NewIdleManager(1, 1)
-	toKafka.isdf, err = sinkforward.NewDataForward(vertexInstance, fromStep, toKafka, fetchWatermark, publishWatermark["testVertex"], idleManager)
 	assert.NoError(t, err)
 	toKafka.name = "Test"
 	toKafka.topic = "topic-1"
@@ -123,7 +91,6 @@ func TestWriteFailureToKafka(t *testing.T) {
 	producer.ExpectInputAndFail(fmt.Errorf("test1"))
 	toKafka.producer = producer
 	toKafka.connected = true
-	toKafka.Start()
 	msgs := []isb.Message{
 		{
 			Header: isb.Header{
@@ -146,6 +113,5 @@ func TestWriteFailureToKafka(t *testing.T) {
 	}
 	assert.Equal(t, "test", errs[0].Error())
 	assert.Equal(t, "test1", errs[1].Error())
-	toKafka.Stop()
 
 }
diff --git a/pkg/sinks/logger/log_test.go b/pkg/sinks/logger/log_test.go
index 1ca5e8a6..8bbda186 100644
--- a/pkg/sinks/logger/log_test.go
+++ b/pkg/sinks/logger/log_test.go
@@ -24,20 +24,10 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
-	sinkforward "github.com/numaproj/numaflow/pkg/sinks/forward"
-	"github.com/numaproj/numaflow/pkg/watermark/generic"
-	"github.com/numaproj/numaflow/pkg/watermark/wmb"
-)
-
-var (
-	testStartTime = time.Unix(1636470000, 0).UTC()
 )
 
 func TestToLog_Start(t *testing.T) {
-	fromStep := simplebuffer.NewInMemoryBuffer("from", 25, 0)
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
 	defer cancel()
 
@@ -56,88 +46,14 @@ func TestToLog_Start(t *testing.T) {
 		Vertex:  vertex,
 		Replica: 0,
 	}
-	fetchWatermark, publishWatermark := generic.BuildNoOpWatermarkProgressorsFromBufferList([]string{vertex.Spec.Name})
-	idleManager, _ := wmb.NewIdleManager(1, 1)
-	s, err := NewToLog(vertexInstance, fromStep, fetchWatermark, publishWatermark[vertex.Spec.Name], idleManager)
+	s, err := NewToLog(ctx, vertexInstance)
 	assert.NoError(t, err)
 
-	stopped := s.Start()
 	// write some data
-	_, errs := fromStep.Write(ctx, writeMessages[0:5])
+	_, errs := s.Write(ctx, writeMessages[0:5])
 	assert.Equal(t, make([]error, 5), errs)
 
 	// write some data
-	_, errs = fromStep.Write(ctx, writeMessages[5:20])
+	_, errs = s.Write(ctx, writeMessages[5:20])
 	assert.Equal(t, make([]error, 15), errs)
-
-	s.Stop()
-
-	<-stopped
-}
-
-// TestToLog_Forward writes to a vertex which has a logger sink
-func TestToLog_Forward(t *testing.T) {
-	tests := []struct {
-		name      string
-		batchSize int64
-	}{
-		{
-			name:      "batch_forward",
-			batchSize: 1,
-		},
-		{
-			name:      "batch_forward",
-			batchSize: 5,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			batchSize := tt.batchSize
-			ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
-			defer cancel()
-
-			fromStep := simplebuffer.NewInMemoryBuffer("from", 5*batchSize, 0)
-			to1 := simplebuffer.NewInMemoryBuffer("sinks.logger1", 5*batchSize, 0)
-
-			vertex1 := &dfv1.Vertex{Spec: dfv1.VertexSpec{
-				AbstractVertex: dfv1.AbstractVertex{
-					Name: "sinks.logger1",
-					Sink: &dfv1.Sink{
-						Log: &dfv1.Log{},
-					},
-				},
-			}}
-			vertexInstance1 := &dfv1.VertexInstance{
-				Vertex:  vertex1,
-				Replica: 0,
-			}
-
-			fetchWatermark1, publishWatermark1 := generic.BuildNoOpWatermarkProgressorsFromBufferList([]string{vertex1.Spec.Name})
-			idleManager0, _ := wmb.NewIdleManager(1, 1)
-			logger1, _ := NewToLog(vertexInstance1, to1, fetchWatermark1, publishWatermark1[vertex1.Spec.Name], idleManager0)
-			logger1Stopped := logger1.Start()
-
-			toSteps := map[string][]isb.BufferWriter{
-				"sinks.logger1": {to1},
-			}
-
-			writeMessages := testutils.BuildTestWriteMessages(int64(20), testStartTime, nil)
-
-			fetchWatermark, publishWatermark := generic.BuildNoOpWatermarkProgressorsFromBufferMap(toSteps)
-			idleManager1, _ := wmb.NewIdleManager(1, 1)
-			f, err := sinkforward.NewDataForward(vertexInstance1, fromStep, to1, fetchWatermark, publishWatermark["sinks.logger1"], idleManager1, sinkforward.WithReadBatchSize(batchSize))
-			assert.NoError(t, err)
-
-			stopped := f.Start()
-			// write some data
-			_, errs := fromStep.Write(ctx, writeMessages[0:batchSize])
-			assert.Equal(t, make([]error, batchSize), errs)
-			f.Stop()
-			<-stopped
-			// downstream should be stopped only after upstream is stopped
-			logger1.Stop()
-
-			<-logger1Stopped
-		})
-	}
 }