Repository: numaproj/numaflow
Pull request #208 (2022-10-12 14:14:35)

feat: reducer for stream aggregation without fault tolerance
Introduces a separate forwarder for reduce operation, Readloop to write the data to PBQ and drive the processing using ProcessAndForward which invokes the Reduce operation.

TODO in subsequent PRs
- introduce consistency option, 
- end to end test
- integrate with gRPC reducer


Patch:
diff --git a/docs/APIs.md b/docs/APIs.md
index a2db610f..866928ec 100644
--- a/docs/APIs.md
+++ b/docs/APIs.md
@@ -3217,6 +3217,9 @@ StoreType (<code>string</code> alias)
 </p>
 </h3>
 <p>
+<p>
+PBQ store’s backend type.
+</p>
 </p>
 <h3 id="numaflow.numaproj.io/v1alpha1.TLS">
 TLS
@@ -3768,6 +3771,12 @@ means no delay.
 </tr>
 </tbody>
 </table>
+<h3 id="numaflow.numaproj.io/v1alpha1.WindowType">
+WindowType (<code>string</code> alias)
+</p>
+</h3>
+<p>
+</p>
 <hr/>
 <p>
 <em> Generated with <code>gen-crd-api-reference-docs</code>. </em>
diff --git a/pkg/apis/numaflow/v1alpha1/const.go b/pkg/apis/numaflow/v1alpha1/const.go
index 1bc2de93..35c59410 100644
--- a/pkg/apis/numaflow/v1alpha1/const.go
+++ b/pkg/apis/numaflow/v1alpha1/const.go
@@ -121,10 +121,23 @@ const (
 	DefaultStoreSize          = 1000000         // Default persistent store size
 	DefaultStoreMaxBufferSize = 100000          // Default buffer size for pbq in bytes
 
+	// Default window options
+	DefaultWindowType     = FixedType
+	DefaultWindowDuration = 0
+
+	// Default reduce forward options
+	DefaultReadBatchSize = 100
 )
 
+// PBQ store's backend type.
 type StoreType string
 
+const (
+	InMemoryType   StoreType = "in-memory"
+	FileSystemType StoreType = "file-system"
+	NoOpType       StoreType = "no-op"
+)
+
 func (st StoreType) String() string {
 	switch st {
 	case InMemoryType:
@@ -138,12 +151,27 @@ func (st StoreType) String() string {
 	}
 }
 
+type WindowType string
+
 const (
-	InMemoryType   StoreType = "in-memory"
-	FileSystemType StoreType = "file-system"
-	NoOpType       StoreType = "no-op"
+	FixedType   WindowType = "fixed"
+	SlidingType WindowType = "sliding"
+	SessionType WindowType = "session"
 )
 
+func (wt WindowType) String() string {
+	switch wt {
+	case FixedType:
+		return string(FixedType)
+	case SlidingType:
+		return string(SlidingType)
+	case SessionType:
+		return string(SessionType)
+	default:
+		return "unknownWindowType"
+	}
+}
+
 var (
 	MessageKeyDrop = fmt.Sprintf("%U__DROP__", '\\') // U+005C__DROP__
 	MessageKeyAll  = fmt.Sprintf("%U__ALL__", '\\')  // U+005C__ALL__
diff --git a/pkg/messages/interfaces.go b/pkg/messages/interfaces.go
new file mode 100644
index 00000000..8707ba93
--- /dev/null
+++ b/pkg/messages/interfaces.go
@@ -0,0 +1,43 @@
+package messages
+
+import (
+	"encoding"
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/watermark/processor"
+	"time"
+)
+
+// Message is the representation of a message flowing through the numaflow
+type Message interface {
+	ReadMessage
+	encoding.BinaryMarshaler
+	encoding.BinaryUnmarshaler
+	// SetID sets the ID for a message
+	SetID(id string)
+	// ID is the unique identifier for a message
+	ID() string
+	// SetEventTime represents the timestamp of the message depending on the time characteristics of the pipeline.
+	// please note that EventTime does not mean only the event time.
+	// This is intended to be set once and should be immutable. Therefore, an error could be thrown if
+	// an attempt is made to overwrite the time on a message.
+	SetEventTime(t time.Time)
+	// EventTime returns the timestamp of the message. Calling Time() on a message should give a meaningful error
+	// for the user to understand that a time has not been set yet.
+	EventTime() time.Time
+	// Key returns the key of the message
+	Key() string
+	// Payload returns the payload received from the source it was read.
+	Payload() []byte
+}
+
+// ReadMessage exposes functions to set and get in-vertex stream characteristics of a NumaMessage.
+type ReadMessage interface {
+	// Offset returns the offset of the message from the source it was read.
+	Offset() isb.Offset
+	// SetWatermark sets the watermark on a message read from a source.
+	SetWatermark(watermark processor.Watermark)
+	// Watermark returns the watermark for a message.
+	Watermark() processor.Watermark
+	// Ack acknowledges a message
+	Ack()
+}
diff --git a/pkg/pbq/interfaces.go b/pkg/pbq/interfaces.go
index e94628ac..6abd4a82 100644
--- a/pkg/pbq/interfaces.go
+++ b/pkg/pbq/interfaces.go
@@ -15,7 +15,7 @@ type ReadWriteCloser interface {
 // Reader provides methods to read from PBQ.
 type Reader interface {
 	// ReadCh exposes channel to read from PBQ
-	ReadCh() <-chan *isb.Message
+	ReadCh() <-chan *isb.ReadMessage
 	// GC does garbage collection, it deletes all the persisted data from the store
 	GC() error
 }
@@ -24,7 +24,7 @@ type Reader interface {
 // No data can be written to PBQ after cob.
 type WriteCloser interface {
 	// Write writes message to PBQ
-	Write(ctx context.Context, msg *isb.Message) error
+	Write(ctx context.Context, msg *isb.ReadMessage) error
 	// CloseOfBook (cob) closes PBQ, no writes will be accepted after cob
 	CloseOfBook()
 	// Close to handle context close on writer
diff --git a/pkg/pbq/partition/partition.go b/pkg/pbq/partition/partition.go
new file mode 100644
index 00000000..11fe0036
--- /dev/null
+++ b/pkg/pbq/partition/partition.go
@@ -0,0 +1,17 @@
+package partition
+
+import (
+	"fmt"
+	"time"
+)
+
+// ID uniquely identifies a partition
+type ID struct {
+	Start time.Time
+	End   time.Time
+	Key   string
+}
+
+func (p ID) String() string {
+	return fmt.Sprintf("%v-%v-%s", p.Start.Unix(), p.End.Unix(), p.Key)
+}
diff --git a/pkg/pbq/pbq.go b/pkg/pbq/pbq.go
index b4040b52..4d3cf042 100644
--- a/pkg/pbq/pbq.go
+++ b/pkg/pbq/pbq.go
@@ -5,19 +5,20 @@ import (
 	"errors"
 
 	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/pbq/store"
 	"go.uber.org/zap"
 )
 
-var COBErr error = errors.New("error while writing to pbq, pbq is closed")
+var COBErr = errors.New("error while writing to pbq, pbq is closed")
 
 // PBQ Buffer queue which is backed with a persisted store, each partition
 // will have a PBQ associated with it
 type PBQ struct {
 	store       store.Store
-	output      chan *isb.Message
+	output      chan *isb.ReadMessage
 	cob         bool // cob to avoid panic in case writes happen after close of book
-	partitionID string
+	PartitionID partition.ID
 	options     *options
 	manager     *Manager
 	log         *zap.SugaredLogger
@@ -26,10 +27,10 @@ type PBQ struct {
 var _ ReadWriteCloser = (*PBQ)(nil)
 
 // Write writes message to pbq and persistent store
-func (p *PBQ) Write(ctx context.Context, message *isb.Message) error {
+func (p *PBQ) Write(ctx context.Context, message *isb.ReadMessage) error {
 	// if cob we should return
 	if p.cob {
-		p.log.Errorw("failed to write message to pbq, pbq is closed", zap.Any("partitionID", p.partitionID), zap.Any("header", message.Header))
+		p.log.Errorw("failed to write message to pbq, pbq is closed", zap.Any("ID", p.PartitionID), zap.Any("header", message.Header))
 		return COBErr
 	}
 	var writeErr error
@@ -62,27 +63,28 @@ func (p *PBQ) Close() error {
 
 // ReadCh exposes read channel to read messages from PBQ
 // close on read channel indicates COB
-func (p *PBQ) ReadCh() <-chan *isb.Message {
+func (p *PBQ) ReadCh() <-chan *isb.ReadMessage {
 	return p.output
 }
 
-// GC is invoked after the Reader (ProcessAndForward) has finished
-// forwarding the output to ISB.
+// GC cleans up the PBQ and also the store associated with it. GC is invoked after the Reader (ProcessAndForward) has
+// finished forwarding the output to ISB.
 func (p *PBQ) GC() error {
 	err := p.store.GC()
 	p.store = nil
-	p.manager.deregister(p.partitionID)
+	p.manager.deregister(p.PartitionID)
 	return err
 }
 
-// replayRecordsFromStore replays store messages when replay flag is set during start up time
+// replayRecordsFromStore replays store messages when replay flag is set during start up time. It replays by reading from
+// the store and writing to the PBQ channel.
 func (p *PBQ) replayRecordsFromStore(ctx context.Context) {
 	size := p.options.readBatchSize
 readLoop:
 	for {
-		readMessages, eof, err := p.store.Read(int64(size))
+		readMessages, eof, err := p.store.Read(size)
 		if err != nil {
-			p.log.Errorw("error while replaying records from store", zap.Any("partitionID", p.partitionID), zap.Error(err))
+			p.log.Errorw("error while replaying records from store", zap.Any("ID", p.PartitionID), zap.Error(err))
 		}
 		for _, msg := range readMessages {
 			// select to avoid infinite blocking while writing to output channel
diff --git a/pkg/pbq/pbqmanager.go b/pkg/pbq/pbqmanager.go
index 5487110d..b7c133ed 100644
--- a/pkg/pbq/pbqmanager.go
+++ b/pkg/pbq/pbqmanager.go
@@ -4,11 +4,13 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math"
 	"sync"
 	"time"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/pbq/store"
 	"github.com/numaproj/numaflow/pkg/pbq/store/memory"
 	"github.com/numaproj/numaflow/pkg/pbq/store/noop"
@@ -52,30 +54,32 @@ func NewManager(ctx context.Context, opts ...PBQOption) (*Manager, error) {
 }
 
 // CreateNewPBQ creates new pbq for a partition
-func (m *Manager) CreateNewPBQ(ctx context.Context, partitionID string) (ReadWriteCloser, error) {
+func (m *Manager) CreateNewPBQ(ctx context.Context, partitionID partition.ID) (ReadWriteCloser, error) {
 
 	var persistentStore store.Store
 	var err error
 
-	switch m.storeOptions.PbqStoreType() {
+	switch m.storeOptions.PBQStoreType() {
 	case dfv1.NoOpType:
 		persistentStore, _ = noop.NewPBQNoOpStore()
 	case dfv1.InMemoryType:
 		persistentStore, err = memory.NewMemoryStore(ctx, partitionID, m.storeOptions)
 		if err != nil {
-			m.log.Errorw("Error while creating persistent store", zap.Any("partitionID", partitionID), zap.Any("storeType", m.storeOptions.PbqStoreType()), zap.Error(err))
+			m.log.Errorw("Error while creating persistent store", zap.Any("ID", partitionID), zap.Any("storeType", m.storeOptions.PBQStoreType()), zap.Error(err))
 			return nil, err
 		}
 	case dfv1.FileSystemType:
-		return nil, errors.New("not implemented")
+		return nil, fmt.Errorf("not implemented, %s", dfv1.FileSystemType)
+	default:
+		return nil, errors.New("not implemented (default)")
 	}
 
 	// output channel is buffered to support bulk reads
 	p := &PBQ{
 		store:       persistentStore,
-		output:      make(chan *isb.Message, m.pbqOptions.channelBufferSize),
+		output:      make(chan *isb.ReadMessage, m.pbqOptions.channelBufferSize),
 		cob:         false,
-		partitionID: partitionID,
+		PartitionID: partitionID,
 		options:     m.pbqOptions,
 		manager:     m,
 		log:         logging.FromContext(ctx).With("PBQ", partitionID),
@@ -85,6 +89,20 @@ func (m *Manager) CreateNewPBQ(ctx context.Context, partitionID string) (ReadWri
 	return p, nil
 }
 
+// discoverPartitions discovers partitions.
+func (m *Manager) discoverPartitions(ctx context.Context) ([]partition.ID, error) {
+	switch m.storeOptions.PBQStoreType() {
+	case dfv1.NoOpType:
+		return noop.DiscoverPartitions(ctx, m.storeOptions)
+	case dfv1.InMemoryType:
+		return memory.DiscoverPartitions(ctx, m.storeOptions)
+	case dfv1.FileSystemType:
+		return nil, fmt.Errorf("not implemented, %s", dfv1.FileSystemType)
+	default:
+		return nil, errors.New("not implemented (default)")
+	}
+}
+
 // ListPartitions returns all the pbq instances
 func (m *Manager) ListPartitions() []*PBQ {
 	m.RLock()
@@ -100,32 +118,70 @@ func (m *Manager) ListPartitions() []*PBQ {
 	return pbqList
 }
 
-// GetPBQ returns pbq for the given partitionID
-func (m *Manager) GetPBQ(partitionID string) ReadWriteCloser {
+// GetPBQ returns pbq for the given ID
+func (m *Manager) GetPBQ(partitionID partition.ID) ReadWriteCloser {
 	m.RLock()
 	defer m.RUnlock()
 
-	if pbqInstance, ok := m.pbqMap[partitionID]; ok {
+	if pbqInstance, ok := m.pbqMap[partitionID.String()]; ok {
 		return pbqInstance
 	}
 
 	return nil
 }
 
-// StartUp restores the state of the pbqManager
-func (m *Manager) StartUp(_ context.Context) error {
+// StartUp restores the state of the pbqManager. It reads from the PBQs store to get the persisted partitions
+// and builds the PBQ Map.
+func (m *Manager) StartUp(ctx context.Context) {
+	var ctxClosedErr error
+	var partitionIDs []partition.ID
 
-	switch m.storeOptions.PbqStoreType() {
-	case dfv1.NoOpType:
-		return nil
-	case dfv1.InMemoryType:
-		return nil
-	case dfv1.FileSystemType:
-		return fmt.Errorf("not implemented store type %s", m.storeOptions.PbqStoreType())
-	default:
-		return fmt.Errorf("unknown store type %s", m.storeOptions.PbqStoreType())
+	var discoverPartitionsBackoff = wait.Backoff{
+		Steps:    math.MaxInt64,
+		Duration: 100 * time.Millisecond,
+		Factor:   1,
+		Jitter:   0.1,
 	}
 
+	ctxClosedErr = wait.ExponentialBackoffWithContext(ctx, discoverPartitionsBackoff, func() (done bool, err error) {
+		var attempt int
+
+		partitionIDs, err = m.discoverPartitions(ctx)
+		if err != nil {
+			attempt += 1
+			m.log.Errorw("Failed to discover partitions during startup, retrying", zap.Any("attempt", attempt), zap.Error(err))
+			return false, nil
+		}
+		return true, nil
+	})
+	if ctxClosedErr != nil {
+		m.log.Errorw("Context closed while discovering partitions", zap.Error(ctxClosedErr))
+		return
+	}
+
+	var createPBQBackoff = wait.Backoff{
+		Steps:    math.MaxInt64,
+		Duration: 100 * time.Millisecond,
+		Factor:   1,
+		Jitter:   0.1,
+	}
+
+	for _, partitionID := range partitionIDs {
+		ctxClosedErr = wait.ExponentialBackoffWithContext(ctx, createPBQBackoff, func() (done bool, err error) {
+			var attempt int
+			_, err = m.CreateNewPBQ(ctx, partitionID)
+
+			if err != nil {
+				attempt += 1
+				m.log.Errorw("Failed to create pbq during startup, retrying", zap.Any("attempt", attempt), zap.Any("partitionID", partitionID.String()), zap.Error(err))
+				return false, nil
+			}
+			return true, nil
+		})
+		if ctxClosedErr != nil {
+			m.log.Errorw("Context closed while creating new pbq", zap.Any("partitionID", partitionID.String()), zap.Error(ctxClosedErr))
+		}
+	}
 }
 
 // ShutDown for clean shut down, flushes pending messages to store and closes the store
@@ -135,31 +191,30 @@ func (m *Manager) ShutDown(ctx context.Context) {
 	var wg sync.WaitGroup
 
 	var PBQCloseBackOff = wait.Backoff{
-		Steps:    5,
-		Duration: 1 * time.Second,
-		Factor:   1.5,
+		Steps:    math.MaxInt64,
+		Duration: 100 * time.Millisecond,
+		Factor:   1,
 		Jitter:   0.1,
-		Cap:      5 * time.Second,
 	}
 
 	for _, v := range m.pbqMap {
 		wg.Add(1)
 		go func(q *PBQ) {
 			defer wg.Done()
-			var closeErr error
+			var ctxClosedErr error
 			var attempt int
-			closeErr = wait.ExponentialBackoffWithContext(ctx, PBQCloseBackOff, func() (done bool, err error) {
-				closeErr = q.Close()
+			ctxClosedErr = wait.ExponentialBackoffWithContext(ctx, PBQCloseBackOff, func() (done bool, err error) {
+				closeErr := q.Close()
 				if closeErr != nil {
 					attempt += 1
-					m.log.Warnw("Failed to close pbq, retrying", zap.Any("attempt", attempt), zap.Any("partitionID", q.partitionID), zap.Error(closeErr))
+					m.log.Errorw("Failed to close pbq, retrying", zap.Any("attempt", attempt), zap.Any("ID", q.PartitionID), zap.Error(closeErr))
 					// exponential backoff will return if err is not nil
 					return false, nil
 				}
 				return true, nil
 			})
-			if closeErr != nil {
-				m.log.Errorw("Failed to close pbq, no retries left", zap.Any("partitionID", q.partitionID), zap.Error(closeErr))
+			if ctxClosedErr != nil {
+				m.log.Errorw("Context closed while closing pbq", zap.Any("ID", q.PartitionID), zap.Error(ctxClosedErr))
 			}
 		}(v)
 	}
@@ -168,30 +223,30 @@ func (m *Manager) ShutDown(ctx context.Context) {
 }
 
 // register is intended to be used by PBQ to register itself with the manager.
-func (m *Manager) register(partitionID string, p *PBQ) {
+func (m *Manager) register(partitionID partition.ID, p *PBQ) {
 	m.Lock()
 	defer m.Unlock()
 
-	if _, ok := m.pbqMap[partitionID]; !ok {
-		m.pbqMap[partitionID] = p
+	if _, ok := m.pbqMap[partitionID.String()]; !ok {
+		m.pbqMap[partitionID.String()] = p
 	}
 }
 
 // deregister is intended to be used by PBQ to deregister itself after GC is called.
-func (m *Manager) deregister(partitionID string) {
+func (m *Manager) deregister(partitionID partition.ID) {
 	m.Lock()
 	defer m.Unlock()
 
-	delete(m.pbqMap, partitionID)
+	delete(m.pbqMap, partitionID.String())
 }
 
-// Replay replays messages which are persisted in pbq store
+// Replay replays messages which are persisted in pbq store.
 func (m *Manager) Replay(ctx context.Context) {
 	var wg sync.WaitGroup
 
 	for _, val := range m.pbqMap {
 		wg.Add(1)
-		m.log.Info("Replaying records from store", zap.Any("PBQ", val.partitionID))
+		m.log.Info("Replaying records from store", zap.Any("PBQ", val.PartitionID))
 		go func(ctx context.Context, p *PBQ) {
 			defer wg.Done()
 			p.replayRecordsFromStore(ctx)
diff --git a/pkg/pbq/store/errors.go b/pkg/pbq/store/errors.go
index 56a6eed4..8c73e7a2 100644
--- a/pkg/pbq/store/errors.go
+++ b/pkg/pbq/store/errors.go
@@ -2,6 +2,6 @@ package store
 
 import "errors"
 
-var WriteStoreFullErr error = errors.New("error while writing to store, store is full")
-var WriteStoreClosedErr error = errors.New("error while writing to store, store is closed")
-var ReadStoreEmptyErr error = errors.New("error while reading from store, store is empty")
+var WriteStoreFullErr error = errors.New("error writing, store is full")
+var WriteStoreClosedErr error = errors.New("error writing, store is closed")
+var ReadStoreEmptyErr error = errors.New("error reading, store is empty")
diff --git a/pkg/pbq/store/interfaces.go b/pkg/pbq/store/interfaces.go
index d60fd8ab..032e8b77 100644
--- a/pkg/pbq/store/interfaces.go
+++ b/pkg/pbq/store/interfaces.go
@@ -7,9 +7,9 @@ import (
 // Store provides methods to read, write and delete data from the store.
 type Store interface {
 	// Read returns upto N(size) messages from the persisted store
-	Read(size int64) ([]*isb.Message, bool, error)
+	Read(size int64) ([]*isb.ReadMessage, bool, error)
 	// Write writes message to persistence store
-	Write(msg *isb.Message) error
+	Write(msg *isb.ReadMessage) error
 	// Close closes store
 	Close() error
 	// GC does garbage collection and deletes all the messages that are persisted
diff --git a/pkg/pbq/store/memory/discover.go b/pkg/pbq/store/memory/discover.go
new file mode 100644
index 00000000..ba44fb15
--- /dev/null
+++ b/pkg/pbq/store/memory/discover.go
@@ -0,0 +1,24 @@
+package memory
+
+import (
+	"context"
+
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/pbq/store"
+)
+
+var discoverer func(*store.StoreOptions) ([]partition.ID, error)
+
+// DiscoverPartitions discovers the in-memory partitions. This is very useful during testing.
+func DiscoverPartitions(ctx context.Context, options *store.StoreOptions) ([]partition.ID, error) {
+	if discoverer == nil {
+		return []partition.ID{}, nil
+	}
+	return discoverer(options)
+}
+
+// SetDiscoverer sets the function that returns the initial list of partitions
+// Only for testing purposes
+func SetDiscoverer(d func(*store.StoreOptions) ([]partition.ID, error)) {
+	discoverer = d
+}
diff --git a/pkg/pbq/store/memory/store.go b/pkg/pbq/store/memory/store.go
index 0a64745f..2e6093c4 100644
--- a/pkg/pbq/store/memory/store.go
+++ b/pkg/pbq/store/memory/store.go
@@ -2,9 +2,11 @@ package memory
 
 import (
 	"context"
+	"math"
+
 	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/pbq/store"
-	"github.com/numaproj/numaflow/pkg/pbq/util"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"go.uber.org/zap"
 )
@@ -14,20 +16,20 @@ type memoryStore struct {
 	closed      bool
 	writePos    int64
 	readPos     int64
-	storage     []*isb.Message
+	storage     []*isb.ReadMessage
 	options     *store.StoreOptions
 	log         *zap.SugaredLogger
-	partitionID string
+	partitionID partition.ID
 }
 
 // NewMemoryStore returns new memory store
-func NewMemoryStore(ctx context.Context, partitionID string, options *store.StoreOptions) (store.Store, error) {
+func NewMemoryStore(ctx context.Context, partitionID partition.ID, options *store.StoreOptions) (store.Store, error) {
 
 	memStore := &memoryStore{
 		writePos:    0,
 		readPos:     0,
 		closed:      false,
-		storage:     make([]*isb.Message, options.StoreSize()),
+		storage:     make([]*isb.ReadMessage, options.StoreSize()),
 		options:     options,
 		log:         logging.FromContext(ctx).With("PBQ store", "Memory store").With("Partition ID", partitionID),
 		partitionID: partitionID,
@@ -38,22 +40,22 @@ func NewMemoryStore(ctx context.Context, partitionID string, options *store.Stor
 
 // ReadFromStore will return upto N messages persisted in store
 // this function will be invoked during bootstrap if there is a restart
-func (m *memoryStore) Read(size int64) ([]*isb.Message, bool, error) {
+func (m *memoryStore) Read(size int64) ([]*isb.ReadMessage, bool, error) {
 	if m.isEmpty() || m.readPos >= m.writePos {
 		m.log.Errorw(store.ReadStoreEmptyErr.Error())
-		return []*isb.Message{}, true, nil
+		return []*isb.ReadMessage{}, true, nil
 	}
 
 	// if size is greater than the number of messages in the store
 	// we will assign size with the number of messages in the store
-	size = util.Min(size, m.writePos-m.readPos)
+	size = int64(math.Min(float64(size), float64(m.writePos-m.readPos)))
 	readMessages := m.storage[m.readPos : m.readPos+size]
 	m.readPos += size
 	return readMessages, false, nil
 }
 
 // WriteToStore writes a message to store
-func (m *memoryStore) Write(msg *isb.Message) error {
+func (m *memoryStore) Write(msg *isb.ReadMessage) error {
 	if m.writePos >= m.options.StoreSize() {
 		m.log.Errorw(store.WriteStoreFullErr.Error(), zap.Any("msg header", msg.Header))
 		return store.WriteStoreFullErr
diff --git a/pkg/pbq/store/noop/discover.go b/pkg/pbq/store/noop/discover.go
new file mode 100644
index 00000000..22eccf2f
--- /dev/null
+++ b/pkg/pbq/store/noop/discover.go
@@ -0,0 +1,12 @@
+package noop
+
+import (
+	"context"
+
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/pbq/store"
+)
+
+func DiscoverPartitions(_ context.Context, _ *store.StoreOptions) ([]partition.ID, error) {
+	return []partition.ID{}, nil
+}
diff --git a/pkg/pbq/store/noop/store.go b/pkg/pbq/store/noop/store.go
index e87bf260..aebafb2b 100644
--- a/pkg/pbq/store/noop/store.go
+++ b/pkg/pbq/store/noop/store.go
@@ -15,11 +15,11 @@ func NewPBQNoOpStore() (*PBQNoOpStore, error) {
 	return &PBQNoOpStore{}, nil
 }
 
-func (p *PBQNoOpStore) Read(size int64) ([]*isb.Message, bool, error) {
-	return []*isb.Message{}, true, nil
+func (p *PBQNoOpStore) Read(size int64) ([]*isb.ReadMessage, bool, error) {
+	return []*isb.ReadMessage{}, true, nil
 }
 
-func (p *PBQNoOpStore) Write(msg *isb.Message) error {
+func (p *PBQNoOpStore) Write(msg *isb.ReadMessage) error {
 	return nil
 }
 
diff --git a/pkg/pbq/store/options.go b/pkg/pbq/store/options.go
index da6b393f..a00e1a25 100644
--- a/pkg/pbq/store/options.go
+++ b/pkg/pbq/store/options.go
@@ -1,8 +1,9 @@
 package store
 
 import (
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"time"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 )
 
 type StoreOptions struct {
@@ -20,7 +21,7 @@ func (o *StoreOptions) StoreSize() int64 {
 	return o.storeSize
 }
 
-func (o *StoreOptions) PbqStoreType() dfv1.StoreType {
+func (o *StoreOptions) PBQStoreType() dfv1.StoreType {
 	return o.pbqStoreType
 }
 
diff --git a/pkg/pbq/util/min.go b/pkg/pbq/util/min.go
deleted file mode 100644
index 0d836fe5..00000000
--- a/pkg/pbq/util/min.go
+++ /dev/null
@@ -1,9 +0,0 @@
-package util
-
-// Min returns the min value
-func Min(x, y int64) int64 {
-	if x > y {
-		return y
-	}
-	return x
-}
diff --git a/pkg/reduce/options.go b/pkg/reduce/options.go
new file mode 100644
index 00000000..3f7cfe61
--- /dev/null
+++ b/pkg/reduce/options.go
@@ -0,0 +1,44 @@
+package reduce
+
+import (
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/window"
+)
+
+// Options for forwarding the message
+type Options struct {
+	// readBatchSize is the default batch size
+	readBatchSize int64
+	// windowOpts Options for window
+	windowOpts *window.Options
+}
+
+type Option func(*Options) error
+
+func DefaultOptions() *Options {
+	return &Options{
+		readBatchSize: dfv1.DefaultReadBatchSize,
+		windowOpts:    window.DefaultOptions(),
+	}
+}
+
+// WithReadBatchSize sets the read batch size
+func WithReadBatchSize(f int64) Option {
+	return func(o *Options) error {
+		o.readBatchSize = f
+		return nil
+	}
+}
+
+// WithWindowOptions sets different window options
+func WithWindowOptions(opts ...window.Option) Option {
+	return func(options *Options) error {
+		for _, opt := range opts {
+			err := opt(options.windowOpts)
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+}
diff --git a/pkg/reduce/pnf/processandforward.go b/pkg/reduce/pnf/processandforward.go
new file mode 100644
index 00000000..a63ad745
--- /dev/null
+++ b/pkg/reduce/pnf/processandforward.go
@@ -0,0 +1,203 @@
+// 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
+// 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
+
+import (
+	"context"
+	"errors"
+	"math"
+	"sync"
+	"time"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/isb/forward"
+	"github.com/numaproj/numaflow/pkg/shared/logging"
+	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
+	"github.com/numaproj/numaflow/pkg/watermark/processor"
+	"github.com/numaproj/numaflow/pkg/watermark/publish"
+	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/util/wait"
+
+	functionsdk "github.com/numaproj/numaflow-go/pkg/function"
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/pbq"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
+	udfreducer "github.com/numaproj/numaflow/pkg/udf/reducer"
+	"google.golang.org/grpc/metadata"
+)
+
+// 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 {
+	PartitionID      partition.ID
+	UDF              udfreducer.Reducer
+	result           []*isb.Message
+	pbqReader        pbq.Reader
+	log              *zap.SugaredLogger
+	toBuffers        map[string]isb.BufferWriter
+	whereToDecider   forward.ToWhichStepDecider
+	publishWatermark map[string]publish.Publisher
+}
+
+// NewProcessAndForward will return a new ProcessAndForward instance
+func NewProcessAndForward(ctx context.Context,
+	partitionID partition.ID,
+	udf udfreducer.Reducer,
+	pbqReader pbq.Reader,
+	toBuffers map[string]isb.BufferWriter,
+	whereToDecider forward.ToWhichStepDecider, pw map[string]publish.Publisher) *ProcessAndForward {
+	return &ProcessAndForward{
+		PartitionID:      partitionID,
+		UDF:              udf,
+		pbqReader:        pbqReader,
+		log:              logging.FromContext(ctx),
+		toBuffers:        toBuffers,
+		whereToDecider:   whereToDecider,
+		publishWatermark: pw,
+	}
+}
+
+// Process method reads messages from the supplied PBQ, invokes UDF to reduce the result.
+func (p *ProcessAndForward) Process(ctx context.Context) error {
+	var wg sync.WaitGroup
+	var err error
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		// FIXME: we need to fix https://github.com/numaproj/numaflow-go/blob/main/pkg/function/service.go#L101
+		ctx = metadata.NewOutgoingContext(ctx, metadata.New(map[string]string{functionsdk.DatumKey: p.PartitionID.Key}))
+		p.result, err = p.UDF.Reduce(ctx, p.pbqReader.ReadCh())
+	}()
+
+	// wait for the reduce method to return
+	wg.Wait()
+	return err
+}
+
+// Forward writes messages to the ISBs, publishes watermark, and invokes GC on PBQ.
+func (p *ProcessAndForward) Forward(ctx context.Context) error {
+	// extract window end time from the partitionID, which will be used for watermark
+	processorWM := processor.Watermark(p.PartitionID.End)
+
+	// decide which ISB to write to
+	to, err := p.whereToDecider.WhereTo(p.PartitionID.Key)
+	if err != nil {
+		return err
+	}
+	messagesToStep := p.whereToStep(to)
+
+	//store write offsets to publish watermark
+	writeOffsets := make(map[string][]isb.Offset)
+
+	// parallel writes to each isb
+	var wg sync.WaitGroup
+	var mu sync.Mutex
+	success := true
+	for key, messages := range messagesToStep {
+		bufferID := key
+		if len(messages) == 0 {
+			continue
+		}
+		wg.Add(1)
+		resultMessages := messages
+		go func() {
+			defer wg.Done()
+			offsets, ctxClosedErr := p.writeToBuffer(ctx, bufferID, resultMessages)
+			if ctxClosedErr != nil {
+				success = false
+				p.log.Errorw("Context closed while waiting to write the message to ISB", zap.Error(ctxClosedErr), zap.Any("partitionID", p.PartitionID))
+				return
+			}
+			mu.Lock()
+			// TODO: do we need lock? isn't each buffer isolated since we do sequential per ISB?
+			writeOffsets[bufferID] = offsets
+			mu.Unlock()
+		}()
+	}
+
+	// wait until all the writer go routines return
+	wg.Wait()
+	// even if one write go routines fails, don't GC just return
+	if !success {
+		return errors.New("failed to forward the messages to isb")
+	}
+
+	p.publishWM(processorWM, writeOffsets)
+	// delete the persisted messages
+	err = p.pbqReader.GC()
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// whereToStep assigns a message to the ISBs based on the Message.Key.
+// TODO: we have to introduce support for shuffle, output of a reducer can be input to the next reducer.
+func (p *ProcessAndForward) whereToStep(to []string) map[string][]isb.Message {
+	// writer doesn't accept array of pointers
+	messagesToStep := make(map[string][]isb.Message)
+	writeMessages := make([]isb.Message, len(p.result))
+	for idx, msg := range p.result {
+		writeMessages[idx] = *msg
+	}
+
+	// if a message is mapped to an isb, all the messages will be mapped to same isb (key is same)
+	switch {
+	case sharedutil.StringSliceContains(to, dfv1.MessageKeyAll):
+		for bufferID := range p.toBuffers {
+			messagesToStep[bufferID] = writeMessages
+		}
+	case sharedutil.StringSliceContains(to, dfv1.MessageKeyDrop):
+	default:
+		for _, bufferID := range to {
+			messagesToStep[bufferID] = writeMessages
+		}
+
+	}
+	return messagesToStep
+}
+
+// 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) {
+	var ISBWriteBackoff = wait.Backoff{
+		Steps:    math.MaxInt64,
+		Duration: 100 * time.Millisecond,
+		Factor:   1,
+		Jitter:   0.1,
+	}
+
+	// write to isb with infinite exponential backoff (until shutdown is triggered)
+	var failedMessages []isb.Message
+	var offsets []isb.Offset
+	ctxClosedErr := wait.ExponentialBackoffWithContext(ctx, ISBWriteBackoff, func() (done bool, err error) {
+		var writeErrs []error
+		offsets, writeErrs = p.toBuffers[bufferID].Write(ctx, resultMessages)
+		for i, message := range resultMessages {
+			if writeErrs[i] != nil {
+				failedMessages = append(failedMessages, message)
+			}
+		}
+		// retry only the failed messages
+		if len(failedMessages) > 0 {
+			resultMessages = failedMessages
+			return false, nil
+		}
+		return true, nil
+	})
+	return offsets, ctxClosedErr
+}
+
+// publishWM publishes the watermark to each edge.
+func (p *ProcessAndForward) publishWM(wm processor.Watermark, writeOffsets map[string][]isb.Offset) {
+	for bufferName, offsets := range writeOffsets {
+		if publisher, ok := p.publishWatermark[bufferName]; ok {
+			if len(offsets) > 0 {
+				publisher.PublishWatermark(wm, offsets[len(offsets)-1])
+			}
+		}
+	}
+}
diff --git a/pkg/reduce/readloop/ordered.go b/pkg/reduce/readloop/ordered.go
new file mode 100644
index 00000000..bc9b228a
--- /dev/null
+++ b/pkg/reduce/readloop/ordered.go
@@ -0,0 +1,169 @@
+package readloop
+
+import (
+	"container/list"
+	"context"
+	"sync"
+	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/isb/forward"
+	"github.com/numaproj/numaflow/pkg/watermark/publish"
+
+	"github.com/numaproj/numaflow/pkg/pbq"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/reduce/pnf"
+	"github.com/numaproj/numaflow/pkg/shared/logging"
+	udfreducer "github.com/numaproj/numaflow/pkg/udf/reducer"
+)
+
+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 {
+	sync.RWMutex
+	taskDone  chan struct{}
+	taskQueue *list.List
+	log       *zap.SugaredLogger
+}
+
+// newOrderedForwarder returns an orderedForwarder.
+func newOrderedForwarder(ctx context.Context) *orderedForwarder {
+	return &orderedForwarder{
+		taskDone:  make(chan struct{}),
+		taskQueue: list.New(),
+		log:       logging.FromContext(ctx),
+	}
+}
+
+// startUp starts forwarder.
+func (of *orderedForwarder) startUp(ctx context.Context) {
+	go of.forward(ctx)
+}
+
+// schedulePnF creates and schedules the PnF routine.
+func (of *orderedForwarder) schedulePnF(ctx context.Context,
+	udf udfreducer.Reducer,
+	pbq pbq.Reader,
+	partitionID partition.ID,
+	toBuffers map[string]isb.BufferWriter,
+	whereToDecider forward.ToWhichStepDecider,
+	pw map[string]publish.Publisher) {
+
+	pf := pnf.NewProcessAndForward(ctx, partitionID, udf, pbq, toBuffers, whereToDecider, pw)
+	doneCh := make(chan struct{})
+	t := &task{
+		doneCh: doneCh,
+		pf:     pf,
+	}
+
+	of.Lock()
+	defer of.Unlock()
+	of.taskQueue.PushBack(t)
+
+	// invoke the reduce function
+	go of.reduceOp(ctx, 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) {
+	// TODO: better logging with partitionID?
+	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() {
+			of.log.Infow("ReduceOp exiting", zap.Error(ctx.Err()))
+			return
+		}
+		of.log.Errorw("Process failed", zap.Error(err))
+		time.Sleep(retryDelay)
+	}
+	// after retrying indicate that we are done with processing the package. the processing can move on
+	close(t.doneCh)
+	of.log.Debugw("Process->Reduce call took ", zap.Int64("duration(ms)", time.Since(start).Milliseconds()))
+	// notify that some work has been completed
+	select {
+	case of.taskDone <- struct{}{}:
+	case <-ctx.Done():
+		return
+	}
+
+	// TODO: remove this?
+	of.log.Debugw("Post to task done chan took ", zap.Int64("duration(ms)", time.Since(start).Milliseconds()))
+}
+
+// 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.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)
+				of.Unlock()
+			case <-ctx.Done():
+				of.log.Infow("Forward exiting while waiting on the head of the queue task", 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
new file mode 100644
index 00000000..a857ed32
--- /dev/null
+++ b/pkg/reduce/readloop/readloop.go
@@ -0,0 +1,247 @@
+// 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"
+	"time"
+
+	"github.com/numaproj/numaflow/pkg/isb/forward"
+	"github.com/numaproj/numaflow/pkg/shared/logging"
+	udfReducer "github.com/numaproj/numaflow/pkg/udf/reducer"
+	"github.com/numaproj/numaflow/pkg/watermark/publish"
+	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/util/wait"
+
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/pbq"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/watermark/processor"
+	"github.com/numaproj/numaflow/pkg/window"
+	"github.com/numaproj/numaflow/pkg/window/keyed"
+	"github.com/numaproj/numaflow/pkg/window/strategy/fixed"
+)
+
+// ReadLoop is responsible for reading and forwarding the message from ISB to PBQ.
+type ReadLoop struct {
+	UDF               udfReducer.Reducer
+	pbqManager        *pbq.Manager
+	windowingStrategy window.Windower
+	aw                *fixed.ActiveWindows
+	op                *orderedForwarder
+	log               *zap.SugaredLogger
+	toBuffers         map[string]isb.BufferWriter
+	whereToDecider    forward.ToWhichStepDecider
+	publishWatermark  map[string]publish.Publisher
+}
+
+// NewReadLoop initializes  and returns ReadLoop.
+func NewReadLoop(ctx context.Context,
+	udf udfReducer.Reducer,
+	pbqManager *pbq.Manager,
+	windowingStrategy window.Windower,
+	toBuffers map[string]isb.BufferWriter,
+	whereToDecider forward.ToWhichStepDecider,
+	pw map[string]publish.Publisher,
+	_ *window.Options) *ReadLoop {
+
+	op := newOrderedForwarder(ctx)
+
+	rl := &ReadLoop{
+		UDF:               udf,
+		windowingStrategy: windowingStrategy,
+		// TODO: pass window type
+		aw:               fixed.NewWindows(),
+		pbqManager:       pbqManager,
+		op:               op,
+		log:              logging.FromContext(ctx),
+		toBuffers:        toBuffers,
+		whereToDecider:   whereToDecider,
+		publishWatermark: pw,
+	}
+	op.startUp(ctx)
+	return rl
+}
+
+// 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) {
+	// start the PBQManager which discovers and builds the state from persistent store of the PBQ.
+	rl.pbqManager.StartUp(ctx)
+	// gets the partitions from the state
+	partitions := rl.pbqManager.ListPartitions()
+
+	for _, p := range partitions {
+		// Create keyed window for a given partition
+		// so that the window can be closed when the watermark
+		// crosses the window.
+		id := p.PartitionID
+		intervalWindow := &window.IntervalWindow{
+			Start: id.Start,
+			End:   id.End,
+		}
+		// These windows have to be recreated as they are completely in-memory
+		rl.aw.CreateKeyedWindow(intervalWindow)
+
+		// create and invoke process and forward for the partition
+		rl.associatePBQAndPnF(ctx, p.PartitionID)
+	}
+
+	// replays the data (replay internally writes the data from persistent store on to the PBQ)
+	rl.pbqManager.Replay(ctx)
+}
+
+// Process is one iteration of the read loop.
+func (rl *ReadLoop) Process(ctx context.Context, messages []*isb.ReadMessage) {
+
+	// There is no Cap on backoff because setting a Cap will result in
+	// backoff stopped once the duration exceeds the Cap
+	var pbqWriteBackoff = wait.Backoff{
+		Steps:    math.MaxInt64,
+		Duration: 1 * time.Second,
+		Factor:   1.5,
+		Jitter:   0.1,
+	}
+
+	for _, m := range messages {
+		// identify and add window for the message
+		var ctxClosedErr error
+		windows := rl.upsertWindowsAndKeys(m)
+		// for each window we will have a PBQ. A message could belong to multiple windows (e.g., sliding).
+		for _, kw := range windows {
+			// identify partition for message
+			partitionID := partition.ID{
+				Start: kw.IntervalWindow.Start,
+				End:   kw.IntervalWindow.End,
+				Key:   m.Key,
+			}
+
+			q := rl.associatePBQAndPnF(ctx, partitionID)
+
+			// write the message to PBQ
+			writeFn := func(ctx context.Context, m *isb.ReadMessage) error {
+				return q.Write(ctx, m)
+			}
+			ctxClosedErr = rl.executeWithBackOff(ctx, writeFn, "failed to Write Message", pbqWriteBackoff, m, partitionID)
+			if ctxClosedErr != nil {
+				rl.log.Errorw("Error while writing the message to PBQ", zap.Error(ctxClosedErr))
+				return
+			}
+
+			// Ack the message to ISB
+			ackFn := func(_ context.Context, m *isb.ReadMessage) error {
+				return m.ReadOffset.AckIt()
+			}
+			ctxClosedErr = rl.executeWithBackOff(ctx, ackFn, "failed to Ack Message", pbqWriteBackoff, m, partitionID)
+			if ctxClosedErr != nil {
+				rl.log.Errorw("Error while acknowledging the message", zap.Error(ctxClosedErr))
+				return
+			}
+		}
+
+		// close any windows that need to be closed.
+		// FIXME(p0): why are we re-reading the watermark? isb.ReadMessage contains watermark.
+		wm := rl.waterMark(m)
+		closedWindows := rl.aw.RemoveWindow(time.Time(wm))
+		rl.log.Debugw("closing windows", zap.Int("length", len(closedWindows)), zap.Time("watermark", time.Time(wm)))
+
+		for _, cw := range closedWindows {
+			partitions := cw.Partitions()
+			rl.closePartitions(partitions)
+		}
+	}
+}
+
+// executeWithBackOff executes a function infinitely until it succeeds using ExponentialBackoffWithContext.
+func (rl *ReadLoop) executeWithBackOff(ctx context.Context, retryableFn func(ctx context.Context, message *isb.ReadMessage) error, errMsg string, pbqWriteBackoff wait.Backoff, m *isb.ReadMessage, partitionID partition.ID) error {
+	attempt := 0
+	ctxClosedErr := wait.ExponentialBackoffWithContext(ctx, pbqWriteBackoff, func() (done bool, err error) {
+		rErr := retryableFn(ctx, m)
+		attempt += 1
+		if rErr != nil {
+			rl.log.Errorw(errMsg, zap.Any("msgOffSet", m.ReadOffset.String()), zap.Any("partitionID", partitionID.String()), zap.Any("attempt", attempt), zap.Error(rErr))
+			return false, nil
+		}
+		return true, nil
+	})
+
+	return ctxClosedErr
+}
+
+// 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) 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.MaxInt64,
+			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)
+			if pbqErr != nil {
+				attempt += 1
+				rl.log.Warnw("Failed to create pbq during startup, retrying", zap.Any("attempt", attempt), zap.Any("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
+		rl.op.schedulePnF(ctx, rl.UDF, q, partitionID, rl.toBuffers, rl.whereToDecider, rl.publishWatermark)
+	}
+	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) []*keyed.KeyedWindow {
+	processingWindows := rl.windowingStrategy.AssignWindow(m.EventTime)
+	var kWindows []*keyed.KeyedWindow
+	for _, win := range processingWindows {
+		kw := rl.aw.GetKeyedWindow(win)
+		if kw == nil {
+			kw = rl.aw.CreateKeyedWindow(win)
+		}
+		// track the key to window relationship
+		kw.AddKey(m.Key)
+		kWindows = append(kWindows, kw)
+	}
+	return kWindows
+}
+
+// FIXME: this shouldn't be required.
+func (rl *ReadLoop) waterMark(message *isb.ReadMessage) processor.Watermark {
+	return processor.Watermark(message.Watermark)
+}
+
+// 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)
+		q.CloseOfBook()
+	}
+}
diff --git a/pkg/reduce/reduce.go b/pkg/reduce/reduce.go
new file mode 100644
index 00000000..63dfb912
--- /dev/null
+++ b/pkg/reduce/reduce.go
@@ -0,0 +1,94 @@
+// Package reduce reads messages from isb
+// attaches watermark to read messages
+// invoke the read-loop with the read messages
+package reduce
+
+import (
+	"context"
+	"time"
+
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/isb/forward"
+	"github.com/numaproj/numaflow/pkg/pbq"
+	"github.com/numaproj/numaflow/pkg/reduce/readloop"
+	"github.com/numaproj/numaflow/pkg/shared/logging"
+	udfReducer "github.com/numaproj/numaflow/pkg/udf/reducer"
+	"github.com/numaproj/numaflow/pkg/watermark/fetch"
+	"github.com/numaproj/numaflow/pkg/watermark/publish"
+	"github.com/numaproj/numaflow/pkg/window"
+	"go.uber.org/zap"
+)
+
+// DataForward reads data from isb and forwards them to readloop
+type DataForward struct {
+	fromBuffer        isb.BufferReader
+	readloop          *readloop.ReadLoop
+	fetchWatermark    fetch.Fetcher
+	windowingStrategy window.Windower
+	opts              *Options
+	log               *zap.SugaredLogger
+}
+
+func NewDataForward(ctx context.Context,
+	udf udfReducer.Reducer,
+	fromBuffer isb.BufferReader,
+	toBuffers map[string]isb.BufferWriter,
+	pbqManager *pbq.Manager,
+	whereToDecider forward.ToWhichStepDecider,
+	fw fetch.Fetcher,
+	publishWatermark map[string]publish.Publisher,
+	windowingStrategy window.Windower, opts ...Option) (*DataForward, error) {
+
+	options := DefaultOptions()
+
+	for _, opt := range opts {
+		if err := opt(options); err != nil {
+			return nil, err
+		}
+	}
+
+	rl := readloop.NewReadLoop(ctx, udf, pbqManager, windowingStrategy, toBuffers, whereToDecider, publishWatermark, options.windowOpts)
+	return &DataForward{
+		fromBuffer:        fromBuffer,
+		readloop:          rl,
+		fetchWatermark:    fw,
+		windowingStrategy: windowingStrategy,
+		log:               logging.FromContext(ctx),
+		opts:              options}, nil
+}
+
+// Start starts forwarding messages to readloop
+func (d *DataForward) Start(ctx context.Context) {
+	d.readloop.Startup(ctx)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			d.forwardAChunk(ctx)
+		}
+	}
+}
+
+// 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)
+
+	if err != nil {
+		d.log.Errorw("failed to read from isb", zap.Error(err))
+	}
+
+	if len(readMessages) == 0 {
+		return
+	}
+
+	// fetch watermark if available
+	// let's track only the last element's watermark
+	processorWM := d.fetchWatermark.GetWatermark(readMessages[len(readMessages)-1].ReadOffset)
+	for _, m := range readMessages {
+		m.Watermark = time.Time(processorWM)
+	}
+
+	d.readloop.Process(ctx, readMessages)
+}
diff --git a/pkg/sinks/kafka/kafka.go b/pkg/sinks/kafka/kafka.go
index ef4a5969..74860dcc 100644
--- a/pkg/sinks/kafka/kafka.go
+++ b/pkg/sinks/kafka/kafka.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/Shopify/sarama"
+	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	"go.uber.org/zap"
@@ -15,7 +16,6 @@ import (
 	metricspkg "github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"github.com/numaproj/numaflow/pkg/shared/util"
-	"github.com/numaproj/numaflow/pkg/udf/applier"
 )
 
 // ToKafka produce the output to a kafka sinks.
diff --git a/pkg/sinks/logger/log.go b/pkg/sinks/logger/log.go
index 16a1ef6c..8253e190 100644
--- a/pkg/sinks/logger/log.go
+++ b/pkg/sinks/logger/log.go
@@ -2,6 +2,7 @@ package logger
 
 import (
 	"context"
+	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	"log"
@@ -14,7 +15,6 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb/forward"
 	metricspkg "github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
-	"github.com/numaproj/numaflow/pkg/udf/applier"
 )
 
 // ToLog prints the output to a log sinks.
diff --git a/pkg/sinks/udsink/sink.go b/pkg/sinks/udsink/sink.go
index 2bf9478c..c73f108c 100644
--- a/pkg/sinks/udsink/sink.go
+++ b/pkg/sinks/udsink/sink.go
@@ -3,6 +3,7 @@ package udsink
 import (
 	"context"
 	"fmt"
+	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"time"
 
 	sinkpb "github.com/numaproj/numaflow-go/pkg/apis/proto/sink/v1"
@@ -10,7 +11,6 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/forward"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
-	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	"go.uber.org/zap"
diff --git a/pkg/sources/generator/tickgen.go b/pkg/sources/generator/tickgen.go
index 732e1675..e6c6282d 100644
--- a/pkg/sources/generator/tickgen.go
+++ b/pkg/sources/generator/tickgen.go
@@ -6,6 +6,7 @@ import (
 	"context"
 	"encoding/binary"
 	"encoding/json"
+	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"strconv"
 	"sync/atomic"
 	"time"
@@ -17,7 +18,6 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb/forward"
 	metricspkg "github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
-	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
diff --git a/pkg/sources/http/http.go b/pkg/sources/http/http.go
index c4868548..7e8ef949 100644
--- a/pkg/sources/http/http.go
+++ b/pkg/sources/http/http.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"io"
 	"net/http"
 	"strconv"
@@ -19,7 +20,6 @@ import (
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedtls "github.com/numaproj/numaflow/pkg/shared/tls"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
-	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
diff --git a/pkg/sources/kafka/reader.go b/pkg/sources/kafka/reader.go
index 923d8238..b282ab76 100644
--- a/pkg/sources/kafka/reader.go
+++ b/pkg/sources/kafka/reader.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"strconv"
 	"strings"
 	"sync"
@@ -18,7 +19,6 @@ import (
 	metricspkg "github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
-	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
diff --git a/pkg/udf/applier/interface.go b/pkg/udf/applier/applier.go
similarity index 92%
rename from pkg/udf/applier/interface.go
rename to pkg/udf/applier/applier.go
index 84802fd3..a56609a2 100644
--- a/pkg/udf/applier/interface.go
+++ b/pkg/udf/applier/applier.go
@@ -12,7 +12,7 @@ type Applier interface {
 	Apply(ctx context.Context, message *isb.ReadMessage) ([]*isb.Message, error)
 }
 
-// ApplyFunc untility function used to create a Applier implementation
+// ApplyFunc utility function used to create an Applier implementation
 type ApplyFunc func(context.Context, *isb.ReadMessage) ([]*isb.Message, error)
 
 func (a ApplyFunc) Apply(ctx context.Context, message *isb.ReadMessage) ([]*isb.Message, error) {
diff --git a/pkg/udf/applier/errors.go b/pkg/udf/function/errors.go
similarity index 97%
rename from pkg/udf/applier/errors.go
rename to pkg/udf/function/errors.go
index 944382e6..6b84da24 100644
--- a/pkg/udf/applier/errors.go
+++ b/pkg/udf/function/errors.go
@@ -1,4 +1,4 @@
-package applier
+package function
 
 import "fmt"
 
diff --git a/pkg/udf/function/reducemock.go b/pkg/udf/function/reducemock.go
new file mode 100644
index 00000000..0a63028c
--- /dev/null
+++ b/pkg/udf/function/reducemock.go
@@ -0,0 +1,151 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1 (interfaces: UserDefinedFunction_ReduceFnClient)
+
+// Package funcmock is a generated GoMock package.
+package function
+
+import (
+	context "context"
+	reflect "reflect"
+
+	gomock "github.com/golang/mock/gomock"
+	v1 "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
+	metadata "google.golang.org/grpc/metadata"
+)
+
+// MockUserDefinedFunction_ReduceFnClient is a mock of UserDefinedFunction_ReduceFnClient interface.
+type MockUserDefinedFunction_ReduceFnClient struct {
+	ctrl     *gomock.Controller
+	recorder *MockUserDefinedFunction_ReduceFnClientMockRecorder
+}
+
+// MockUserDefinedFunction_ReduceFnClientMockRecorder is the mock recorder for MockUserDefinedFunction_ReduceFnClient.
+type MockUserDefinedFunction_ReduceFnClientMockRecorder struct {
+	mock *MockUserDefinedFunction_ReduceFnClient
+}
+
+// NewMockUserDefinedFunction_ReduceFnClient creates a new mock instance.
+func NewMockUserDefinedFunction_ReduceFnClient(ctrl *gomock.Controller) *MockUserDefinedFunction_ReduceFnClient {
+	mock := &MockUserDefinedFunction_ReduceFnClient{ctrl: ctrl}
+	mock.recorder = &MockUserDefinedFunction_ReduceFnClientMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockUserDefinedFunction_ReduceFnClient) EXPECT() *MockUserDefinedFunction_ReduceFnClientMockRecorder {
+	return m.recorder
+}
+
+// CloseAndRecv mocks base method.
+func (m *MockUserDefinedFunction_ReduceFnClient) CloseAndRecv() (*v1.DatumList, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CloseAndRecv")
+	ret0, _ := ret[0].(*v1.DatumList)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// CloseAndRecv indicates an expected call of CloseAndRecv.
+func (mr *MockUserDefinedFunction_ReduceFnClientMockRecorder) CloseAndRecv() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CloseAndRecv", reflect.TypeOf((*MockUserDefinedFunction_ReduceFnClient)(nil).CloseAndRecv))
+}
+
+// CloseSend mocks base method.
+func (m *MockUserDefinedFunction_ReduceFnClient) CloseSend() error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CloseSend")
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// CloseSend indicates an expected call of CloseSend.
+func (mr *MockUserDefinedFunction_ReduceFnClientMockRecorder) CloseSend() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CloseSend", reflect.TypeOf((*MockUserDefinedFunction_ReduceFnClient)(nil).CloseSend))
+}
+
+// Context mocks base method.
+func (m *MockUserDefinedFunction_ReduceFnClient) Context() context.Context {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Context")
+	ret0, _ := ret[0].(context.Context)
+	return ret0
+}
+
+// Context indicates an expected call of Context.
+func (mr *MockUserDefinedFunction_ReduceFnClientMockRecorder) Context() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Context", reflect.TypeOf((*MockUserDefinedFunction_ReduceFnClient)(nil).Context))
+}
+
+// Header mocks base method.
+func (m *MockUserDefinedFunction_ReduceFnClient) Header() (metadata.MD, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Header")
+	ret0, _ := ret[0].(metadata.MD)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Header indicates an expected call of Header.
+func (mr *MockUserDefinedFunction_ReduceFnClientMockRecorder) Header() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Header", reflect.TypeOf((*MockUserDefinedFunction_ReduceFnClient)(nil).Header))
+}
+
+// RecvMsg mocks base method.
+func (m *MockUserDefinedFunction_ReduceFnClient) RecvMsg(arg0 interface{}) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "RecvMsg", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// RecvMsg indicates an expected call of RecvMsg.
+func (mr *MockUserDefinedFunction_ReduceFnClientMockRecorder) RecvMsg(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecvMsg", reflect.TypeOf((*MockUserDefinedFunction_ReduceFnClient)(nil).RecvMsg), arg0)
+}
+
+// Send mocks base method.
+func (m *MockUserDefinedFunction_ReduceFnClient) Send(arg0 *v1.Datum) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Send", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Send indicates an expected call of Send.
+func (mr *MockUserDefinedFunction_ReduceFnClientMockRecorder) Send(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Send", reflect.TypeOf((*MockUserDefinedFunction_ReduceFnClient)(nil).Send), arg0)
+}
+
+// SendMsg mocks base method.
+func (m *MockUserDefinedFunction_ReduceFnClient) SendMsg(arg0 interface{}) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "SendMsg", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// SendMsg indicates an expected call of SendMsg.
+func (mr *MockUserDefinedFunction_ReduceFnClientMockRecorder) SendMsg(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendMsg", reflect.TypeOf((*MockUserDefinedFunction_ReduceFnClient)(nil).SendMsg), arg0)
+}
+
+// Trailer mocks base method.
+func (m *MockUserDefinedFunction_ReduceFnClient) Trailer() metadata.MD {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Trailer")
+	ret0, _ := ret[0].(metadata.MD)
+	return ret0
+}
+
+// Trailer indicates an expected call of Trailer.
+func (mr *MockUserDefinedFunction_ReduceFnClientMockRecorder) Trailer() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Trailer", reflect.TypeOf((*MockUserDefinedFunction_ReduceFnClient)(nil).Trailer))
+}
diff --git a/pkg/udf/applier/uds_grpc.go b/pkg/udf/function/uds_grpc.go
similarity index 53%
rename from pkg/udf/applier/uds_grpc.go
rename to pkg/udf/function/uds_grpc.go
index 63c30bd2..ca45172a 100644
--- a/pkg/udf/applier/uds_grpc.go
+++ b/pkg/udf/function/uds_grpc.go
@@ -1,8 +1,10 @@
-package applier
+package function
 
 import (
 	"context"
 	"fmt"
+	"github.com/numaproj/numaflow/pkg/udf/applier"
+	"sync"
 	"time"
 
 	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
@@ -19,17 +21,22 @@ type udsGRPCBasedUDF struct {
 	client functionsdk.Client
 }
 
-var _ Applier = (*udsGRPCBasedUDF)(nil)
+var _ applier.Applier = (*udsGRPCBasedUDF)(nil)
 
 // NewUDSGRPCBasedUDF returns a new udsGRPCBasedUDF object.
 func NewUDSGRPCBasedUDF() (*udsGRPCBasedUDF, error) {
-	c, err := client.New()
+	c, err := client.New() // Can we pass this as a parameter to the function?
 	if err != nil {
 		return nil, fmt.Errorf("failed to create a new gRPC client: %w", err)
 	}
 	return &udsGRPCBasedUDF{c}, nil
 }
 
+// NewUdsGRPCBasedUDFWithClient need this for testing
+func NewUdsGRPCBasedUDFWithClient(client functionsdk.Client) *udsGRPCBasedUDF {
+	return &udsGRPCBasedUDF{client: client}
+}
+
 // CloseConn closes the gRPC client connection.
 func (u *udsGRPCBasedUDF) CloseConn(ctx context.Context) error {
 	return u.client.CloseConn(ctx)
@@ -92,3 +99,86 @@ func (u *udsGRPCBasedUDF) Apply(ctx context.Context, readMessage *isb.ReadMessag
 	}
 	return writeMessages, nil
 }
+
+// should we pass metadata information ?
+
+// Reduce accepts a channel of isbMessages and returns the aggregated result
+func (u *udsGRPCBasedUDF) Reduce(ctx context.Context, messageStream <-chan *isb.ReadMessage) ([]*isb.Message, error) {
+	datumCh := make(chan *functionpb.Datum)
+	var wg sync.WaitGroup
+	var result []*functionpb.Datum
+	var err error
+
+	// invoke the reduceFn method with datumCh channel
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		// TODO handle this error here itself
+		result, err = u.client.ReduceFn(ctx, datumCh)
+	}()
+
+readLoop:
+	for {
+		select {
+		case msg, ok := <-messageStream:
+			if msg != nil {
+				d := createDatum(msg)
+				select {
+				case datumCh <- d:
+				case <-ctx.Done():
+					return nil, ctx.Err()
+				}
+			}
+			if !ok {
+				break readLoop
+			}
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}
+
+	// close the datumCh, let the reduceFn know that there are no more messages
+	close(datumCh)
+
+	wg.Wait()
+
+	if err != nil {
+		return nil, ApplyUDFErr{
+			UserUDFErr: false,
+			Message:    fmt.Sprintf("gRPC client.ReduceFn failed, %s", err),
+			InternalErr: InternalErr{
+				Flag:        true,
+				MainCarDown: false,
+			},
+		}
+	}
+
+	writeMessages := make([]*isb.Message, 0)
+	for _, datum := range result {
+		key := datum.Key
+		writeMessage := &isb.Message{
+			Header: isb.Header{
+				Key: key,
+			},
+			Body: isb.Body{
+				Payload: datum.Value,
+			},
+		}
+		writeMessages = append(writeMessages, writeMessage)
+	}
+	return writeMessages, nil
+}
+
+func createDatum(readMessage *isb.ReadMessage) *functionpb.Datum {
+	key := readMessage.Key
+	payload := readMessage.Body.Payload
+	parentPaneInfo := readMessage.PaneInfo
+
+	var d = &functionpb.Datum{
+		Key:       key,
+		Value:     payload,
+		EventTime: &functionpb.EventTime{EventTime: timestamppb.New(parentPaneInfo.EventTime)},
+		Watermark: &functionpb.Watermark{Watermark: timestamppb.New(readMessage.Watermark)},
+	}
+	return d
+}
diff --git a/pkg/udf/reducer/reducer.go b/pkg/udf/reducer/reducer.go
new file mode 100644
index 00000000..fd3e3ec7
--- /dev/null
+++ b/pkg/udf/reducer/reducer.go
@@ -0,0 +1,19 @@
+package reducer
+
+import (
+	"context"
+	"github.com/numaproj/numaflow/pkg/isb"
+)
+
+// Reducer applies the HTTPBasedUDF on the read message and gives back a new message. Any UserError will be retried here, while
+// InternalErr can be returned and could be retried by the callee.
+type Reducer interface {
+	Reduce(ctx context.Context, messageStream <-chan *isb.ReadMessage) ([]*isb.Message, error)
+}
+
+// ReduceFunc utility function used to create a Reducer implementation
+type ReduceFunc func(context.Context, <-chan *isb.ReadMessage) ([]*isb.Message, error)
+
+func (a ReduceFunc) Reduce(ctx context.Context, messageStream <-chan *isb.ReadMessage) ([]*isb.Message, error) {
+	return a(ctx, messageStream)
+}
diff --git a/pkg/udf/udf.go b/pkg/udf/udf.go
index 47d38d50..80ae5156 100644
--- a/pkg/udf/udf.go
+++ b/pkg/udf/udf.go
@@ -6,9 +6,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/numaproj/numaflow/pkg/watermark/generic"
-	"go.uber.org/zap"
-
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/forward"
@@ -19,8 +16,11 @@ import (
 	redisclient "github.com/numaproj/numaflow/pkg/shared/clients/redis"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
-	"github.com/numaproj/numaflow/pkg/udf/applier"
+
+	"github.com/numaproj/numaflow/pkg/udf/function"
+	"github.com/numaproj/numaflow/pkg/watermark/generic"
 	"github.com/numaproj/numaflow/pkg/watermark/generic/jetstream"
+	"go.uber.org/zap"
 )
 
 type UDFProcessor struct {
@@ -122,7 +122,7 @@ func (u *UDFProcessor) Start(ctx context.Context) error {
 	})
 
 	log = log.With("protocol", "uds-grpc-udf")
-	udfHandler, err := applier.NewUDSGRPCBasedUDF()
+	udfHandler, err := function.NewUDSGRPCBasedUDF()
 	if err != nil {
 		return fmt.Errorf("failed to create gRPC client, %w", err)
 	}
diff --git a/pkg/watermark/fetch/processor_to_fetch.go b/pkg/watermark/fetch/processor_to_fetch.go
index 2a2f62bf..804b2bd1 100644
--- a/pkg/watermark/fetch/processor_to_fetch.go
+++ b/pkg/watermark/fetch/processor_to_fetch.go
@@ -112,7 +112,7 @@ func (p *ProcessorToFetch) startTimeLineWatcher() {
 			case store.KVPut:
 				epoch, skip, err := p.entity.ParseOTWatcherKey(value.Key())
 				if err != nil {
-					p.log.Errorw("Unable to convert value.Key() to int64", zap.String("received", value.Key()), zap.Error(err))
+					p.log.Errorw("Unable to convert value.PartitionID() to int64", zap.String("received", value.Key()), zap.Error(err))
 					continue
 				}
 				// if skip is set to true, it means the key update we received is for a different processor (sharing of bucket)
diff --git a/pkg/window/keyed/keyed.go b/pkg/window/keyed/keyed.go
index b5d169c9..4c5114d6 100644
--- a/pkg/window/keyed/keyed.go
+++ b/pkg/window/keyed/keyed.go
@@ -1,20 +1,20 @@
+// Package keyed ....
+// TODO: document
 package keyed
 
 import (
-	"fmt"
 	"sync"
 
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/window"
 )
 
-// PartitionId uniquely identifies a partition
-type PartitionId string
-
 // KeyedWindow maintains association between keys and a window.
 // In a keyed stream, we need to close all the partitions when the watermark is past the window.
 type KeyedWindow struct {
 	*window.IntervalWindow
-	Keys []string
+	// TODO: can this be map[string]struct{} ?
+	Keys map[string]string
 	lock sync.RWMutex
 }
 
@@ -22,7 +22,7 @@ type KeyedWindow struct {
 func NewKeyedWindow(window *window.IntervalWindow) *KeyedWindow {
 	kw := &KeyedWindow{
 		IntervalWindow: window,
-		Keys:           make([]string, 0),
+		Keys:           make(map[string]string),
 		lock:           sync.RWMutex{},
 	}
 	return kw
@@ -32,23 +32,22 @@ func NewKeyedWindow(window *window.IntervalWindow) *KeyedWindow {
 func (kw *KeyedWindow) AddKey(key string) {
 	kw.lock.Lock()
 	defer kw.lock.Unlock()
-	kw.Keys = append(kw.Keys, key)
+	if _, ok := kw.Keys[key]; !ok {
+		kw.Keys[key] = key
+	}
 }
 
 // Partitions returns an array of partitions for a window
-func (kw *KeyedWindow) Partitions() []PartitionId {
+func (kw *KeyedWindow) Partitions() []partition.ID {
 	kw.lock.RLock()
 	defer kw.lock.RUnlock()
 
-	partitions := make([]PartitionId, len(kw.Keys))
-	for i, key := range kw.Keys {
-		partitions[i] = Partition(kw.IntervalWindow, key)
+	partitions := make([]partition.ID, len(kw.Keys))
+	idx := 0
+	for k := range kw.Keys {
+		partitions[idx] = partition.ID{Start: kw.IntervalWindow.Start, End: kw.IntervalWindow.End, Key: k}
+		idx++
 	}
 
 	return partitions
 }
-
-// Partition returns the partitionId for a given window and a key
-func Partition(window *window.IntervalWindow, key string) PartitionId {
-	return PartitionId(fmt.Sprintf("%v-%v-%s", window.Start.Unix(), window.End.Unix(), key))
-}
diff --git a/pkg/window/options.go b/pkg/window/options.go
new file mode 100644
index 00000000..d63ba00e
--- /dev/null
+++ b/pkg/window/options.go
@@ -0,0 +1,38 @@
+package window
+
+import (
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"time"
+)
+
+type Options struct {
+	// windowType to specify the window type(fixed, sliding or session)
+	windowType dfv1.WindowType
+	// windowDuration to specify the duration of the window
+	windowDuration time.Duration
+}
+
+func DefaultOptions() *Options {
+	return &Options{
+		windowType:     dfv1.DefaultWindowType,
+		windowDuration: dfv1.DefaultWindowDuration,
+	}
+}
+
+type Option func(options *Options) error
+
+// WithWindowType sets the window type
+func WithWindowType(wt dfv1.WindowType) Option {
+	return func(o *Options) error {
+		o.windowType = wt
+		return nil
+	}
+}
+
+// WithWindowDuration sets the window duration
+func WithWindowDuration(wd time.Duration) Option {
+	return func(o *Options) error {
+		o.windowDuration = wd
+		return nil
+	}
+}


Test patch:
diff --git a/pkg/isb/forward/forward_test.go b/pkg/isb/forward/forward_test.go
index 28a8f3ba..e292a35b 100644
--- a/pkg/isb/forward/forward_test.go
+++ b/pkg/isb/forward/forward_test.go
@@ -17,7 +17,7 @@ import (
 
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
-	udfapplier "github.com/numaproj/numaflow/pkg/udf/applier"
+	udfapplier "github.com/numaproj/numaflow/pkg/udf/function"
 	"github.com/stretchr/testify/assert"
 )
 
diff --git a/pkg/pbq/pbq_test.go b/pkg/pbq/pbq_test.go
index d6e7f83b..6e5432bf 100644
--- a/pkg/pbq/pbq_test.go
+++ b/pkg/pbq/pbq_test.go
@@ -9,6 +9,7 @@ import (
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/pbq/store"
 	"github.com/stretchr/testify/assert"
 )
@@ -29,12 +30,18 @@ func TestPBQ_ReadWrite(t *testing.T) {
 	// write 10 isb messages to persisted store
 	msgCount := 10
 	startTime := time.Now()
-	writeMessages := testutils.BuildTestWriteMessages(int64(msgCount), startTime)
+	writeMessages := testutils.BuildTestReadMessages(int64(msgCount), startTime)
 
-	pq, err := qManager.CreateNewPBQ(ctx, "partition-13")
+	partitionID := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Key:   "new-partition",
+	}
+
+	pq, err := qManager.CreateNewPBQ(ctx, partitionID)
 	assert.NoError(t, err)
 
-	var readMessages []*isb.Message
+	var readMessages []*isb.ReadMessage
 	// run a parallel go routine which reads from pbq
 	var wg sync.WaitGroup
 	wg.Add(1)
@@ -87,13 +94,18 @@ func Test_PBQReadWithCanceledContext(t *testing.T) {
 	//write 10 isb messages to persisted store
 	msgCount := 10
 	startTime := time.Now()
-	writeMessages := testutils.BuildTestWriteMessages(int64(msgCount), startTime)
+	writeMessages := testutils.BuildTestReadMessages(int64(msgCount), startTime)
 
+	partitionID := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Key:   "new-partition",
+	}
 	var pq ReadWriteCloser
-	pq, err = qManager.CreateNewPBQ(ctx, "partition-14")
+	pq, err = qManager.CreateNewPBQ(ctx, partitionID)
 	assert.NoError(t, err)
 
-	var readMessages []*isb.Message
+	var readMessages []*isb.ReadMessage
 	// run a parallel go routine which reads from pbq
 	var wg sync.WaitGroup
 	wg.Add(1)
@@ -148,10 +160,15 @@ func TestPBQ_WriteWithStoreFull(t *testing.T) {
 	// write 101 isb messages to pbq, but the store size is 100, we should get store is full error
 	msgCount := 101
 	startTime := time.Now()
-	writeMessages := testutils.BuildTestWriteMessages(int64(msgCount), startTime)
+	writeMessages := testutils.BuildTestReadMessages(int64(msgCount), startTime)
+	partitionID := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Key:   "new-partition",
+	}
 
 	var pq ReadWriteCloser
-	pq, err = qManager.CreateNewPBQ(ctx, "partition-10")
+	pq, err = qManager.CreateNewPBQ(ctx, partitionID)
 	assert.NoError(t, err)
 
 	for _, msg := range writeMessages {
diff --git a/pkg/pbq/pbqmanager_test.go b/pkg/pbq/pbqmanager_test.go
index 04890250..d6367f9c 100644
--- a/pkg/pbq/pbqmanager_test.go
+++ b/pkg/pbq/pbqmanager_test.go
@@ -9,7 +9,9 @@ import (
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/pbq/store"
+	"github.com/numaproj/numaflow/pkg/pbq/store/memory"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -24,11 +26,21 @@ func TestManager_ListPartitions(t *testing.T) {
 	assert.NoError(t, err)
 
 	// create a new pbq using pbq manager
+	testPartition := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Key:   "partition-1",
+	}
+	partitionTwo := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Key:   "partition-2",
+	}
 	var pq1, pq2 ReadWriteCloser
-	pq1, err = pbqManager.CreateNewPBQ(ctx, "partition-1")
+	pq1, err = pbqManager.CreateNewPBQ(ctx, testPartition)
 	assert.NoError(t, err)
 
-	pq2, err = pbqManager.CreateNewPBQ(ctx, "partition-2")
+	pq2, err = pbqManager.CreateNewPBQ(ctx, partitionTwo)
 	assert.NoError(t, err)
 
 	assert.Len(t, pbqManager.ListPartitions(), 2)
@@ -52,11 +64,16 @@ func TestManager_GetPBQ(t *testing.T) {
 	assert.NoError(t, err)
 
 	// create a new pbq using CreateNewPBQ PBQ
-	pb1, err = pbqManager.CreateNewPBQ(ctx, "partition-3")
+	testPartition := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Key:   "partition-1",
+	}
+	pb1, err = pbqManager.CreateNewPBQ(ctx, testPartition)
 	assert.NoError(t, err)
 
 	// get the created pbq
-	pb2 = pbqManager.GetPBQ("partition-3")
+	pb2 = pbqManager.GetPBQ(testPartition)
 
 	assert.Equal(t, pb1, pb2)
 }
@@ -69,9 +86,14 @@ func TestPBQFlow(t *testing.T) {
 	pbqManager, err := NewManager(ctx, WithPBQStoreOptions(store.WithStoreSize(int64(size)), store.WithPbqStoreType(dfv1.InMemoryType)),
 		WithReadTimeout(1*time.Second), WithChannelBufferSize(10))
 	assert.NoError(t, err)
+	testPartition := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Key:   "partition-1",
+	}
 
 	var pq ReadWriteCloser
-	pq, err = pbqManager.CreateNewPBQ(ctx, "partition-4")
+	pq, err = pbqManager.CreateNewPBQ(ctx, testPartition)
 	assert.NoError(t, err)
 
 	msgsCount := 5
@@ -79,7 +101,7 @@ func TestPBQFlow(t *testing.T) {
 	wg.Add(1)
 
 	// read messages from pbq
-	var readMessages []*isb.Message
+	var readMessages []*isb.ReadMessage
 
 	go func() {
 	readLoop:
@@ -100,7 +122,7 @@ func TestPBQFlow(t *testing.T) {
 	}()
 
 	// write messages to pbq
-	writeMessages := testutils.BuildTestWriteMessages(int64(msgsCount), time.Now())
+	writeMessages := testutils.BuildTestReadMessages(int64(msgsCount), time.Now())
 	for _, msg := range writeMessages {
 		err := pq.Write(ctx, &msg)
 		assert.NoError(t, err)
@@ -125,14 +147,19 @@ func TestPBQFlowWithNoOpStore(t *testing.T) {
 	pbqManager, err := NewManager(ctx, WithPBQStoreOptions(store.WithStoreSize(int64(size)), store.WithPbqStoreType(dfv1.NoOpType)),
 		WithReadTimeout(1*time.Second), WithChannelBufferSize(10))
 	assert.NoError(t, err)
+	testPartition := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Key:   "partition-1",
+	}
 
 	// create a pbq backed with no op store
 	var pq ReadWriteCloser
-	pq, err = pbqManager.CreateNewPBQ(ctx, "partition-6")
+	pq, err = pbqManager.CreateNewPBQ(ctx, testPartition)
 	msgsCount := 50
 	var wg sync.WaitGroup
 
-	var readMessages []*isb.Message
+	var readMessages []*isb.ReadMessage
 
 	// read messages from pbq(with no op store)
 	wg.Add(1)
@@ -155,7 +182,7 @@ func TestPBQFlowWithNoOpStore(t *testing.T) {
 	}()
 
 	// write messages to pbq
-	writeMessages := testutils.BuildTestWriteMessages(int64(msgsCount), time.Now())
+	writeMessages := testutils.BuildTestReadMessages(int64(msgsCount), time.Now())
 	for _, msg := range writeMessages {
 		err := pq.Write(ctx, &msg)
 		assert.NoError(t, err)
@@ -178,14 +205,19 @@ func TestManager_Replay(t *testing.T) {
 	pbqManager, err := NewManager(ctx, WithPBQStoreOptions(store.WithStoreSize(int64(size)), store.WithPbqStoreType(dfv1.InMemoryType)),
 		WithReadTimeout(1*time.Second), WithChannelBufferSize(10), WithReadBatchSize(10))
 	assert.NoError(t, err)
+	testPartition := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Key:   "partition-1",
+	}
 
 	var pq ReadWriteCloser
-	pq, err = pbqManager.CreateNewPBQ(ctx, "partition-5")
+	pq, err = pbqManager.CreateNewPBQ(ctx, testPartition)
 	assert.NoError(t, err)
 	var wg sync.WaitGroup
 	wg.Add(1)
 
-	var readMessages []*isb.Message
+	var readMessages []*isb.ReadMessage
 	// go routine which reads the messages from pbq
 	go func() {
 	readLoop:
@@ -207,7 +239,7 @@ func TestManager_Replay(t *testing.T) {
 
 	// write 50 messages to pbq
 	msgsCount := 50
-	writeMessages := testutils.BuildTestWriteMessages(int64(msgsCount), time.Now())
+	writeMessages := testutils.BuildTestReadMessages(int64(msgsCount), time.Now())
 
 	for _, msg := range writeMessages {
 		err := pq.Write(ctx, &msg)
@@ -223,3 +255,32 @@ func TestManager_Replay(t *testing.T) {
 	// since we replayed the messages which are persisted in store
 	assert.Len(t, readMessages, 2*msgsCount)
 }
+
+func TestManager_StartUp(t *testing.T) {
+	size := 100
+
+	ctx := context.Background()
+	pbqManager, err := NewManager(ctx, WithPBQStoreOptions(store.WithStoreSize(int64(size)), store.WithPbqStoreType(dfv1.InMemoryType)),
+		WithReadTimeout(1*time.Second), WithChannelBufferSize(10), WithReadBatchSize(10))
+	assert.NoError(t, err)
+	pID1 := partition.ID{
+		Start: time.Now(),
+		End:   time.Now(),
+		Key:   "test-partition-1",
+	}
+
+	pID2 := partition.ID{
+		Start: time.Now(),
+		End:   time.Now(),
+		Key:   "test-partition-2",
+	}
+	dp := func(storeOptions *store.StoreOptions) ([]partition.ID, error) {
+		return []partition.ID{pID1,
+			pID2}, nil
+	}
+	memory.SetDiscoverer(dp)
+	pbqManager.StartUp(ctx)
+	assert.Len(t, pbqManager.ListPartitions(), 2)
+	assert.NotNil(t, pbqManager.GetPBQ(pID1))
+	assert.NotNil(t, pbqManager.GetPBQ(pID2))
+}
diff --git a/pkg/pbq/store/memory/store_test.go b/pkg/pbq/store/memory/store_test.go
index 520586d7..f630562b 100644
--- a/pkg/pbq/store/memory/store_test.go
+++ b/pkg/pbq/store/memory/store_test.go
@@ -2,13 +2,15 @@ package memory
 
 import (
 	"context"
+	"testing"
+	"time"
+
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/pbq/store"
 	"github.com/stretchr/testify/assert"
-	"testing"
-	"time"
 )
 
 func TestMemoryStore_WriteToStore(t *testing.T) {
@@ -19,13 +21,19 @@ func TestMemoryStore_WriteToStore(t *testing.T) {
 	_ = store.WithStoreSize(int64(storeSize))(options)
 	ctx := context.Background()
 
-	memStore, err := NewMemoryStore(ctx, "new-partition", options)
+	partitionID := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Key:   "new-partition",
+	}
+
+	memStore, err := NewMemoryStore(ctx, partitionID, options)
 	assert.NoError(t, err)
 
 	//write 10 isb messages to persisted store
 	msgCount := 10
 	startTime := time.Now()
-	writeMessages := testutils.BuildTestWriteMessages(int64(msgCount), startTime)
+	writeMessages := testutils.BuildTestReadMessages(int64(msgCount), startTime)
 
 	for _, msg := range writeMessages {
 		err := memStore.Write(&msg)
@@ -41,19 +49,25 @@ func TestMemoryStore_ReadFromStore(t *testing.T) {
 	_ = store.WithStoreSize(int64(storeSize))(options)
 	ctx := context.Background()
 
-	memStore, err := NewMemoryStore(ctx, "new-partition-2", options)
+	partitionID := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Key:   "new-partition",
+	}
+
+	memStore, err := NewMemoryStore(ctx, partitionID, options)
 	assert.NoError(t, err)
 
 	//write 10 isb messages to persisted store
 	msgCount := 10
 	startTime := time.Now()
-	writeMessages := testutils.BuildTestWriteMessages(int64(msgCount), startTime)
+	writeMessages := testutils.BuildTestReadMessages(int64(msgCount), startTime)
 
 	for _, msg := range writeMessages {
 		err := memStore.Write(&msg)
 		assert.NoError(t, err)
 	}
-	var readMessages []*isb.Message
+	var readMessages []*isb.ReadMessage
 	readMessages, _, err = memStore.Read(int64(msgCount))
 	assert.NoError(t, err)
 	// number of read messages should be equal to msgCount
@@ -67,7 +81,13 @@ func TestEmptyStore_Read(t *testing.T) {
 	_ = store.WithPbqStoreType(dfv1.InMemoryType)(options)
 	ctx := context.Background()
 
-	memStore, err := NewMemoryStore(ctx, "new-partition-3", options)
+	partitionID := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Key:   "new-partition",
+	}
+
+	memStore, err := NewMemoryStore(ctx, partitionID, options)
 	assert.NoError(t, err)
 	var eof bool
 	_, eof, err = memStore.Read(int64(storeSize))
@@ -85,13 +105,19 @@ func TestFullStore_Write(t *testing.T) {
 	_ = store.WithStoreSize(int64(storeSize))(options)
 	ctx := context.Background()
 
-	memStore, err := NewMemoryStore(ctx, "new-partition-4", options)
+	partitionID := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Key:   "new-partition",
+	}
+
+	memStore, err := NewMemoryStore(ctx, partitionID, options)
 	assert.NoError(t, err)
 
 	//write 100 isb messages to persisted store
 	msgCount := 100
 	startTime := time.Now()
-	writeMessages := testutils.BuildTestWriteMessages(int64(msgCount), startTime)
+	writeMessages := testutils.BuildTestReadMessages(int64(msgCount), startTime)
 
 	for _, msg := range writeMessages {
 		err := memStore.Write(&msg)
diff --git a/pkg/reduce/pnf/processandforward_test.go b/pkg/reduce/pnf/processandforward_test.go
new file mode 100644
index 00000000..1559abd2
--- /dev/null
+++ b/pkg/reduce/pnf/processandforward_test.go
@@ -0,0 +1,249 @@
+package pnf
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
+	"github.com/numaproj/numaflow/pkg/watermark/generic"
+
+	"github.com/golang/mock/gomock"
+	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
+	"github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1/funcmock"
+	"github.com/numaproj/numaflow-go/pkg/function/clienttest"
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/isb/testutils"
+	"github.com/numaproj/numaflow/pkg/pbq"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/pbq/store"
+	udfcall "github.com/numaproj/numaflow/pkg/udf/function"
+	"github.com/stretchr/testify/assert"
+)
+
+type myForwardTest struct {
+}
+
+func (f myForwardTest) WhereTo(key string) ([]string, error) {
+	if strings.Compare(key, "test-forward-one") == 0 {
+		return []string{"buffer1"}, nil
+	} else if strings.Compare(key, "test-forward-all") == 0 {
+		return []string{dfv1.MessageKeyAll}, nil
+	}
+	return []string{dfv1.MessageKeyDrop}, nil
+}
+
+func (f myForwardTest) Apply(ctx context.Context, message *isb.ReadMessage) ([]*isb.Message, error) {
+	return testutils.CopyUDFTestApply(ctx, message)
+}
+
+type PayloadForTest struct {
+	Key   string
+	Value int64
+}
+
+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
+	// 3. mock the grpc client methods
+	// 4. assert to check if the process method is returning the result
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	defer cancel()
+	go func() {
+		<-ctx.Done()
+		if ctx.Err() == context.DeadlineExceeded {
+			t.Log(t.Name(), "test timeout")
+		}
+	}()
+
+	size := 100
+	testPartition := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Key:   "partition-1",
+	}
+	var err error
+	var pbqManager *pbq.Manager
+
+	pbqManager, err = pbq.NewManager(ctx, pbq.WithPBQStoreOptions(store.WithStoreSize(int64(size)), store.WithPbqStoreType(dfv1.InMemoryType)),
+		pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
+	assert.NoError(t, err)
+
+	// create a pbq for a partition
+	var simplePbq pbq.ReadWriteCloser
+	simplePbq, err = pbqManager.CreateNewPBQ(ctx, testPartition)
+	assert.NoError(t, err)
+
+	// write messages to pbq
+	go func() {
+		writeMessages := testutils.BuildTestReadMessages(10, time.Now())
+		for index := range writeMessages {
+			err := simplePbq.Write(ctx, &writeMessages[index])
+			assert.NoError(t, err)
+		}
+		// done writing, cob
+		simplePbq.CloseOfBook()
+	}()
+
+	// mock grpc reducer
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockClient := funcmock.NewMockUserDefinedFunctionClient(ctrl)
+	mockReduceClient := udfcall.NewMockUserDefinedFunction_ReduceFnClient(ctrl)
+
+	mockReduceClient.EXPECT().Send(gomock.Any()).Return(nil).AnyTimes()
+	mockReduceClient.EXPECT().CloseAndRecv().Return(&functionpb.DatumList{
+		Elements: []*functionpb.Datum{
+			{
+				Key:   "reduced_result_key",
+				Value: []byte(`forward_message`),
+			},
+		},
+	}, nil)
+
+	mockClient.EXPECT().ReduceFn(gomock.Any(), gomock.Any()).Return(mockReduceClient, nil)
+
+	c, _ := clienttest.New(mockClient)
+	client := udfcall.NewUdsGRPCBasedUDFWithClient(c)
+
+	assert.NoError(t, err)
+	_, publishWatermark := generic.BuildNoOpWatermarkProgressorsFromBufferMap(make(map[string]isb.BufferWriter))
+	// create pf using key and reducer
+	pf := NewProcessAndForward(ctx, testPartition, client, simplePbq, make(map[string]isb.BufferWriter), myForwardTest{}, publishWatermark)
+
+	err = pf.Process(ctx)
+	assert.NoError(t, err)
+	assert.Len(t, pf.result, 1)
+}
+
+func TestProcessAndForward_Forward(t *testing.T) {
+	ctx := context.Background()
+
+	var pbqManager *pbq.Manager
+
+	pbqManager, _ = pbq.NewManager(ctx)
+
+	test1Buffer1 := simplebuffer.NewInMemoryBuffer("buffer1", 10)
+	test1Buffer2 := simplebuffer.NewInMemoryBuffer("buffer2", 10)
+
+	toBuffers1 := map[string]isb.BufferWriter{
+		"buffer1": test1Buffer1,
+		"buffer2": test1Buffer2,
+	}
+
+	test2Buffer1 := simplebuffer.NewInMemoryBuffer("buffer1", 10)
+	test2Buffer2 := simplebuffer.NewInMemoryBuffer("buffer2", 10)
+
+	toBuffers2 := map[string]isb.BufferWriter{
+		"buffer1": test2Buffer1,
+		"buffer2": test2Buffer2,
+	}
+
+	test3Buffer1 := simplebuffer.NewInMemoryBuffer("buffer1", 10)
+	test3Buffer2 := simplebuffer.NewInMemoryBuffer("buffer2", 10)
+
+	toBuffers3 := map[string]isb.BufferWriter{
+		"buffer1": test3Buffer1,
+		"buffer2": test3Buffer2,
+	}
+
+	tests := []struct {
+		name     string
+		id       partition.ID
+		buffers  []*simplebuffer.InMemoryBuffer
+		pf       ProcessAndForward
+		expected []bool
+	}{
+		{
+			name: "test-forward-one",
+			id: partition.ID{
+				Start: time.Unix(60, 0),
+				End:   time.Unix(120, 0),
+				Key:   "test-forward-one",
+			},
+			buffers:  []*simplebuffer.InMemoryBuffer{test1Buffer1, test1Buffer2},
+			pf:       createProcessAndForward(ctx, "test-forward-one", pbqManager, toBuffers1),
+			expected: []bool{false, true},
+		},
+		{
+			name: "test-forward-all",
+			id: partition.ID{
+				Start: time.Unix(60, 0),
+				End:   time.Unix(120, 0),
+				Key:   "test-forward-all",
+			},
+			buffers:  []*simplebuffer.InMemoryBuffer{test2Buffer1, test2Buffer2},
+			pf:       createProcessAndForward(ctx, "test-forward-all", pbqManager, toBuffers2),
+			expected: []bool{false, false},
+		},
+		{
+			name: "test-drop-all",
+			id: partition.ID{
+				Start: time.Unix(60, 0),
+				End:   time.Unix(120, 0),
+				Key:   "test-drop-all",
+			},
+			buffers:  []*simplebuffer.InMemoryBuffer{test3Buffer1, test3Buffer2},
+			pf:       createProcessAndForward(ctx, "test-drop-all", pbqManager, toBuffers3),
+			expected: []bool{true, true},
+		},
+	}
+
+	for _, value := range tests {
+		t.Run(value.name, func(t *testing.T) {
+			err := value.pf.Forward(ctx)
+			assert.NoError(t, err)
+			assert.Equal(t, []bool{value.buffers[0].IsEmpty(), value.buffers[1].IsEmpty()}, value.expected)
+			assert.Equal(t, pbqManager.GetPBQ(value.id), nil)
+		})
+	}
+}
+
+func createProcessAndForward(ctx context.Context, key string, pbqManager *pbq.Manager, toBuffers map[string]isb.BufferWriter) ProcessAndForward {
+
+	testPartition := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Key:   key,
+	}
+
+	// create a pbq for a partition
+	_, pw := generic.BuildNoOpWatermarkProgressorsFromBufferMap(toBuffers)
+	var simplePbq pbq.Reader
+	simplePbq, _ = pbqManager.CreateNewPBQ(ctx, testPartition)
+
+	resultPayload, _ := json.Marshal(PayloadForTest{
+		Key:   "result_payload",
+		Value: 100,
+	})
+
+	var result = []*isb.Message{
+		{
+			Header: isb.Header{
+				PaneInfo: isb.PaneInfo{
+					EventTime: time.Unix(60, 0),
+				},
+				ID: "1",
+			},
+			Body: isb.Body{Payload: resultPayload},
+		},
+	}
+
+	pf := ProcessAndForward{
+		PartitionID:      testPartition,
+		UDF:              nil,
+		result:           result,
+		pbqReader:        simplePbq,
+		log:              nil,
+		toBuffers:        toBuffers,
+		whereToDecider:   myForwardTest{},
+		publishWatermark: pw,
+	}
+
+	return pf
+}
diff --git a/pkg/reduce/readloop/ordered_test.go b/pkg/reduce/readloop/ordered_test.go
new file mode 100644
index 00000000..035061ed
--- /dev/null
+++ b/pkg/reduce/readloop/ordered_test.go
@@ -0,0 +1,165 @@
+package readloop
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
+	"github.com/numaproj/numaflow/pkg/isb/testutils"
+	"github.com/numaproj/numaflow/pkg/watermark/generic"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/pbq"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/pbq/store"
+	"github.com/numaproj/numaflow/pkg/udf/reducer"
+	"github.com/stretchr/testify/assert"
+)
+
+type myForwardTest struct {
+}
+
+func (f myForwardTest) WhereTo(_ string) ([]string, error) {
+	return []string{dfv1.MessageKeyDrop}, nil
+}
+
+func (f myForwardTest) Apply(ctx context.Context, message *isb.ReadMessage) ([]*isb.Message, error) {
+	return testutils.CopyUDFTestApply(ctx, message)
+}
+
+func TestOrderedProcessing(t *testing.T) {
+
+	// Test Reducer returns the messages as is
+	identityReducer := reducer.ReduceFunc(func(ctx context.Context, input <-chan *isb.ReadMessage) ([]*isb.Message, error) {
+		messages := make([]*isb.Message, 0)
+		for msg := range input {
+			messages = append(messages, &msg.Message)
+		}
+		return messages, nil
+	})
+	to1 := simplebuffer.NewInMemoryBuffer("to1", 100)
+	toSteps := map[string]isb.BufferWriter{
+		"to1": to1,
+	}
+	_, pw := generic.BuildNoOpWatermarkProgressorsFromBufferMap(make(map[string]isb.BufferWriter))
+
+	ctx := context.Background()
+
+	tests := []struct {
+		name           string
+		partitions     []partition.ID
+		reduceOrder    []partition.ID
+		expectedBefore int
+		expectedAfter  []partition.ID
+	}{
+		{
+			name:           "single-task-finished",
+			partitions:     partitions(1),
+			expectedBefore: 1,
+			reduceOrder:    partitionsFor([]int{0}),
+			expectedAfter:  []partition.ID{},
+		},
+		{
+			name:           "middle-task-finished-in-multiple-tasks",
+			partitions:     partitions(4),
+			expectedBefore: 4,
+			reduceOrder:    partitionsFor([]int{2}),
+			expectedAfter:  partitionsFor([]int{0, 1, 2, 3}),
+		},
+		{
+			name:           "tasks-in-order",
+			partitions:     partitions(4),
+			expectedBefore: 4,
+			reduceOrder:    partitionsFor([]int{0, 1, 2, 3}),
+			expectedAfter:  partitionsFor([]int{}),
+		},
+		{
+			name:           "first-task-multiple-tasks",
+			partitions:     partitions(4),
+			expectedBefore: 4,
+			reduceOrder:    partitionsFor([]int{0}),
+			expectedAfter:  partitionsFor([]int{1, 2, 3}),
+		},
+	}
+	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)
+			op.startUp(ctx)
+			// although this could be declared outside, since we are using common naming scheme for partitions,
+			// things will go haywire.
+			pbqManager, _ := pbq.NewManager(ctx, pbq.WithPBQStoreOptions(store.WithStoreSize(int64(100)), store.WithPbqStoreType(dfv1.InMemoryType)),
+				pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
+			cCtx, cancelFn := context.WithCancel(ctx)
+			defer cancelFn()
+			for _, _partition := range tt.partitions {
+				p, _ := pbqManager.CreateNewPBQ(ctx, _partition)
+				op.schedulePnF(cCtx, identityReducer, p, _partition, toSteps, myForwardTest{}, pw)
+			}
+			assert.Equal(t, op.taskQueue.Len(), tt.expectedBefore)
+			count := 0
+			for e := op.taskQueue.Front(); e != nil; e = e.Next() {
+				pfTask := e.Value.(*task)
+				partitionKey := pfTask.pf.PartitionID
+				assert.Equal(t, partition.ID{Key: fmt.Sprintf("partition-%d", count)}, partitionKey)
+				count = count + 1
+			}
+
+			for _, id := range tt.reduceOrder {
+				p := pbqManager.GetPBQ(id)
+				p.CloseOfBook()
+				// wait for reduce operation to be done.
+				pfTask := taskForPartition(op, id)
+				// it might so happen that this task got processed and removed
+				// before we could even inspect it.
+				if pfTask == nil {
+					continue
+				}
+				<-pfTask.doneCh
+			}
+
+			if len(tt.expectedAfter) == 0 {
+				time.Sleep(100 * time.Millisecond)
+			}
+
+			for _, partitionId := range tt.expectedAfter {
+				pfTask := taskForPartition(op, partitionId)
+				assert.NotNil(t, pfTask)
+			}
+
+		})
+	}
+
+}
+
+func partitions(count int) []partition.ID {
+	partitions := make([]partition.ID, count)
+	for i := 0; i < count; i++ {
+		partitions[i] = partition.ID{Key: fmt.Sprintf("partition-%d", i)}
+	}
+	return partitions
+}
+
+func partitionsFor(partitionIdx []int) []partition.ID {
+	partitions := make([]partition.ID, len(partitionIdx))
+	for i, idx := range partitionIdx {
+		partitions[i] = partition.ID{Key: fmt.Sprintf("partition-%d", idx)}
+	}
+	return partitions
+}
+
+func taskForPartition(op *orderedForwarder, partitionId partition.ID) *task {
+	op.RLock()
+	defer op.RUnlock()
+	for e := op.taskQueue.Front(); e != nil; e = e.Next() {
+		pfTask := e.Value.(*task)
+		partitionKey := pfTask.pf.PartitionID
+		if partitionKey == partitionId {
+			return pfTask
+		}
+	}
+	return nil
+}
diff --git a/pkg/reduce/reduce_test.go b/pkg/reduce/reduce_test.go
new file mode 100644
index 00000000..0dc3ae44
--- /dev/null
+++ b/pkg/reduce/reduce_test.go
@@ -0,0 +1,192 @@
+package reduce
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"testing"
+	"time"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
+	"github.com/numaproj/numaflow/pkg/pbq"
+	"github.com/numaproj/numaflow/pkg/pbq/store"
+	"github.com/numaproj/numaflow/pkg/watermark/processor"
+	"github.com/numaproj/numaflow/pkg/watermark/publish"
+	"github.com/numaproj/numaflow/pkg/window/strategy/fixed"
+	"github.com/stretchr/testify/assert"
+)
+
+type EventTypeWMProgressor struct{}
+
+func (e EventTypeWMProgressor) PublishWatermark(watermark processor.Watermark, offset isb.Offset) {
+
+}
+
+func (e EventTypeWMProgressor) GetLatestWatermark() processor.Watermark {
+	return processor.Watermark{}
+}
+
+func (e EventTypeWMProgressor) StopPublisher() {
+}
+
+func (e EventTypeWMProgressor) GetWatermark(_ isb.Offset) processor.Watermark {
+	return processor.Watermark(time.Now())
+}
+
+func (e EventTypeWMProgressor) GetHeadWatermark() processor.Watermark {
+	return processor.Watermark{}
+}
+
+// PayloadForTest is a dummy payload for testing.
+type PayloadForTest struct {
+	Key   string
+	Value int
+}
+
+type forwardReduceTest struct {
+}
+
+// Reduce returns a result, with last message's event time
+func (f forwardReduceTest) Reduce(ctx context.Context, messageStream <-chan *isb.ReadMessage) ([]*isb.Message, error) {
+	count := 0
+	for range messageStream {
+		count += 1
+	}
+
+	payload := PayloadForTest{Key: "count", Value: count}
+	b, _ := json.Marshal(payload)
+	ret := &isb.Message{
+		Header: isb.Header{
+			PaneInfo: isb.PaneInfo{
+				EventTime: time.Now(),
+			},
+			ID:  "msgID",
+			Key: "result",
+		},
+		Body: isb.Body{Payload: b},
+	}
+	return []*isb.Message{
+		ret,
+	}, nil
+}
+
+func (f forwardReduceTest) WhereTo(s string) ([]string, error) {
+	return []string{"to"}, nil
+}
+
+// read from simple buffer
+// mock reduce op to return result
+// assert to check if the result is forwarded to toBuffers
+func TestDataForward_Start(t *testing.T) {
+	windowTime := 2 * time.Second
+	parentCtx := context.Background()
+	child, cancelFn := context.WithTimeout(parentCtx, windowTime*2)
+	defer cancelFn()
+	fromBufferSize := int64(1000)
+	toBufferSize := int64(10)
+	fromBuffer := simplebuffer.NewInMemoryBuffer("from", fromBufferSize)
+	to := simplebuffer.NewInMemoryBuffer("to", toBufferSize)
+
+	// keep on writing <count> messages every 1 second for the supplied key
+	go writeMessages(child, 10, "test-1", fromBuffer)
+
+	toBuffer := map[string]isb.BufferWriter{
+		"to": to,
+	}
+
+	var err error
+	var pbqManager *pbq.Manager
+
+	// create pbqManager
+	pbqManager, err = pbq.NewManager(child, pbq.WithPBQStoreOptions(store.WithStoreSize(100), store.WithPbqStoreType(dfv1.InMemoryType)),
+		pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
+	assert.NoError(t, err)
+
+	publisher := map[string]publish.Publisher{
+		"to": EventTypeWMProgressor{},
+	}
+
+	// create new fixed window of (windowTime)
+	window := fixed.NewFixed(windowTime)
+
+	var reduceDataForwarder *DataForward
+	reduceDataForwarder, err = NewDataForward(child, forwardReduceTest{}, fromBuffer, toBuffer, pbqManager, forwardReduceTest{}, EventTypeWMProgressor{}, publisher, window, WithReadBatchSize(10))
+	assert.NoError(t, err)
+
+	go reduceDataForwarder.Start(child)
+
+	for to.IsEmpty() {
+		select {
+		case <-child.Done():
+			assert.Fail(t, child.Err().Error())
+		default:
+			time.Sleep(100 * time.Millisecond)
+		}
+	}
+
+	// we are reading only one message here but the count should be equal to
+	// the number of keyed windows that closed
+	msgs, readErr := to.Read(child, 1)
+
+	assert.Nil(t, readErr)
+	assert.Len(t, msgs, 1)
+
+	// assert the output of reduce (count of messages)
+	var readMessagePayload PayloadForTest
+	_ = json.Unmarshal(msgs[0].Payload, &readMessagePayload)
+	// here the count will always be equal to the number of messages written per key
+	assert.Equal(t, 10, readMessagePayload.Value)
+	assert.Equal(t, "count", readMessagePayload.Key)
+
+}
+
+func writeMessages(ctx context.Context, count int, key string, fromBuffer *simplebuffer.InMemoryBuffer) {
+	generateTime := 1 * time.Second
+	ticker := time.NewTicker(generateTime)
+	i := 1
+	for {
+		select {
+		case <-ticker.C:
+			// build 10 messages from the start time, time difference between the messages is 1 second
+			messages := buildMessagesForReduce(count, key+strconv.Itoa(i))
+			i++
+
+			// TODO use watermark in memory store when it is ready to synthesize offsets
+			// write the messages to fromBuffer, so that it will be available for consuming
+			fromBuffer.Write(ctx, messages)
+		case <-ctx.Done():
+			ticker.Stop()
+			return
+		}
+	}
+
+}
+
+// buildMessagesForReduce builds test isb.Message which can be used for testing reduce.
+func buildMessagesForReduce(count int, key string) []isb.Message {
+	var messages = make([]isb.Message, 0, count)
+	for i := 0; i < count; i++ {
+		tmpTime := time.Now()
+		result, _ := json.Marshal(PayloadForTest{
+			Key:   key,
+			Value: i,
+		})
+		messages = append(messages,
+			isb.Message{
+				Header: isb.Header{
+					PaneInfo: isb.PaneInfo{
+						EventTime: tmpTime,
+					},
+					ID:  fmt.Sprintf("%d", i),
+					Key: key,
+				},
+				Body: isb.Body{Payload: result},
+			},
+		)
+	}
+
+	return messages
+}
diff --git a/pkg/sinks/kafka/kafka_test.go b/pkg/sinks/kafka/kafka_test.go
index 58e62abd..4e7c6375 100644
--- a/pkg/sinks/kafka/kafka_test.go
+++ b/pkg/sinks/kafka/kafka_test.go
@@ -3,6 +3,7 @@ package kafka
 import (
 	"context"
 	"fmt"
+	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"testing"
 
 	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
@@ -16,7 +17,6 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/forward"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
-	"github.com/numaproj/numaflow/pkg/udf/applier"
 )
 
 func TestWriteSuccessToKafka(t *testing.T) {
diff --git a/pkg/sinks/logger/log_test.go b/pkg/sinks/logger/log_test.go
index 73218e2d..6bc59513 100644
--- a/pkg/sinks/logger/log_test.go
+++ b/pkg/sinks/logger/log_test.go
@@ -2,6 +2,7 @@ package logger
 
 import (
 	"context"
+	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"testing"
 	"time"
 
@@ -12,7 +13,6 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/forward"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
-	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"github.com/stretchr/testify/assert"
 )
 
diff --git a/pkg/udf/applier/uds_grpc_test.go b/pkg/udf/function/uds_grpc_test.go
similarity index 55%
rename from pkg/udf/applier/uds_grpc_test.go
rename to pkg/udf/function/uds_grpc_test.go
index 1eef624b..0774bc34 100644
--- a/pkg/udf/applier/uds_grpc_test.go
+++ b/pkg/udf/function/uds_grpc_test.go
@@ -1,8 +1,9 @@
-package applier
+package function
 
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"testing"
 	"time"
@@ -240,3 +241,201 @@ func TestHGRPCBasedUDF_ApplyWithMockClient(t *testing.T) {
 	assert.Equal(t, expectedResults, results)
 	assert.Equal(t, expectedKeys, resultKeys)
 }
+
+func TestGRPCBasedUDF_BasicReduceWithMockClient(t *testing.T) {
+	t.Run("test success", func(t *testing.T) {
+
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+
+		mockClient := funcmock.NewMockUserDefinedFunctionClient(ctrl)
+		mockReduceClient := NewMockUserDefinedFunction_ReduceFnClient(ctrl)
+
+		mockReduceClient.EXPECT().Send(gomock.Any()).Return(nil).AnyTimes()
+		mockReduceClient.EXPECT().CloseAndRecv().Return(&functionpb.DatumList{
+			Elements: []*functionpb.Datum{
+				{
+					Key:   "reduced_result_key",
+					Value: []byte(`forward_message`),
+				},
+			},
+		}, nil)
+
+		messageCh := make(chan *isb.ReadMessage)
+
+		mockClient.EXPECT().ReduceFn(gomock.Any(), gomock.Any()).Return(mockReduceClient, nil)
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		go func() {
+			<-ctx.Done()
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Log(t.Name(), "test timeout")
+			}
+		}()
+
+		u := NewMockUDSGRPCBasedUDF(mockClient)
+		messages := testutils.BuildTestReadMessages(10, time.Now())
+
+		go func() {
+			for index := range messages {
+				messageCh <- &messages[index]
+			}
+			close(messageCh)
+		}()
+
+		got, err := u.Reduce(ctx, messageCh)
+
+		assert.Len(t, got, 1)
+		assert.NoError(t, err)
+	})
+
+	t.Run("test error", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+
+		mockClient := funcmock.NewMockUserDefinedFunctionClient(ctrl)
+		mockReduceClient := NewMockUserDefinedFunction_ReduceFnClient(ctrl)
+		mockReduceClient.EXPECT().Send(gomock.Any()).Return(nil).AnyTimes()
+		mockReduceClient.EXPECT().CloseAndRecv().Return(&functionpb.DatumList{
+			Elements: []*functionpb.Datum{
+				{
+					Key:   "reduced_result_key",
+					Value: []byte(`forward_message`),
+				},
+			},
+		}, errors.New("mock error for reduce"))
+
+		mockClient.EXPECT().ReduceFn(gomock.Any(), gomock.Any()).Return(mockReduceClient, nil)
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		go func() {
+			<-ctx.Done()
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Log(t.Name(), "test timeout")
+			}
+		}()
+
+		u := NewMockUDSGRPCBasedUDF(mockClient)
+
+		messageCh := make(chan *isb.ReadMessage)
+		messages := testutils.BuildTestReadMessages(10, time.Now())
+
+		go func() {
+			for index := range messages {
+				messageCh <- &messages[index]
+			}
+			close(messageCh)
+		}()
+
+		_, err := u.Reduce(ctx, messageCh)
+
+		assert.ErrorIs(t, err, ApplyUDFErr{
+			UserUDFErr: false,
+			Message:    fmt.Sprintf("%s", err),
+			InternalErr: InternalErr{
+				Flag:        true,
+				MainCarDown: false,
+			},
+		})
+	})
+
+	t.Run("test context close", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+
+		mockClient := funcmock.NewMockUserDefinedFunctionClient(ctrl)
+		mockReduceClient := NewMockUserDefinedFunction_ReduceFnClient(ctrl)
+		mockReduceClient.EXPECT().Send(gomock.Any()).Return(nil).AnyTimes()
+		mockReduceClient.EXPECT().CloseAndRecv().Return(&functionpb.DatumList{
+			Elements: []*functionpb.Datum{
+				{
+					Key:   "reduced_result_key",
+					Value: []byte(`forward_message`),
+				},
+			},
+		}, nil).AnyTimes()
+
+		mockClient.EXPECT().ReduceFn(gomock.Any(), gomock.Any()).Return(mockReduceClient, nil)
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		u := NewMockUDSGRPCBasedUDF(mockClient)
+
+		messageCh := make(chan *isb.ReadMessage)
+
+		_, err := u.Reduce(ctx, messageCh)
+
+		assert.Error(t, err, ctx.Err())
+	})
+}
+
+func TestHGRPCBasedUDF_Reduce(t *testing.T) {
+	sumFunc := func(dataStreamCh <-chan *functionpb.Datum) interface{} {
+		var sum testutils.PayloadForTest
+		for datum := range dataStreamCh {
+			var payLoad testutils.PayloadForTest
+			_ = json.Unmarshal(datum.GetValue(), &payLoad)
+			sum.Value += payLoad.Value
+		}
+		return sum
+	}
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	messageCh := make(chan *isb.ReadMessage, 10)
+	datumStreamCh := make(chan *functionpb.Datum, 10)
+	messages := testutils.BuildTestReadMessages(10, time.Now())
+
+	go func() {
+		for index := range messages {
+			messageCh <- &messages[index]
+			datumStreamCh <- createDatum(&messages[index])
+		}
+		close(messageCh)
+		close(datumStreamCh)
+	}()
+
+	mockClient := funcmock.NewMockUserDefinedFunctionClient(ctrl)
+	mockReduceClient := NewMockUserDefinedFunction_ReduceFnClient(ctrl)
+	mockReduceClient.EXPECT().Send(gomock.Any()).Return(nil).AnyTimes()
+	mockReduceClient.EXPECT().CloseAndRecv().DoAndReturn(
+		func() (*functionpb.DatumList, error) {
+			result := sumFunc(datumStreamCh)
+			sumValue, _ := json.Marshal(result.(testutils.PayloadForTest))
+			var elements []*functionpb.Datum
+			elements = append(elements, &functionpb.Datum{
+				Key:   "sum",
+				Value: sumValue,
+			})
+			datumList := &functionpb.DatumList{
+				Elements: elements,
+			}
+			return datumList, nil
+		}).AnyTimes()
+
+	mockClient.EXPECT().ReduceFn(gomock.Any(), gomock.Any()).Return(mockReduceClient, nil)
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
+	go func() {
+		<-ctx.Done()
+		if ctx.Err() == context.DeadlineExceeded {
+			t.Log(t.Name(), "test timeout")
+		}
+	}()
+
+	u := NewMockUDSGRPCBasedUDF(mockClient)
+
+	result, err := u.Reduce(ctx, messageCh)
+
+	var resultPayload testutils.PayloadForTest
+	_ = json.Unmarshal(result[0].Payload, &resultPayload)
+
+	assert.NoError(t, err)
+	assert.Equal(t, result[0].Key, "sum")
+	assert.Equal(t, resultPayload.Value, int64(45))
+}
diff --git a/pkg/window/keyed/keyed_test.go b/pkg/window/keyed/keyed_test.go
index a58322af..8785ccca 100644
--- a/pkg/window/keyed/keyed_test.go
+++ b/pkg/window/keyed/keyed_test.go
@@ -4,6 +4,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
+
 	"github.com/numaproj/numaflow/pkg/window"
 	"github.com/stretchr/testify/assert"
 )
@@ -18,30 +20,35 @@ func TestKeyedWindow_AddKey(t *testing.T) {
 		name         string
 		given        *KeyedWindow
 		input        string
-		expectedKeys []string
+		expectedKeys map[string]string
 	}{
 		{
 			name:         "no_keys",
 			given:        &KeyedWindow{},
 			input:        "key1",
-			expectedKeys: []string{"key1"},
+			expectedKeys: map[string]string{"key1": "key1"},
 		},
 		{
 			name: "with_some_existing_keys",
 			given: &KeyedWindow{
-				Keys: []string{"key2", "key3"},
+				Keys: map[string]string{"key2": "key2", "key3": "key3"},
 			},
 			input:        "key4",
-			expectedKeys: []string{"key2", "key3", "key4"},
+			expectedKeys: map[string]string{"key2": "key2", "key3": "key3", "key4": "key4"},
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			kw = NewKeyedWindow(iw)
-			kw.Keys = append(kw.Keys, tt.given.Keys...)
+			for k := range tt.given.Keys {
+				kw.AddKey(k)
+			}
 			kw.AddKey(tt.input)
-			assert.ElementsMatch(t, kw.Keys, tt.expectedKeys)
+			for k := range tt.expectedKeys {
+				_, ok := kw.Keys[k]
+				assert.True(t, ok)
+			}
 		})
 	}
 }
@@ -56,21 +63,29 @@ func TestKeyedWindow_Partitions(t *testing.T) {
 		name     string
 		given    *KeyedWindow
 		input    string
-		expected []string
+		expected []partition.ID
 	}{
 		{
 			name:     "no_keys",
 			given:    &KeyedWindow{},
-			expected: []string{},
+			expected: []partition.ID{},
 		},
 		{
 			name: "with_some_existing_keys",
 			given: &KeyedWindow{
-				Keys: []string{"key2", "key3"},
+				Keys: map[string]string{"key2": "key2", "key3": "key3"},
 			},
-			expected: []string{
-				"60-120-key2",
-				"60-120-key3",
+			expected: []partition.ID{
+				{
+					Key:   "key2",
+					Start: time.Unix(60, 0),
+					End:   time.Unix(120, 0),
+				},
+				{
+					Key:   "key3",
+					Start: time.Unix(60, 0),
+					End:   time.Unix(120, 0),
+				},
 			},
 		},
 	}