Repository: numaproj/numaflow
Pull request #736 (2023-05-13 19:57:35)

test: make Test* works for all e2e test cases
Explain what this PR does.

<!--

Before you push your changes:

* Run `make pre-push -B` to fix codegen and lint problems.

Then, you MUST:

* Sign-off your commit (otherwise the DCO check will fail).
* Use [a conventional commit message](https://www.conventionalcommits.org/en/v1.0.0/) (otherwise the commit message check will fail).

If you did not do this, reset all your commit and replace them with a single commit:

```
git reset HEAD~1 ;# change 1 to how many commits you made
git commit --signoff -m 'feat: my feat. Fixes #1234'
```

When creating your PR: 

* Make sure that "Fixes #" or "Closes #" is in both the PR title (for release notes) and description (to automatically link and close the issue).
* Say how you tested your changes. If you changed the UI, attach screenshots.
* Set your PR as a draft initially.
* Your PR needs to pass the required checks before it can be approved. 
* Once required tests have passed, mark your PR "Ready for review".

If changes were requested, once you've made them, you MUST dismiss the review to get it reviewed again.

-->



Patch:
diff --git a/Makefile b/Makefile
index 0c414479..69ba5a11 100644
--- a/Makefile
+++ b/Makefile
@@ -140,7 +140,7 @@ Test%:
 	kubectl -n numaflow-system delete po -lapp.kubernetes.io/component=controller-manager,app.kubernetes.io/part-of=numaflow
 	kubectl -n numaflow-system delete po e2e-api-pod  --ignore-not-found=true
 	cat test/manifests/e2e-api-pod.yaml |  sed 's@quay.io/numaproj/@$(IMAGE_NAMESPACE)/@' | sed 's/:$(BASE_VERSION)/:$(VERSION)/' | kubectl -n numaflow-system apply -f -
-	-go test -v -timeout 10m -count 1 --tags test -p 1 ./test/e2e-suite-1  -run='.*/$*'
+	-go test -v -timeout 10m -count 1 --tags test -p 1 ./test/$(shell grep $(*) -R ./test | head -1 | awk -F\/ '{print $$3}' ) -run='.*/$*'
 	$(MAKE) cleanup-e2e
 
 .PHONY: ui-build
diff --git a/pkg/reduce/data_forward.go b/pkg/reduce/data_forward.go
index 583d883b..549c0539 100644
--- a/pkg/reduce/data_forward.go
+++ b/pkg/reduce/data_forward.go
@@ -14,55 +14,69 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// Package reduce reads messages from isb
-// attaches watermark to read messages
-// invoke the read-loop with the read messages
+// Package reduce reads messages from isb and is responsible for the first part of reduce subsystem.
+// It is responsible for feeding in the data to the second part of the reduce subsystem called processAndForward
+// and processes the read data from the ISB, writes to the outbound channel called PBQ so the message can be
+// asynchronously processed by `processAndForward`, and then closes the partition if possible based on watermark
+// progression. To write to the outbound channel (PBQ), it has to partition the message first. Hence,
+// also contains partitioning logic. Partitioner identifies a set of elements with a common key and time, and buckets
+// them in to a common window. A partition is uniquely identified using a tuple {window, key}. Type of window does not
+// matter. Partitioner is responsible for managing the persistence and processing of each partition. It uses PBQ for
+// durable persistence of elements that belong to a partition and orchestrates the processing of elements using processAndForward
+// function. Partitioner tracks active partitions, closes the partitions based on watermark progression and co-ordinates the
+// materialization and forwarding the results to the next vertex in the pipeline.
 package reduce
 
 import (
 	"context"
+	"math"
 	"strconv"
+	"sync"
 	"time"
 
 	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/util/wait"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/forward"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/metrics"
-	"github.com/numaproj/numaflow/pkg/reduce/applier"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq"
-	"github.com/numaproj/numaflow/pkg/reduce/readloop"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/reduce/pnf"
 	"github.com/numaproj/numaflow/pkg/shared/idlehandler"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 	"github.com/numaproj/numaflow/pkg/window"
+	"github.com/numaproj/numaflow/pkg/window/keyed"
 )
 
-// DataForward reads data from isb and forwards them to readloop
+// DataForward is responsible for reading and forwarding the message from ISB to PBQ.
 type DataForward struct {
-	ctx                 context.Context
-	vertexName          string
-	pipelineName        string
-	vertexReplica       int32
-	fromBuffer          isb.BufferReader
-	toBuffers           map[string]isb.BufferWriter
-	readloop            *readloop.ReadLoop
-	watermarkFetcher    fetch.Fetcher
-	watermarkPublishers map[string]publish.Publisher
-	windowingStrategy   window.Windower
-	keyed               bool
-	idleManager         *wmb.IdleManager
-	wmbChecker          wmb.WMBChecker
-	pbqManager          *pbq.Manager
-	opts                *Options
-	log                 *zap.SugaredLogger
+	ctx                   context.Context
+	vertexName            string
+	pipelineName          string
+	vertexReplica         int32
+	fromBuffer            isb.BufferReader
+	toBuffers             map[string]isb.BufferWriter
+	watermarkFetcher      fetch.Fetcher
+	watermarkPublishers   map[string]publish.Publisher
+	windower              window.Windower
+	keyed                 bool
+	idleManager           *wmb.IdleManager
+	wmbChecker            wmb.WMBChecker
+	pbqManager            *pbq.Manager
+	whereToDecider        forward.ToWhichStepDecider
+	udfInvocationTracking map[partition.ID]*pnf.ForwardTask
+	of                    *pnf.OrderedProcessor
+	opts                  *Options
+	log                   *zap.SugaredLogger
 }
 
+// NewDataForward creates a new DataForward
 func NewDataForward(ctx context.Context,
-	udf applier.ReduceApplier,
 	vertexInstance *dfv1.VertexInstance,
 	fromBuffer isb.BufferReader,
 	toBuffers map[string]isb.BufferWriter,
@@ -71,6 +85,8 @@ func NewDataForward(ctx context.Context,
 	fw fetch.Fetcher,
 	watermarkPublishers map[string]publish.Publisher,
 	windowingStrategy window.Windower,
+	idleManager *wmb.IdleManager,
+	of *pnf.OrderedProcessor,
 	opts ...Option) (*DataForward, error) {
 
 	options := DefaultOptions()
@@ -81,116 +97,120 @@ func NewDataForward(ctx context.Context,
 		}
 	}
 
-	idleManager := wmb.NewIdleManager(len(toBuffers))
-
-	rl, err := readloop.NewReadLoop(ctx, vertexInstance.Vertex.Spec.Name, vertexInstance.Vertex.Spec.PipelineName,
-		vertexInstance.Replica, udf, pbqManager, windowingStrategy, toBuffers, whereToDecider, watermarkPublishers,
-		idleManager, options.allowedLateness)
-
-	df := &DataForward{
-		ctx:                 ctx,
-		vertexName:          vertexInstance.Vertex.Spec.Name,
-		pipelineName:        vertexInstance.Vertex.Spec.PipelineName,
-		vertexReplica:       vertexInstance.Replica,
-		fromBuffer:          fromBuffer,
-		toBuffers:           toBuffers,
-		readloop:            rl,
-		watermarkFetcher:    fw,
-		watermarkPublishers: watermarkPublishers,
-		windowingStrategy:   windowingStrategy,
-		keyed:               vertexInstance.Vertex.Spec.UDF.GroupBy.Keyed,
-		idleManager:         idleManager,
-		pbqManager:          pbqManager,
-		wmbChecker:          wmb.NewWMBChecker(2), // TODO: make configurable
-		log:                 logging.FromContext(ctx),
-		opts:                options}
-
-	return df, err
+	rl := &DataForward{
+		ctx:                   ctx,
+		vertexName:            vertexInstance.Vertex.Spec.Name,
+		pipelineName:          vertexInstance.Vertex.Spec.PipelineName,
+		vertexReplica:         vertexInstance.Replica,
+		fromBuffer:            fromBuffer,
+		toBuffers:             toBuffers,
+		watermarkFetcher:      fw,
+		watermarkPublishers:   watermarkPublishers,
+		windower:              windowingStrategy,
+		keyed:                 vertexInstance.Vertex.Spec.UDF.GroupBy.Keyed,
+		idleManager:           idleManager,
+		pbqManager:            pbqManager,
+		whereToDecider:        whereToDecider,
+		udfInvocationTracking: make(map[partition.ID]*pnf.ForwardTask),
+		of:                    of,
+		wmbChecker:            wmb.NewWMBChecker(2), // TODO: make configurable
+		log:                   logging.FromContext(ctx),
+		opts:                  options}
+
+	return rl, nil
 }
 
-// Start starts forwarding messages to readloop
-func (d *DataForward) Start() {
+// Start starts reading messages from ISG
+func (df *DataForward) Start() {
 	for {
 		select {
-		case <-d.ctx.Done():
-			d.log.Infow("Stopping reduce data forwarder...")
-
-			if err := d.fromBuffer.Close(); err != nil {
-				d.log.Errorw("Failed to close buffer reader, shutdown anyways...", zap.Error(err))
-			} else {
-				d.log.Infow("Closed buffer reader", zap.String("bufferFrom", d.fromBuffer.GetName()))
-			}
+		case <-df.ctx.Done():
 
 			// hard shutdown after timeout
 			cctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 			defer cancel()
-			// allow readloop to clean itself up.
-			d.readloop.ShutDown(cctx)
-
-			for _, v := range d.toBuffers {
-				if err := v.Close(); err != nil {
-					d.log.Errorw("Failed to close buffer writer, shutdown anyways...", zap.Error(err), zap.String("bufferTo", v.GetName()))
-				} else {
-					d.log.Infow("Closed buffer writer", zap.String("bufferTo", v.GetName()))
-				}
-			}
+			// allow to clean itself up.
+			df.ShutDown(cctx)
 
-			// stop watermark fetcher
-			if err := d.watermarkFetcher.Close(); err != nil {
-				d.log.Errorw("Failed to close watermark fetcher", zap.Error(err))
-			}
-
-			// stop watermark publisher
-			for _, publisher := range d.watermarkPublishers {
-				if err := publisher.Close(); err != nil {
-					d.log.Errorw("Failed to close watermark publisher", zap.Error(err))
-				}
-			}
 			return
 		default:
-			// pass the child context so that the reader can be closed before the readloop
+			// pass the child context so that the reader can be closed.
 			// this way we can avoid the race condition and have all the read messages persisted
 			// and acked.
-			d.forwardAChunk(d.ctx)
+			df.forwardAChunk(df.ctx)
 		}
 	}
 }
 
+// ReplayPersistedMessages replays persisted messages, because during boot up, it has to replay the data from the persistent store of
+// PBQ before it can start reading from ISB. ReplayPersistedMessages will return only after the replay has been completed.
+func (df *DataForward) ReplayPersistedMessages(ctx context.Context) error {
+	// start the PBQManager which discovers and builds the state from persistent store of the PBQ.
+	partitions, err := df.pbqManager.GetExistingPartitions(ctx)
+	if err != nil {
+		return err
+	}
+
+	df.log.Infow("Partitions to be replayed ", zap.Int("count", len(partitions)), zap.Any("partitions", partitions))
+
+	for _, p := range partitions {
+		// Create keyed window for a given partition
+		// so that the window can be closed when the watermark
+		// crosses the window.
+
+		alignedKeyedWindow := keyed.NewKeyedWindow(p.Start, p.End)
+
+		// insert the window to the list of active windows, since the active window list is in-memory
+		keyedWindow, _ := df.windower.InsertIfNotPresent(alignedKeyedWindow)
+
+		// add slots to the window, so that when a new message with the watermark greater than
+		// the window end time comes, slots will not be lost and the windows will be closed as expected
+		keyedWindow.AddSlot(p.Slot)
+
+		// create and invoke process and forward for the partition
+		df.associatePBQAndPnF(ctx, p, keyedWindow)
+	}
+
+	// replays the data (replay internally writes the data from persistent store on to the PBQ)
+	df.pbqManager.Replay(ctx)
+	return nil
+}
+
 // forwardAChunk reads a chunk of messages from isb and assigns watermark to messages
-// and forwards the messages to readloop
-func (d *DataForward) forwardAChunk(ctx context.Context) {
-	readMessages, err := d.fromBuffer.Read(ctx, d.opts.readBatchSize)
+// and writes the messages to pbq
+func (df *DataForward) forwardAChunk(ctx context.Context) {
+	readMessages, err := df.fromBuffer.Read(ctx, df.opts.readBatchSize)
 	totalBytes := 0
 	if err != nil {
-		d.log.Errorw("Failed to read from isb", zap.Error(err))
+		df.log.Errorw("Failed to read from isb", zap.Error(err))
 		readMessagesError.With(map[string]string{
-			metrics.LabelVertex:             d.vertexName,
-			metrics.LabelPipeline:           d.pipelineName,
-			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(d.vertexReplica)),
-			"buffer":                        d.fromBuffer.GetName()}).Inc()
+			metrics.LabelVertex:             df.vertexName,
+			metrics.LabelPipeline:           df.pipelineName,
+			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+			"buffer":                        df.fromBuffer.GetName()}).Inc()
 	}
 
 	if len(readMessages) == 0 {
 		// we use the HeadWMB as the watermark for the idle
-		var processorWMB = d.watermarkFetcher.GetHeadWMB()
-		if !d.wmbChecker.ValidateHeadWMB(processorWMB) {
+		var processorWMB = df.watermarkFetcher.GetHeadWMB()
+		if !df.wmbChecker.ValidateHeadWMB(processorWMB) {
 			// validation failed, skip publishing
-			d.log.Debugw("skip publishing idle watermark",
-				zap.Int("counter", d.wmbChecker.GetCounter()),
+			df.log.Debugw("skip publishing idle watermark",
+				zap.Int("counter", df.wmbChecker.GetCounter()),
 				zap.Int64("offset", processorWMB.Offset),
 				zap.Int64("watermark", processorWMB.Watermark),
 				zap.Bool("Idle", processorWMB.Idle))
 			return
 		}
 
-		nextWin := d.pbqManager.NextWindowToBeClosed()
+		nextWin := df.pbqManager.NextWindowToBeClosed()
 		if nextWin == nil {
 			// if all the windows are closed already, and the len(readBatch) == 0
 			// then it means there's an idle situation
 			// in this case, send idle watermark to all the toBuffers
-			for _, toBuffer := range d.toBuffers {
-				if publisher, ok := d.watermarkPublishers[toBuffer.GetName()]; ok {
-					idlehandler.PublishIdleWatermark(ctx, toBuffer, publisher, d.idleManager, d.log, dfv1.VertexTypeReduceUDF, wmb.Watermark(time.UnixMilli(processorWMB.Watermark)))
+			for _, toBuffer := range df.toBuffers {
+				if publisher, ok := df.watermarkPublishers[toBuffer.GetName()]; ok {
+					idlehandler.PublishIdleWatermark(ctx, toBuffer, publisher, df.idleManager, df.log, dfv1.VertexTypeReduceUDF, wmb.Watermark(time.UnixMilli(processorWMB.Watermark)))
 				}
 			}
 		} else {
@@ -198,17 +218,17 @@ func (d *DataForward) forwardAChunk(ctx context.Context) {
 			// then it means the overall dataflow of the pipeline has already reached a later time point
 			// so we can close the window and process the data in this window
 			if watermark := time.UnixMilli(processorWMB.Watermark).Add(-1 * time.Millisecond); nextWin.EndTime().Before(watermark) {
-				closedWindows := d.windowingStrategy.RemoveWindows(watermark)
+				closedWindows := df.windower.RemoveWindows(watermark)
 				for _, win := range closedWindows {
-					d.readloop.ClosePartitions(win.Partitions())
+					df.ClosePartitions(win.Partitions())
 				}
 			} else {
 				// if toBeClosed window exists, but the watermark we fetch is still within the endTime of the window
 				// then we can't close the window because there could still be data after the idling situation ends
 				// so in this case, we publish an idle watermark
-				for _, toBuffer := range d.toBuffers {
-					if publisher, ok := d.watermarkPublishers[toBuffer.GetName()]; ok {
-						idlehandler.PublishIdleWatermark(ctx, toBuffer, publisher, d.idleManager, d.log, dfv1.VertexTypeReduceUDF, wmb.Watermark(watermark))
+				for _, toBuffer := range df.toBuffers {
+					if publisher, ok := df.watermarkPublishers[toBuffer.GetName()]; ok {
+						idlehandler.PublishIdleWatermark(ctx, toBuffer, publisher, df.idleManager, df.log, dfv1.VertexTypeReduceUDF, wmb.Watermark(watermark))
 					}
 				}
 			}
@@ -216,16 +236,16 @@ func (d *DataForward) forwardAChunk(ctx context.Context) {
 		return
 	}
 	readMessagesCount.With(map[string]string{
-		metrics.LabelVertex:             d.vertexName,
-		metrics.LabelPipeline:           d.pipelineName,
-		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(d.vertexReplica)),
-		"buffer":                        d.fromBuffer.GetName(),
+		metrics.LabelVertex:             df.vertexName,
+		metrics.LabelPipeline:           df.pipelineName,
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+		"buffer":                        df.fromBuffer.GetName(),
 	}).Add(float64(len(readMessages)))
 	// fetch watermark using the first element's watermark, because we assign the watermark to all other
 	// elements in the batch based on the watermark we fetch from 0th offset.
-	processorWM := d.watermarkFetcher.GetWatermark(readMessages[0].ReadOffset)
+	processorWM := df.watermarkFetcher.GetWatermark(readMessages[0].ReadOffset)
 	for _, m := range readMessages {
-		if !d.keyed {
+		if !df.keyed {
 			m.Keys = []string{dfv1.DefaultKeyForNonKeyedData}
 			m.Message.Keys = []string{dfv1.DefaultKeyForNonKeyedData}
 		}
@@ -233,12 +253,359 @@ func (d *DataForward) forwardAChunk(ctx context.Context) {
 		totalBytes += len(m.Payload)
 	}
 	readBytesCount.With(map[string]string{
-		metrics.LabelVertex:             d.vertexName,
-		metrics.LabelPipeline:           d.pipelineName,
-		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(d.vertexReplica)),
-		"buffer":                        d.fromBuffer.GetName(),
+		metrics.LabelVertex:             df.vertexName,
+		metrics.LabelPipeline:           df.pipelineName,
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+		"buffer":                        df.fromBuffer.GetName(),
 	}).Add(float64(totalBytes))
 
 	// readMessages has to be written to PBQ, acked, etc.
-	d.readloop.Process(ctx, readMessages)
+	df.Process(ctx, readMessages)
+}
+
+// associatePBQAndPnF associates a PBQ with the partition if a PBQ exists, else creates a new one and then associates
+// it to the partition.
+func (df *DataForward) associatePBQAndPnF(ctx context.Context, partitionID partition.ID, kw window.AlignedKeyedWindower) pbq.ReadWriteCloser {
+	// look for existing pbq
+	q := df.pbqManager.GetPBQ(partitionID)
+
+	// if we do not have already created PBQ, we have to create a new one.
+	if q == nil {
+		var pbqErr error
+		var infiniteBackoff = wait.Backoff{
+			Steps:    math.MaxInt,
+			Duration: 1 * time.Second,
+			Factor:   1.5,
+			Jitter:   0.1,
+		}
+		pbqErr = wait.ExponentialBackoffWithContext(ctx, infiniteBackoff, func() (done bool, err error) {
+			var attempt int
+			q, pbqErr = df.pbqManager.CreateNewPBQ(ctx, partitionID, kw)
+			if pbqErr != nil {
+				attempt += 1
+				df.log.Warnw("Failed to create pbq during startup, retrying", zap.Any("attempt", attempt), zap.String("partitionID", partitionID.String()), zap.Error(pbqErr))
+				return false, nil
+			}
+			return true, nil
+		})
+		// since we created a brand new PBQ it means there is no PnF listening on this PBQ.
+		// we should create and attach the read side of the loop (PnF) to the partition and then
+		// start process-and-forward (pnf) loop
+		t := df.of.SchedulePnF(ctx, partitionID)
+		df.udfInvocationTracking[partitionID] = t
+		df.log.Debugw("Successfully Created/Found pbq and started PnF", zap.String("partitionID", partitionID.String()))
+	}
+
+	return q
+}
+
+// Process is one iteration of the read loop which writes the messages to the PBQs followed by acking the messages, and
+// then closing the windows that can closed.
+func (df *DataForward) Process(ctx context.Context, messages []*isb.ReadMessage) {
+	var dataMessages = make([]*isb.ReadMessage, 0, len(messages))
+	var ctrlMessages = make([]*isb.ReadMessage, 0) // for a high TPS pipeline, 0 is the most optimal value
+
+	for _, message := range messages {
+		if message.Kind == isb.Data {
+			dataMessages = append(dataMessages, message)
+		} else {
+			ctrlMessages = append(ctrlMessages, message)
+		}
+	}
+
+	// write messages to windows based by PBQs.
+	successfullyWrittenMessages, err := df.writeMessagesToWindows(ctx, dataMessages)
+	if err != nil {
+		df.log.Errorw("Failed to write messages", zap.Int("totalMessages", len(messages)), zap.Int("writtenMessage", len(successfullyWrittenMessages)))
+	}
+
+	// ack the control messages
+	if len(ctrlMessages) != 0 {
+		df.ackMessages(ctx, ctrlMessages)
+	}
+
+	if len(successfullyWrittenMessages) == 0 {
+		return
+	}
+	// ack successful messages
+	df.ackMessages(ctx, successfullyWrittenMessages)
+
+	// close any windows that need to be closed.
+	// since the watermark will be same for all the messages in the batch
+	// we can invoke remove windows only once per batch
+	var closedWindows []window.AlignedKeyedWindower
+	wm := wmb.Watermark(successfullyWrittenMessages[0].Watermark)
+
+	closedWindows = df.windower.RemoveWindows(time.Time(wm).Add(-1 * df.opts.allowedLateness))
+
+	df.log.Debugw("Windows eligible for closing", zap.Int("length", len(closedWindows)), zap.Time("watermark", time.Time(wm)))
+
+	for _, cw := range closedWindows {
+		partitions := cw.Partitions()
+		df.ClosePartitions(partitions)
+		df.log.Debugw("Closing Window", zap.Int64("windowStart", cw.StartTime().UnixMilli()), zap.Int64("windowEnd", cw.EndTime().UnixMilli()))
+	}
+
+	// solve Reduce withholding of watermark where we do not send WM until the window is closed.
+	if nextWin := df.pbqManager.NextWindowToBeClosed(); nextWin != nil {
+		// minus 1 ms because if it's the same as the end time the window would have already been closed
+		if watermark := time.Time(wm).Add(-1 * time.Millisecond); nextWin.EndTime().After(watermark) {
+			// publish idle watermark so that the next vertex doesn't need to wait for the window to close to
+			// start processing data whose watermark is smaller than the endTime of the toBeClosed window
+
+			// this is to minimize watermark latency
+			for _, toBuffer := range df.toBuffers {
+				if publisher, ok := df.watermarkPublishers[toBuffer.GetName()]; ok {
+					idlehandler.PublishIdleWatermark(ctx, toBuffer, publisher, df.idleManager, df.log, dfv1.VertexTypeReduceUDF, wmb.Watermark(watermark))
+				}
+			}
+		}
+	}
+}
+
+// writeMessagesToWindows write the messages to each window that message belongs to. Each window is backed by a PBQ.
+func (df *DataForward) writeMessagesToWindows(ctx context.Context, messages []*isb.ReadMessage) ([]*isb.ReadMessage, error) {
+	var err error
+	var writtenMessages = make([]*isb.ReadMessage, 0, len(messages))
+
+messagesLoop:
+	for _, message := range messages {
+		// drop the late messages only if there is no window open
+		if message.IsLate {
+			// we should be able to get the late message in as long as there is an open window
+			nextWin := df.pbqManager.NextWindowToBeClosed()
+			if nextWin != nil && message.EventTime.Before(nextWin.StartTime()) {
+				df.log.Warnw("Dropping the late message", zap.Time("eventTime", message.EventTime), zap.Time("watermark", message.Watermark), zap.Time("nextWindowToBeClosed", nextWin.StartTime()))
+				droppedMessagesCount.With(map[string]string{
+					metrics.LabelVertex:             df.vertexName,
+					metrics.LabelPipeline:           df.pipelineName,
+					metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+					LabelReason:                     "late"}).Inc()
+
+				// mark it as a successfully written message as the message will be acked to avoid subsequent retries
+				writtenMessages = append(writtenMessages, message)
+				continue
+			} else {
+				var startTime time.Time
+				// bit of an overkill, but this is an unlikely path
+				if nextWin == nil {
+					startTime = time.Time{}
+				} else {
+					startTime = nextWin.StartTime()
+				}
+				df.log.Debugw("Keeping the late message for next condition check because COB has not happened yet", zap.Int64("eventTime", message.EventTime.UnixMilli()), zap.Int64("watermark", message.Watermark.UnixMilli()), zap.Int64("nextWindowToBeClosed.startTime", startTime.UnixMilli()))
+			}
+		}
+
+		// NOTE(potential bug): if we get a message where the event-time is < (watermark-allowedLateness), skip processing the message.
+		// This could be due to a couple of problem, eg. ack was not registered, etc.
+		// Please do not confuse this with late data! This is a platform related problem causing the watermark inequality
+		// to be violated.
+		if message.EventTime.Before(message.Watermark.Add(-1 * df.opts.allowedLateness)) {
+			// TODO: track as a counter metric
+			df.log.Errorw("An old message just popped up", zap.Any("msgOffSet", message.ReadOffset.String()), zap.Int64("eventTime", message.EventTime.UnixMilli()), zap.Int64("watermark", message.Watermark.UnixMilli()), zap.Any("message", message.Message))
+			// mark it as a successfully written message as the message will be acked to avoid subsequent retries
+			writtenMessages = append(writtenMessages, message)
+			// let's not continue processing this message, most likely the window has already been closed and the message
+			// won't be processed anyways.
+			droppedMessagesCount.With(map[string]string{
+				metrics.LabelVertex:             df.vertexName,
+				metrics.LabelPipeline:           df.pipelineName,
+				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+				LabelReason:                     "watermark_issue"}).Inc()
+			continue
+		}
+
+		// identify and add window for the message
+		windows := df.upsertWindowsAndKeys(message)
+
+		// for each window we will have a PBQ. A message could belong to multiple windows (e.g., sliding).
+		// We need to write the messages to these PBQs.
+		for _, kw := range windows {
+
+			for _, partitionID := range kw.Partitions() {
+
+				err := df.writeToPBQ(ctx, message, partitionID, kw)
+				// there is no point continuing because we are seeing an error.
+				// this error will ONLY BE set if we are in a erroring loop and ctx.Done() has been invoked.
+				if err != nil {
+					df.log.Errorw("Failed to write message, asked to stop trying", zap.Any("msgOffSet", message.ReadOffset.String()), zap.String("partitionID", partitionID.String()), zap.Error(err))
+					break messagesLoop
+				}
+			}
+		}
+
+		writtenMessages = append(writtenMessages, message)
+	}
+
+	return writtenMessages, err
+}
+
+// writeToPBQ writes to the PBQ. It will return error only if it is not failing to write to PBQ and is in a continuous
+// error loop, and we have received ctx.Done() via SIGTERM.
+func (df *DataForward) writeToPBQ(ctx context.Context, m *isb.ReadMessage, p partition.ID, kw window.AlignedKeyedWindower) error {
+	startTime := time.Now()
+	defer pbqWriteTime.With(map[string]string{
+		metrics.LabelVertex:             df.vertexName,
+		metrics.LabelPipeline:           df.pipelineName,
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+	}).Observe(float64(time.Since(startTime).Milliseconds()))
+
+	var pbqWriteBackoff = wait.Backoff{
+		Steps:    math.MaxInt,
+		Duration: 1 * time.Second,
+		Factor:   1.5,
+		Jitter:   0.1,
+	}
+
+	q := df.associatePBQAndPnF(ctx, p, kw)
+
+	err := wait.ExponentialBackoff(pbqWriteBackoff, func() (done bool, err error) {
+		rErr := q.Write(context.Background(), m)
+		if rErr != nil {
+			df.log.Errorw("Failed to write message", zap.Any("msgOffSet", m.ReadOffset.String()), zap.String("partitionID", p.String()), zap.Error(rErr))
+			pbqWriteErrorCount.With(map[string]string{
+				metrics.LabelVertex:             df.vertexName,
+				metrics.LabelPipeline:           df.pipelineName,
+				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+			}).Inc()
+			// no point retrying if ctx.Done has been invoked
+			select {
+			case <-ctx.Done():
+				// no point in retrying after we have been asked to stop.
+				return false, ctx.Err()
+			default:
+				// keep retrying
+				return false, nil
+			}
+
+		}
+		// happy path
+		pbqWriteMessagesCount.With(map[string]string{
+			metrics.LabelVertex:             df.vertexName,
+			metrics.LabelPipeline:           df.pipelineName,
+			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+		}).Inc()
+		df.log.Debugw("Successfully wrote the message", zap.String("msgOffSet", m.ReadOffset.String()), zap.String("partitionID", p.String()))
+		return true, nil
+	})
+
+	return err
+}
+
+// ackMessages acks messages. Retries until it can succeed or ctx.Done() happens.
+func (df *DataForward) ackMessages(ctx context.Context, messages []*isb.ReadMessage) {
+	var ackBackoff = wait.Backoff{
+		Steps:    math.MaxInt,
+		Duration: 1 * time.Second,
+		Factor:   1.5,
+		Jitter:   0.1,
+	}
+	var wg sync.WaitGroup
+
+	// Ack the message to ISB
+	for _, m := range messages {
+		wg.Add(1)
+
+		go func(o isb.Offset) {
+			defer wg.Done()
+			attempt := 0
+			ackErr := wait.ExponentialBackoff(ackBackoff, func() (done bool, err error) {
+				rErr := o.AckIt()
+				attempt += 1
+				if rErr != nil {
+					ackMessageError.With(map[string]string{
+						metrics.LabelVertex:             df.vertexName,
+						metrics.LabelPipeline:           df.pipelineName,
+						metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+					}).Inc()
+
+					df.log.Errorw("Failed to ack message, retrying", zap.String("msgOffSet", o.String()), zap.Error(rErr), zap.Int("attempt", attempt))
+					// no point retrying if ctx.Done has been invoked
+					select {
+					case <-ctx.Done():
+						// no point in retrying after we have been asked to stop.
+						return false, ctx.Err()
+					default:
+						// keep retrying
+						return false, nil
+					}
+				}
+				df.log.Debugw("Successfully acked message", zap.String("msgOffSet", o.String()))
+				ackMessagesCount.With(map[string]string{
+					metrics.LabelVertex:             df.vertexName,
+					metrics.LabelPipeline:           df.pipelineName,
+					metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+				}).Inc()
+				return true, nil
+			})
+			// no point trying the rest of the message, most likely that will also fail
+			if ackErr != nil {
+				df.log.Errorw("Context closed while waiting to ack messages inside data forward", zap.Error(ackErr), zap.String("offset", o.String()))
+
+			}
+		}(m.ReadOffset)
+
+	}
+	wg.Wait()
+}
+
+// ShutDown shutdowns the read-loop.
+func (df *DataForward) ShutDown(ctx context.Context) {
+
+	df.log.Infow("Stopping reduce data forwarder...")
+
+	if err := df.fromBuffer.Close(); err != nil {
+		df.log.Errorw("Failed to close buffer reader, shutdown anyways...", zap.Error(err))
+	} else {
+		df.log.Infow("Closed buffer reader", zap.String("bufferFrom", df.fromBuffer.GetName()))
+	}
+
+	// flush pending messages to persistent storage
+	df.pbqManager.ShutDown(ctx)
+
+	// stop watermark fetcher
+	if err := df.watermarkFetcher.Close(); err != nil {
+		df.log.Errorw("Failed to close watermark fetcher", zap.Error(err))
+	}
+
+	// stop watermark publisher
+	for _, publisher := range df.watermarkPublishers {
+		if err := publisher.Close(); err != nil {
+			df.log.Errorw("Failed to close watermark publisher", zap.Error(err))
+		}
+	}
+}
+
+// upsertWindowsAndKeys will create or assigns (if already present) a window to the message. It is an upsert operation
+// because windows are created out of order, but they will be closed in-order.
+func (df *DataForward) upsertWindowsAndKeys(m *isb.ReadMessage) []window.AlignedKeyedWindower {
+
+	processingWindows := df.windower.AssignWindow(m.EventTime)
+	var kWindows []window.AlignedKeyedWindower
+	for _, win := range processingWindows {
+		w, isPresent := df.windower.InsertIfNotPresent(win)
+		if !isPresent {
+			df.log.Debugw("Creating new keyed window", zap.String("msg.offset", m.ID), zap.Int64("startTime", w.StartTime().UnixMilli()), zap.Int64("endTime", w.EndTime().UnixMilli()))
+		} else {
+			df.log.Debugw("Found an existing window", zap.String("msg.offset", m.ID), zap.Int64("startTime", w.StartTime().UnixMilli()), zap.Int64("endTime", w.EndTime().UnixMilli()))
+		}
+		// track the key to window relationship
+		// FIXME: create a slot from m.key
+		w.AddSlot("slot-0")
+		kWindows = append(kWindows, w)
+	}
+	return kWindows
+}
+
+// ClosePartitions closes the partitions by invoking close-of-book (COB).
+func (df *DataForward) ClosePartitions(partitions []partition.ID) {
+	for _, p := range partitions {
+		q := df.pbqManager.GetPBQ(p)
+		df.log.Infow("Close of book", zap.String("partitionID", p.String()))
+		// schedule the task for ordered processing.
+		df.of.InsertTask(df.udfInvocationTracking[p])
+		q.CloseOfBook()
+		delete(df.udfInvocationTracking, p)
+	}
 }
diff --git a/pkg/reduce/metrics.go b/pkg/reduce/metrics.go
index 8c566563..9a73282a 100644
--- a/pkg/reduce/metrics.go
+++ b/pkg/reduce/metrics.go
@@ -22,6 +22,53 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 )
 
+const (
+	LabelReason = "reason"
+)
+
+// droppedMessagesCount is used to indicate the number of messages dropped
+var droppedMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "reduce_data_forward",
+	Name:      "dropped_total",
+	Help:      "Total number of Messages Dropped",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, LabelReason})
+
+// pbqWriteErrorCount is used to indicate the number of errors while writing to pbq
+var pbqWriteErrorCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "reduce_pbq",
+	Name:      "write_error_total",
+	Help:      "Total number of PBQ Write Errors",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex})
+
+// pbqWriteMessagesCount is used to indicate the number of messages written to pbq
+var pbqWriteMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "reduce_pbq",
+	Name:      "write_total",
+	Help:      "Total number of Messages Written to PBQ",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex})
+
+// pbqWriteTime pbq write latency
+var pbqWriteTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Subsystem: "reduce_pbq",
+	Name:      "write_time",
+	Help:      "Entry write time (1 to 5000 microseconds)",
+	Buckets:   prometheus.ExponentialBucketsRange(1, 5000, 5),
+}, []string{metrics.LabelPipeline, metrics.LabelVertex, metrics.LabelVertexReplicaIndex})
+
+// ackMessagesCount is used to indicate the number of messages acknowledged
+var ackMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "reduce_isb_reader",
+	Name:      "ack_total",
+	Help:      "Total number of Messages Acknowledged",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex})
+
+// ackMessageError is used to indicate the errors in the number of  messages acknowledged
+var ackMessageError = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "reduce_isb_reader",
+	Name:      "ack_error_total",
+	Help:      "Total number of Acknowledged Errors",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex})
+
 // readMessagesCount is used to indicate the number of messages read
 var readMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "reduce_isb_reader",
diff --git a/pkg/reduce/pnf/metrics.go b/pkg/reduce/pnf/metrics.go
index e6ad7a15..3277ca63 100644
--- a/pkg/reduce/pnf/metrics.go
+++ b/pkg/reduce/pnf/metrics.go
@@ -65,7 +65,7 @@ var platformError = promauto.NewCounterVec(prometheus.CounterOpts{
 	Help:      "Total number of platform Errors",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex})
 
-// reduceProcessTime reduce task processing latency
+// reduceProcessTime reduce ForwardTask processing latency
 var reduceProcessTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
 	Subsystem: "reduce_pnf",
 	Name:      "process_time",
@@ -80,3 +80,17 @@ var reduceForwardTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
 	Help:      "Reduce forward time (1 to 100000 microseconds)",
 	Buckets:   prometheus.ExponentialBucketsRange(1, 100000, 5),
 }, []string{metrics.LabelPipeline, metrics.LabelVertex, metrics.LabelVertexReplicaIndex})
+
+// partitionsInFlight is used to indicate the partitions in flight
+var partitionsInFlight = promauto.NewGaugeVec(prometheus.GaugeOpts{
+	Subsystem: "reduce_pnf",
+	Name:      "partitions_inflight",
+	Help:      "Total number of partitions in flight",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex})
+
+// udfError is used to indicate the number of UDF errors
+var udfError = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "reduce_udf",
+	Name:      "error_total",
+	Help:      "Total number of UDF Errors",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex})
diff --git a/pkg/reduce/pnf/ordered.go b/pkg/reduce/pnf/ordered.go
new file mode 100644
index 00000000..e7d9eaf3
--- /dev/null
+++ b/pkg/reduce/pnf/ordered.go
@@ -0,0 +1,243 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package pnf
+
+import (
+	"container/list"
+	"context"
+	"strconv"
+	"sync"
+	"time"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
+	"go.uber.org/zap"
+
+	"github.com/numaproj/numaflow/pkg/forward"
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/metrics"
+	"github.com/numaproj/numaflow/pkg/reduce/applier"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/shared/logging"
+	"github.com/numaproj/numaflow/pkg/watermark/publish"
+)
+
+var retryDelay = 1 * time.Second
+
+// ForwardTask wraps the `processAndForward`.
+type ForwardTask struct {
+	// doneCh is used to notify when the ForwardTask has been completed.
+	doneCh chan struct{}
+	pf     *processAndForward
+}
+
+// OrderedProcessor orders the forwarding of the result of the execution of the tasks, even though the tasks itself are
+// run concurrently in an out of ordered fashion.
+type OrderedProcessor struct {
+	vertexName    string
+	pipelineName  string
+	vertexReplica int32
+	sync.RWMutex
+	taskDone            chan struct{}
+	taskQueue           *list.List
+	pbqManager          *pbq.Manager
+	udf                 applier.ReduceApplier
+	toBuffers           map[string]isb.BufferWriter
+	whereToDecider      forward.ToWhichStepDecider
+	watermarkPublishers map[string]publish.Publisher
+	idleManager         *wmb.IdleManager
+	log                 *zap.SugaredLogger
+}
+
+// NewOrderedProcessor returns an OrderedProcessor.
+func NewOrderedProcessor(ctx context.Context,
+	vertexInstance *dfv1.VertexInstance,
+	udf applier.ReduceApplier,
+	toBuffers map[string]isb.BufferWriter,
+	pbqManager *pbq.Manager,
+	whereToDecider forward.ToWhichStepDecider,
+	watermarkPublishers map[string]publish.Publisher,
+	idleManager *wmb.IdleManager) *OrderedProcessor {
+	of := &OrderedProcessor{
+		vertexName:          vertexInstance.Vertex.Spec.Name,
+		pipelineName:        vertexInstance.Vertex.Spec.PipelineName,
+		vertexReplica:       vertexInstance.Replica,
+		taskDone:            make(chan struct{}),
+		taskQueue:           list.New(),
+		pbqManager:          pbqManager,
+		udf:                 udf,
+		toBuffers:           toBuffers,
+		whereToDecider:      whereToDecider,
+		watermarkPublishers: watermarkPublishers,
+		idleManager:         idleManager,
+		log:                 logging.FromContext(ctx),
+	}
+
+	go of.forward(ctx)
+
+	return of
+}
+
+func (op *OrderedProcessor) InsertTask(t *ForwardTask) {
+	op.Lock()
+	defer op.Unlock()
+	op.taskQueue.PushBack(t)
+}
+
+// SchedulePnF creates and schedules the PnF routine.
+func (op *OrderedProcessor) SchedulePnF(
+	ctx context.Context,
+	partitionID partition.ID) *ForwardTask {
+
+	pbq := op.pbqManager.GetPBQ(partitionID)
+
+	pf := newProcessAndForward(ctx, op.vertexName, op.pipelineName, op.vertexReplica, partitionID,
+		op.udf, pbq, op.toBuffers, op.whereToDecider, op.watermarkPublishers, op.idleManager)
+
+	doneCh := make(chan struct{})
+	t := &ForwardTask{
+		doneCh: doneCh,
+		pf:     pf,
+	}
+	partitionsInFlight.With(map[string]string{
+		metrics.LabelVertex:             op.vertexName,
+		metrics.LabelPipeline:           op.pipelineName,
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(op.vertexReplica)),
+	}).Inc()
+
+	// invoke the reduce function
+	go op.reduceOp(ctx, t)
+	return t
+}
+
+// reduceOp invokes the reduce function. The reducer is a long-running function since we stream in the data and it has
+// to wait for the close-of-book on the PBQ to materialize the result.
+func (op *OrderedProcessor) reduceOp(ctx context.Context, t *ForwardTask) {
+	start := time.Now()
+	for {
+		// FIXME: this error handling won't work with streams. We cannot do infinite retries
+		//  because whatever is written to the stream is lost between retries.
+		err := t.pf.Process(ctx)
+		if err == nil {
+			break
+		} else if err == ctx.Err() {
+			udfError.With(map[string]string{
+				metrics.LabelVertex:             op.vertexName,
+				metrics.LabelPipeline:           op.pipelineName,
+				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(op.vertexReplica)),
+			}).Inc()
+			op.log.Infow("ReduceOp exiting", zap.String("partitionID", t.pf.PartitionID.String()), zap.Error(ctx.Err()))
+			return
+		}
+		op.log.Errorw("Process failed", zap.String("partitionID", t.pf.PartitionID.String()), zap.Error(err))
+		time.Sleep(retryDelay)
+	}
+
+	// indicate that we are done with reduce UDF invocation.
+	close(t.doneCh)
+	op.log.Debugw("Process->Reduce call took ", zap.String("partitionID", t.pf.PartitionID.String()), zap.Int64("duration(ms)", time.Since(start).Milliseconds()))
+
+	// notify that some work has been completed
+	select {
+	case op.taskDone <- struct{}{}:
+	case <-ctx.Done():
+		return
+	}
+}
+
+// forward monitors the ForwardTask queue, as soon as the ForwardTask at the head of the queue has been completed, the result is
+// forwarded to the next ISB. It keeps doing this for forever or until ctx.Done() happens.
+func (op *OrderedProcessor) forward(ctx context.Context) {
+	var currElement *list.Element
+	var t *ForwardTask
+
+outerLoop:
+	for {
+		start := time.Now()
+		// block till we have some work
+		select {
+		case <-op.taskDone:
+		case <-ctx.Done():
+			op.log.Infow("forward exiting while waiting for ForwardTask completion event", zap.Error(ctx.Err()))
+			break outerLoop
+		}
+		op.log.Debugw("Time waited for a completion event to happen ", zap.Int64("duration(ms)", time.Since(start).Milliseconds()))
+
+		// a signal does not mean that we have any pending work to be done because
+		// for every signal, we try to empty out the ForwardTask-queue.
+		op.RLock()
+		n := op.taskQueue.Len()
+		op.log.Debugw("Received a signal in ForwardTask queue ", zap.Int("ForwardTask count", n))
+
+		op.RUnlock()
+		// n could be 0 because we have emptied the queue
+		if n == 0 {
+			continue
+		}
+
+		// now that we know there is at least an element, let's start from the front.
+		op.RLock()
+		currElement = op.taskQueue.Front()
+		op.RUnlock()
+
+		// empty out the entire ForwardTask-queue everytime there has been some work done
+		startLoop := time.Now()
+		for i := 0; i < n; i++ {
+			t = currElement.Value.(*ForwardTask)
+			select {
+			case <-t.doneCh:
+				for {
+					err := t.pf.Forward(ctx)
+					if err != nil {
+						logging.FromContext(ctx).Error(err)
+						time.Sleep(retryDelay)
+					} else {
+						break
+					}
+				}
+				op.Lock()
+				rm := currElement
+				currElement = currElement.Next()
+				op.taskQueue.Remove(rm)
+				partitionsInFlight.With(map[string]string{
+					metrics.LabelVertex:             op.vertexName,
+					metrics.LabelPipeline:           op.pipelineName,
+					metrics.LabelVertexReplicaIndex: strconv.Itoa(int(op.vertexReplica)),
+				}).Dec()
+
+				op.log.Debugw("Removing ForwardTask post forward call", zap.String("partitionID", t.pf.PartitionID.String()))
+				op.Unlock()
+			case <-ctx.Done():
+				op.log.Infow("Forward exiting while waiting on the head op the queue ForwardTask", zap.String("partitionID", t.pf.PartitionID.String()), zap.Error(ctx.Err()))
+				break outerLoop
+			}
+		}
+		op.log.Debugw("One iteration op the ordered tasks queue loop took ", zap.Int64("duration(ms)", time.Since(startLoop).Milliseconds()), zap.Int("elements", n))
+	}
+	op.Shutdown()
+}
+
+func (op *OrderedProcessor) Shutdown() {
+	for _, v := range op.toBuffers {
+		if err := v.Close(); err != nil {
+			op.log.Errorw("Failed to close buffer writer, shutdown anyways...", zap.Error(err), zap.String("bufferTo", v.GetName()))
+		} else {
+			op.log.Infow("Closed buffer writer", zap.String("bufferTo", v.GetName()))
+		}
+	}
+}
diff --git a/pkg/reduce/pnf/processandforward.go b/pkg/reduce/pnf/processandforward.go
index b49bf632..dfbcb3be 100644
--- a/pkg/reduce/pnf/processandforward.go
+++ b/pkg/reduce/pnf/processandforward.go
@@ -15,7 +15,7 @@ limitations under the License.
 */
 
 // Package pnf processes and then forwards messages belonging to a window. It reads the data from PBQ (which is populated
-// by the `readloop`), calls the UDF reduce function, and then forwards to the next ISB. After a successful forward, it
+// by the `data forwarder`), calls the UDF reduce function, and then forwards to the next ISB. After a successful forward, it
 // invokes `GC` to clean up the PBQ. Since pnf is a reducer, it mutates the watermark. The watermark after the pnf will
 // be the end time of the window.
 package pnf
@@ -44,9 +44,9 @@ import (
 	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
-// ProcessAndForward reads messages from pbq, invokes udf using grpc, forwards the results to ISB, and then publishes
+// processAndForward reads messages from pbq, invokes udf using grpc, forwards the results to ISB, and then publishes
 // the watermark for that partition.
-type ProcessAndForward struct {
+type processAndForward struct {
 	vertexName       string
 	pipelineName     string
 	vertexReplica    int32
@@ -61,8 +61,8 @@ type ProcessAndForward struct {
 	idleManager      *wmb.IdleManager
 }
 
-// NewProcessAndForward will return a new ProcessAndForward instance
-func NewProcessAndForward(ctx context.Context,
+// newProcessAndForward will return a new processAndForward instance
+func newProcessAndForward(ctx context.Context,
 	vertexName string,
 	pipelineName string,
 	vr int32,
@@ -72,8 +72,8 @@ func NewProcessAndForward(ctx context.Context,
 	toBuffers map[string]isb.BufferWriter,
 	whereToDecider forward.ToWhichStepDecider,
 	pw map[string]publish.Publisher,
-	idleManager *wmb.IdleManager) *ProcessAndForward {
-	return &ProcessAndForward{
+	idleManager *wmb.IdleManager) *processAndForward {
+	return &processAndForward{
 		vertexName:       vertexName,
 		pipelineName:     pipelineName,
 		vertexReplica:    vr,
@@ -89,7 +89,7 @@ func NewProcessAndForward(ctx context.Context,
 }
 
 // Process method reads messages from the supplied PBQ, invokes UDF to reduce the result.
-func (p *ProcessAndForward) Process(ctx context.Context) error {
+func (p *processAndForward) Process(ctx context.Context) error {
 	var err error
 	startTime := time.Now()
 	defer reduceProcessTime.With(map[string]string{
@@ -104,7 +104,7 @@ func (p *ProcessAndForward) Process(ctx context.Context) error {
 }
 
 // Forward writes messages to the ISBs, publishes watermark, and invokes GC on PBQ.
-func (p *ProcessAndForward) Forward(ctx context.Context) error {
+func (p *processAndForward) Forward(ctx context.Context) error {
 	// extract window end time from the partitionID, which will be used for watermark
 	startTime := time.Now()
 	defer reduceForwardTime.With(map[string]string{
@@ -164,7 +164,7 @@ func (p *ProcessAndForward) Forward(ctx context.Context) error {
 }
 
 // whereToStep assigns a message to the ISBs based on the Message.Keys.
-func (p *ProcessAndForward) whereToStep() map[string][]isb.Message {
+func (p *processAndForward) whereToStep() map[string][]isb.Message {
 	// writer doesn't accept array of pointers
 	messagesToStep := make(map[string][]isb.Message)
 
@@ -199,7 +199,7 @@ func (p *ProcessAndForward) whereToStep() map[string][]isb.Message {
 
 // writeToBuffer writes to the ISBs.
 // TODO: is there any point in returning an error here? this is an infinite loop and the only error is ctx.Done!
-func (p *ProcessAndForward) writeToBuffer(ctx context.Context, bufferID string, resultMessages []isb.Message) ([]isb.Offset, error) {
+func (p *processAndForward) writeToBuffer(ctx context.Context, bufferID string, resultMessages []isb.Message) ([]isb.Offset, error) {
 	var (
 		writeCount int
 		writeBytes float64
@@ -275,7 +275,7 @@ func (p *ProcessAndForward) writeToBuffer(ctx context.Context, bufferID string,
 }
 
 // publishWM publishes the watermark to each edge.
-func (p *ProcessAndForward) publishWM(ctx context.Context, wm wmb.Watermark, writeOffsets map[string][]isb.Offset) {
+func (p *processAndForward) publishWM(ctx context.Context, wm wmb.Watermark, writeOffsets map[string][]isb.Offset) {
 	// activeWatermarkBuffers records the buffers that the publisher has published
 	// a watermark in this batch processing cycle.
 	// it's used to determine which buffers should receive an idle watermark.
diff --git a/pkg/reduce/readloop/metrics.go b/pkg/reduce/readloop/metrics.go
deleted file mode 100644
index c41e896b..00000000
--- a/pkg/reduce/readloop/metrics.go
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
-Copyright 2022 The Numaproj Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package readloop
-
-import (
-	"github.com/numaproj/numaflow/pkg/metrics"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
-)
-
-const (
-	LabelReason = "reason"
-)
-
-// droppedMessagesCount is used to indicate the number of messages dropped
-var droppedMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
-	Subsystem: "reduce_readloop",
-	Name:      "dropped_total",
-	Help:      "Total number of Messages Dropped",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, LabelReason})
-
-// pbqWriteErrorCount is used to indicate the number of errors while writing to pbq
-var pbqWriteErrorCount = promauto.NewCounterVec(prometheus.CounterOpts{
-	Subsystem: "reduce_pbq",
-	Name:      "write_error_total",
-	Help:      "Total number of PBQ Write Errors",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex})
-
-// pbqWriteMessagesCount is used to indicate the number of messages written to pbq
-var pbqWriteMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
-	Subsystem: "reduce_pbq",
-	Name:      "write_total",
-	Help:      "Total number of Messages Written to PBQ",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex})
-
-// partitionsInFlight is used to indicate the partitions in flight
-var partitionsInFlight = promauto.NewGaugeVec(prometheus.GaugeOpts{
-	Subsystem: "reduce_readloop",
-	Name:      "partitions_inflight",
-	Help:      "Total number of partitions in flight",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex})
-
-// pbqWriteTime pbq write latency
-var pbqWriteTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
-	Subsystem: "reduce_pbq",
-	Name:      "write_time",
-	Help:      "Entry write time (1 to 5000 microseconds)",
-	Buckets:   prometheus.ExponentialBucketsRange(1, 5000, 5),
-}, []string{metrics.LabelPipeline, metrics.LabelVertex, metrics.LabelVertexReplicaIndex})
-
-// ackMessagesCount is used to indicate the number of  messages acknowledged
-var ackMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
-	Subsystem: "reduce_isb_reader",
-	Name:      "ack_total",
-	Help:      "Total number of Messages Acknowledged",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex})
-
-// ackMessageError is used to indicate the errors in the number of  messages acknowledged
-var ackMessageError = promauto.NewCounterVec(prometheus.CounterOpts{
-	Subsystem: "reduce_isb_reader",
-	Name:      "ack_error_total",
-	Help:      "Total number of Acknowledged Errors",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex})
-
-// udfError is used to indicate the number of UDF errors
-var udfError = promauto.NewCounterVec(prometheus.CounterOpts{
-	Subsystem: "reduce_udf",
-	Name:      "error_total",
-	Help:      "Total number of UDF Errors",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex})
diff --git a/pkg/reduce/readloop/ordered.go b/pkg/reduce/readloop/ordered.go
deleted file mode 100644
index c49e7b76..00000000
--- a/pkg/reduce/readloop/ordered.go
+++ /dev/null
@@ -1,213 +0,0 @@
-/*
-Copyright 2022 The Numaproj Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package readloop
-
-import (
-	"container/list"
-	"context"
-	"strconv"
-	"sync"
-	"time"
-
-	"github.com/numaproj/numaflow/pkg/watermark/wmb"
-	"go.uber.org/zap"
-
-	"github.com/numaproj/numaflow/pkg/forward"
-	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/metrics"
-	"github.com/numaproj/numaflow/pkg/reduce/applier"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
-	"github.com/numaproj/numaflow/pkg/reduce/pnf"
-	"github.com/numaproj/numaflow/pkg/shared/logging"
-	"github.com/numaproj/numaflow/pkg/watermark/publish"
-)
-
-var retryDelay = 1 * time.Second
-
-// task wraps the `ProcessAndForward`.
-type task struct {
-	// doneCh is used to notify when the task has been completed.
-	doneCh chan struct{}
-	pf     *pnf.ProcessAndForward
-}
-
-// orderedForwarder orders the forwarding of the result of the execution of the tasks, even though the tasks itself are
-// run concurrently in an out of ordered fashion.
-type orderedForwarder struct {
-	vertexName    string
-	pipelineName  string
-	vertexReplica int32
-	sync.RWMutex
-	taskDone  chan struct{}
-	taskQueue *list.List
-	log       *zap.SugaredLogger
-}
-
-// newOrderedForwarder returns an orderedForwarder.
-func newOrderedForwarder(ctx context.Context, vertexName string, pipelineName string, vr int32) *orderedForwarder {
-	of := &orderedForwarder{
-		vertexName:    vertexName,
-		pipelineName:  pipelineName,
-		vertexReplica: vr,
-		taskDone:      make(chan struct{}),
-		taskQueue:     list.New(),
-		log:           logging.FromContext(ctx),
-	}
-
-	go of.forward(ctx)
-
-	return of
-}
-
-func (of *orderedForwarder) insertTask(t *task) {
-	of.Lock()
-	defer of.Unlock()
-	of.taskQueue.PushBack(t)
-}
-
-// schedulePnF creates and schedules the PnF routine.
-func (of *orderedForwarder) schedulePnF(
-	ctx context.Context,
-	udf applier.ReduceApplier,
-	pbq pbq.Reader,
-	partitionID partition.ID,
-	toBuffers map[string]isb.BufferWriter,
-	whereToDecider forward.ToWhichStepDecider,
-	pw map[string]publish.Publisher,
-	idleManager *wmb.IdleManager) *task {
-
-	pf := pnf.NewProcessAndForward(ctx, of.vertexName, of.pipelineName, of.vertexReplica, partitionID, udf, pbq, toBuffers, whereToDecider, pw, idleManager)
-	doneCh := make(chan struct{})
-	t := &task{
-		doneCh: doneCh,
-		pf:     pf,
-	}
-	partitionsInFlight.With(map[string]string{
-		metrics.LabelVertex:             of.vertexName,
-		metrics.LabelPipeline:           of.pipelineName,
-		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(of.vertexReplica)),
-	}).Inc()
-
-	// invoke the reduce function
-	go of.reduceOp(ctx, t)
-	return t
-}
-
-// reduceOp invokes the reduce function. The reducer is a long-running function since we stream in the data and it has
-// to wait for the close-of-book on the PBQ to materialize the result.
-func (of *orderedForwarder) reduceOp(ctx context.Context, t *task) {
-	start := time.Now()
-	for {
-		// FIXME: this error handling won't work with streams. We cannot do infinite retries
-		//  because whatever is written to the stream is lost between retries.
-		err := t.pf.Process(ctx)
-		if err == nil {
-			break
-		} else if err == ctx.Err() {
-			udfError.With(map[string]string{
-				metrics.LabelVertex:             of.vertexName,
-				metrics.LabelPipeline:           of.pipelineName,
-				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(of.vertexReplica)),
-			}).Inc()
-			of.log.Infow("ReduceOp exiting", zap.String("partitionID", t.pf.PartitionID.String()), zap.Error(ctx.Err()))
-			return
-		}
-		of.log.Errorw("Process failed", zap.String("partitionID", t.pf.PartitionID.String()), zap.Error(err))
-		time.Sleep(retryDelay)
-	}
-	// indicate that we are done with reduce UDF invocation.
-	close(t.doneCh)
-	of.log.Debugw("Process->Reduce call took ", zap.String("partitionID", t.pf.PartitionID.String()), zap.Int64("duration(ms)", time.Since(start).Milliseconds()))
-
-	// notify that some work has been completed
-	select {
-	case of.taskDone <- struct{}{}:
-	case <-ctx.Done():
-		return
-	}
-}
-
-// forward monitors the task queue, as soon as the task at the head of the queue has been completed, the result is
-// forwarded to the next ISB. It keeps doing this for forever or until ctx.Done() happens.
-func (of *orderedForwarder) forward(ctx context.Context) {
-	var currElement *list.Element
-	var t *task
-
-	for {
-		start := time.Now()
-		// block till we have some work
-		select {
-		case <-of.taskDone:
-		case <-ctx.Done():
-			of.log.Infow("Forward exiting while waiting for task completion event", zap.Error(ctx.Err()))
-			return
-		}
-		of.log.Debugw("Time waited for a completion event to happen ", zap.Int64("duration(ms)", time.Since(start).Milliseconds()))
-
-		// a signal does not mean that we have any pending work to be done because
-		// for every signal, we try to empty out the task-queue.
-		of.RLock()
-		n := of.taskQueue.Len()
-		of.log.Debugw("Received a signal in task queue ", zap.Int("task count", n))
-
-		of.RUnlock()
-		// n could be 0 because we have emptied the queue
-		if n == 0 {
-			continue
-		}
-
-		// now that we know there is at least an element, let's start from the front.
-		of.RLock()
-		currElement = of.taskQueue.Front()
-		of.RUnlock()
-
-		// empty out the entire task-queue everytime there has been some work done
-		startLoop := time.Now()
-		for i := 0; i < n; i++ {
-			t = currElement.Value.(*task)
-			select {
-			case <-t.doneCh:
-				for {
-					err := t.pf.Forward(ctx)
-					if err != nil {
-						logging.FromContext(ctx).Error(err)
-						time.Sleep(retryDelay)
-					} else {
-						break
-					}
-				}
-				of.Lock()
-				rm := currElement
-				currElement = currElement.Next()
-				of.taskQueue.Remove(rm)
-				partitionsInFlight.With(map[string]string{
-					metrics.LabelVertex:             of.vertexName,
-					metrics.LabelPipeline:           of.pipelineName,
-					metrics.LabelVertexReplicaIndex: strconv.Itoa(int(of.vertexReplica)),
-				}).Dec()
-
-				of.log.Debugw("Removing task post forward call", zap.String("partitionID", t.pf.PartitionID.String()))
-				of.Unlock()
-			case <-ctx.Done():
-				of.log.Infow("Forward exiting while waiting on the head of the queue task", zap.String("partitionID", t.pf.PartitionID.String()), zap.Error(ctx.Err()))
-				return
-			}
-		}
-		of.log.Debugw("One iteration of the ordered tasks queue loop took ", zap.Int64("duration(ms)", time.Since(startLoop).Milliseconds()), zap.Int("elements", n))
-	}
-}
diff --git a/pkg/reduce/readloop/readloop.go b/pkg/reduce/readloop/readloop.go
deleted file mode 100644
index 09d27d1c..00000000
--- a/pkg/reduce/readloop/readloop.go
+++ /dev/null
@@ -1,468 +0,0 @@
-/*
-Copyright 2022 The Numaproj Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-// Package readloop is responsible for the first part of reduce subsystem. It is responsible for feeding in the data
-// to the second part of the reduce subsystem called ProcessAndForward. `readloop` processes the read data from the ISB,
-// writes to the outbound channel called PBQ so the message can be asynchronously processed by `ProcessAndForward`, and
-// then closes the partition if possible based on watermark progression. To write to the outbound channel (PBQ),
-// `readloop` has to partition the message first. Hence, `readloop` also contains partitioning logic.
-// Partitioner identifies a set of elements with a common key and time, and buckets them in to a common window. A
-// partition is uniquely identified using a tuple {window, key}. Type of window does not matter. Partitioner is
-// responsible for managing the persistence and processing of each partition. It uses PBQ for durable persistence of
-// elements that belong to a partition and orchestrates the processing of elements using ProcessAndForward function.
-// Partitioner tracks active partitions, closes the partitions based on watermark progression and co-ordinates the
-// materialization and forwarding the results to the next vertex in the pipeline.
-package readloop
-
-import (
-	"context"
-	"math"
-	"strconv"
-	"sync"
-	"time"
-
-	"go.uber.org/zap"
-	"k8s.io/apimachinery/pkg/util/wait"
-
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/forward"
-	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/metrics"
-	"github.com/numaproj/numaflow/pkg/reduce/applier"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
-	"github.com/numaproj/numaflow/pkg/shared/idlehandler"
-	"github.com/numaproj/numaflow/pkg/shared/logging"
-	"github.com/numaproj/numaflow/pkg/watermark/publish"
-	"github.com/numaproj/numaflow/pkg/watermark/wmb"
-	"github.com/numaproj/numaflow/pkg/window"
-	"github.com/numaproj/numaflow/pkg/window/keyed"
-)
-
-// ReadLoop is responsible for reading and forwarding the message from ISB to PBQ.
-type ReadLoop struct {
-	vertexName            string
-	pipelineName          string
-	vertexReplica         int32
-	UDF                   applier.ReduceApplier
-	pbqManager            *pbq.Manager
-	windower              window.Windower
-	op                    *orderedForwarder
-	log                   *zap.SugaredLogger
-	toBuffers             map[string]isb.BufferWriter
-	whereToDecider        forward.ToWhichStepDecider
-	publishWatermark      map[string]publish.Publisher
-	udfInvocationTracking map[partition.ID]*task
-	idleManager           *wmb.IdleManager
-	allowedLateness       time.Duration
-}
-
-// NewReadLoop initializes and returns ReadLoop.
-func NewReadLoop(ctx context.Context,
-	vertexName string,
-	pipelineName string,
-	vr int32,
-	udf applier.ReduceApplier,
-	pbqManager *pbq.Manager,
-	windowingStrategy window.Windower,
-	toBuffers map[string]isb.BufferWriter,
-	whereToDecider forward.ToWhichStepDecider,
-	pw map[string]publish.Publisher,
-	idleManager *wmb.IdleManager,
-	allowedLateness time.Duration,
-) (*ReadLoop, error) {
-	op := newOrderedForwarder(ctx, vertexName, pipelineName, vr)
-
-	rl := &ReadLoop{
-		vertexName:            vertexName,
-		pipelineName:          pipelineName,
-		vertexReplica:         vr,
-		UDF:                   udf,
-		windower:              windowingStrategy,
-		pbqManager:            pbqManager,
-		op:                    op,
-		log:                   logging.FromContext(ctx),
-		toBuffers:             toBuffers,
-		whereToDecider:        whereToDecider,
-		publishWatermark:      pw,
-		udfInvocationTracking: make(map[partition.ID]*task),
-		idleManager:           idleManager,
-		allowedLateness:       allowedLateness,
-	}
-
-	err := rl.Startup(ctx)
-
-	return rl, err
-}
-
-// Startup starts up the read-loop, because during boot up, it has to replay the data from the persistent store of
-// PBQ before it can start reading from ISB. Startup will return only after the replay has been completed.
-func (rl *ReadLoop) Startup(ctx context.Context) error {
-	// start the PBQManager which discovers and builds the state from persistent store of the PBQ.
-	partitions, err := rl.pbqManager.GetExistingPartitions(ctx)
-	if err != nil {
-		return err
-	}
-
-	rl.log.Infow("Partitions to be replayed ", zap.Int("count", len(partitions)), zap.Any("partitions", partitions))
-
-	for _, p := range partitions {
-		// Create keyed window for a given partition
-		// so that the window can be closed when the watermark
-		// crosses the window.
-
-		alignedKeyedWindow := keyed.NewKeyedWindow(p.Start, p.End)
-
-		// insert the window to the list of active windows, since the active window list is in-memory
-		keyedWindow, _ := rl.windower.InsertIfNotPresent(alignedKeyedWindow)
-
-		// add slots to the window, so that when a new message with the watermark greater than
-		// the window end time comes, slots will not be lost and the windows will be closed as expected
-		keyedWindow.AddSlot(p.Slot)
-
-		// create and invoke process and forward for the partition
-		rl.associatePBQAndPnF(ctx, p, keyedWindow)
-	}
-
-	// replays the data (replay internally writes the data from persistent store on to the PBQ)
-	rl.pbqManager.Replay(ctx)
-	return nil
-}
-
-// Process is one iteration of the read loop which writes the messages to the PBQs followed by acking the messages, and
-// then closing the windows that can closed.
-func (rl *ReadLoop) Process(ctx context.Context, messages []*isb.ReadMessage) {
-	var dataMessages = make([]*isb.ReadMessage, 0, len(messages))
-	var ctrlMessages = make([]*isb.ReadMessage, 0) // for a high TPS pipeline, 0 is the most optimal value
-
-	for _, message := range messages {
-		if message.Kind == isb.Data {
-			dataMessages = append(dataMessages, message)
-		} else {
-			ctrlMessages = append(ctrlMessages, message)
-		}
-	}
-
-	// write messages to windows based by PBQs.
-	successfullyWrittenMessages, err := rl.writeMessagesToWindows(ctx, dataMessages)
-	if err != nil {
-		rl.log.Errorw("Failed to write messages", zap.Int("totalMessages", len(messages)), zap.Int("writtenMessage", len(successfullyWrittenMessages)))
-	}
-
-	// ack the control messages
-	if len(ctrlMessages) != 0 {
-		rl.ackMessages(ctx, ctrlMessages)
-	}
-
-	if len(successfullyWrittenMessages) == 0 {
-		return
-	}
-	// ack successful messages
-	rl.ackMessages(ctx, successfullyWrittenMessages)
-
-	// close any windows that need to be closed.
-	// since the watermark will be same for all the messages in the batch
-	// we can invoke remove windows only once per batch
-	var closedWindows []window.AlignedKeyedWindower
-	wm := wmb.Watermark(successfullyWrittenMessages[0].Watermark)
-
-	closedWindows = rl.windower.RemoveWindows(time.Time(wm).Add(-1 * rl.allowedLateness))
-
-	rl.log.Debugw("Windows eligible for closing", zap.Int("length", len(closedWindows)), zap.Time("watermark", time.Time(wm)))
-
-	for _, cw := range closedWindows {
-		partitions := cw.Partitions()
-		rl.ClosePartitions(partitions)
-		rl.log.Debugw("Closing Window", zap.Int64("windowStart", cw.StartTime().UnixMilli()), zap.Int64("windowEnd", cw.EndTime().UnixMilli()))
-	}
-
-	// solve Reduce withholding of watermark where we do not send WM until the window is closed.
-	if nextWin := rl.pbqManager.NextWindowToBeClosed(); nextWin != nil {
-		// minus 1 ms because if it's the same as the end time the window would have already been closed
-		if watermark := time.Time(wm).Add(-1 * time.Millisecond); nextWin.EndTime().After(watermark) {
-			// publish idle watermark so that the next vertex doesn't need to wait for the window to close to
-			// start processing data whose watermark is smaller than the endTime of the toBeClosed window
-
-			// this is to minimize watermark latency
-			for _, toBuffer := range rl.toBuffers {
-				if publisher, ok := rl.publishWatermark[toBuffer.GetName()]; ok {
-					idlehandler.PublishIdleWatermark(ctx, toBuffer, publisher, rl.idleManager, rl.log, dfv1.VertexTypeReduceUDF, wmb.Watermark(watermark))
-				}
-			}
-		}
-	}
-}
-
-// writeMessagesToWindows write the messages to each window that message belongs to. Each window is backed by a PBQ.
-func (rl *ReadLoop) writeMessagesToWindows(ctx context.Context, messages []*isb.ReadMessage) ([]*isb.ReadMessage, error) {
-	var err error
-	var writtenMessages = make([]*isb.ReadMessage, 0, len(messages))
-
-messagesLoop:
-	for _, message := range messages {
-		// drop the late messages only if there is no window open
-		if message.IsLate {
-			// we should be able to get the late message in as long as there is an open window
-			nextWin := rl.pbqManager.NextWindowToBeClosed()
-			if nextWin != nil && message.EventTime.Before(nextWin.StartTime()) {
-				rl.log.Warnw("Dropping the late message", zap.Time("eventTime", message.EventTime), zap.Time("watermark", message.Watermark), zap.Time("nextWindowToBeClosed", nextWin.StartTime()))
-				droppedMessagesCount.With(map[string]string{
-					metrics.LabelVertex:             rl.vertexName,
-					metrics.LabelPipeline:           rl.pipelineName,
-					metrics.LabelVertexReplicaIndex: strconv.Itoa(int(rl.vertexReplica)),
-					LabelReason:                     "late"}).Inc()
-
-				// mark it as a successfully written message as the message will be acked to avoid subsequent retries
-				writtenMessages = append(writtenMessages, message)
-				continue
-			} else {
-				var startTime time.Time
-				// bit of an overkill, but this is an unlikely path
-				if nextWin == nil {
-					startTime = time.Time{}
-				} else {
-					startTime = nextWin.StartTime()
-				}
-				rl.log.Debugw("Keeping the late message for next condition check because COB has not happened yet", zap.Int64("eventTime", message.EventTime.UnixMilli()), zap.Int64("watermark", message.Watermark.UnixMilli()), zap.Int64("nextWindowToBeClosed.startTime", startTime.UnixMilli()))
-			}
-		}
-
-		// NOTE(potential bug): if we get a message where the event-time is < (watermark-allowedLateness), skip processing the message.
-		// This could be due to a couple of problem, eg. ack was not registered, etc.
-		// Please do not confuse this with late data! This is a platform related problem causing the watermark inequality
-		// to be violated.
-		if message.EventTime.Before(message.Watermark.Add(-1 * rl.allowedLateness)) {
-			// TODO: track as a counter metric
-			rl.log.Errorw("An old message just popped up", zap.Any("msgOffSet", message.ReadOffset.String()), zap.Int64("eventTime", message.EventTime.UnixMilli()), zap.Int64("watermark", message.Watermark.UnixMilli()), zap.Any("message", message.Message))
-			// mark it as a successfully written message as the message will be acked to avoid subsequent retries
-			writtenMessages = append(writtenMessages, message)
-			// let's not continue processing this message, most likely the window has already been closed and the message
-			// won't be processed anyways.
-			droppedMessagesCount.With(map[string]string{
-				metrics.LabelVertex:             rl.vertexName,
-				metrics.LabelPipeline:           rl.pipelineName,
-				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(rl.vertexReplica)),
-				LabelReason:                     "watermark_issue"}).Inc()
-			continue
-		}
-
-		// identify and add window for the message
-		windows := rl.upsertWindowsAndKeys(message)
-
-		// for each window we will have a PBQ. A message could belong to multiple windows (e.g., sliding).
-		// We need to write the messages to these PBQs.
-		for _, kw := range windows {
-
-			for _, partitionID := range kw.Partitions() {
-
-				err := rl.writeToPBQ(ctx, message, partitionID, kw)
-				// there is no point continuing because we are seeing an error.
-				// this error will ONLY BE set if we are in a erroring loop and ctx.Done() has been invoked.
-				if err != nil {
-					rl.log.Errorw("Failed to write message, asked to stop trying", zap.Any("msgOffSet", message.ReadOffset.String()), zap.String("partitionID", partitionID.String()), zap.Error(err))
-					break messagesLoop
-				}
-			}
-		}
-
-		writtenMessages = append(writtenMessages, message)
-	}
-
-	return writtenMessages, err
-}
-
-// writeToPBQ writes to the PBQ. It will return error only if it is not failing to write to PBQ and is in a continuous
-// error loop, and we have received ctx.Done() via SIGTERM.
-func (rl *ReadLoop) writeToPBQ(ctx context.Context, m *isb.ReadMessage, p partition.ID, kw window.AlignedKeyedWindower) error {
-	startTime := time.Now()
-	defer pbqWriteTime.With(map[string]string{
-		metrics.LabelVertex:             rl.vertexName,
-		metrics.LabelPipeline:           rl.pipelineName,
-		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(rl.vertexReplica)),
-	}).Observe(float64(time.Since(startTime).Milliseconds()))
-
-	var pbqWriteBackoff = wait.Backoff{
-		Steps:    math.MaxInt,
-		Duration: 1 * time.Second,
-		Factor:   1.5,
-		Jitter:   0.1,
-	}
-
-	q := rl.associatePBQAndPnF(ctx, p, kw)
-
-	err := wait.ExponentialBackoff(pbqWriteBackoff, func() (done bool, err error) {
-		rErr := q.Write(context.Background(), m)
-		if rErr != nil {
-			rl.log.Errorw("Failed to write message", zap.Any("msgOffSet", m.ReadOffset.String()), zap.String("partitionID", p.String()), zap.Error(rErr))
-			pbqWriteErrorCount.With(map[string]string{
-				metrics.LabelVertex:             rl.vertexName,
-				metrics.LabelPipeline:           rl.pipelineName,
-				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(rl.vertexReplica)),
-			}).Inc()
-			// no point retrying if ctx.Done has been invoked
-			select {
-			case <-ctx.Done():
-				// no point in retrying after we have been asked to stop.
-				return false, ctx.Err()
-			default:
-				// keep retrying
-				return false, nil
-			}
-
-		}
-		// happy path
-		pbqWriteMessagesCount.With(map[string]string{
-			metrics.LabelVertex:             rl.vertexName,
-			metrics.LabelPipeline:           rl.pipelineName,
-			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(rl.vertexReplica)),
-		}).Inc()
-		rl.log.Debugw("Successfully wrote the message", zap.String("msgOffSet", m.ReadOffset.String()), zap.String("partitionID", p.String()))
-		return true, nil
-	})
-
-	return err
-}
-
-// ackMessages acks messages. Retries until it can succeed or ctx.Done() happens.
-func (rl *ReadLoop) ackMessages(ctx context.Context, messages []*isb.ReadMessage) {
-	var ackBackoff = wait.Backoff{
-		Steps:    math.MaxInt,
-		Duration: 1 * time.Second,
-		Factor:   1.5,
-		Jitter:   0.1,
-	}
-	var wg sync.WaitGroup
-
-	// Ack the message to ISB
-	for _, m := range messages {
-		wg.Add(1)
-
-		go func(o isb.Offset) {
-			defer wg.Done()
-			attempt := 0
-			ackErr := wait.ExponentialBackoff(ackBackoff, func() (done bool, err error) {
-				rErr := o.AckIt()
-				attempt += 1
-				if rErr != nil {
-					ackMessageError.With(map[string]string{
-						metrics.LabelVertex:             rl.vertexName,
-						metrics.LabelPipeline:           rl.pipelineName,
-						metrics.LabelVertexReplicaIndex: strconv.Itoa(int(rl.vertexReplica)),
-					}).Inc()
-
-					rl.log.Errorw("Failed to ack message, retrying", zap.String("msgOffSet", o.String()), zap.Error(rErr), zap.Int("attempt", attempt))
-					// no point retrying if ctx.Done has been invoked
-					select {
-					case <-ctx.Done():
-						// no point in retrying after we have been asked to stop.
-						return false, ctx.Err()
-					default:
-						// keep retrying
-						return false, nil
-					}
-				}
-				rl.log.Debugw("Successfully acked message", zap.String("msgOffSet", o.String()))
-				ackMessagesCount.With(map[string]string{
-					metrics.LabelVertex:             rl.vertexName,
-					metrics.LabelPipeline:           rl.pipelineName,
-					metrics.LabelVertexReplicaIndex: strconv.Itoa(int(rl.vertexReplica)),
-				}).Inc()
-				return true, nil
-			})
-			// no point trying the rest of the message, most likely that will also fail
-			if ackErr != nil {
-				rl.log.Errorw("Context closed while waiting to ack messages inside readloop", zap.Error(ackErr), zap.String("offset", o.String()))
-
-			}
-		}(m.ReadOffset)
-
-	}
-	wg.Wait()
-}
-
-// associatePBQAndPnF associates a PBQ with the partition if a PBQ exists, else creates a new one and then associates
-// it to the partition.
-func (rl *ReadLoop) associatePBQAndPnF(ctx context.Context, partitionID partition.ID, kw window.AlignedKeyedWindower) pbq.ReadWriteCloser {
-	// look for existing pbq
-	q := rl.pbqManager.GetPBQ(partitionID)
-
-	// if we do not have already created PBQ, we have to create a new one.
-	if q == nil {
-		var pbqErr error
-		var infiniteBackoff = wait.Backoff{
-			Steps:    math.MaxInt,
-			Duration: 1 * time.Second,
-			Factor:   1.5,
-			Jitter:   0.1,
-		}
-		pbqErr = wait.ExponentialBackoffWithContext(ctx, infiniteBackoff, func() (done bool, err error) {
-			var attempt int
-			q, pbqErr = rl.pbqManager.CreateNewPBQ(ctx, partitionID, kw)
-			if pbqErr != nil {
-				attempt += 1
-				rl.log.Warnw("Failed to create pbq during startup, retrying", zap.Any("attempt", attempt), zap.String("partitionID", partitionID.String()), zap.Error(pbqErr))
-				return false, nil
-			}
-			return true, nil
-		})
-		// since we created a brand new PBQ it means there is no PnF listening on this PBQ.
-		// we should create and attach the read side of the loop (PnF) to the partition and then
-		// start process-and-forward (pnf) loop
-		t := rl.op.schedulePnF(ctx, rl.UDF, q, partitionID, rl.toBuffers, rl.whereToDecider, rl.publishWatermark, rl.idleManager)
-		rl.udfInvocationTracking[partitionID] = t
-		rl.log.Debugw("Successfully Created/Found pbq and started PnF", zap.String("partitionID", partitionID.String()))
-	}
-
-	return q
-}
-
-// ShutDown shutdowns the read-loop.
-func (rl *ReadLoop) ShutDown(ctx context.Context) {
-	rl.pbqManager.ShutDown(ctx)
-}
-
-// upsertWindowsAndKeys will create or assigns (if already present) a window to the message. It is an upsert operation
-// because windows are created out of order, but they will be closed in-order.
-func (rl *ReadLoop) upsertWindowsAndKeys(m *isb.ReadMessage) []window.AlignedKeyedWindower {
-
-	processingWindows := rl.windower.AssignWindow(m.EventTime)
-	var kWindows []window.AlignedKeyedWindower
-	for _, win := range processingWindows {
-		w, isPresent := rl.windower.InsertIfNotPresent(win)
-		if !isPresent {
-			rl.log.Debugw("Creating new keyed window", zap.String("msg.offset", m.ID), zap.Int64("startTime", w.StartTime().UnixMilli()), zap.Int64("endTime", w.EndTime().UnixMilli()))
-		} else {
-			rl.log.Debugw("Found an existing window", zap.String("msg.offset", m.ID), zap.Int64("startTime", w.StartTime().UnixMilli()), zap.Int64("endTime", w.EndTime().UnixMilli()))
-		}
-		// track the key to window relationship
-		// FIXME: create a slot from m.key
-		w.AddSlot("slot-0")
-		kWindows = append(kWindows, w)
-	}
-	return kWindows
-}
-
-// ClosePartitions closes the partitions by invoking close-of-book (COB).
-func (rl *ReadLoop) ClosePartitions(partitions []partition.ID) {
-	for _, p := range partitions {
-		q := rl.pbqManager.GetPBQ(p)
-		rl.log.Infow("Close of book", zap.String("partitionID", p.String()))
-		// schedule the task for ordered processing.
-		rl.op.insertTask(rl.udfInvocationTracking[p])
-		q.CloseOfBook()
-		delete(rl.udfInvocationTracking, p)
-	}
-}
diff --git a/pkg/udf/reduce_udf.go b/pkg/udf/reduce_udf.go
index 6dcde350..333def7d 100644
--- a/pkg/udf/reduce_udf.go
+++ b/pkg/udf/reduce_udf.go
@@ -22,27 +22,27 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/numaproj/numaflow-go/pkg/function/client"
 	"go.uber.org/zap"
 
-	"github.com/numaproj/numaflow/pkg/forward"
-	"github.com/numaproj/numaflow/pkg/metrics"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq/store/wal"
-
-	"github.com/numaproj/numaflow/pkg/reduce/pbq"
-	"github.com/numaproj/numaflow/pkg/window/strategy/fixed"
-	"github.com/numaproj/numaflow/pkg/window/strategy/sliding"
-
+	"github.com/numaproj/numaflow-go/pkg/function/client"
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/forward"
 	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/reduce"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/store/wal"
+	"github.com/numaproj/numaflow/pkg/reduce/pnf"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"github.com/numaproj/numaflow/pkg/shuffle"
 	"github.com/numaproj/numaflow/pkg/udf/function"
 	"github.com/numaproj/numaflow/pkg/watermark/generic"
 	"github.com/numaproj/numaflow/pkg/watermark/generic/jetstream"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 	"github.com/numaproj/numaflow/pkg/window"
+	"github.com/numaproj/numaflow/pkg/window/strategy/fixed"
+	"github.com/numaproj/numaflow/pkg/window/strategy/sliding"
 )
 
 type ReduceUDFProcessor struct {
@@ -191,12 +191,22 @@ func (u *ReduceUDFProcessor) Start(ctx context.Context) error {
 	if allowedLateness := u.VertexInstance.Vertex.Spec.UDF.GroupBy.AllowedLateness; allowedLateness != nil {
 		opts = append(opts, reduce.WithAllowedLateness(allowedLateness.Duration))
 	}
+	idleManager := wmb.NewIdleManager(len(writers))
+
+	op := pnf.NewOrderedProcessor(ctx, u.VertexInstance, udfHandler, writers, pbqManager, conditionalForwarder, publishWatermark, idleManager)
 
-	dataForwarder, err := reduce.NewDataForward(ctx, udfHandler, u.VertexInstance, reader, writers, pbqManager, conditionalForwarder, fetchWatermark, publishWatermark, windower, opts...)
+	dataForwarder, err := reduce.NewDataForward(ctx, u.VertexInstance, reader, writers, pbqManager, conditionalForwarder, fetchWatermark, publishWatermark, windower, idleManager, op, opts...)
 	if err != nil {
 		return fmt.Errorf("failed get a new DataForward, %w", err)
 	}
 
+	// read the persisted messages before reading the messages from ISB
+	err = dataForwarder.ReplayPersistedMessages(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to read and process persisted messages, %w", err)
+	}
+
+	// start reading the ISB messages.
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
 	go func() {


Test patch:
diff --git a/pkg/reduce/data_forward_test.go b/pkg/reduce/data_forward_test.go
index 6038f79d..3ed0ba1e 100644
--- a/pkg/reduce/data_forward_test.go
+++ b/pkg/reduce/data_forward_test.go
@@ -26,6 +26,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/reduce/pnf"
 	"github.com/stretchr/testify/assert"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
@@ -267,9 +268,12 @@ func TestDataForward_StartWithNoOpWM(t *testing.T) {
 	// create new fixed window of (windowTime)
 	window := fixed.NewFixed(windowTime)
 
+	idleManager := wmb.NewIdleManager(len(toBuffer))
+	op := pnf.NewOrderedProcessor(child, keyedVertex, CounterReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publisher, idleManager)
+
 	var reduceDataForwarder *DataForward
-	reduceDataForwarder, err = NewDataForward(child, CounterReduceTest{}, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, wmpublisher, publisher,
-		window, WithReadBatchSize(10))
+	reduceDataForwarder, err = NewDataForward(child, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, wmpublisher, publisher,
+		window, idleManager, op, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
 	go reduceDataForwarder.Start()
@@ -351,10 +355,12 @@ func TestReduceDataForward_IdleWM(t *testing.T) {
 
 	// create a fixed window of 5s
 	window := fixed.NewFixed(5 * time.Second)
+	idleManager := wmb.NewIdleManager(len(toBuffers))
+	op := pnf.NewOrderedProcessor(ctx, keyedVertex, CounterReduceTest{}, toBuffers, pbqManager, CounterReduceTest{}, publisherMap, idleManager)
 
 	var reduceDataForward *DataForward
-	reduceDataForward, err = NewDataForward(ctx, CounterReduceTest{}, keyedVertex, fromBuffer, toBuffers, pbqManager, CounterReduceTest{}, f, publisherMap,
-		window, WithReadBatchSize(10))
+	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffers, pbqManager, CounterReduceTest{}, f, publisherMap,
+		window, idleManager, op, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
 	// start the forwarder
@@ -549,10 +555,12 @@ func TestReduceDataForward_Count(t *testing.T) {
 
 	// create a fixed window of 60s
 	window := fixed.NewFixed(60 * time.Second)
+	idleManager := wmb.NewIdleManager(len(toBuffer))
+	op := pnf.NewOrderedProcessor(ctx, keyedVertex, CounterReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publisherMap, idleManager)
 
 	var reduceDataForward *DataForward
-	reduceDataForward, err = NewDataForward(ctx, CounterReduceTest{}, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publisherMap,
-		window, WithReadBatchSize(10))
+	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publisherMap,
+		window, idleManager, op, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
 	// start the forwarder
@@ -623,10 +631,13 @@ func TestReduceDataForward_AllowedLatencyCount(t *testing.T) {
 	// create a fixed window of 10s
 	window := fixed.NewFixed(5 * time.Second)
 
+	idleManager := wmb.NewIdleManager(len(toBuffer))
+	op := pnf.NewOrderedProcessor(ctx, keyedVertex, CounterReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publisherMap, idleManager)
+
 	var reduceDataForward *DataForward
 	allowedLatency := 1000
-	reduceDataForward, err = NewDataForward(ctx, CounterReduceTest{}, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publisherMap,
-		window, WithReadBatchSize(int64(batchSize)),
+	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publisherMap,
+		window, idleManager, op, WithReadBatchSize(int64(batchSize)),
 		WithAllowedLateness(time.Duration(allowedLatency)*time.Millisecond),
 	)
 	assert.NoError(t, err)
@@ -697,10 +708,12 @@ func TestReduceDataForward_Sum(t *testing.T) {
 	publishersMap, _ := buildPublisherMapAndOTStore(ctx, toBuffer, pipelineName)
 	// create a fixed window of 2 minutes
 	window := fixed.NewFixed(2 * time.Minute)
+	idleManager := wmb.NewIdleManager(len(toBuffer))
+	op := pnf.NewOrderedProcessor(ctx, keyedVertex, SumReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publishersMap, idleManager)
 
 	var reduceDataForward *DataForward
-	reduceDataForward, err = NewDataForward(ctx, SumReduceTest{}, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publishersMap,
-		window, WithReadBatchSize(10))
+	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publishersMap,
+		window, idleManager, op, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
 	// start the forwarder
@@ -771,10 +784,12 @@ func TestReduceDataForward_Max(t *testing.T) {
 
 	// create a fixed window of 5 minutes
 	window := fixed.NewFixed(5 * time.Minute)
+	idleManager := wmb.NewIdleManager(len(toBuffer))
+	op := pnf.NewOrderedProcessor(ctx, keyedVertex, MaxReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publishersMap, idleManager)
 
 	var reduceDataForward *DataForward
-	reduceDataForward, err = NewDataForward(ctx, MaxReduceTest{}, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publishersMap,
-		window, WithReadBatchSize(10))
+	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publishersMap,
+		window, idleManager, op, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
 	// start the forwarder
@@ -845,10 +860,12 @@ func TestReduceDataForward_SumWithDifferentKeys(t *testing.T) {
 	// create a fixed window of 5 minutes
 	window := fixed.NewFixed(5 * time.Minute)
 
-	var reduceDataForward *DataForward
+	idleManager := wmb.NewIdleManager(len(toBuffer))
+	op := pnf.NewOrderedProcessor(ctx, keyedVertex, SumReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publishersMap, idleManager)
 
-	reduceDataForward, err = NewDataForward(ctx, SumReduceTest{}, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publishersMap,
-		window, WithReadBatchSize(10))
+	var reduceDataForward *DataForward
+	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publishersMap,
+		window, idleManager, op, WithReadBatchSize(10))
 
 	assert.NoError(t, err)
 
@@ -940,9 +957,12 @@ func TestReduceDataForward_NonKeyed(t *testing.T) {
 	// create a fixed window of 5 minutes
 	window := fixed.NewFixed(5 * time.Minute)
 
+	idleManager := wmb.NewIdleManager(len(toBuffer))
+	op := pnf.NewOrderedProcessor(ctx, keyedVertex, SumReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publishersMap, idleManager)
+
 	var reduceDataForward *DataForward
-	reduceDataForward, err = NewDataForward(ctx, SumReduceTest{}, nonKeyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publishersMap,
-		window, WithReadBatchSize(10))
+	reduceDataForward, err = NewDataForward(ctx, nonKeyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publishersMap,
+		window, idleManager, op, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
 	// start the forwarder
@@ -1021,9 +1041,12 @@ func TestDataForward_WithContextClose(t *testing.T) {
 	// create a fixed window of 5 minutes
 	window := fixed.NewFixed(5 * time.Minute)
 
+	idleManager := wmb.NewIdleManager(len(toBuffer))
+	op := pnf.NewOrderedProcessor(ctx, keyedVertex, SumReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publishersMap, idleManager)
+
 	var reduceDataForward *DataForward
-	reduceDataForward, err = NewDataForward(cctx, SumReduceTest{}, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publishersMap,
-		window, WithReadBatchSize(1))
+	reduceDataForward, err = NewDataForward(cctx, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publishersMap,
+		window, idleManager, op, WithReadBatchSize(1))
 	assert.NoError(t, err)
 
 	// start the forwarder
@@ -1284,7 +1307,7 @@ func publishMessagesAllowedLatency(ctx context.Context, startTime int, message i
 
 	// dummy messages to
 	//  - start a new window
-	//  - make sure readloop can
+	//  - make sure data forward can
 	//    read a message,
 	//    get the `6000000` watermark,
 	//    and close the first window.
diff --git a/pkg/reduce/readloop/ordered_test.go b/pkg/reduce/pnf/ordered_test.go
similarity index 88%
rename from pkg/reduce/readloop/ordered_test.go
rename to pkg/reduce/pnf/ordered_test.go
index 4dc239b6..67865cb0 100644
--- a/pkg/reduce/readloop/ordered_test.go
+++ b/pkg/reduce/pnf/ordered_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package readloop
+package pnf
 
 import (
 	"context"
@@ -47,6 +47,18 @@ func (f myForwardTest) Apply(ctx context.Context, message *isb.ReadMessage) ([]*
 	return testutils.CopyUDFTestApply(ctx, message)
 }
 
+var keyedVertex = &dfv1.VertexInstance{
+	Vertex: &dfv1.Vertex{Spec: dfv1.VertexSpec{
+		PipelineName: "testPipeline",
+		AbstractVertex: dfv1.AbstractVertex{
+			Name: "testVertex",
+			UDF:  &dfv1.UDF{GroupBy: &dfv1.GroupBy{Keyed: true}},
+		},
+	}},
+	Hostname: "test-host",
+	Replica:  0,
+}
+
 func TestOrderedProcessing(t *testing.T) {
 	// Test Reducer returns the messages as is
 	identityReducer := applier.ApplyReduceFunc(func(ctx context.Context, partitionID *partition.ID, input <-chan *isb.ReadMessage) ([]*isb.WriteMessage, error) {
@@ -104,25 +116,26 @@ func TestOrderedProcessing(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			// clean out the task queue before we start a run
-			op := newOrderedForwarder(ctx, "reduce", "test-pipeline", 0)
 			// although this could be declared outside, since we are using common naming scheme for partitions,
 			// things will go haywire.
 			pbqManager, _ := pbq.NewManager(ctx, "reduce", "test-pipeline", 0, memory.NewMemoryStores(memory.WithStoreSize(100)),
 				pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
+
+			op := NewOrderedProcessor(ctx, keyedVertex, identityReducer, toSteps, pbqManager, myForwardTest{}, pw, idleManager)
 			cCtx, cancelFn := context.WithCancel(ctx)
 			defer cancelFn()
 			for _, _partition := range tt.partitions {
 				kw := keyed.NewKeyedWindow(_partition.Start, _partition.End)
 				kw.AddSlot(_partition.Slot)
 
-				p, _ := pbqManager.CreateNewPBQ(ctx, _partition, kw)
-				t := op.schedulePnF(cCtx, identityReducer, p, _partition, toSteps, myForwardTest{}, pw, idleManager)
-				op.insertTask(t)
+				_, _ = pbqManager.CreateNewPBQ(ctx, _partition, kw)
+				t := op.SchedulePnF(cCtx, _partition)
+				op.InsertTask(t)
 			}
 			assert.Equal(t, op.taskQueue.Len(), tt.expectedBefore)
 			count := 0
 			for e := op.taskQueue.Front(); e != nil; e = e.Next() {
-				pfTask := e.Value.(*task)
+				pfTask := e.Value.(*ForwardTask)
 				assert.Equal(t, partitionFor(count), pfTask.pf.PartitionID)
 				count = count + 1
 			}
@@ -174,11 +187,11 @@ func partitionFor(i int) partition.ID {
 	return partition.ID{Start: time.UnixMilli(int64(base * i)), End: time.UnixMilli(int64(base * (i + 1)))}
 }
 
-func taskForPartition(op *orderedForwarder, partitionId partition.ID) *task {
+func taskForPartition(op *OrderedProcessor, partitionId partition.ID) *ForwardTask {
 	op.RLock()
 	defer op.RUnlock()
 	for e := op.taskQueue.Front(); e != nil; e = e.Next() {
-		pfTask := e.Value.(*task)
+		pfTask := e.Value.(*ForwardTask)
 		partitionKey := pfTask.pf.PartitionID
 		if partitionKey == partitionId {
 			return pfTask
diff --git a/pkg/reduce/pnf/processandforward_test.go b/pkg/reduce/pnf/processandforward_test.go
index 81760906..45db4afd 100644
--- a/pkg/reduce/pnf/processandforward_test.go
+++ b/pkg/reduce/pnf/processandforward_test.go
@@ -57,11 +57,11 @@ const (
 	publisherOTKeyspace = testPipelineName + "_" + testProcessorEntity + "_%s_" + "OT"
 )
 
-type myForwardTest struct {
+type forwardTest struct {
 	buffers []string
 }
 
-func (f myForwardTest) WhereTo(keys []string, _ []string) ([]string, error) {
+func (f forwardTest) WhereTo(keys []string, _ []string) ([]string, error) {
 	if strings.Compare(keys[len(keys)-1], "test-forward-one") == 0 {
 		return []string{"buffer1"}, nil
 	} else if strings.Compare(keys[len(keys)-1], "test-forward-all") == 0 {
@@ -70,7 +70,7 @@ func (f myForwardTest) WhereTo(keys []string, _ []string) ([]string, error) {
 	return []string{}, nil
 }
 
-func (f myForwardTest) Apply(ctx context.Context, message *isb.ReadMessage) ([]*isb.WriteMessage, error) {
+func (f forwardTest) Apply(ctx context.Context, message *isb.ReadMessage) ([]*isb.WriteMessage, error) {
 	return testutils.CopyUDFTestApply(ctx, message)
 }
 
@@ -81,7 +81,7 @@ type PayloadForTest struct {
 
 func TestProcessAndForward_Process(t *testing.T) {
 	// 1. create a pbq which has to be passed to the process method
-	// 2. pass the pbqReader interface and create a new ProcessAndForward instance
+	// 2. pass the pbqReader interface and create a new processAndForward instance
 	// 3. mock the grpc client methods
 	// 4. assert to check if the process method is returning the result
 
@@ -160,7 +160,7 @@ func TestProcessAndForward_Process(t *testing.T) {
 	_, publishWatermark := generic.BuildNoOpWatermarkProgressorsFromBufferMap(make(map[string]isb.BufferWriter))
 
 	// create pf using key and reducer
-	pf := NewProcessAndForward(ctx, "reduce", "test-pipeline", 0, testPartition, client, simplePbq, make(map[string]isb.BufferWriter, 1), myForwardTest{}, publishWatermark, wmb.NewIdleManager(1))
+	pf := newProcessAndForward(ctx, "reduce", "test-pipeline", 0, testPartition, client, simplePbq, make(map[string]isb.BufferWriter, 1), forwardTest{}, publishWatermark, wmb.NewIdleManager(1))
 
 	err = pf.Process(ctx)
 	assert.NoError(t, err)
@@ -208,7 +208,7 @@ func TestProcessAndForward_Forward(t *testing.T) {
 		name       string
 		id         partition.ID
 		buffers    []*simplebuffer.InMemoryBuffer
-		pf         ProcessAndForward
+		pf         processAndForward
 		otStores   map[string]wmstore.WatermarkKVStorer
 		expected   []bool
 		wmExpected map[string]wmb.WMB
@@ -352,7 +352,7 @@ func TestWriteToBuffer(t *testing.T) {
 	}
 }
 
-func createProcessAndForwardAndOTStore(ctx context.Context, key string, pbqManager *pbq.Manager, toBuffers map[string]isb.BufferWriter) (ProcessAndForward, map[string]wmstore.WatermarkKVStorer) {
+func createProcessAndForwardAndOTStore(ctx context.Context, key string, pbqManager *pbq.Manager, toBuffers map[string]isb.BufferWriter) (processAndForward, map[string]wmstore.WatermarkKVStorer) {
 
 	testPartition := partition.ID{
 		Start: time.UnixMilli(60000),
@@ -392,11 +392,11 @@ func createProcessAndForwardAndOTStore(ctx context.Context, key string, pbqManag
 	for k := range toBuffers {
 		buffers = append(buffers, k)
 	}
-	whereto := &myForwardTest{
+	whereto := &forwardTest{
 		buffers: buffers,
 	}
 
-	pf := ProcessAndForward{
+	pf := processAndForward{
 		PartitionID:      testPartition,
 		UDF:              nil,
 		result:           result,
diff --git a/pkg/reduce/readloop/readloop_test.go b/pkg/reduce/readloop/readloop_test.go
deleted file mode 100644
index f000446b..00000000
--- a/pkg/reduce/readloop/readloop_test.go
+++ /dev/null
@@ -1,228 +0,0 @@
-/*
-Copyright 2022 The Numaproj Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package readloop
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/assert"
-
-	"github.com/numaproj/numaflow/pkg/watermark/wmb"
-
-	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq/store/memory"
-	"github.com/numaproj/numaflow/pkg/watermark/generic"
-	"github.com/numaproj/numaflow/pkg/window/strategy/fixed"
-)
-
-// PayloadForTest is a dummy payload for testing.
-type PayloadForTest struct {
-	Key   string
-	Value int
-}
-
-type SumReduceTest struct {
-}
-
-func (s *SumReduceTest) WhereTo(_ []string, _ []string) ([]string, error) {
-	return []string{"reduce-buffer"}, nil
-}
-
-func (s SumReduceTest) ApplyReduce(_ context.Context, partitionID *partition.ID, messageStream <-chan *isb.ReadMessage) ([]*isb.WriteMessage, error) {
-	sums := make(map[string]int)
-
-	for msg := range messageStream {
-		var payload PayloadForTest
-		_ = json.Unmarshal(msg.Payload, &payload)
-		keys := msg.Keys
-		sums[keys[0]] += payload.Value
-	}
-
-	msgs := make([]*isb.WriteMessage, 0)
-
-	for k, s := range sums {
-		payload := PayloadForTest{Key: k, Value: s}
-		b, _ := json.Marshal(payload)
-		msg := &isb.WriteMessage{
-			Message: isb.Message{
-				Header: isb.Header{
-					MessageInfo: isb.MessageInfo{
-						EventTime: partitionID.End,
-					},
-					ID:   "msgID",
-					Keys: []string{k},
-				},
-				Body: isb.Body{Payload: b},
-			},
-			Tags: nil,
-		}
-		msgs = append(msgs, msg)
-	}
-
-	return msgs, nil
-}
-
-// testing startup code with replay included using in-memory pbq
-func TestReadLoop_Startup(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
-	defer cancel()
-
-	// partitions to be replayed
-	partitionIds := []partition.ID{
-		{
-			Start: time.Unix(60, 0),
-			End:   time.Unix(120, 0),
-			Slot:  "window",
-		},
-		{
-			Start: time.Unix(60, 0),
-			End:   time.Unix(120, 0),
-			Slot:  "window",
-		},
-		{
-			Start: time.Unix(120, 0),
-			End:   time.Unix(180, 0),
-			Slot:  "window",
-		},
-		{
-			Start: time.Unix(180, 0),
-			End:   time.Unix(240, 0),
-			Slot:  "window",
-		},
-	}
-
-	memStoreProvider := memory.NewMemoryStores(memory.WithStoreSize(100))
-
-	for _, id := range partitionIds {
-		memStore, err := memStoreProvider.CreateStore(ctx, id)
-		assert.NoError(t, err)
-
-		var msgVal int
-		if id.Slot == "even" {
-			msgVal = 2
-		} else {
-			msgVal = 3
-		}
-
-		// write messages to the store, which will be replayed
-		storeMessages := createStoreMessages(ctx, id.Slot, msgVal, id.Start, 10)
-		for _, msg := range storeMessages {
-			err = memStore.Write(msg)
-			assert.NoError(t, err)
-		}
-	}
-
-	pManager, _ := pbq.NewManager(ctx, "reduce", "test-pipeline", 0, memStoreProvider, pbq.WithChannelBufferSize(10))
-
-	to1 := simplebuffer.NewInMemoryBuffer("reduce-buffer", 3)
-	toSteps := map[string]isb.BufferWriter{
-		"reduce-buffer": to1,
-	}
-
-	_, pw := generic.BuildNoOpWatermarkProgressorsFromBufferMap(toSteps)
-
-	window := fixed.NewFixed(60 * time.Second)
-
-	idleManager := wmb.NewIdleManager(len(toSteps))
-
-	rl, err := NewReadLoop(ctx, "reduce", "test-pipeline", 0, &SumReduceTest{}, pManager, window, toSteps, &SumReduceTest{}, pw, idleManager, time.Duration(0))
-	assert.NoError(t, err)
-	err = rl.Startup(ctx)
-	assert.NoError(t, err)
-
-	// send a message with the higher watermark so that the windows will be closed
-	latestMessage := &isb.ReadMessage{
-		Message: isb.Message{
-			Header: isb.Header{
-				MessageInfo: isb.MessageInfo{
-					EventTime: time.Unix(300, 0),
-					IsLate:    false,
-				},
-				ID:   "",
-				Keys: []string{""},
-			},
-			Body: isb.Body{},
-		},
-		ReadOffset: isb.SimpleStringOffset(func() string { return "simple-offset" }),
-		Watermark:  time.Unix(300, 0),
-	}
-
-	rl.Process(ctx, []*isb.ReadMessage{latestMessage})
-
-	for !to1.IsFull() {
-		select {
-		case <-ctx.Done():
-			assert.Fail(t, ctx.Err().Error())
-			return
-		default:
-			time.Sleep(100 * time.Millisecond)
-		}
-	}
-
-	msgs, readErr := to1.Read(ctx, 3)
-	assert.Nil(t, readErr)
-	assert.Len(t, msgs, 3)
-	// since we have 3 partitions we should have 3 different outputs
-	var readMessagePayload1 PayloadForTest
-	var readMessagePayload2 PayloadForTest
-	var readMessagePayload3 PayloadForTest
-	_ = json.Unmarshal(msgs[0].Payload, &readMessagePayload1)
-	_ = json.Unmarshal(msgs[1].Payload, &readMessagePayload2)
-	_ = json.Unmarshal(msgs[2].Payload, &readMessagePayload3)
-	// since we had 10 messages in the store with value 2 and 3
-	// the expected value is 20 and 30, since the reduce operation is sum
-	assert.Contains(t, []int{20, 30, 20, 30, 60}, readMessagePayload1.Value)
-	assert.Contains(t, []int{20, 30, 20, 30}, readMessagePayload2.Value)
-	assert.Contains(t, []int{20, 30, 20, 30}, readMessagePayload3.Value)
-	assert.Equal(t, "window", readMessagePayload1.Key)
-	assert.Equal(t, "window", readMessagePayload2.Key)
-	assert.Equal(t, "window", readMessagePayload3.Key)
-
-}
-
-func createStoreMessages(_ context.Context, key string, value int, eventTime time.Time, count int) []*isb.ReadMessage {
-	readMessages := make([]*isb.ReadMessage, count)
-	for j := 0; j < count; j++ {
-		result, _ := json.Marshal(PayloadForTest{
-			Key:   key,
-			Value: value,
-		})
-		readMessage := &isb.ReadMessage{
-			Message: isb.Message{
-				Header: isb.Header{
-					MessageInfo: isb.MessageInfo{
-						EventTime: eventTime,
-					},
-					ID:   fmt.Sprintf("%d", value+1),
-					Keys: []string{key},
-				},
-				Body: isb.Body{Payload: result},
-			},
-			ReadOffset: isb.SimpleStringOffset(func() string { return "simple-offset" }),
-		}
-		eventTime = eventTime.Add(time.Second)
-		readMessages[j] = readMessage
-	}
-	return readMessages
-}