Repository: numaproj/numaflow
Pull request #1501 (2024-02-12 05:42:30)

chore: minor fixes
* fix dedup for aligned reduce (TODO: unaligned)
* increase the nats ping interval from 1 second to 3 seconds to avoid a reconnect storm, when the server goes down.
* fix seg fault when the watermark is disabled in the pipeline
* adding ISB latency metrics for better debugging in case of an ISB issue
* adding a metric to track the number of control messages

vigith: @yhl25 please describe in the description about all the minor fixes 😄 
yhl25: > this PR will increase the total metrics count (PPS, points per second). should we think about having tiered metrics, eg., P0 (must-haves), P1, P2, etc. and ability to turn on on-demand.

yeah, I think we should.

Patch:
diff --git a/go.mod b/go.mod
index da248ce7..04216e29 100644
--- a/go.mod
+++ b/go.mod
@@ -29,7 +29,7 @@ require (
 	github.com/imdario/mergo v0.3.13
 	github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe
 	github.com/nats-io/nats-server/v2 v2.10.4
-	github.com/nats-io/nats.go v1.31.0
+	github.com/nats-io/nats.go v1.32.0
 	github.com/numaproj/numaflow-go v0.5.3-0.20231213060340-dbd9016bbcb6
 	github.com/prometheus/client_golang v1.14.0
 	github.com/prometheus/common v0.37.0
@@ -44,7 +44,7 @@ require (
 	go.uber.org/goleak v1.2.1
 	go.uber.org/multierr v1.7.0
 	go.uber.org/zap v1.24.0
-	golang.org/x/crypto v0.17.0
+	golang.org/x/crypto v0.18.0
 	golang.org/x/net v0.17.0
 	golang.org/x/oauth2 v0.7.0
 	golang.org/x/sync v0.3.0
@@ -153,7 +153,7 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/nats-io/jwt/v2 v2.5.2 // indirect
-	github.com/nats-io/nkeys v0.4.6 // indirect
+	github.com/nats-io/nkeys v0.4.7 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/oklog/ulid v1.3.1 // indirect
 	github.com/pelletier/go-toml v1.9.4 // indirect
@@ -191,8 +191,8 @@ require (
 	go.mongodb.org/mongo-driver v1.7.3 // indirect
 	golang.org/x/arch v0.3.0 // indirect
 	golang.org/x/mod v0.8.0 // indirect
-	golang.org/x/sys v0.15.0 // indirect
-	golang.org/x/term v0.15.0 // indirect
+	golang.org/x/sys v0.16.0 // indirect
+	golang.org/x/term v0.16.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
 	golang.org/x/tools v0.6.0 // indirect
diff --git a/go.sum b/go.sum
index 9d41d67f..836b0a8b 100644
--- a/go.sum
+++ b/go.sum
@@ -674,10 +674,10 @@ github.com/nats-io/jwt/v2 v2.5.2 h1:DhGH+nKt+wIkDxM6qnVSKjokq5t59AZV5HRcFW0zJwU=
 github.com/nats-io/jwt/v2 v2.5.2/go.mod h1:24BeQtRwxRV8ruvC4CojXlx/WQ/VjuwlYiH+vu/+ibI=
 github.com/nats-io/nats-server/v2 v2.10.4 h1:uB9xcwon3tPXWAdmTJqqqC6cie3yuPWHJjjTBgaPNus=
 github.com/nats-io/nats-server/v2 v2.10.4/go.mod h1:eWm2JmHP9Lqm2oemB6/XGi0/GwsZwtWf8HIPUsh+9ns=
-github.com/nats-io/nats.go v1.31.0 h1:/WFBHEc/dOKBF6qf1TZhrdEfTmOZ5JzdJ+Y3m6Y/p7E=
-github.com/nats-io/nats.go v1.31.0/go.mod h1:di3Bm5MLsoB4Bx61CBTsxuarI36WbhAwOm8QrW39+i8=
-github.com/nats-io/nkeys v0.4.6 h1:IzVe95ru2CT6ta874rt9saQRkWfe2nFj1NtvYSLqMzY=
-github.com/nats-io/nkeys v0.4.6/go.mod h1:4DxZNzenSVd1cYQoAa8948QY3QDjrHfcfVADymtkpts=
+github.com/nats-io/nats.go v1.32.0 h1:Bx9BZS+aXYlxW08k8Gd3yR2s73pV5XSoAQUyp1Kwvp0=
+github.com/nats-io/nats.go v1.32.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
+github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
+github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
@@ -936,8 +936,8 @@ golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5y
 golang.org/x/crypto v0.0.0-20210817164053-32db794688a5/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20220214200702-86341886e292/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k=
-golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
+golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
+golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
@@ -1150,13 +1150,13 @@ golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.15.0 h1:h48lPFYpsTvQJZF4EKyI4aLHaev3CxivZmv7yZig9pc=
-golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
+golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
-golang.org/x/term v0.15.0 h1:y/Oo/a/q3IXu26lQgl04j/gjuBDOBlx7X6Om1j2CPW4=
-golang.org/x/term v0.15.0/go.mod h1:BDl952bC7+uMoWR75FIrCDx79TPU9oHkTZ9yRbYOrX0=
+golang.org/x/term v0.16.0 h1:m+B6fahuftsE9qjo0VWp2FW0mB3MTJvR0BaMQrq0pmE=
+golang.org/x/term v0.16.0/go.mod h1:yn7UURbUtPyrVJPGPq404EukNFxcm/foM+bV/bfcDsY=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
diff --git a/pkg/isb/stores/jetstream/metrics.go b/pkg/isb/stores/jetstream/metrics.go
index 495378d5..d9558ff3 100644
--- a/pkg/isb/stores/jetstream/metrics.go
+++ b/pkg/isb/stores/jetstream/metrics.go
@@ -83,3 +83,27 @@ var isbWriteTimeout = promauto.NewCounterVec(prometheus.CounterOpts{
 	Name:      "write_timeout_total",
 	Help:      "Total number of jetstream write timeouts",
 }, []string{"buffer"})
+
+// isbWriteTime is a histogram to Observe isb write time for a buffer
+var isbWriteTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Subsystem: "isb_jetstream",
+	Name:      "write_time_total",
+	Help:      "Processing times of Writes for jetstream",
+	Buckets:   prometheus.ExponentialBucketsRange(100, 60000000*2, 10),
+}, []string{"buffer"})
+
+// isbReadTime is a histogram to Observe isb read time for a buffer
+var isbReadTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Subsystem: "isb_jetstream",
+	Name:      "read_time_total",
+	Help:      "Processing times of reads for jetstream",
+	Buckets:   prometheus.ExponentialBucketsRange(100, 60000000*2, 10),
+}, []string{"buffer"})
+
+// isbAckTime is a histogram to Observe isb ack time for a buffer
+var isbAckTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Subsystem: "isb_jetstream",
+	Name:      "ack_time_total",
+	Help:      "Processing times of acks for jetstream",
+	Buckets:   prometheus.ExponentialBucketsRange(100, 60000000*2, 10),
+}, []string{"buffer"})
diff --git a/pkg/isb/stores/jetstream/reader.go b/pkg/isb/stores/jetstream/reader.go
index 09a8bdaa..72e643e9 100644
--- a/pkg/isb/stores/jetstream/reader.go
+++ b/pkg/isb/stores/jetstream/reader.go
@@ -36,7 +36,7 @@ type jetStreamReader struct {
 	name                   string
 	stream                 string
 	subject                string
-	client                 *jsclient.NATSClient
+	client                 *jsclient.Client
 	sub                    *nats.Subscription
 	opts                   *readOptions
 	inProgressTickDuration time.Duration
@@ -45,7 +45,7 @@ type jetStreamReader struct {
 }
 
 // NewJetStreamBufferReader is used to provide a new JetStream buffer reader connection
-func NewJetStreamBufferReader(ctx context.Context, client *jsclient.NATSClient, name, stream, subject string, partitionIdx int32, opts ...ReadOption) (isb.BufferReader, error) {
+func NewJetStreamBufferReader(ctx context.Context, client *jsclient.Client, name, stream, subject string, partitionIdx int32, opts ...ReadOption) (isb.BufferReader, error) {
 	log := logging.FromContext(ctx).With("bufferReader", name).With("stream", stream).With("subject", subject)
 	o := defaultReadOptions()
 	for _, opt := range opts {
@@ -113,6 +113,10 @@ func (jr *jetStreamReader) Pending(_ context.Context) (int64, error) {
 }
 
 func (jr *jetStreamReader) Read(_ context.Context, count int64) ([]*isb.ReadMessage, error) {
+	labels := map[string]string{"buffer": jr.GetName()}
+	defer func(t time.Time) {
+		isbReadTime.With(labels).Observe(float64(time.Since(t).Microseconds()))
+	}(time.Now())
 	var err error
 	var result []*isb.ReadMessage
 	msgs, err := jr.sub.Fetch(int(count), nats.MaxWait(jr.opts.readTimeOut))
@@ -144,6 +148,10 @@ func (jr *jetStreamReader) Read(_ context.Context, count int64) ([]*isb.ReadMess
 }
 
 func (jr *jetStreamReader) Ack(_ context.Context, offsets []isb.Offset) []error {
+	labels := map[string]string{"buffer": jr.GetName()}
+	defer func(t time.Time) {
+		isbAckTime.With(labels).Observe(float64(time.Since(t).Microseconds()))
+	}(time.Now())
 	errs := make([]error, len(offsets))
 	done := make(chan struct{})
 	wg := &sync.WaitGroup{}
diff --git a/pkg/isb/stores/jetstream/writer.go b/pkg/isb/stores/jetstream/writer.go
index 50e300e1..a680f785 100644
--- a/pkg/isb/stores/jetstream/writer.go
+++ b/pkg/isb/stores/jetstream/writer.go
@@ -38,7 +38,7 @@ type jetStreamWriter struct {
 	partitionIdx int32
 	stream       string
 	subject      string
-	client       *jsclient.NATSClient
+	client       *jsclient.Client
 	js           nats.JetStreamContext
 	opts         *writeOptions
 	isFull       *atomic.Bool
@@ -46,7 +46,7 @@ type jetStreamWriter struct {
 }
 
 // NewJetStreamBufferWriter is used to provide a new instance of JetStreamBufferWriter
-func NewJetStreamBufferWriter(ctx context.Context, client *jsclient.NATSClient, name, stream, subject string, partitionIdx int32, opts ...WriteOption) (isb.BufferWriter, error) {
+func NewJetStreamBufferWriter(ctx context.Context, client *jsclient.Client, name, stream, subject string, partitionIdx int32, opts ...WriteOption) (isb.BufferWriter, error) {
 	o := defaultWriteOptions()
 	for _, opt := range opts {
 		if opt != nil {
@@ -197,7 +197,13 @@ func (jw *jetStreamWriter) asyncWrite(_ context.Context, messages []isb.Message,
 			Subject: jw.subject,
 			Data:    payload,
 		}
-		if future, err := jw.js.PublishMsgAsync(m, nats.MsgId(message.Header.ID)); err != nil { // nats.MsgId() is for exactly-once writing
+		var pubOpts []nats.PubOpt
+		// nats.MsgId() is for exactly-once writing
+		// we don't need to set MsgId for control message
+		if message.Header.Kind != isb.WMB {
+			pubOpts = append(pubOpts, nats.MsgId(message.Header.ID))
+		}
+		if future, err := jw.js.PublishMsgAsync(m, pubOpts...); err != nil { // nats.MsgId() is for exactly-once writing
 			errs[index] = err
 		} else {
 			futures[index] = future
@@ -244,6 +250,9 @@ func (jw *jetStreamWriter) asyncWrite(_ context.Context, messages []isb.Message,
 }
 
 func (jw *jetStreamWriter) syncWrite(_ context.Context, messages []isb.Message, errs []error, metricsLabels map[string]string) ([]isb.Offset, []error) {
+	defer func(t time.Time) {
+		isbWriteTime.With(metricsLabels).Observe(float64(time.Since(t).Microseconds()))
+	}(time.Now())
 	var writeOffsets = make([]isb.Offset, len(messages))
 	wg := new(sync.WaitGroup)
 	for index, msg := range messages {
@@ -259,7 +268,13 @@ func (jw *jetStreamWriter) syncWrite(_ context.Context, messages []isb.Message,
 				Subject: jw.subject,
 				Data:    payload,
 			}
-			if pubAck, err := jw.js.PublishMsg(m, nats.MsgId(message.Header.ID), nats.AckWait(2*time.Second)); err != nil { // nats.MsgId() is for exactly-once writing
+			pubOpts := []nats.PubOpt{nats.AckWait(2 * time.Second)}
+			// nats.MsgId() is for exactly-once writing
+			// we don't need to set MsgId for control message
+			if message.Header.Kind != isb.WMB {
+				pubOpts = append(pubOpts, nats.MsgId(message.Header.ID))
+			}
+			if pubAck, err := jw.js.PublishMsg(m, pubOpts...); err != nil {
 				errs[idx] = err
 				isbWriteErrors.With(metricsLabels).Inc()
 			} else {
diff --git a/pkg/isbsvc/jetstream_service.go b/pkg/isbsvc/jetstream_service.go
index 10f64fac..19b0150e 100644
--- a/pkg/isbsvc/jetstream_service.go
+++ b/pkg/isbsvc/jetstream_service.go
@@ -34,9 +34,8 @@ import (
 
 type jetStreamSvc struct {
 	pipelineName string
-
-	jsClient *jsclient.NATSClient
-	js       nats.JetStreamContext
+	jsClient     *jsclient.Client
+	js           nats.JetStreamContext
 }
 
 func NewISBJetStreamSvc(pipelineName string, opts ...JSServiceOption) (ISBService, error) {
@@ -51,7 +50,7 @@ func NewISBJetStreamSvc(pipelineName string, opts ...JSServiceOption) (ISBServic
 
 type JSServiceOption func(*jetStreamSvc) error
 
-func WithJetStreamClient(jsClient *jsclient.NATSClient) JSServiceOption {
+func WithJetStreamClient(jsClient *jsclient.Client) JSServiceOption {
 	return func(j *jetStreamSvc) error {
 		j.jsClient = jsClient
 		return nil
diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go
index 52a49181..dc3f4dc6 100644
--- a/pkg/metrics/metrics.go
+++ b/pkg/metrics/metrics.go
@@ -271,3 +271,13 @@ var (
 		Help:      "Total number of closed windows",
 	}, []string{LabelVertex, LabelPipeline, LabelVertexReplicaIndex})
 )
+
+// Ctrl Message Metric
+var (
+	// CtrlMessagesCount is used to indicate the number of total ctrl messages sent.
+	CtrlMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+		Subsystem: "idlemanager",
+		Name:      "ctrl_msg_total",
+		Help:      "Total number of ctrl Messages sent",
+	}, []string{LabelVertex, LabelPipeline, LabelVertexType, LabelVertexReplicaIndex, LabelPartitionName})
+)
diff --git a/pkg/reduce/data_forward.go b/pkg/reduce/data_forward.go
index c3fa8ca7..c93e13be 100644
--- a/pkg/reduce/data_forward.go
+++ b/pkg/reduce/data_forward.go
@@ -214,7 +214,7 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 			for toVertexName, toVertexBuffer := range df.toBuffers {
 				if publisher, ok := df.wmPublishers[toVertexName]; ok {
 					for _, bufferPartition := range toVertexBuffer {
-						idlehandler.PublishIdleWatermark(ctx, bufferPartition, publisher, df.idleManager, df.log, dfv1.VertexTypeReduceUDF, wmb.Watermark(time.UnixMilli(processorWMB.Watermark)))
+						idlehandler.PublishIdleWatermark(ctx, bufferPartition, publisher, df.idleManager, df.log, df.vertexName, df.pipelineName, dfv1.VertexTypeReduceUDF, df.vertexReplica, wmb.Watermark(time.UnixMilli(processorWMB.Watermark)))
 					}
 				}
 			}
@@ -237,7 +237,7 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 				for toVertexName, toVertexBuffer := range df.toBuffers {
 					if publisher, ok := df.wmPublishers[toVertexName]; ok {
 						for _, bufferPartition := range toVertexBuffer {
-							idlehandler.PublishIdleWatermark(ctx, bufferPartition, publisher, df.idleManager, df.log, dfv1.VertexTypeReduceUDF, wmb.Watermark(watermark))
+							idlehandler.PublishIdleWatermark(ctx, bufferPartition, publisher, df.idleManager, df.log, df.vertexName, df.pipelineName, dfv1.VertexTypeReduceUDF, df.vertexReplica, wmb.Watermark(watermark))
 						}
 					}
 				}
@@ -388,7 +388,7 @@ func (df *DataForward) process(ctx context.Context, messages []*isb.ReadMessage)
 			for toVertexName, toVertexBuffer := range df.toBuffers {
 				if publisher, ok := df.wmPublishers[toVertexName]; ok {
 					for _, bufferPartition := range toVertexBuffer {
-						idlehandler.PublishIdleWatermark(ctx, bufferPartition, publisher, df.idleManager, df.log, dfv1.VertexTypeReduceUDF, wmb.Watermark(watermark))
+						idlehandler.PublishIdleWatermark(ctx, bufferPartition, publisher, df.idleManager, df.log, df.vertexName, df.pipelineName, dfv1.VertexTypeReduceUDF, df.vertexReplica, wmb.Watermark(watermark))
 					}
 				}
 			}
diff --git a/pkg/reduce/pnf/processandforward.go b/pkg/reduce/pnf/processandforward.go
index 42ad8220..c2be8e83 100644
--- a/pkg/reduce/pnf/processandforward.go
+++ b/pkg/reduce/pnf/processandforward.go
@@ -392,7 +392,7 @@ func (p *processAndForward) publishWM(ctx context.Context, endTime time.Time) {
 		for index, activePartition := range activeWatermarkBuffers[toVertexName] {
 			if !activePartition {
 				if publisher, ok := p.wmPublishers[toVertexName]; ok {
-					idlehandler.PublishIdleWatermark(ctx, p.toBuffers[toVertexName][index], publisher, p.idleManager, p.log, dfv1.VertexTypeReduceUDF, wm)
+					idlehandler.PublishIdleWatermark(ctx, p.toBuffers[toVertexName][index], publisher, p.idleManager, p.log, p.vertexName, p.pipelineName, dfv1.VertexTypeReduceUDF, p.vertexReplica, wm)
 				}
 			}
 		}
diff --git a/pkg/shared/clients/nats/client_pool.go b/pkg/shared/clients/nats/client_pool.go
index bcc67ce6..eea25a09 100644
--- a/pkg/shared/clients/nats/client_pool.go
+++ b/pkg/shared/clients/nats/client_pool.go
@@ -50,7 +50,7 @@ func NewClientPool(ctx context.Context, opts ...Option) (*ClientPool, error) {
 
 // NextAvailableClient returns the next available NATS client. This code need not be optimized because this is
 // not in hot code path. It is only during connection creation/startup.
-func (p *ClientPool) NextAvailableClient() *NATSClient {
+func (p *ClientPool) NextAvailableClient() *Client {
 	p.mutex.Lock()
 	defer p.mutex.Unlock()
 
@@ -61,7 +61,7 @@ func (p *ClientPool) NextAvailableClient() *NATSClient {
 	// get the first client and move it to the back of the list
 	front := p.clients.Front()
 	p.clients.MoveToBack(front)
-	return front.Value.(*NATSClient)
+	return front.Value.(*Client)
 }
 
 // CloseAll closes all the clients in the pool
@@ -70,6 +70,6 @@ func (p *ClientPool) CloseAll() {
 	defer p.mutex.Unlock()
 
 	for e := p.clients.Front(); e != nil; e = e.Next() {
-		e.Value.(*NATSClient).Close()
+		e.Value.(*Client).Close()
 	}
 }
diff --git a/pkg/shared/clients/nats/nats_client.go b/pkg/shared/clients/nats/nats_client.go
index 443718de..9f6ccff8 100644
--- a/pkg/shared/clients/nats/nats_client.go
+++ b/pkg/shared/clients/nats/nats_client.go
@@ -34,8 +34,8 @@ import (
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 )
 
-// NATSClient is a client for NATS server which be shared by multiple connections (reader, writer, kv, buffer management, etc.)
-type NATSClient struct {
+// Client is a client for NATS server which be shared by multiple connections (reader, writer, kv, buffer management, etc.)
+type Client struct {
 	sync.Mutex
 	nc    *nats.Conn
 	jsCtx nats.JetStreamContext
@@ -43,7 +43,7 @@ type NATSClient struct {
 }
 
 // NewNATSClient Create a new NATS client
-func NewNATSClient(ctx context.Context, natsOptions ...nats.Option) (*NATSClient, error) {
+func NewNATSClient(ctx context.Context, natsOptions ...nats.Option) (*Client, error) {
 	log := logging.FromContext(ctx)
 	var jsCtx nats.JetStreamContext
 	opts := []nats.Option{
@@ -52,7 +52,7 @@ func NewNATSClient(ctx context.Context, natsOptions ...nats.Option) (*NATSClient
 		nats.MaxReconnects(-1),
 		// every one second we will try to ping the server, if we don't get a pong back
 		// after two attempts, we will consider the connection lost and try to reconnect
-		nats.PingInterval(1 * time.Second),
+		nats.PingInterval(3 * time.Second),
 		// error handler for the connection
 		nats.ErrorHandler(func(nc *nats.Conn, sub *nats.Subscription, err error) {
 			log.Errorw("Nats default: error occurred for subscription", zap.Error(err))
@@ -73,6 +73,14 @@ func NewNATSClient(ctx context.Context, natsOptions ...nats.Option) (*NATSClient
 		}),
 		// Write (and flush) timeout
 		nats.FlusherTimeout(10 * time.Second),
+
+		// If the server doesn't respond to 2 pings we will reconnect
+		nats.MaxPingsOutstanding(2),
+
+		// log when the server enters lame duck mode
+		nats.LameDuckModeHandler(func(nc *nats.Conn) {
+			log.Info("Nats default: entering lame duck mode to avoid reconnect storm")
+		}),
 	}
 
 	url, existing := os.LookupEnv(dfv1.EnvISBSvcJetStreamURL)
@@ -103,12 +111,12 @@ func NewNATSClient(ctx context.Context, natsOptions ...nats.Option) (*NATSClient
 		if err != nil {
 			return nil, fmt.Errorf("failed to create to nats jetstream context: %w", err)
 		}
-		return &NATSClient{nc: nc, jsCtx: jsCtx, log: logging.FromContext(ctx)}, nil
+		return &Client{nc: nc, jsCtx: jsCtx, log: logging.FromContext(ctx)}, nil
 	}
 }
 
 // Subscribe returns a subscription for the given subject and stream
-func (c *NATSClient) Subscribe(subject string, stream string, opts ...nats.SubOpt) (*nats.Subscription, error) {
+func (c *Client) Subscribe(subject string, stream string, opts ...nats.SubOpt) (*nats.Subscription, error) {
 	var (
 		err       error
 		jsContext nats.JetStreamContext
@@ -123,7 +131,7 @@ func (c *NATSClient) Subscribe(subject string, stream string, opts ...nats.SubOp
 }
 
 // BindKVStore lookup and bind to an existing KeyValue store and return the KeyValue interface
-func (c *NATSClient) BindKVStore(kvName string) (nats.KeyValue, error) {
+func (c *Client) BindKVStore(kvName string) (nats.KeyValue, error) {
 	var (
 		err       error
 		jsContext nats.JetStreamContext
@@ -138,7 +146,7 @@ func (c *NATSClient) BindKVStore(kvName string) (nats.KeyValue, error) {
 }
 
 // PendingForStream returns the number of pending messages for the given consumer and stream
-func (c *NATSClient) PendingForStream(consumer string, stream string) (int64, error) {
+func (c *Client) PendingForStream(consumer string, stream string) (int64, error) {
 	var (
 		err   error
 		cInfo *nats.ConsumerInfo
@@ -155,21 +163,21 @@ func (c *NATSClient) PendingForStream(consumer string, stream string) (int64, er
 }
 
 // JetStreamContext returns a new JetStreamContext
-func (c *NATSClient) JetStreamContext(opts ...nats.JSOpt) (nats.JetStreamContext, error) {
+func (c *Client) JetStreamContext(opts ...nats.JSOpt) (nats.JetStreamContext, error) {
 	return c.nc.JetStream(opts...)
 }
 
 // Close closes the NATS client
-func (c *NATSClient) Close() {
+func (c *Client) Close() {
 	c.nc.Close()
 }
 
 // NewTestClient creates a new NATS client for testing
 // only use this for testing
-func NewTestClient(t *testing.T, url string) *NATSClient {
+func NewTestClient(t *testing.T, url string) *Client {
 	nc, err := nats.Connect(url)
 	if err != nil {
 		panic(err)
 	}
-	return &NATSClient{nc: nc}
+	return &Client{nc: nc}
 }
diff --git a/pkg/shared/idlehandler/idlehandler.go b/pkg/shared/idlehandler/idlehandler.go
index 90381d97..1ef2c444 100644
--- a/pkg/shared/idlehandler/idlehandler.go
+++ b/pkg/shared/idlehandler/idlehandler.go
@@ -18,17 +18,19 @@ package idlehandler
 
 import (
 	"context"
+	"strconv"
 
 	"go.uber.org/zap"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 // PublishIdleWatermark publishes a ctrl message with isb.Kind set to WMB. We only send one ctrl message when
-func PublishIdleWatermark(ctx context.Context, toBufferPartition isb.BufferWriter, wmPublisher publish.Publisher, idleManager wmb.IdleManager, logger *zap.SugaredLogger, vertexType dfv1.VertexType, wm wmb.Watermark) {
+func PublishIdleWatermark(ctx context.Context, toBufferPartition isb.BufferWriter, wmPublisher publish.Publisher, idleManager wmb.IdleManager, logger *zap.SugaredLogger, vertexName string, pipelineName string, vertexType dfv1.VertexType, vertexReplica int32, wm wmb.Watermark) {
 
 	var toPartitionName = toBufferPartition.GetName()
 	var toVertexPartition = toBufferPartition.GetPartitionIdx()
@@ -53,6 +55,8 @@ func PublishIdleWatermark(ctx context.Context, toBufferPartition isb.BufferWrite
 				// we only write one ctrl message, so there's only one offset in the array, use index=0 to get the offset
 				idleManager.Update(toPartitionName, writeOffsets[0])
 			}
+			metrics.CtrlMessagesCount.With(map[string]string{metrics.LabelVertex: vertexName, metrics.LabelPipeline: pipelineName, metrics.LabelVertexType: string(vertexType), metrics.LabelVertexReplicaIndex: strconv.Itoa(int(vertexReplica)), metrics.LabelPartitionName: toPartitionName}).Add(1)
+
 		}
 	}
 
diff --git a/pkg/shared/kvs/jetstream/kv_store.go b/pkg/shared/kvs/jetstream/kv_store.go
index 19180b50..28d4f836 100644
--- a/pkg/shared/kvs/jetstream/kv_store.go
+++ b/pkg/shared/kvs/jetstream/kv_store.go
@@ -37,7 +37,7 @@ import (
 type jetStreamStore struct {
 	ctx               context.Context
 	kvName            string
-	client            *jsclient.NATSClient
+	client            *jsclient.Client
 	kv                nats.KeyValue
 	previousFetchTime time.Time
 	doneCh            chan struct{}
@@ -49,7 +49,7 @@ type jetStreamStore struct {
 var _ kvs.KVStorer = (*jetStreamStore)(nil)
 
 // NewKVJetStreamKVStore returns KVJetStreamStore.
-func NewKVJetStreamKVStore(ctx context.Context, kvName string, client *jsclient.NATSClient, opts ...Option) (kvs.KVStorer, error) {
+func NewKVJetStreamKVStore(ctx context.Context, kvName string, client *jsclient.Client, opts ...Option) (kvs.KVStorer, error) {
 	var err error
 
 	kvOpts := defaultOptions()
diff --git a/pkg/sideinputs/initializer/initializer.go b/pkg/sideinputs/initializer/initializer.go
index be0051dc..789145e7 100644
--- a/pkg/sideinputs/initializer/initializer.go
+++ b/pkg/sideinputs/initializer/initializer.go
@@ -56,7 +56,7 @@ func NewSideInputsInitializer(isbSvcType dfv1.ISBSvcType, pipelineName, sideInpu
 // and update the values on the disk. This would exit once all the side inputs are initialized.
 func (sii *sideInputsInitializer) Run(ctx context.Context) error {
 	var (
-		natsClient     *jsclient.NATSClient
+		natsClient     *jsclient.Client
 		err            error
 		sideInputStore kvs.KVStorer
 	)
diff --git a/pkg/sideinputs/manager/manager.go b/pkg/sideinputs/manager/manager.go
index c1c75179..bcbfcd70 100644
--- a/pkg/sideinputs/manager/manager.go
+++ b/pkg/sideinputs/manager/manager.go
@@ -58,7 +58,7 @@ func (sim *sideInputsManager) Start(ctx context.Context) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	var natsClient *jsclient.NATSClient
+	var natsClient *jsclient.Client
 	var err error
 	var siStore kvs.KVStorer
 	switch sim.isbSvcType {
diff --git a/pkg/sideinputs/synchronizer/synchronizer.go b/pkg/sideinputs/synchronizer/synchronizer.go
index 8b2d359b..884e294b 100644
--- a/pkg/sideinputs/synchronizer/synchronizer.go
+++ b/pkg/sideinputs/synchronizer/synchronizer.go
@@ -57,7 +57,7 @@ func NewSideInputsSynchronizer(isbSvcType dfv1.ISBSvcType, pipelineName, sideInp
 // and keeps on watching for updates for all the side inputs while writing the new values to the disk.
 func (sis *sideInputsSynchronizer) Start(ctx context.Context) error {
 	var (
-		natsClient     *jsclient.NATSClient
+		natsClient     *jsclient.Client
 		err            error
 		sideInputStore kvs.KVStorer
 	)
diff --git a/pkg/sinks/forward/forward.go b/pkg/sinks/forward/forward.go
index ad4cf643..7d3023ca 100644
--- a/pkg/sinks/forward/forward.go
+++ b/pkg/sinks/forward/forward.go
@@ -198,7 +198,7 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 		}
 
 		// if the validation passed, we will publish the watermark to all the toBuffer partitions.
-		idlehandler.PublishIdleWatermark(ctx, df.toBuffer, df.wmPublisher, df.idleManager, df.opts.logger, dfv1.VertexTypeSink, wmb.Watermark(time.UnixMilli(processorWMB.Watermark)))
+		idlehandler.PublishIdleWatermark(ctx, df.toBuffer, df.wmPublisher, df.idleManager, df.opts.logger, df.vertexName, df.pipelineName, dfv1.VertexTypeSink, df.vertexReplica, wmb.Watermark(time.UnixMilli(processorWMB.Watermark)))
 		return
 	}
 
diff --git a/pkg/sources/forward/data_forward.go b/pkg/sources/forward/data_forward.go
index bc3b93d2..c20155a8 100644
--- a/pkg/sources/forward/data_forward.go
+++ b/pkg/sources/forward/data_forward.go
@@ -236,7 +236,7 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 							publisher = df.createToVertexWatermarkPublisher(toVertexName, sp)
 							vertexPublishers[sp] = publisher
 						}
-						idlehandler.PublishIdleWatermark(ctx, df.toBuffers[toVertexName][index], publisher, df.idleManager, df.opts.logger, dfv1.VertexTypeSource, fetchedWm)
+						idlehandler.PublishIdleWatermark(ctx, df.toBuffers[toVertexName][index], publisher, df.idleManager, df.opts.logger, df.vertexName, df.pipelineName, dfv1.VertexTypeSource, df.vertexReplica, fetchedWm)
 					}
 				}
 			}
@@ -327,10 +327,6 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 		}
 	}
 
-	// print the latest event time for each partition
-	for partition, latestEt := range latestEtMap {
-		df.opts.logger.Infow("Latest event time for partition - ", zap.Int32("partition", partition), zap.Int64("latestEt", int64(time.Since(time.Unix(0, latestEt)).Minutes())))
-	}
 	// publish source watermark
 	df.srcWMPublisher.PublishSourceWatermarks(transformedReadMessages)
 	// update the watermark configs for lastTimestampSrcWMUpdated, lastFetchedSrcWatermark and lastTimestampIdleWMFound.
@@ -420,7 +416,7 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 							publisher = df.createToVertexWatermarkPublisher(toVertexName, sp)
 							vertexPublishers[sp] = publisher
 						}
-						idlehandler.PublishIdleWatermark(ctx, df.toBuffers[toVertexName][index], publisher, df.idleManager, df.opts.logger, dfv1.VertexTypeSource, processorWM)
+						idlehandler.PublishIdleWatermark(ctx, df.toBuffers[toVertexName][index], publisher, df.idleManager, df.opts.logger, df.vertexName, df.pipelineName, dfv1.VertexTypeSource, df.vertexReplica, processorWM)
 					}
 				}
 			}
diff --git a/pkg/sources/source.go b/pkg/sources/source.go
index af354a9a..55ccad99 100644
--- a/pkg/sources/source.go
+++ b/pkg/sources/source.go
@@ -115,9 +115,34 @@ func (sp *SourceProcessor) Start(ctx context.Context) error {
 			writersMap[e.To] = bufferWriters
 		}
 	case dfv1.ISBSvcTypeJetStream:
-		if sp.VertexInstance.Vertex.Spec.Watermark.Disabled {
-			// use default no op sourcePublisherStores, toVertexWatermarkStores, fetcher, idleManager
-		} else {
+		for _, e := range sp.VertexInstance.Vertex.Spec.ToEdges {
+			writeOpts := []jetstreamisb.WriteOption{
+				jetstreamisb.WithBufferFullWritingStrategy(e.BufferFullWritingStrategy()),
+			}
+			if x := e.ToVertexLimits; x != nil && x.BufferMaxLength != nil {
+				writeOpts = append(writeOpts, jetstreamisb.WithMaxLength(int64(*x.BufferMaxLength)))
+			}
+			if x := e.ToVertexLimits; x != nil && x.BufferUsageLimit != nil {
+				writeOpts = append(writeOpts, jetstreamisb.WithBufferUsageLimit(float64(*x.BufferUsageLimit)/100))
+			}
+			var bufferWriters []isb.BufferWriter
+			partitionedBuffers := dfv1.GenerateBufferNames(sp.VertexInstance.Vertex.Namespace, sp.VertexInstance.Vertex.Spec.PipelineName, e.To, e.GetToVertexPartitionCount())
+			// create a writer for each partition.
+			for partitionIdx, partition := range partitionedBuffers {
+				streamName := isbsvc.JetStreamName(partition)
+				jetStreamClient := natsClientPool.NextAvailableClient()
+				writer, err := jetstreamisb.NewJetStreamBufferWriter(ctx, jetStreamClient, partition, streamName, streamName, int32(partitionIdx), writeOpts...)
+				if err != nil {
+					return err
+				}
+				bufferWriters = append(bufferWriters, writer)
+			}
+			writersMap[e.To] = bufferWriters
+		}
+
+		// created watermark related components only if watermark is enabled
+		// otherwise no op will used
+		if !sp.VertexInstance.Vertex.Spec.Watermark.Disabled {
 			// build watermark stores for from vertex
 			sourceWmStores, err = jetstream.BuildFromVertexWatermarkStores(ctx, sp.VertexInstance, natsClientPool.NextAvailableClient())
 			if err != nil {
@@ -138,31 +163,6 @@ func (sp *SourceProcessor) Start(ctx context.Context) error {
 			if err != nil {
 				return err
 			}
-			for _, e := range sp.VertexInstance.Vertex.Spec.ToEdges {
-				writeOpts := []jetstreamisb.WriteOption{
-					jetstreamisb.WithBufferFullWritingStrategy(e.BufferFullWritingStrategy()),
-				}
-				if x := e.ToVertexLimits; x != nil && x.BufferMaxLength != nil {
-					writeOpts = append(writeOpts, jetstreamisb.WithMaxLength(int64(*x.BufferMaxLength)))
-				}
-				if x := e.ToVertexLimits; x != nil && x.BufferUsageLimit != nil {
-					writeOpts = append(writeOpts, jetstreamisb.WithBufferUsageLimit(float64(*x.BufferUsageLimit)/100))
-				}
-				var bufferWriters []isb.BufferWriter
-				partitionedBuffers := dfv1.GenerateBufferNames(sp.VertexInstance.Vertex.Namespace, sp.VertexInstance.Vertex.Spec.PipelineName, e.To, e.GetToVertexPartitionCount())
-				// create a writer for each partition.
-				for partitionIdx, partition := range partitionedBuffers {
-					streamName := isbsvc.JetStreamName(partition)
-					jetStreamClient := natsClientPool.NextAvailableClient()
-					writer, err := jetstreamisb.NewJetStreamBufferWriter(ctx, jetStreamClient, partition, streamName, streamName, int32(partitionIdx), writeOpts...)
-					if err != nil {
-						return err
-					}
-					bufferWriters = append(bufferWriters, writer)
-				}
-
-				writersMap[e.To] = bufferWriters
-			}
 			idleManager = wmb.NewIdleManager(len(writersMap))
 		}
 	default:
diff --git a/pkg/udf/forward/forward.go b/pkg/udf/forward/forward.go
index b0c51031..d28db484 100644
--- a/pkg/udf/forward/forward.go
+++ b/pkg/udf/forward/forward.go
@@ -228,7 +228,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		for toVertexName, toVertexBuffer := range isdf.toBuffers {
 			for _, partition := range toVertexBuffer {
 				if p, ok := isdf.wmPublishers[toVertexName]; ok {
-					idlehandler.PublishIdleWatermark(ctx, partition, p, isdf.idleManager, isdf.opts.logger, dfv1.VertexTypeMapUDF, wmb.Watermark(time.UnixMilli(processorWMB.Watermark)))
+					idlehandler.PublishIdleWatermark(ctx, partition, p, isdf.idleManager, isdf.opts.logger, isdf.vertexName, isdf.pipelineName, dfv1.VertexTypeMapUDF, isdf.vertexReplica, wmb.Watermark(time.UnixMilli(processorWMB.Watermark)))
 				}
 			}
 		}
@@ -378,7 +378,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 					// 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.wmPublishers[bufferName]; ok {
-						idlehandler.PublishIdleWatermark(ctx, isdf.toBuffers[bufferName][index], p, isdf.idleManager, isdf.opts.logger, dfv1.VertexTypeMapUDF, processorWM)
+						idlehandler.PublishIdleWatermark(ctx, isdf.toBuffers[bufferName][index], p, isdf.idleManager, isdf.opts.logger, isdf.vertexName, isdf.pipelineName, dfv1.VertexTypeMapUDF, isdf.vertexReplica, processorWM)
 					}
 				}
 			}
diff --git a/pkg/udf/map_udf.go b/pkg/udf/map_udf.go
index 9d221f03..37c119ad 100644
--- a/pkg/udf/map_udf.go
+++ b/pkg/udf/map_udf.go
@@ -88,14 +88,16 @@ func (u *MapUDFProcessor) Start(ctx context.Context) error {
 			return err
 		}
 	case dfv1.ISBSvcTypeJetStream:
-
-		// build watermark progressors
 		// multiple go routines can share the same set of writers since nats conn is thread safe
 		// https://github.com/nats-io/nats.go/issues/241
+		readers, writers, err = buildJetStreamBufferIO(ctx, u.VertexInstance, natsClientPool)
+		if err != nil {
+			return err
+		}
 
-		if u.VertexInstance.Vertex.Spec.Watermark.Disabled {
-			// use default no op fetcher, publisher, idleManager
-		} else {
+		// created watermark related components only if watermark is enabled
+		// otherwise no op will used
+		if !u.VertexInstance.Vertex.Spec.Watermark.Disabled {
 			// create from vertex watermark stores
 			fromVertexWmStores, err = jetstream.BuildFromVertexWatermarkStores(ctx, u.VertexInstance, natsClientPool.NextAvailableClient())
 			if err != nil {
@@ -115,10 +117,6 @@ func (u *MapUDFProcessor) Start(ctx context.Context) error {
 			// create watermark publisher using watermark stores
 			publishWatermark = jetstream.BuildPublishersFromStores(ctx, u.VertexInstance, toVertexWmStores)
 
-			readers, writers, err = buildJetStreamBufferIO(ctx, u.VertexInstance, natsClientPool)
-			if err != nil {
-				return err
-			}
 			idleManager = wmb.NewIdleManager(len(writers))
 		}
 	default:
@@ -243,13 +241,13 @@ func (u *MapUDFProcessor) Start(ctx context.Context) error {
 			}
 		}
 		// create a forwarder for each partition
-		forwarder, err := forward.NewInterStepDataForward(u.VertexInstance, readers[index], writers, conditionalForwarder, mapHandler, mapStreamHandler, fetchWatermark, publishWatermark, idleManager, opts...)
+		df, err := forward.NewInterStepDataForward(u.VertexInstance, readers[index], writers, conditionalForwarder, mapHandler, mapStreamHandler, fetchWatermark, publishWatermark, idleManager, opts...)
 		if err != nil {
 			return err
 		}
 		finalWg.Add(1)
 
-		// start the forwarder for each partition using a go routine
+		// start the df for each partition using a go routine
 		go func(fromBufferPartitionName string, isdf *forward.InterStepDataForward) {
 			defer finalWg.Done()
 			log.Infow("Start processing udf messages", zap.String("isbsvc", string(u.ISBSvcType)), zap.String("from", fromBufferPartitionName), zap.Any("to", u.VertexInstance.Vertex.GetToBuffers()))
@@ -271,7 +269,7 @@ func (u *MapUDFProcessor) Start(ctx context.Context) error {
 			isdf.Stop()
 			wg.Wait()
 			log.Info("Exited for partition...", zap.String("partition", fromBufferPartitionName))
-		}(bufferPartition, forwarder)
+		}(bufferPartition, df)
 	}
 	// create lag readers from buffer readers
 	var lagReaders []isb.LagReader
diff --git a/pkg/udf/reduce_udf.go b/pkg/udf/reduce_udf.go
index 8a7606b3..5f896905 100644
--- a/pkg/udf/reduce_udf.go
+++ b/pkg/udf/reduce_udf.go
@@ -108,7 +108,7 @@ func (u *ReduceUDFProcessor) Start(ctx context.Context) error {
 			return fmt.Errorf("failed to create a new reducer gRPC client: %w", err)
 		}
 
-		reduceHandler := rpc.NewUDSgRPCAlignedReduce(client)
+		reduceHandler := rpc.NewUDSgRPCAlignedReduce(u.VertexInstance.Vertex.Name, u.VertexInstance.Replica, client)
 		// Readiness check
 		if err := reduceHandler.WaitUntilReady(ctx); err != nil {
 			return fmt.Errorf("failed on udf readiness check, %w", err)
@@ -185,10 +185,14 @@ func (u *ReduceUDFProcessor) Start(ctx context.Context) error {
 			return err
 		}
 	case dfv1.ISBSvcTypeJetStream:
-		// build watermark progressors
-		if u.VertexInstance.Vertex.Spec.Watermark.Disabled {
-			// use　default no op fetcher, publisher, idleManager
-		} else {
+		readers, writers, err = buildJetStreamBufferIO(ctx, u.VertexInstance, natsClientPool)
+		if err != nil {
+			return err
+		}
+
+		// created watermark related components only if watermark is enabled
+		// otherwise no op will used
+		if !u.VertexInstance.Vertex.Spec.Watermark.Disabled {
 			// create from vertex watermark stores
 			fromVertexWmStores, err = jetstream.BuildFromVertexWatermarkStores(ctx, u.VertexInstance, natsClientPool.NextAvailableClient())
 			if err != nil {
@@ -208,10 +212,6 @@ func (u *ReduceUDFProcessor) Start(ctx context.Context) error {
 			// create watermark publisher using watermark stores
 			publishWatermark = jetstream.BuildPublishersFromStores(ctx, u.VertexInstance, toVertexWmStores)
 
-			readers, writers, err = buildJetStreamBufferIO(ctx, u.VertexInstance, natsClientPool)
-			if err != nil {
-				return err
-			}
 			idleManager = wmb.NewIdleManager(len(writers))
 		}
 	default:
diff --git a/pkg/udf/rpc/grpc_aligned_reduce.go b/pkg/udf/rpc/grpc_aligned_reduce.go
index bb363c4a..d4522dbf 100644
--- a/pkg/udf/rpc/grpc_aligned_reduce.go
+++ b/pkg/udf/rpc/grpc_aligned_reduce.go
@@ -40,11 +40,17 @@ import (
 
 // GRPCBasedAlignedReduce is a reduce applier that uses gRPC client to invoke the aligned reduce UDF. It implements the applier.ReduceApplier interface.
 type GRPCBasedAlignedReduce struct {
-	client reducer.Client
+	vertexName    string
+	vertexReplica int32
+	client        reducer.Client
 }
 
-func NewUDSgRPCAlignedReduce(client reducer.Client) *GRPCBasedAlignedReduce {
-	return &GRPCBasedAlignedReduce{client: client}
+func NewUDSgRPCAlignedReduce(vertexName string, vertexReplica int32, client reducer.Client) *GRPCBasedAlignedReduce {
+	return &GRPCBasedAlignedReduce{
+		vertexName:    vertexName,
+		vertexReplica: vertexReplica,
+		client:        client,
+	}
 }
 
 // IsHealthy checks if the map udf is healthy.
@@ -94,6 +100,7 @@ func (u *GRPCBasedAlignedReduce) ApplyReduce(ctx context.Context, partitionID *p
 	// invoke the AsyncReduceFn method with reduceRequests channel and send the result to responseCh channel
 	// and any error to errCh channel
 	go func() {
+		index := 0
 		resultCh, reduceErrCh := u.client.ReduceFn(grpcCtx, reduceRequests)
 		for {
 			select {
@@ -104,7 +111,10 @@ func (u *GRPCBasedAlignedReduce) ApplyReduce(ctx context.Context, partitionID *p
 					close(responseCh)
 					return
 				}
-				responseCh <- parseReduceResponse(result)
+				// create a unique message id for each response message which will be used for deduplication
+				msgId := fmt.Sprintf("%s-%d-%s-%d", u.vertexName, u.vertexReplica, partitionID.String(), index)
+				index++
+				responseCh <- parseReduceResponse(result, msgId)
 			case err := <-reduceErrCh:
 				// ctx.Done() event will be handled by the AsyncReduceFn method
 				// so we don't need a separate case for ctx.Done() here
@@ -230,7 +240,7 @@ func convertToUdfError(err error) ApplyUDFErr {
 }
 
 // parseReduceResponse parse the SDK response to TimedWindowResponse
-func parseReduceResponse(response *reducepb.ReduceResponse) *window.TimedWindowResponse {
+func parseReduceResponse(response *reducepb.ReduceResponse, msgId string) *window.TimedWindowResponse {
 	taggedMessage := &isb.WriteMessage{
 		Message: isb.Message{
 			Header: isb.Header{
@@ -239,6 +249,7 @@ func parseReduceResponse(response *reducepb.ReduceResponse) *window.TimedWindowR
 					IsLate:    false,
 				},
 				Keys: response.GetResult().GetKeys(),
+				ID:   msgId,
 			},
 			Body: isb.Body{
 				Payload: response.GetResult().GetValue(),
diff --git a/pkg/udf/rpc/grpc_unaligned_reduce.go b/pkg/udf/rpc/grpc_unaligned_reduce.go
index 6fce6d9b..c95c115f 100644
--- a/pkg/udf/rpc/grpc_unaligned_reduce.go
+++ b/pkg/udf/rpc/grpc_unaligned_reduce.go
@@ -211,6 +211,7 @@ func parseSessionReduceResponse(response *sessionreducepb.SessionReduceResponse)
 			Body: isb.Body{
 				Payload: result.GetValue(),
 			},
+			// FIXME: dedup for unaligned windows
 		},
 		Tags: result.GetTags(),
 	}
diff --git a/pkg/watermark/generic/jetstream/generic.go b/pkg/watermark/generic/jetstream/generic.go
index 66debe30..1359b705 100644
--- a/pkg/watermark/generic/jetstream/generic.go
+++ b/pkg/watermark/generic/jetstream/generic.go
@@ -30,7 +30,7 @@ import (
 )
 
 // BuildFromVertexWatermarkStores creates a map of WatermarkStores for all the incoming edges of the given Vertex.
-func BuildFromVertexWatermarkStores(ctx context.Context, vertexInstance *v1alpha1.VertexInstance, client *jsclient.NATSClient) (map[string]store.WatermarkStore, error) {
+func BuildFromVertexWatermarkStores(ctx context.Context, vertexInstance *v1alpha1.VertexInstance, client *jsclient.Client) (map[string]store.WatermarkStore, error) {
 	var wmStores = make(map[string]store.WatermarkStore)
 	vertex := vertexInstance.Vertex
 
@@ -58,7 +58,7 @@ func BuildFromVertexWatermarkStores(ctx context.Context, vertexInstance *v1alpha
 }
 
 // BuildToVertexWatermarkStores creates a map of WatermarkStore for all the to buckets of the given vertex.
-func BuildToVertexWatermarkStores(ctx context.Context, vertexInstance *v1alpha1.VertexInstance, client *jsclient.NATSClient) (map[string]store.WatermarkStore, error) {
+func BuildToVertexWatermarkStores(ctx context.Context, vertexInstance *v1alpha1.VertexInstance, client *jsclient.Client) (map[string]store.WatermarkStore, error) {
 	var wmStores = make(map[string]store.WatermarkStore)
 	vertex := vertexInstance.Vertex
 
@@ -111,7 +111,7 @@ func BuildPublishersFromStores(ctx context.Context, vertexInstance *v1alpha1.Ver
 }
 
 // BuildSourcePublisherStores builds the watermark stores for source publisher.
-func BuildSourcePublisherStores(ctx context.Context, vertexInstance *v1alpha1.VertexInstance, client *jsclient.NATSClient) (store.WatermarkStore, error) {
+func BuildSourcePublisherStores(ctx context.Context, vertexInstance *v1alpha1.VertexInstance, client *jsclient.Client) (store.WatermarkStore, error) {
 	if !vertexInstance.Vertex.IsASource() {
 		return nil, fmt.Errorf("not a source vertex")
 	}
diff --git a/pkg/watermark/store/watermark_store.go b/pkg/watermark/store/watermark_store.go
index d568e476..4a493870 100644
--- a/pkg/watermark/store/watermark_store.go
+++ b/pkg/watermark/store/watermark_store.go
@@ -69,7 +69,7 @@ func BuildInmemWatermarkStore(ctx context.Context, bucket string) (WatermarkStor
 }
 
 // BuildJetStreamWatermarkStore returns a JetStream WatermarkStore instance
-func BuildJetStreamWatermarkStore(ctx context.Context, bucket string, client *jsclient.NATSClient) (WatermarkStore, error) {
+func BuildJetStreamWatermarkStore(ctx context.Context, bucket string, client *jsclient.Client) (WatermarkStore, error) {
 	// build heartBeat store
 	hbKVName := JetStreamProcessorKVName(bucket)
 	hbStore, err := jetstream.NewKVJetStreamKVStore(ctx, hbKVName, client)


Test patch:
diff --git a/pkg/shared/clients/nats/test/client.go b/pkg/shared/clients/nats/test/client.go
index b58427ab..01bfef5a 100644
--- a/pkg/shared/clients/nats/test/client.go
+++ b/pkg/shared/clients/nats/test/client.go
@@ -25,6 +25,6 @@ import (
 )
 
 // JetStreamClient is used to get a testing JetStream client instance
-func JetStreamClient(t *testing.T, s *server.Server) *nats.NATSClient {
+func JetStreamClient(t *testing.T, s *server.Server) *nats.Client {
 	return nats.NewTestClient(t, s.ClientURL())
 }
diff --git a/pkg/udf/rpc/grpc_aligned_reduce_test.go b/pkg/udf/rpc/grpc_aligned_reduce_test.go
index c90e7b76..1288c21b 100644
--- a/pkg/udf/rpc/grpc_aligned_reduce_test.go
+++ b/pkg/udf/rpc/grpc_aligned_reduce_test.go
@@ -38,7 +38,7 @@ import (
 
 func NewMockUDSGRPCBasedReduce(mockClient *reducemock.MockReduceClient) *GRPCBasedAlignedReduce {
 	c, _ := reducer.NewFromClient(mockClient)
-	return &GRPCBasedAlignedReduce{c}
+	return &GRPCBasedAlignedReduce{"test", 0, c}
 }
 
 func TestGRPCBasedReduce_WaitUntilReadyWithMockClient(t *testing.T) {
diff --git a/test/http-e2e/testdata/http-source.yaml b/test/http-e2e/testdata/http-source.yaml
index f2bc6281..e807c9fc 100644
--- a/test/http-e2e/testdata/http-source.yaml
+++ b/test/http-e2e/testdata/http-source.yaml
@@ -3,6 +3,8 @@ kind: Pipeline
 metadata:
   name: http-source
 spec:
+  watermark:
+    disabled: true
   vertices:
     - name: in
       source: