Repository: numaproj/numaflow
Pull request #1508 (2024-02-14 04:37:52)

chore: refactor store to support unaligned wal


Patch:
diff --git a/examples/11-join-on-reduce.yaml b/examples/11-join-on-reduce.yaml
index 0496921f..7336ba9d 100644
--- a/examples/11-join-on-reduce.yaml
+++ b/examples/11-join-on-reduce.yaml
@@ -20,7 +20,7 @@ spec:
       udf:
         container:
           # Compute the sum, see https://github.com/numaproj/numaflow-go/tree/main/pkg/reducer/examples/sum
-          image: quay.io/numaio/numaflow-go/reduce-sum:v0.5.0
+          image: quay.io/numaio/numaflow-go/reduce-sum:v0.6.0
         groupBy:
           window:
             fixed:
diff --git a/examples/11-join-on-sink.yaml b/examples/11-join-on-sink.yaml
index 101f99a5..f7ffe6b2 100644
--- a/examples/11-join-on-sink.yaml
+++ b/examples/11-join-on-sink.yaml
@@ -13,7 +13,7 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.5.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
     - name: even-cat
       udf:
         builtin:
diff --git a/examples/2-even-odd-pipeline.yaml b/examples/2-even-odd-pipeline.yaml
index c733ec37..f0bed70b 100644
--- a/examples/2-even-odd-pipeline.yaml
+++ b/examples/2-even-odd-pipeline.yaml
@@ -13,7 +13,7 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.5.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
     - name: even-sink
       scale:
         min: 1
diff --git a/examples/6-reduce-fixed-window.yaml b/examples/6-reduce-fixed-window.yaml
index 391dae2b..e9027f22 100644
--- a/examples/6-reduce-fixed-window.yaml
+++ b/examples/6-reduce-fixed-window.yaml
@@ -13,12 +13,12 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.5.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
     - name: compute-sum
       udf:
         container:
           # compute the sum
-          image: quay.io/numaio/numaflow-go/reduce-sum:v0.5.0
+          image: quay.io/numaio/numaflow-go/reduce-sum:v0.6.0
         groupBy:
           window:
             fixed:
diff --git a/examples/7-reduce-sliding-window.yaml b/examples/7-reduce-sliding-window.yaml
index 8ab75470..c6f696e5 100644
--- a/examples/7-reduce-sliding-window.yaml
+++ b/examples/7-reduce-sliding-window.yaml
@@ -24,13 +24,13 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.5.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
     - name: reduce-sliding
       partitions: 2
       udf:
         container:
           # compute the sum
-          image: quay.io/numaio/numaflow-go/reduce-sum:v0.5.0
+          image: quay.io/numaio/numaflow-go/reduce-sum:v0.6.0
         groupBy:
           window:
             sliding:
diff --git a/examples/8-reduce-complex-pipeline.yaml b/examples/8-reduce-complex-pipeline.yaml
index a5458edd..8d3a174f 100644
--- a/examples/8-reduce-complex-pipeline.yaml
+++ b/examples/8-reduce-complex-pipeline.yaml
@@ -17,13 +17,13 @@ spec:
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/evenodd
-          image: quay.io/numaio/numaflow-go/map-even-odd:v0.5.0
+          image: quay.io/numaio/numaflow-go/map-even-odd:v0.6.0
     - name: keyed-fixed-sum
       partitions: 2
       udf:
         container:
           # compute the sum
-          image: quay.io/numaio/numaflow-go/reduce-sum:v0.5.0
+          image: quay.io/numaio/numaflow-go/reduce-sum:v0.6.0
         groupBy:
           window:
             fixed:
@@ -31,13 +31,13 @@ spec:
           keyed: true
           storage:
             persistentVolumeClaim:
-              volumeSize: 10Gi
+              volumeSize: 2Gi
               accessMode: ReadWriteOnce
     - name: non-keyed-fixed-sum
       udf:
         container:
           # compute the sum
-          image: quay.io/numaio/numaflow-go/reduce-sum:v0.5.0
+          image: quay.io/numaio/numaflow-go/reduce-sum:v0.6.0
         groupBy:
           window:
             fixed:
@@ -45,13 +45,13 @@ spec:
           keyed: false
           storage:
             persistentVolumeClaim:
-              volumeSize: 10Gi
+              volumeSize: 2Gi
               accessMode: ReadWriteOnce
     - name: non-keyed-sliding-sum
       udf:
         container:
           # compute the sum
-          image: quay.io/numaio/numaflow-go/reduce-sum:v0.5.0
+          image: quay.io/numaio/numaflow-go/reduce-sum:v0.6.0
         groupBy:
           window:
             sliding:
@@ -60,7 +60,7 @@ spec:
           keyed: false
           storage:
             persistentVolumeClaim:
-              volumeSize: 10Gi
+              volumeSize: 2Gi
               accessMode: ReadWriteOnce
     - name: sink
       scale:
diff --git a/pkg/reduce/data_forward.go b/pkg/reduce/data_forward.go
index c93e13be..86e7968a 100644
--- a/pkg/reduce/data_forward.go
+++ b/pkg/reduce/data_forward.go
@@ -35,6 +35,7 @@ import (
 	"time"
 
 	"go.uber.org/zap"
+	"golang.org/x/sync/errgroup"
 	"k8s.io/apimachinery/pkg/util/wait"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
@@ -43,6 +44,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/store"
 	"github.com/numaproj/numaflow/pkg/reduce/pnf"
 	"github.com/numaproj/numaflow/pkg/shared/idlehandler"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
@@ -65,13 +67,13 @@ type DataForward struct {
 	windower            window.TimedWindower
 	keyed               bool
 	idleManager         wmb.IdleManager
-	// wmbChecker checks if the idle watermark is valid when the len(readMessage) is 0.
-	wmbChecker     wmb.WMBChecker
-	pbqManager     *pbq.Manager
-	whereToDecider forwarder.ToWhichStepDecider
-	of             *pnf.Manager
-	opts           *Options
-	log            *zap.SugaredLogger
+	wmbChecker          wmb.WMBChecker // wmbChecker checks if the idle watermark is valid when the len(readMessage) is 0.
+	pbqManager          *pbq.Manager
+	whereToDecider      forwarder.ToWhichStepDecider
+	storeManager        store.Manager
+	of                  *pnf.Manager
+	opts                *Options
+	log                 *zap.SugaredLogger
 }
 
 // NewDataForward creates a new DataForward
@@ -80,6 +82,7 @@ func NewDataForward(ctx context.Context,
 	fromBuffer isb.BufferReader,
 	toBuffers map[string][]isb.BufferWriter,
 	pbqManager *pbq.Manager,
+	storeManager store.Manager,
 	whereToDecider forwarder.ToWhichStepDecider,
 	fw fetch.Fetcher,
 	watermarkPublishers map[string]publish.Publisher,
@@ -109,6 +112,7 @@ func NewDataForward(ctx context.Context,
 		keyed:               vertexInstance.Vertex.Spec.UDF.GroupBy.Keyed,
 		idleManager:         idleManager,
 		pbqManager:          pbqManager,
+		storeManager:        storeManager,
 		whereToDecider:      whereToDecider,
 		of:                  of,
 		wmbChecker:          wmb.NewWMBChecker(2), // TODO: make configurable
@@ -148,18 +152,14 @@ func (df *DataForward) ReplayPersistedMessages(ctx context.Context) error {
 	if df.windower.Type() == window.Unaligned {
 		return nil
 	}
-	// start the PBQManager which discovers and builds the state from persistent store of the PBQ.
-	partitions, err := df.pbqManager.GetExistingPartitions(ctx)
+
+	existingStores, err := df.storeManager.DiscoverStores(ctx)
 	if err != nil {
 		return err
 	}
 
-	df.log.Infow("Partitions to be replayed ", zap.Int("count", len(partitions)), zap.Any("partitions", partitions))
-
-	// for aligned windows, we can create/assign a partition based on the timestamp in the payload, but this is not
-	// necessarily true for unaligned windows.
-
-	for _, p := range partitions {
+	for _, s := range existingStores {
+		p := s.PartitionID()
 		// create a window for each partition and insert it to the windower
 		// so that the window can be closed when the watermark
 		// crosses the window.
@@ -169,11 +169,48 @@ func (df *DataForward) ReplayPersistedMessages(ctx context.Context) error {
 
 		df.associatePBQAndPnF(ctx, &p)
 	}
-
-	// replays the data (replay internally writes the data from persistent store on to the PBQ)
-	df.pbqManager.Replay(ctx)
-
-	return nil
+	eg := errgroup.Group{}
+	df.log.Info("Number of partitions to replay: ", len(existingStores))
+	// replay the messages from each store in parallel
+	for _, sr := range existingStores {
+		df.log.Info("Replaying messages from partition: ", sr.PartitionID().String())
+		func(ctx context.Context, s store.Store) {
+			eg.Go(func() error {
+				pid := s.PartitionID()
+				readCh, errCh := s.Replay()
+				for {
+					select {
+					case <-ctx.Done():
+						return nil
+					case err := <-errCh:
+						if err != nil {
+							return err
+						}
+					case msg, ok := <-readCh:
+						if !ok {
+							return nil
+						}
+						// TODO: support unaligned windows
+						tw := window.NewAlignedTimedWindow(pid.Start, pid.End, pid.Slot)
+						request := &window.TimedWindowRequest{
+							ReadMessage: msg,
+							Operation:   window.Append,
+							Windows:     []window.TimedWindow{tw},
+							ID:          &pid,
+						}
+						// we don't want to persist the messages again
+						// because they are already persisted in the store
+						// so we set persist to false
+						err = df.writeToPBQ(ctx, request, false)
+						if err != nil {
+							return err
+						}
+					}
+				}
+			})
+		}(ctx, sr)
+	}
+	return eg.Wait()
 }
 
 // forwardAChunk reads a chunk of messages from isb and assigns watermark to messages
@@ -225,7 +262,7 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 			if watermark := time.UnixMilli(processorWMB.Watermark).Add(-1 * time.Millisecond); oldestWindowEndTime.Before(watermark) {
 				windowOperations := df.windower.CloseWindows(watermark)
 				for _, op := range windowOperations {
-					err = df.writeToPBQ(ctx, op)
+					err = df.writeToPBQ(ctx, op, true)
 					if err != nil {
 						df.log.Errorw("Failed to write to PBQ", zap.Error(err))
 					}
@@ -370,7 +407,7 @@ func (df *DataForward) process(ctx context.Context, messages []*isb.ReadMessage)
 
 	// write the close window operations to PBQ
 	for _, winOp := range closedWindowOps {
-		err = df.writeToPBQ(ctx, winOp)
+		err = df.writeToPBQ(ctx, winOp, true)
 		if err != nil {
 			df.log.Errorw("Failed to write close signal to PBQ", zap.Error(err))
 		}
@@ -416,7 +453,7 @@ messagesLoop:
 		// We need to write the messages to these PBQs
 		for _, winOp := range windowOperations {
 
-			err := df.writeToPBQ(ctx, winOp)
+			err := df.writeToPBQ(ctx, winOp, true)
 			// there is no point continuing because we are seeing an error.
 			// this error will ONLY BE set if we are in an error loop and ctx.Done() has been invoked.
 			if err != nil {
@@ -480,7 +517,7 @@ func (df *DataForward) shouldDropMessage(message *isb.ReadMessage) bool {
 
 // writeToPBQ writes to the PBQ. It will return error only if it is not failing to write to PBQ and is in a continuous
 // error loop, and we have received ctx.Done() via SIGTERM.
-func (df *DataForward) writeToPBQ(ctx context.Context, winOp *window.TimedWindowRequest) error {
+func (df *DataForward) writeToPBQ(ctx context.Context, winOp *window.TimedWindowRequest, persist bool) error {
 	defer func(t time.Time) {
 		metrics.PBQWriteTime.With(map[string]string{
 			metrics.LabelVertex:             df.vertexName,
@@ -499,7 +536,7 @@ func (df *DataForward) writeToPBQ(ctx context.Context, winOp *window.TimedWindow
 	q := df.associatePBQAndPnF(ctx, winOp.ID)
 
 	err := wait.ExponentialBackoff(pbqWriteBackoff, func() (done bool, err error) {
-		rErr := q.Write(context.Background(), winOp)
+		rErr := q.Write(context.Background(), winOp, persist)
 		if rErr != nil {
 			df.log.Errorw("Failed to write message", zap.String("msgOffSet", winOp.ReadMessage.ReadOffset.String()), zap.String("partitionID", winOp.ID.String()), zap.Error(rErr))
 			metrics.PBQWriteErrorCount.With(map[string]string{
diff --git a/pkg/reduce/pbq/interfaces.go b/pkg/reduce/pbq/interfaces.go
index c58fb38b..21509c3f 100644
--- a/pkg/reduce/pbq/interfaces.go
+++ b/pkg/reduce/pbq/interfaces.go
@@ -40,7 +40,9 @@ type Reader interface {
 // No data can be written to PBQ after cob.
 type WriteCloser interface {
 	// Write writes the TimedWindowRequest to PBQ
-	Write(ctx context.Context, msg *window.TimedWindowRequest) error
+	// persist to decide if the data should be persisted or not
+	// during replay persist will be false
+	Write(ctx context.Context, msg *window.TimedWindowRequest, persist bool) error
 	// CloseOfBook (cob) closes PBQ, no writes will be accepted after cob
 	CloseOfBook()
 	// Close to handle context close on writer
diff --git a/pkg/reduce/pbq/pbq.go b/pkg/reduce/pbq/pbq.go
index 76673f4b..a99d18f2 100644
--- a/pkg/reduce/pbq/pbq.go
+++ b/pkg/reduce/pbq/pbq.go
@@ -26,7 +26,7 @@ import (
 
 	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq/store/aligned"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/store"
 	"github.com/numaproj/numaflow/pkg/window"
 )
 
@@ -36,7 +36,7 @@ type PBQ struct {
 	vertexName    string
 	pipelineName  string
 	vertexReplica int32
-	store         aligned.Store
+	store         store.Store
 	output        chan *window.TimedWindowRequest
 	cob           bool // cob to avoid panic in case writes happen after close of book
 	PartitionID   partition.ID
@@ -52,7 +52,7 @@ var _ ReadWriteCloser = (*PBQ)(nil)
 // Write accepts a window request and writes it to the PBQ, only the isb message is written to the store.
 // The other metadata like operation etc are recomputed from WAL.
 // request can never be nil.
-func (p *PBQ) Write(ctx context.Context, request *window.TimedWindowRequest) error {
+func (p *PBQ) Write(ctx context.Context, request *window.TimedWindowRequest, persist bool) error {
 	// if cob we should return
 	if p.cob {
 		p.log.Errorw("Failed to write request to pbq, pbq is closed", zap.Any("ID", p.PartitionID), zap.Any("request", request))
@@ -75,7 +75,10 @@ func (p *PBQ) Write(ctx context.Context, request *window.TimedWindowRequest) err
 			if p.windowType == window.Unaligned {
 				return nil
 			}
-			writeErr = p.store.Write(request.ReadMessage)
+			// during replay we do not have to persist
+			if persist {
+				writeErr = p.store.Write(request.ReadMessage)
+			}
 		case window.Close, window.Merge:
 		// these do not have request.ReadMessage, only metadata fields are used
 		default:
@@ -129,41 +132,3 @@ func (p *PBQ) GC() error {
 	p.store = nil
 	return p.manager.deregister(p.PartitionID)
 }
-
-// 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.
-// FIXME: works only for fixed and sliding window
-func (p *PBQ) replayRecordsFromStore(ctx context.Context) {
-	size := p.options.readBatchSize
-readLoop:
-	for {
-		readMessages, eof, err := p.store.Read(size)
-		if err != nil {
-			p.log.Errorw("Error while replaying records from store", zap.Any("ID", p.PartitionID), zap.Error(err))
-		}
-		for _, msg := range readMessages {
-			// FIXME: support for session window
-			var w window.TimedWindow
-			if p.windowType == window.Aligned {
-				w = window.NewAlignedTimedWindow(p.PartitionID.Start, p.PartitionID.End, p.PartitionID.Slot)
-			} else {
-				p.log.Errorw("unAligned window strategy not supported", zap.Any("ID", p.PartitionID), zap.Any("strategy", p.windowType))
-			}
-			// select to avoid infinite blocking while writing to output channel
-			select {
-			case p.output <- &window.TimedWindowRequest{
-				ReadMessage: msg,
-				Operation:   window.Append,
-				Windows:     []window.TimedWindow{w},
-				ID:          &p.PartitionID,
-			}:
-			case <-ctx.Done():
-				break readLoop
-			}
-		}
-		// after replaying all the messages from store, unset replay flag
-		if eof {
-			break
-		}
-	}
-}
diff --git a/pkg/reduce/pbq/pbqmanager.go b/pkg/reduce/pbq/pbqmanager.go
index 2c94d31e..511cb5d7 100644
--- a/pkg/reduce/pbq/pbqmanager.go
+++ b/pkg/reduce/pbq/pbqmanager.go
@@ -29,7 +29,7 @@ import (
 
 	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq/store/aligned"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/store"
 	"github.com/numaproj/numaflow/pkg/window"
 
 	"github.com/numaproj/numaflow/pkg/shared/logging"
@@ -40,7 +40,7 @@ type Manager struct {
 	vertexName    string
 	pipelineName  string
 	vertexReplica int32
-	storeProvider aligned.StoreProvider
+	storeProvider store.Manager
 	pbqOptions    *options
 	pbqMap        map[string]*PBQ
 	log           *zap.SugaredLogger
@@ -53,7 +53,7 @@ type Manager struct {
 
 // NewManager returns new instance of manager
 // We don't intend this to be called by multiple routines.
-func NewManager(ctx context.Context, vertexName string, pipelineName string, vr int32, storeProvider aligned.StoreProvider, windowType window.Type, opts ...PBQOption) (*Manager, error) {
+func NewManager(ctx context.Context, vertexName string, pipelineName string, vr int32, storeProvider store.Manager, windowType window.Type, opts ...PBQOption) (*Manager, error) {
 	pbqOpts := DefaultOptions()
 	for _, opt := range opts {
 		if opt != nil {
@@ -129,38 +129,6 @@ func (m *Manager) GetPBQ(partitionID partition.ID) ReadWriteCloser {
 	return nil
 }
 
-// GetExistingPartitions 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) GetExistingPartitions(ctx context.Context) ([]partition.ID, error) {
-	var ctxClosedErr error
-	var partitionIDs []partition.ID
-
-	var discoverPartitionsBackoff = wait.Backoff{
-		Steps:    math.MaxInt,
-		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.storeProvider.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 partitionIDs, ctxClosedErr
-	}
-
-	return partitionIDs, nil
-}
-
 // ShutDown for clean shut down, flushes pending messages to store and closes the store
 func (m *Manager) ShutDown(ctx context.Context) {
 	// iterate through the map of pbq
@@ -241,22 +209,3 @@ func (m *Manager) getPBQs() []*PBQ {
 
 	return pbqs
 }
-
-// Replay replays messages which are persisted in pbq store.
-func (m *Manager) Replay(ctx context.Context) {
-	var wg sync.WaitGroup
-	var tm = time.Now()
-	partitionsIds := make([]partition.ID, 0)
-	for _, val := range m.getPBQs() {
-		partitionsIds = append(partitionsIds, val.PartitionID)
-		wg.Add(1)
-		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)
-		}(ctx, val)
-	}
-
-	wg.Wait()
-	m.log.Infow("Finished replaying records from store", zap.Duration("took", time.Since(tm)), zap.Any("partitions", partitionsIds))
-}
diff --git a/pkg/reduce/pbq/store/aligned/wal/bootstrap.go b/pkg/reduce/pbq/store/aligned/fs/bootstrap.go
similarity index 80%
rename from pkg/reduce/pbq/store/aligned/wal/bootstrap.go
rename to pkg/reduce/pbq/store/aligned/fs/bootstrap.go
index 5d5cb624..3780eaf8 100644
--- a/pkg/reduce/pbq/store/aligned/wal/bootstrap.go
+++ b/pkg/reduce/pbq/store/aligned/fs/bootstrap.go
@@ -14,14 +14,13 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package wal
+package fs
 
 import (
 	"encoding/binary"
 	"errors"
 	"fmt"
 	"io"
-	"os"
 	"time"
 
 	"github.com/numaproj/numaflow/pkg/isb"
@@ -44,10 +43,6 @@ func (w *WAL) IsCorrupted() bool {
 }
 
 func (w *WAL) readWALHeader() (*partition.ID, error) {
-	if w.openMode == os.O_WRONLY {
-		return nil, fmt.Errorf("opened using O_WRONLY")
-	}
-
 	if w.rOffset > 0 {
 		return nil, fmt.Errorf("header has already been read, current readoffset is at %d", w.rOffset)
 	}
@@ -96,40 +91,37 @@ func decodeWALHeader(buf io.Reader) (*partition.ID, error) {
 	}, nil
 }
 
-func (w *WAL) Read(size int64) ([]*isb.ReadMessage, bool, error) {
-	if w.openMode == os.O_WRONLY {
-		return nil, false, fmt.Errorf("opened using O_WRONLY")
-	}
-
-	if w.rOffset < w.wOffset {
-		return nil, false, fmt.Errorf("read can only happen at startup not after any new writes")
-	}
-
-	messages := make([]*isb.ReadMessage, 0)
-	// if size is greater than the number of messages in the store
-	// we will assign size with the number of messages in the store
-	start := w.rOffset
-	for size > w.rOffset-start && !w.isEnd() {
-		message, sizeRead, err := decodeReadMessage(w.fp)
-		if err != nil {
-			if errors.Is(err, errChecksumMismatch) {
-				w.corrupted = true
+// Replay replays the WAL messages, returns a channel to read messages and a channel to read errors.
+// channel will be closed after all the messages are read from the WAL.
+func (w *WAL) Replay() (<-chan *isb.ReadMessage, <-chan error) {
+	messages := make(chan *isb.ReadMessage)
+	errs := make(chan error)
+
+	go func() {
+		defer close(messages)
+		defer func() { errs = nil }()
+
+		// decode read message and send it to the channel
+		// dont use Read method
+		for !w.isEnd() {
+			message, sizeRead, err := decodeReadMessage(w.fp)
+			if err != nil {
+				if errors.Is(err, errChecksumMismatch) {
+					w.corrupted = true
+				}
+				errs <- err
+				return
 			}
-			return nil, false, err
-		}
 
-		w.rOffset += sizeRead
-		messages = append(messages, message)
-	}
-	currentTime := time.Now()
-	if w.isEnd() {
+			w.rOffset += sizeRead
+			messages <- message
+		}
 		w.wOffset = w.rOffset
 		w.prevSyncedWOffset = w.wOffset
-		w.prevSyncedTime = currentTime
+		w.prevSyncedTime = time.Now()
 		w.numOfUnsyncedMsgs = 0
-		return messages, true, nil
-	}
-	return messages, false, nil
+	}()
+	return messages, errs
 }
 
 // decodeReadMessage decodes the WALMessage which is encoded by encodeWALMessage.
diff --git a/pkg/reduce/pbq/store/aligned/wal/doc.go b/pkg/reduce/pbq/store/aligned/fs/doc.go
similarity index 89%
rename from pkg/reduce/pbq/store/aligned/wal/doc.go
rename to pkg/reduce/pbq/store/aligned/fs/doc.go
index 3b613302..207229cf 100644
--- a/pkg/reduce/pbq/store/aligned/wal/doc.go
+++ b/pkg/reduce/pbq/store/aligned/fs/doc.go
@@ -14,5 +14,5 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// Package wal implements write-ahead-log.
-package wal
+// Package fs implements write-ahead-log on FileSystem.
+package fs
diff --git a/pkg/reduce/pbq/store/aligned/fs/manager.go b/pkg/reduce/pbq/store/aligned/fs/manager.go
new file mode 100644
index 00000000..92818cb5
--- /dev/null
+++ b/pkg/reduce/pbq/store/aligned/fs/manager.go
@@ -0,0 +1,152 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package fs
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/metrics"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/store"
+)
+
+type fsWAL struct {
+	storePath string
+	// maxBufferSize max size of batch before it's flushed to store
+	maxBatchSize int64
+	// syncDuration timeout to sync to store
+	syncDuration time.Duration
+	pipelineName string
+	vertexName   string
+	replicaIndex int32
+	activeStores map[string]store.Store
+}
+
+// NewFSManager is a FileSystem Stores Manager.
+func NewFSManager(vertexInstance *dfv1.VertexInstance, opts ...Option) store.Manager {
+	s := &fsWAL{
+		storePath:    dfv1.DefaultStorePath,
+		maxBatchSize: dfv1.DefaultStoreMaxBufferSize,
+		syncDuration: dfv1.DefaultStoreSyncDuration,
+		pipelineName: vertexInstance.Vertex.Spec.PipelineName,
+		vertexName:   vertexInstance.Vertex.Spec.AbstractVertex.Name,
+		replicaIndex: vertexInstance.Replica,
+		activeStores: make(map[string]store.Store),
+	}
+	for _, o := range opts {
+		o(s)
+	}
+	return s
+}
+
+// CreateStore creates the FS WAL.
+func (ws *fsWAL) CreateStore(_ context.Context, partitionID partition.ID) (store.Store, error) {
+	// check if the store is already present
+	// during crash recovery, we might have already created the store while replaying
+	if store, ok := ws.activeStores[partitionID.String()]; ok {
+		return store, nil
+	}
+	// Create fs dir if not exist
+	var err error
+	if _, err = os.Stat(ws.storePath); os.IsNotExist(err) {
+		err = os.Mkdir(ws.storePath, 0755)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	filePath := getSegmentFilePath(&partitionID, ws.storePath)
+	// we are interested only in the number of new files created
+	filesCount.With(map[string]string{
+		metrics.LabelPipeline:           ws.pipelineName,
+		metrics.LabelVertex:             ws.vertexName,
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(ws.replicaIndex)),
+	}).Inc()
+
+	return NewWriteOnlyWAL(&partitionID, filePath, ws.maxBatchSize, ws.syncDuration, ws.pipelineName, ws.vertexName, ws.replicaIndex)
+
+}
+
+// DiscoverStores returns all the stores present in the storePath
+func (ws *fsWAL) DiscoverStores(_ context.Context) ([]store.Store, error) {
+	files, err := os.ReadDir(ws.storePath)
+	if os.IsNotExist(err) {
+		return []store.Store{}, nil
+	} else if err != nil {
+		return nil, err
+	}
+	partitions := make([]store.Store, 0)
+
+	for _, f := range files {
+		if strings.HasPrefix(f.Name(), SegmentPrefix) && !f.IsDir() {
+			filePath := filepath.Join(ws.storePath, f.Name())
+			wal, err := NewReadWriteWAL(filePath, ws.maxBatchSize, ws.syncDuration, ws.pipelineName, ws.vertexName, ws.replicaIndex)
+			if err != nil {
+				return nil, err
+			}
+			partitions = append(partitions, wal)
+		}
+	}
+
+	return partitions, nil
+}
+
+// DeleteStore deletes the store for the given partitionID
+func (ws *fsWAL) DeleteStore(partitionID partition.ID) error {
+	var err error
+	defer func() {
+		if err != nil {
+			walErrors.With(map[string]string{
+				metrics.LabelPipeline:           ws.pipelineName,
+				metrics.LabelVertex:             ws.vertexName,
+				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(ws.replicaIndex)),
+				labelErrorKind:                  "gc",
+			}).Inc()
+		}
+	}()
+
+	filePath := getSegmentFilePath(&partitionID, ws.storePath)
+	_, err = os.Stat(filePath)
+
+	if err != nil {
+		return err
+	}
+
+	start := time.Now()
+	// an open file can also be deleted
+	err = os.Remove(filePath)
+
+	if err == nil {
+		garbageCollectingTime.With(map[string]string{
+			metrics.LabelPipeline:           ws.pipelineName,
+			metrics.LabelVertex:             ws.vertexName,
+			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(ws.replicaIndex)),
+		}).Observe(float64(time.Since(start).Microseconds()))
+		activeFilesCount.With(map[string]string{
+			metrics.LabelPipeline:           ws.pipelineName,
+			metrics.LabelVertex:             ws.vertexName,
+			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(ws.replicaIndex)),
+		}).Dec()
+	}
+	return err
+}
diff --git a/pkg/reduce/pbq/store/aligned/wal/metrics.go b/pkg/reduce/pbq/store/aligned/fs/metrics.go
similarity index 93%
rename from pkg/reduce/pbq/store/aligned/wal/metrics.go
rename to pkg/reduce/pbq/store/aligned/fs/metrics.go
index cc49d0f1..2c5739b0 100644
--- a/pkg/reduce/pbq/store/aligned/wal/metrics.go
+++ b/pkg/reduce/pbq/store/aligned/fs/metrics.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package wal
+package fs
 
 import (
 	"github.com/prometheus/client_golang/prometheus"
@@ -44,19 +44,19 @@ var entriesBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 var filesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "pbq_wal",
 	Name:      "wal_files_total",
-	Help:      "Total number of wal files/partitions (including both active and closed)",
+	Help:      "Total number of fs files/partitions (including both active and closed)",
 }, []string{metrics.LabelPipeline, metrics.LabelVertex, metrics.LabelVertexReplicaIndex})
 
 var activeFilesCount = promauto.NewGaugeVec(prometheus.GaugeOpts{
 	Subsystem: "pbq_wal",
 	Name:      "active_wal_files_total",
-	Help:      "Total number of active wal files/partitions",
+	Help:      "Total number of active fs files/partitions",
 }, []string{metrics.LabelPipeline, metrics.LabelVertex, metrics.LabelVertexReplicaIndex})
 
 var garbageCollectingTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
 	Subsystem: "pbq_wal",
 	Name:      "wal_garbage_collecting_time",
-	Help:      "Garbage Collecting time of a pbq wal (100 to 5000 microseconds)",
+	Help:      "Garbage Collecting time of a pbq fs (100 to 5000 microseconds)",
 	Buckets:   prometheus.ExponentialBucketsRange(100, 5000, 5),
 }, []string{metrics.LabelPipeline, metrics.LabelVertex, metrics.LabelVertexReplicaIndex})
 
diff --git a/pkg/reduce/pbq/store/aligned/wal/options.go b/pkg/reduce/pbq/store/aligned/fs/options.go
similarity index 86%
rename from pkg/reduce/pbq/store/aligned/wal/options.go
rename to pkg/reduce/pbq/store/aligned/fs/options.go
index 4afd0a58..7d5189c6 100644
--- a/pkg/reduce/pbq/store/aligned/wal/options.go
+++ b/pkg/reduce/pbq/store/aligned/fs/options.go
@@ -14,29 +14,29 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package wal
+package fs
 
 import "time"
 
-type Option func(stores *walStores)
+type Option func(stores *fsWAL)
 
 // WithStorePath sets the WAL store path
 func WithStorePath(path string) Option {
-	return func(stores *walStores) {
+	return func(stores *fsWAL) {
 		stores.storePath = path
 	}
 }
 
 // WithMaxBufferSize sets the WAL buffer max size option
 func WithMaxBufferSize(size int64) Option {
-	return func(stores *walStores) {
+	return func(stores *fsWAL) {
 		stores.maxBatchSize = size
 	}
 }
 
 // WithSyncDuration sets the WAL sync duration option
 func WithSyncDuration(maxDuration time.Duration) Option {
-	return func(stores *walStores) {
+	return func(stores *fsWAL) {
 		stores.syncDuration = maxDuration
 	}
 }
diff --git a/pkg/reduce/pbq/store/aligned/wal/segment.go b/pkg/reduce/pbq/store/aligned/fs/wal_segment.go
similarity index 57%
rename from pkg/reduce/pbq/store/aligned/wal/segment.go
rename to pkg/reduce/pbq/store/aligned/fs/wal_segment.go
index f7b88ab1..812830fe 100644
--- a/pkg/reduce/pbq/store/aligned/wal/segment.go
+++ b/pkg/reduce/pbq/store/aligned/fs/wal_segment.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package wal
+package fs
 
 import (
 	"bytes"
@@ -29,6 +29,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/store"
 )
 
 const (
@@ -46,41 +47,125 @@ var (
 // infrequent, meaning a read will only happen during a boot up. WAL will only have one segment since these are
 // relatively short-lived.
 type WAL struct {
-	// fp is the file pointer to the WAL segment
-	fp *os.File
-	// wOffset is the write offset as tracked by the writer
-	wOffset int64
-	// rOffset is the read offset as tracked when reading. Reading only
-	// happens during boostrap. It is highly unlikely to reread the data
-	// once boostrap sequence has been completed.
-	rOffset int64
-	// readUpTo is the read offset at which the reader will stop reading.
-	readUpTo int64
-	// openMode denotes which mode we opened the file in. Only during boot up we will open in read-write mode
-	openMode int
-	// createTime is the timestamp when the WAL segment is created.
-	createTime time.Time
-	// closed indicates whether the file has been closed
-	closed bool
-	// corrupted indicates whether the data of the file has been corrupted
-	corrupted   bool
-	partitionID *partition.ID
-	// prevSyncedWOffset is the write offset that is already synced as tracked by the writer
-	prevSyncedWOffset int64
-	// prevSyncedTime is the time when the last sync was made
-	prevSyncedTime    time.Time
-	walStores         *walStores
+	pipelineName      string
+	vertexName        string
+	replicaIndex      int32
+	maxBatchSize      int64         // maxBatchSize is the maximum size of the batch before we sync the file.
+	syncDuration      time.Duration // syncDuration is the duration after which the writer will sync the file.
+	fp                *os.File      // fp is the file pointer to the WAL segment
+	wOffset           int64         // wOffset is the write offset as tracked by the writer
+	rOffset           int64         // rOffset is the read offset as tracked when reading.
+	readUpTo          int64         // readUpTo is the read offset at which the reader will stop reading.
+	createTime        time.Time     // createTime is the timestamp when the WAL segment is created.
+	closed            bool          // closed indicates whether the file has been closed
+	corrupted         bool          // corrupted indicates whether the data of the file has been corrupted
+	partitionID       *partition.ID // partitionID is the partition ID of the WAL segment
+	prevSyncedWOffset int64         // prevSyncedWOffset is the write offset that is already synced as tracked by the writer
+	prevSyncedTime    time.Time     // prevSyncedTime is the time when the last sync was made
 	numOfUnsyncedMsgs int64
 }
 
+// NewWriteOnlyWAL creates a new WAL instance for write-only. This will be used in happy path where we are only
+// writing to the WAL.
+func NewWriteOnlyWAL(id *partition.ID,
+	filePath string,
+	maxBufferSize int64,
+	syncDuration time.Duration,
+	pipelineName string,
+	vertexName string,
+	replica int32) (store.Store, error) {
+
+	wal := &WAL{
+		pipelineName:      pipelineName,
+		vertexName:        vertexName,
+		replicaIndex:      replica,
+		createTime:        time.Now(),
+		wOffset:           0,
+		rOffset:           0,
+		readUpTo:          0,
+		partitionID:       id,
+		prevSyncedWOffset: 0,
+		prevSyncedTime:    time.Time{},
+		numOfUnsyncedMsgs: 0,
+		maxBatchSize:      maxBufferSize,
+		syncDuration:      syncDuration,
+	}
+
+	// here we are explicitly giving O_WRONLY because we will not be using this to read. Our read is only during
+	// boot up.
+	fp, err := os.OpenFile(filePath, os.O_WRONLY|os.O_CREATE, 0644)
+	if err != nil {
+		return nil, err
+	}
+	wal.fp = fp
+	err = wal.writeWALHeader()
+	if err != nil {
+		return nil, err
+	}
+
+	return wal, nil
+}
+
+// NewReadWriteWAL creates a new WAL instance for read-write. This will be used during boot up where we will be replaying
+// the messages from the WAL and then writing to it.
+func NewReadWriteWAL(filePath string,
+	maxBufferSize int64,
+	syncDuration time.Duration,
+	pipelineName string,
+	vertexName string,
+	replica int32) (store.Store, error) {
+	wal := &WAL{
+		pipelineName:      pipelineName,
+		vertexName:        vertexName,
+		replicaIndex:      replica,
+		createTime:        time.Now(),
+		wOffset:           0,
+		rOffset:           0,
+		readUpTo:          0,
+		prevSyncedWOffset: 0,
+		prevSyncedTime:    time.Time{},
+		numOfUnsyncedMsgs: 0,
+		maxBatchSize:      maxBufferSize,
+		syncDuration:      syncDuration,
+	}
+
+	// here we are explicitly giving O_RDWR because we will be using this to read too. Our read is only during
+	// boot up.
+	fp, err := os.OpenFile(filePath, os.O_RDWR, 0644)
+	if err != nil {
+		return nil, err
+	}
+	wal.fp = fp
+
+	// read the partition ID from the WAL header and set it in the WAL.
+	readPartition, err := wal.readWALHeader()
+	if err != nil {
+		return nil, err
+	}
+	wal.partitionID = readPartition
+
+	// set the read up to the end of the file.
+	stat, err := os.Stat(filePath)
+	if err != nil {
+		return nil, err
+	}
+	wal.readUpTo = stat.Size()
+
+	return wal, nil
+}
+
+func (w *WAL) PartitionID() partition.ID {
+	return *w.partitionID
+}
+
 // writeWALHeader writes the WAL header to the file.
 func (w *WAL) writeWALHeader() (err error) {
 	defer func() {
 		if err != nil {
 			walErrors.With(map[string]string{
-				metrics.LabelPipeline:           w.walStores.pipelineName,
-				metrics.LabelVertex:             w.walStores.vertexName,
-				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
+				metrics.LabelPipeline:           w.pipelineName,
+				metrics.LabelVertex:             w.vertexName,
+				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.replicaIndex)),
 				labelErrorKind:                  "writeWALHeader",
 			}).Inc()
 		}
@@ -126,9 +211,9 @@ func (w *WAL) encodeWALHeader(id *partition.ID) (buf *bytes.Buffer, err error) {
 	defer func() {
 		if err != nil {
 			walErrors.With(map[string]string{
-				metrics.LabelPipeline:           w.walStores.pipelineName,
-				metrics.LabelVertex:             w.walStores.vertexName,
-				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
+				metrics.LabelPipeline:           w.pipelineName,
+				metrics.LabelVertex:             w.vertexName,
+				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.replicaIndex)),
 				labelErrorKind:                  "encodeWALHeader",
 			}).Inc()
 		}
@@ -157,9 +242,9 @@ func (w *WAL) encodeWALMessage(message *isb.ReadMessage) (buf *bytes.Buffer, err
 	defer func() {
 		if err != nil {
 			walErrors.With(map[string]string{
-				metrics.LabelPipeline:           w.walStores.pipelineName,
-				metrics.LabelVertex:             w.walStores.vertexName,
-				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
+				metrics.LabelPipeline:           w.pipelineName,
+				metrics.LabelVertex:             w.vertexName,
+				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.replicaIndex)),
 				labelErrorKind:                  "encodeWALMessage",
 			}).Inc()
 		}
@@ -217,9 +302,9 @@ func (w *WAL) encodeWALMessageHeader(message *isb.ReadMessage, messageLen int64,
 	err = binary.Write(buf, binary.LittleEndian, msgHeader)
 	if err != nil {
 		walErrors.With(map[string]string{
-			metrics.LabelPipeline:           w.walStores.pipelineName,
-			metrics.LabelVertex:             w.walStores.vertexName,
-			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
+			metrics.LabelPipeline:           w.pipelineName,
+			metrics.LabelVertex:             w.vertexName,
+			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.replicaIndex)),
 			labelErrorKind:                  "encodeWALMessageHeader",
 		}).Inc()
 		return nil, err
@@ -233,9 +318,9 @@ func (w *WAL) encodeWALMessageBody(readMsg *isb.ReadMessage) ([]byte, error) {
 	msgBinary, err := readMsg.Message.MarshalBinary()
 	if err != nil {
 		walErrors.With(map[string]string{
-			metrics.LabelPipeline:           w.walStores.pipelineName,
-			metrics.LabelVertex:             w.walStores.vertexName,
-			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
+			metrics.LabelPipeline:           w.pipelineName,
+			metrics.LabelVertex:             w.vertexName,
+			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.replicaIndex)),
 			labelErrorKind:                  "encodeWALMessageBody",
 		}).Inc()
 		return nil, fmt.Errorf("encodeWALMessageBody encountered encode err: %w", err)
@@ -259,9 +344,9 @@ func (w *WAL) Write(message *isb.ReadMessage) (err error) {
 	defer func() {
 		if err != nil {
 			walErrors.With(map[string]string{
-				metrics.LabelPipeline:           w.walStores.pipelineName,
-				metrics.LabelVertex:             w.walStores.vertexName,
-				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
+				metrics.LabelPipeline:           w.pipelineName,
+				metrics.LabelVertex:             w.vertexName,
+				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.replicaIndex)),
 				labelErrorKind:                  "write",
 			}).Inc()
 		}
@@ -269,9 +354,9 @@ func (w *WAL) Write(message *isb.ReadMessage) (err error) {
 	encodeStart := time.Now()
 	entry, err := w.encodeWALMessage(message)
 	entryEncodeLatency.With(map[string]string{
-		metrics.LabelPipeline:           w.walStores.pipelineName,
-		metrics.LabelVertex:             w.walStores.vertexName,
-		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
+		metrics.LabelPipeline:           w.pipelineName,
+		metrics.LabelVertex:             w.vertexName,
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.replicaIndex)),
 	}).Observe(float64(time.Since(encodeStart).Milliseconds()))
 	if err != nil {
 		return err
@@ -280,9 +365,9 @@ func (w *WAL) Write(message *isb.ReadMessage) (err error) {
 	writeStart := time.Now()
 	wrote, err := w.fp.WriteAt(entry.Bytes(), w.wOffset)
 	entryWriteLatency.With(map[string]string{
-		metrics.LabelPipeline:           w.walStores.pipelineName,
-		metrics.LabelVertex:             w.walStores.vertexName,
-		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
+		metrics.LabelPipeline:           w.pipelineName,
+		metrics.LabelVertex:             w.vertexName,
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.replicaIndex)),
 	}).Observe(float64(time.Since(writeStart).Milliseconds()))
 	if wrote != entry.Len() {
 		return fmt.Errorf("expected to write %d, but wrote only %d, %w", entry.Len(), wrote, err)
@@ -295,26 +380,26 @@ func (w *WAL) Write(message *isb.ReadMessage) (err error) {
 	// Only increase the write offset when we successfully write for atomicity.
 	w.wOffset += int64(wrote)
 	entriesBytesCount.With(map[string]string{
-		metrics.LabelPipeline:           w.walStores.pipelineName,
-		metrics.LabelVertex:             w.walStores.vertexName,
-		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
+		metrics.LabelPipeline:           w.pipelineName,
+		metrics.LabelVertex:             w.vertexName,
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.replicaIndex)),
 	}).Add(float64(wrote))
 	entriesCount.With(map[string]string{
-		metrics.LabelPipeline:           w.walStores.pipelineName,
-		metrics.LabelVertex:             w.walStores.vertexName,
-		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
+		metrics.LabelPipeline:           w.pipelineName,
+		metrics.LabelVertex:             w.vertexName,
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.replicaIndex)),
 	}).Inc()
 	currentTime := time.Now()
 
-	if w.wOffset-w.prevSyncedWOffset > w.walStores.maxBatchSize || currentTime.Sub(w.prevSyncedTime) > w.walStores.syncDuration {
+	if w.wOffset-w.prevSyncedWOffset > w.maxBatchSize || currentTime.Sub(w.prevSyncedTime) > w.syncDuration {
 		w.prevSyncedWOffset = w.wOffset
 		w.prevSyncedTime = currentTime
 		fSyncStart := time.Now()
 		err = w.fp.Sync()
 		fileSyncWaitTime.With(map[string]string{
-			metrics.LabelPipeline:           w.walStores.pipelineName,
-			metrics.LabelVertex:             w.walStores.vertexName,
-			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
+			metrics.LabelPipeline:           w.pipelineName,
+			metrics.LabelVertex:             w.vertexName,
+			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.replicaIndex)),
 		}).Observe(float64(time.Since(fSyncStart).Milliseconds()))
 		w.numOfUnsyncedMsgs = 0
 		return err
@@ -327,9 +412,9 @@ func (w *WAL) Close() (err error) {
 	defer func() {
 		if err != nil {
 			walErrors.With(map[string]string{
-				metrics.LabelPipeline:           w.walStores.pipelineName,
-				metrics.LabelVertex:             w.walStores.vertexName,
-				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
+				metrics.LabelPipeline:           w.pipelineName,
+				metrics.LabelVertex:             w.vertexName,
+				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.replicaIndex)),
 				labelErrorKind:                  "close",
 			}).Inc()
 		}
@@ -337,9 +422,9 @@ func (w *WAL) Close() (err error) {
 	start := time.Now()
 	err = w.fp.Sync()
 	fileSyncWaitTime.With(map[string]string{
-		metrics.LabelPipeline:           w.walStores.pipelineName,
-		metrics.LabelVertex:             w.walStores.vertexName,
-		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
+		metrics.LabelPipeline:           w.pipelineName,
+		metrics.LabelVertex:             w.vertexName,
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.replicaIndex)),
 	}).Observe(float64(time.Since(start).Milliseconds()))
 
 	if err != nil {
diff --git a/pkg/reduce/pbq/store/aligned/memory/stores.go b/pkg/reduce/pbq/store/aligned/memory/manager.go
similarity index 80%
rename from pkg/reduce/pbq/store/aligned/memory/stores.go
rename to pkg/reduce/pbq/store/aligned/memory/manager.go
index 41354919..3df6823c 100644
--- a/pkg/reduce/pbq/store/aligned/memory/stores.go
+++ b/pkg/reduce/pbq/store/aligned/memory/manager.go
@@ -23,18 +23,18 @@ import (
 
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq/store/aligned"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/store"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 )
 
 type memoryStores struct {
 	storeSize    int64
-	discoverFunc func(ctx context.Context) ([]partition.ID, error)
+	discoverFunc func(ctx context.Context) ([]store.Store, error)
 	partitions   map[partition.ID]*memoryStore
 	sync.RWMutex
 }
 
-func NewMemoryStores(opts ...Option) aligned.StoreProvider {
+func NewMemoryStores(opts ...Option) store.Manager {
 	s := &memoryStores{
 		storeSize:  100000,
 		partitions: make(map[partition.ID]*memoryStore),
@@ -46,7 +46,7 @@ func NewMemoryStores(opts ...Option) aligned.StoreProvider {
 	return s
 }
 
-func (ms *memoryStores) CreateStore(ctx context.Context, partitionID partition.ID) (aligned.Store, error) {
+func (ms *memoryStores) CreateStore(ctx context.Context, partitionID partition.ID) (store.Store, error) {
 	ms.Lock()
 	defer ms.Unlock()
 	if memStore, ok := ms.partitions[partitionID]; ok {
@@ -65,15 +65,15 @@ func (ms *memoryStores) CreateStore(ctx context.Context, partitionID partition.I
 	return memStore, nil
 }
 
-func (ms *memoryStores) DiscoverPartitions(ctx context.Context) ([]partition.ID, error) {
+func (ms *memoryStores) DiscoverStores(ctx context.Context) ([]store.Store, error) {
 	ms.RLock()
 	defer ms.RUnlock()
 	if ms.discoverFunc == nil {
-		partitionsIds := make([]partition.ID, 0)
-		for key := range ms.partitions {
-			partitionsIds = append(partitionsIds, key)
+		s := make([]store.Store, 0)
+		for _, val := range ms.partitions {
+			s = append(s, val)
 		}
-		return partitionsIds, nil
+		return s, nil
 	}
 	return ms.discoverFunc(ctx)
 }
diff --git a/pkg/reduce/pbq/store/aligned/memory/options.go b/pkg/reduce/pbq/store/aligned/memory/options.go
index 32454aaa..b3102841 100644
--- a/pkg/reduce/pbq/store/aligned/memory/options.go
+++ b/pkg/reduce/pbq/store/aligned/memory/options.go
@@ -19,13 +19,13 @@ package memory
 import (
 	"context"
 
-	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/store"
 )
 
 type Option func(stores *memoryStores)
 
 // WithDiscoverer sets the discover func of memorystores
-func WithDiscoverer(f func(ctx context.Context) ([]partition.ID, error)) Option {
+func WithDiscoverer(f func(ctx context.Context) ([]store.Store, error)) Option {
 	return func(stores *memoryStores) {
 		stores.discoverFunc = f
 	}
diff --git a/pkg/reduce/pbq/store/aligned/memory/store.go b/pkg/reduce/pbq/store/aligned/memory/wal.go
similarity index 68%
rename from pkg/reduce/pbq/store/aligned/memory/store.go
rename to pkg/reduce/pbq/store/aligned/memory/wal.go
index effc2b4b..219b877b 100644
--- a/pkg/reduce/pbq/store/aligned/memory/store.go
+++ b/pkg/reduce/pbq/store/aligned/memory/wal.go
@@ -17,8 +17,6 @@ limitations under the License.
 package memory
 
 import (
-	"math"
-
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/store/aligned"
@@ -37,20 +35,19 @@ type memoryStore struct {
 	partitionID partition.ID
 }
 
-// ReadFromStore will return upto N messages persisted in store
+// Replay will replay all the messages persisted in store
 // this function will be invoked during bootstrap if there is a restart
-func (m *memoryStore) Read(size int64) ([]*isb.ReadMessage, bool, error) {
-	if m.isEmpty() || m.readPos >= m.writePos {
-		m.log.Errorw(aligned.ErrReadStoreEmpty.Error())
-		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 = 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
+func (m *memoryStore) Replay() (<-chan *isb.ReadMessage, <-chan error) {
+	msgChan := make(chan *isb.ReadMessage)
+	errChan := make(chan error)
+	go func() {
+		for _, msg := range m.storage {
+			msgChan <- msg
+		}
+		close(msgChan)
+		close(errChan)
+	}()
+	return msgChan, errChan
 }
 
 // WriteToStore writes a message to store
@@ -75,8 +72,6 @@ func (m *memoryStore) Close() error {
 	return nil
 }
 
-// isEmpty check if there are any records persisted in store
-func (m *memoryStore) isEmpty() bool {
-	// is empty should return true when the store is created and no messages are written
-	return m.writePos == 0
+func (m *memoryStore) PartitionID() partition.ID {
+	return m.partitionID
 }
diff --git a/pkg/reduce/pbq/store/aligned/noop/stores.go b/pkg/reduce/pbq/store/aligned/noop/manager.go
similarity index 77%
rename from pkg/reduce/pbq/store/aligned/noop/stores.go
rename to pkg/reduce/pbq/store/aligned/noop/manager.go
index 8a2dbfa5..cecec6db 100644
--- a/pkg/reduce/pbq/store/aligned/noop/stores.go
+++ b/pkg/reduce/pbq/store/aligned/noop/manager.go
@@ -20,22 +20,22 @@ import (
 	"context"
 
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq/store/aligned"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/store"
 )
 
 type noopStores struct {
 }
 
-func NewNoopStores() aligned.StoreProvider {
+func NewNoopStores() store.Manager {
 	return &noopStores{}
 }
 
-func (ns *noopStores) CreateStore(ctx context.Context, partitionID partition.ID) (aligned.Store, error) {
+func (ns *noopStores) CreateStore(ctx context.Context, partitionID partition.ID) (store.Store, error) {
 	return &PBQNoOpStore{}, nil
 }
 
-func (ns *noopStores) DiscoverPartitions(ctx context.Context) ([]partition.ID, error) {
-	return []partition.ID{}, nil
+func (ns *noopStores) DiscoverStores(ctx context.Context) ([]store.Store, error) {
+	return []store.Store{}, nil
 }
 
 func (ns *noopStores) DeleteStore(partitionID partition.ID) error {
diff --git a/pkg/reduce/pbq/store/aligned/noop/store.go b/pkg/reduce/pbq/store/aligned/noop/wal.go
similarity index 75%
rename from pkg/reduce/pbq/store/aligned/noop/store.go
rename to pkg/reduce/pbq/store/aligned/noop/wal.go
index 271036d1..4ffa767d 100644
--- a/pkg/reduce/pbq/store/aligned/noop/store.go
+++ b/pkg/reduce/pbq/store/aligned/noop/wal.go
@@ -18,21 +18,22 @@ package noop
 
 import (
 	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq/store/aligned"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/store"
 )
 
 // PBQNoOpStore is a no-op pbq store which does not do any operation but can be safely invoked.
 type PBQNoOpStore struct {
 }
 
-var _ aligned.Store = (*PBQNoOpStore)(nil)
+var _ store.Store = (*PBQNoOpStore)(nil)
 
 func NewPBQNoOpStore() (*PBQNoOpStore, error) {
 	return &PBQNoOpStore{}, nil
 }
 
-func (p *PBQNoOpStore) Read(size int64) ([]*isb.ReadMessage, bool, error) {
-	return []*isb.ReadMessage{}, true, nil
+func (p *PBQNoOpStore) Replay() (<-chan *isb.ReadMessage, <-chan error) {
+	return nil, nil
 }
 
 func (p *PBQNoOpStore) Write(msg *isb.ReadMessage) error {
@@ -42,3 +43,7 @@ func (p *PBQNoOpStore) Write(msg *isb.ReadMessage) error {
 func (p *PBQNoOpStore) Close() error {
 	return nil
 }
+
+func (p *PBQNoOpStore) PartitionID() partition.ID {
+	return partition.ID{}
+}
diff --git a/pkg/reduce/pbq/store/aligned/wal/stores.go b/pkg/reduce/pbq/store/aligned/wal/stores.go
deleted file mode 100644
index 18a51a2b..00000000
--- a/pkg/reduce/pbq/store/aligned/wal/stores.go
+++ /dev/null
@@ -1,250 +0,0 @@
-/*
-Copyright 2022 The Numaproj Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package wal
-
-import (
-	"context"
-	"fmt"
-	"os"
-	"path/filepath"
-	"strconv"
-	"strings"
-	"time"
-
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/metrics"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq/store/aligned"
-)
-
-type walStores struct {
-	storePath string
-	// maxBufferSize max size of batch before it's flushed to store
-	maxBatchSize int64
-	// syncDuration timeout to sync to store
-	syncDuration time.Duration
-	pipelineName string
-	vertexName   string
-	replicaIndex int32
-}
-
-func NewWALStores(vertexInstance *dfv1.VertexInstance, opts ...Option) aligned.StoreProvider {
-	s := &walStores{
-		storePath:    dfv1.DefaultStorePath,
-		maxBatchSize: dfv1.DefaultStoreMaxBufferSize,
-		syncDuration: dfv1.DefaultStoreSyncDuration,
-		pipelineName: vertexInstance.Vertex.Spec.PipelineName,
-		vertexName:   vertexInstance.Vertex.Spec.AbstractVertex.Name,
-		replicaIndex: vertexInstance.Replica,
-	}
-	for _, o := range opts {
-		o(s)
-	}
-	return s
-}
-
-func (ws *walStores) CreateStore(_ context.Context, partitionID partition.ID) (aligned.Store, error) {
-	// Create wal dir if not exist
-	var err error
-	if _, err = os.Stat(ws.storePath); os.IsNotExist(err) {
-		err = os.Mkdir(ws.storePath, 0755)
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	// let's open or create, initialize and return a new WAL
-	wal, err := ws.openOrCreateWAL(&partitionID)
-
-	return wal, err
-}
-
-// openOrCreateWAL open or creates a new WAL segment.
-func (ws *walStores) openOrCreateWAL(id *partition.ID) (*WAL, error) {
-	var err error
-
-	defer func() {
-		// increment active WAL count only if we are able to successfully create/open one.
-		if err == nil {
-			activeFilesCount.With(map[string]string{
-				metrics.LabelPipeline:           ws.pipelineName,
-				metrics.LabelVertex:             ws.vertexName,
-				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(ws.replicaIndex)),
-			}).Inc()
-		}
-	}()
-
-	filePath := getSegmentFilePath(id, ws.storePath)
-	stat, err := os.Stat(filePath)
-
-	var fp *os.File
-	var wal *WAL
-	if os.IsNotExist(err) {
-		// here we are explicitly giving O_WRONLY because we will not be using this to read. Our read is only during
-		// boot up.
-		fp, err = os.OpenFile(filePath, os.O_WRONLY|os.O_CREATE, 0644)
-		if err != nil {
-			return nil, err
-		}
-		// we are interested only in the number of new files created
-		filesCount.With(map[string]string{
-			metrics.LabelPipeline:           ws.pipelineName,
-			metrics.LabelVertex:             ws.vertexName,
-			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(ws.replicaIndex)),
-		}).Inc()
-
-		wal = &WAL{
-			fp:                fp,
-			openMode:          os.O_WRONLY,
-			createTime:        time.Now(),
-			wOffset:           0,
-			rOffset:           0,
-			readUpTo:          0,
-			partitionID:       id,
-			prevSyncedWOffset: 0,
-			prevSyncedTime:    time.Time{},
-			walStores:         ws,
-			numOfUnsyncedMsgs: 0,
-		}
-
-		err = wal.writeWALHeader()
-		if err != nil {
-			return nil, err
-		}
-	} else if err != nil {
-		return nil, err
-	} else {
-		// here we are explicitly giving O_RDWR because we will be using this to read too. Our read is only during
-		// bootstrap.
-		fp, err = os.OpenFile(filePath, os.O_RDWR, stat.Mode())
-		if err != nil {
-			return nil, err
-		}
-		wal = &WAL{
-			fp:                fp,
-			openMode:          os.O_RDWR,
-			createTime:        time.Now(),
-			wOffset:           0,
-			rOffset:           0,
-			readUpTo:          stat.Size(),
-			partitionID:       id,
-			prevSyncedWOffset: 0,
-			prevSyncedTime:    time.Time{},
-			walStores:         ws,
-			numOfUnsyncedMsgs: 0,
-		}
-		readPartition, err := wal.readWALHeader()
-		if err != nil {
-			return nil, err
-		}
-		if id.Slot != readPartition.Slot {
-			return nil, fmt.Errorf("expected partition key %s, but got %s", id.Slot, readPartition.Slot)
-		}
-	}
-
-	return wal, err
-}
-
-func (ws *walStores) DiscoverPartitions(_ context.Context) ([]partition.ID, error) {
-	files, err := os.ReadDir(ws.storePath)
-	if os.IsNotExist(err) {
-		return []partition.ID{}, nil
-	} else if err != nil {
-		return nil, err
-	}
-	partitions := make([]partition.ID, 0)
-
-	for _, f := range files {
-		if strings.HasPrefix(f.Name(), SegmentPrefix) && !f.IsDir() {
-			filePath := filepath.Join(ws.storePath, f.Name())
-			wal, err := ws.openWAL(filePath)
-			if err != nil {
-				return nil, err
-			}
-			partitions = append(partitions, *wal.partitionID)
-		}
-	}
-
-	return partitions, nil
-}
-
-// openWAL returns a WAL if present
-func (ws *walStores) openWAL(filePath string) (*WAL, error) {
-	stat, err := os.Stat(filePath)
-	if err != nil {
-		if os.IsNotExist(err) {
-			return nil, nil
-		}
-		return nil, fmt.Errorf("filed to open WAL file %q, %w", filePath, err)
-	}
-
-	// here we are explicitly giving O_RDWR because we will be using this to read too. Our read is only during
-	// boot up.
-	fp, err := os.OpenFile(filePath, os.O_RDWR, stat.Mode())
-	if err != nil {
-		return nil, err
-	}
-
-	w := &WAL{
-		fp:        fp,
-		openMode:  os.O_RDWR,
-		walStores: ws,
-	}
-
-	w.partitionID, err = w.readWALHeader()
-
-	return w, err
-}
-
-func (ws *walStores) DeleteStore(partitionID partition.ID) error {
-	var err error
-	defer func() {
-		if err != nil {
-			walErrors.With(map[string]string{
-				metrics.LabelPipeline:           ws.pipelineName,
-				metrics.LabelVertex:             ws.vertexName,
-				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(ws.replicaIndex)),
-				labelErrorKind:                  "gc",
-			}).Inc()
-		}
-	}()
-
-	filePath := getSegmentFilePath(&partitionID, ws.storePath)
-	_, err = os.Stat(filePath)
-
-	if err != nil {
-		return err
-	}
-
-	start := time.Now()
-	// an open file can also be deleted
-	err = os.Remove(filePath)
-
-	if err == nil {
-		garbageCollectingTime.With(map[string]string{
-			metrics.LabelPipeline:           ws.pipelineName,
-			metrics.LabelVertex:             ws.vertexName,
-			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(ws.replicaIndex)),
-		}).Observe(float64(time.Since(start).Microseconds()))
-		activeFilesCount.With(map[string]string{
-			metrics.LabelPipeline:           ws.pipelineName,
-			metrics.LabelVertex:             ws.vertexName,
-			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(ws.replicaIndex)),
-		}).Dec()
-	}
-	return err
-}
diff --git a/pkg/reduce/pbq/store/aligned/interfaces.go b/pkg/reduce/pbq/store/interfaces.go
similarity index 66%
rename from pkg/reduce/pbq/store/aligned/interfaces.go
rename to pkg/reduce/pbq/store/interfaces.go
index 336693d9..cc459606 100644
--- a/pkg/reduce/pbq/store/aligned/interfaces.go
+++ b/pkg/reduce/pbq/store/interfaces.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package aligned
+package store
 
 import (
 	"context"
@@ -25,21 +25,24 @@ 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, it also returns
-	// a boolean flag to indicate if the end of file has been reached.
-	Read(size int64) ([]*isb.ReadMessage, bool, error)
+	// Replay to replay persisted messages during startup
+	// returns a channel to read messages and a channel to read errors
+	Replay() (<-chan *isb.ReadMessage, <-chan error)
 	// Write writes message to persistence store
 	Write(msg *isb.ReadMessage) error
+	// PartitionID returns the partition ID of the store
+	PartitionID() partition.ID
 	// Close closes store
 	Close() error
 }
 
-// StoreProvider defines the functions for store implementation
-type StoreProvider interface {
+// Manager defines the interface to manage the stores.
+type Manager interface {
 	// CreateStore returns a new store instance.
 	CreateStore(context.Context, partition.ID) (Store, error)
-	// DiscoverPartitions discovers all the managed partitions.
-	DiscoverPartitions(context.Context) ([]partition.ID, error)
+	// DiscoverStores discovers all the existing stores.
+	// This is used to recover from a crash and replay all the messages from the store.
+	DiscoverStores(context.Context) ([]Store, error)
 	// DeleteStore deletes the store
 	DeleteStore(partition.ID) error
 }
diff --git a/pkg/udf/reduce_udf.go b/pkg/udf/reduce_udf.go
index 5f896905..657e3ea8 100644
--- a/pkg/udf/reduce_udf.go
+++ b/pkg/udf/reduce_udf.go
@@ -31,7 +31,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/reduce"
 	"github.com/numaproj/numaflow/pkg/reduce/applier"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq"
-	wal2 "github.com/numaproj/numaflow/pkg/reduce/pbq/store/aligned/wal"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/store/aligned/fs"
 	"github.com/numaproj/numaflow/pkg/reduce/pnf"
 	"github.com/numaproj/numaflow/pkg/sdkclient"
 	"github.com/numaproj/numaflow/pkg/sdkclient/reducer"
@@ -286,9 +286,9 @@ func (u *ReduceUDFProcessor) Start(ctx context.Context) error {
 		defer func() { _ = shutdown(context.Background()) }()
 	}
 
-	storeProvider := wal2.NewWALStores(u.VertexInstance, wal2.WithStorePath(dfv1.DefaultStorePath), wal2.WithMaxBufferSize(dfv1.DefaultStoreMaxBufferSize), wal2.WithSyncDuration(dfv1.DefaultStoreSyncDuration))
+	storeManager := fs.NewFSManager(u.VertexInstance, fs.WithStorePath(dfv1.DefaultStorePath), fs.WithMaxBufferSize(dfv1.DefaultStoreMaxBufferSize), fs.WithSyncDuration(dfv1.DefaultStoreSyncDuration))
 
-	pbqManager, err := pbq.NewManager(ctx, u.VertexInstance.Vertex.Spec.Name, u.VertexInstance.Vertex.Spec.PipelineName, u.VertexInstance.Replica, storeProvider, windower.Type())
+	pbqManager, err := pbq.NewManager(ctx, u.VertexInstance.Vertex.Spec.Name, u.VertexInstance.Vertex.Spec.PipelineName, u.VertexInstance.Replica, storeManager, windower.Type())
 	if err != nil {
 		log.Errorw("Failed to create pbq manager", zap.Error(err))
 		return fmt.Errorf("failed to create pbq manager, %w", err)
@@ -307,7 +307,7 @@ func (u *ReduceUDFProcessor) Start(ctx context.Context) error {
 	op := pnf.NewPnFManager(ctx, u.VertexInstance, udfApplier, writers, pbqManager, conditionalForwarder, publishWatermark, idleManager, windower)
 
 	// for reduce, we read only from one partition
-	dataForwarder, err := reduce.NewDataForward(ctx, u.VertexInstance, readers[0], writers, pbqManager, conditionalForwarder, fetchWatermark, publishWatermark, windower, idleManager, op, opts...)
+	dataForwarder, err := reduce.NewDataForward(ctx, u.VertexInstance, readers[0], writers, pbqManager, storeManager, conditionalForwarder, fetchWatermark, publishWatermark, windower, idleManager, op, opts...)
 	if err != nil {
 		return fmt.Errorf("failed get a new DataForward, %w", err)
 	}


Test patch:
diff --git a/pkg/reduce/data_forward_test.go b/pkg/reduce/data_forward_test.go
index 30e80f95..275aa1e9 100644
--- a/pkg/reduce/data_forward_test.go
+++ b/pkg/reduce/data_forward_test.go
@@ -34,6 +34,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/store"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/store/aligned/memory"
 	"github.com/numaproj/numaflow/pkg/reduce/pnf"
 	"github.com/numaproj/numaflow/pkg/shared/kvs"
@@ -386,11 +387,10 @@ func TestDataForward_StartWithNoOpWM(t *testing.T) {
 		toVertexName: {to},
 	}
 
-	var err error
-	var pbqManager *pbq.Manager
-
+	// create store manager
+	storeManager := memory.NewMemoryStores(memory.WithStoreSize(100))
 	// create pbqManager
-	pbqManager, err = pbq.NewManager(child, "reduce", pipelineName, 0, memory.NewMemoryStores(memory.WithStoreSize(100)),
+	pbqManager, err := pbq.NewManager(child, "reduce", pipelineName, 0, storeManager,
 		window.Aligned, pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -412,7 +412,7 @@ func TestDataForward_StartWithNoOpWM(t *testing.T) {
 	op := pnf.NewPnFManager(child, keyedVertex, CounterReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publisher, idleManager, windower)
 
 	var reduceDataForwarder *DataForward
-	reduceDataForwarder, err = NewDataForward(child, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, wmpublisher, publisher,
+	reduceDataForwarder, err = NewDataForward(child, keyedVertex, fromBuffer, toBuffer, pbqManager, storeManager, CounterReduceTest{}, wmpublisher, publisher,
 		windower, idleManager, op, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
@@ -476,9 +476,12 @@ func TestReduceDataForward_IdleWM(t *testing.T) {
 		toVertexName: {toBuffer},
 	}
 
+	// create store manager
+	storeManager := memory.NewMemoryStores(memory.WithStoreSize(1000))
+
 	// create pbq manager
 	var pbqManager *pbq.Manager
-	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, memory.NewMemoryStores(memory.WithStoreSize(1000)),
+	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, storeManager,
 		window.Aligned, pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -505,7 +508,7 @@ func TestReduceDataForward_IdleWM(t *testing.T) {
 	op := pnf.NewPnFManager(ctx, keyedVertex, CounterReduceTest{}, toBuffers, pbqManager, CounterReduceTest{}, publisherMap, idleManager, windower)
 
 	var reduceDataForward *DataForward
-	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffers, pbqManager, CounterReduceTest{}, f, publisherMap,
+	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffers, pbqManager, storeManager, CounterReduceTest{}, f, publisherMap,
 		windower, idleManager, op, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
@@ -688,9 +691,12 @@ func TestReduceDataForward_Count(t *testing.T) {
 		toVertexName: {buffer},
 	}
 
+	// create store manager
+	storeManager := memory.NewMemoryStores(memory.WithStoreSize(1000))
+
 	// create pbq manager
 	var pbqManager *pbq.Manager
-	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, memory.NewMemoryStores(memory.WithStoreSize(1000)),
+	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, storeManager,
 		window.Aligned, pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -711,7 +717,7 @@ func TestReduceDataForward_Count(t *testing.T) {
 	op := pnf.NewPnFManager(ctx, keyedVertex, CounterReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publisherMap, idleManager, windower)
 
 	var reduceDataForward *DataForward
-	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publisherMap,
+	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, storeManager, CounterReduceTest{}, f, publisherMap,
 		windower, idleManager, op, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
@@ -769,9 +775,12 @@ func TestReduceDataForward_AllowedLatencyCount(t *testing.T) {
 		toVertexName: {buffer},
 	}
 
+	// create store manager
+	storeManager := memory.NewMemoryStores(memory.WithStoreSize(1000))
+
 	// create pbq manager
 	var pbqManager *pbq.Manager
-	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, memory.NewMemoryStores(memory.WithStoreSize(1000)),
+	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, storeManager,
 		window.Aligned, pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -794,7 +803,7 @@ func TestReduceDataForward_AllowedLatencyCount(t *testing.T) {
 
 	var reduceDataForward *DataForward
 	allowedLatency := 1000
-	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publisherMap,
+	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, storeManager, CounterReduceTest{}, f, publisherMap,
 		windower, idleManager, op, WithReadBatchSize(int64(batchSize)),
 		WithAllowedLateness(time.Duration(allowedLatency)*time.Millisecond),
 	)
@@ -854,9 +863,12 @@ func TestReduceDataForward_Sum(t *testing.T) {
 		toVertexName: {buffer},
 	}
 
+	// create store manager
+	storeManager := memory.NewMemoryStores(memory.WithStoreSize(1000))
+
 	// create pbq manager
 	var pbqManager *pbq.Manager
-	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, memory.NewMemoryStores(memory.WithStoreSize(1000)),
+	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, storeManager,
 		window.Aligned, pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -877,7 +889,7 @@ func TestReduceDataForward_Sum(t *testing.T) {
 	op := pnf.NewPnFManager(ctx, keyedVertex, SumReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publishersMap, idleManager, windower)
 
 	var reduceDataForward *DataForward
-	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publishersMap,
+	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, storeManager, CounterReduceTest{}, f, publishersMap,
 		windower, idleManager, op, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
@@ -936,9 +948,12 @@ func TestReduceDataForward_Max(t *testing.T) {
 		toVertexName: {buffer},
 	}
 
+	// create store manager
+	storeManager := memory.NewMemoryStores(memory.WithStoreSize(1000))
+
 	// create pbq manager
 	var pbqManager *pbq.Manager
-	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, memory.NewMemoryStores(memory.WithStoreSize(1000)),
+	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, storeManager,
 		window.Aligned, pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -959,7 +974,7 @@ func TestReduceDataForward_Max(t *testing.T) {
 	op := pnf.NewPnFManager(ctx, keyedVertex, MaxReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publishersMap, idleManager, windower)
 
 	var reduceDataForward *DataForward
-	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publishersMap,
+	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, storeManager, CounterReduceTest{}, f, publishersMap,
 		windower, idleManager, op, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
@@ -1018,9 +1033,12 @@ func TestReduceDataForward_FixedSumWithDifferentKeys(t *testing.T) {
 		toVertexName: {buffer},
 	}
 
+	// create store manager
+	storeManager := memory.NewMemoryStores(memory.WithStoreSize(1000))
+
 	// create pbq manager
 	var pbqManager *pbq.Manager
-	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, memory.NewMemoryStores(memory.WithStoreSize(1000)),
+	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, storeManager,
 		window.Aligned, pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -1042,7 +1060,7 @@ func TestReduceDataForward_FixedSumWithDifferentKeys(t *testing.T) {
 	op := pnf.NewPnFManager(ctx, keyedVertex, SumReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publishersMap, idleManager, windower)
 
 	var reduceDataForward *DataForward
-	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publishersMap,
+	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, storeManager, CounterReduceTest{}, f, publishersMap,
 		windower, idleManager, op, WithReadBatchSize(10))
 
 	assert.NoError(t, err)
@@ -1121,9 +1139,12 @@ func TestReduceDataForward_SumWithDifferentKeys(t *testing.T) {
 		toVertexName: {buffer},
 	}
 
+	// create store manager
+	storeManager := memory.NewMemoryStores(memory.WithStoreSize(1000))
+
 	// create pbq manager
 	var pbqManager *pbq.Manager
-	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, memory.NewMemoryStores(memory.WithStoreSize(1000)),
+	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, storeManager,
 		window.Aligned, pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -1145,7 +1166,7 @@ func TestReduceDataForward_SumWithDifferentKeys(t *testing.T) {
 	op := pnf.NewPnFManager(ctx, keyedVertex, SessionSumReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publishersMap, idleManager, windower)
 
 	var reduceDataForward *DataForward
-	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, SessionSumReduceTest{}, f, publishersMap,
+	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, storeManager, SessionSumReduceTest{}, f, publishersMap,
 		windower, idleManager, op, WithReadBatchSize(10))
 
 	assert.NoError(t, err)
@@ -1222,9 +1243,12 @@ func TestReduceDataForward_NonKeyed(t *testing.T) {
 		toVertexName: {buffer},
 	}
 
+	// create store manager
+	storeManager := memory.NewMemoryStores(memory.WithStoreSize(1000))
+
 	// create pbq manager
 	var pbqManager *pbq.Manager
-	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, memory.NewMemoryStores(memory.WithStoreSize(1000)),
+	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, storeManager,
 		window.Aligned, pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -1246,7 +1270,7 @@ func TestReduceDataForward_NonKeyed(t *testing.T) {
 	op := pnf.NewPnFManager(ctx, keyedVertex, SumReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publishersMap, idleManager, windower)
 
 	var reduceDataForward *DataForward
-	reduceDataForward, err = NewDataForward(ctx, nonKeyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publishersMap,
+	reduceDataForward, err = NewDataForward(ctx, nonKeyedVertex, fromBuffer, toBuffer, pbqManager, storeManager, CounterReduceTest{}, f, publishersMap,
 		windower, idleManager, op, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
@@ -1309,12 +1333,12 @@ func TestDataForward_WithContextClose(t *testing.T) {
 		toVertexName: {buffer},
 	}
 
-	// create a store provider
-	storeProvider := memory.NewMemoryStores(memory.WithStoreSize(1000))
+	// create store manager
+	storeManager := memory.NewMemoryStores(memory.WithStoreSize(100))
 
 	// create pbq manager
 	var pbqManager *pbq.Manager
-	pbqManager, err = pbq.NewManager(cctx, "reduce", pipelineName, 0, storeProvider,
+	pbqManager, err = pbq.NewManager(cctx, "reduce", pipelineName, 0, storeManager,
 		window.Aligned, pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -1336,7 +1360,7 @@ func TestDataForward_WithContextClose(t *testing.T) {
 	op := pnf.NewPnFManager(ctx, keyedVertex, SumReduceTest{}, toBuffer, pbqManager, CounterReduceTest{}, publishersMap, idleManager, windower)
 
 	var reduceDataForward *DataForward
-	reduceDataForward, err = NewDataForward(cctx, keyedVertex, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, publishersMap,
+	reduceDataForward, err = NewDataForward(cctx, keyedVertex, fromBuffer, toBuffer, pbqManager, storeManager, CounterReduceTest{}, f, publishersMap,
 		windower, idleManager, op, WithReadBatchSize(1))
 	assert.NoError(t, err)
 
@@ -1360,11 +1384,11 @@ func TestDataForward_WithContextClose(t *testing.T) {
 		}
 	}
 
-	var discoveredPartitions []partition.ID
+	var discoveredStores []store.Store
 	for {
-		discoveredPartitions, _ = storeProvider.DiscoverPartitions(ctx)
+		discoveredStores, _ = storeManager.DiscoverStores(ctx)
 
-		if len(discoveredPartitions) == 1 {
+		if len(discoveredStores) == 1 {
 			break
 		}
 		select {
@@ -1377,7 +1401,7 @@ func TestDataForward_WithContextClose(t *testing.T) {
 	}
 
 	// even though we have 2 different keys
-	assert.Len(t, discoveredPartitions, 1)
+	assert.Len(t, discoveredStores, 1)
 
 }
 
@@ -1406,9 +1430,12 @@ func TestReduceDataForward_SumMultiPartitions(t *testing.T) {
 		toVertexName: {buffer1, buffer2},
 	}
 
+	// create store manager
+	storeManager := memory.NewMemoryStores(memory.WithStoreSize(1000))
+
 	// create pbq manager
 	var pbqManager *pbq.Manager
-	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, memory.NewMemoryStores(memory.WithStoreSize(1000)),
+	pbqManager, err = pbq.NewManager(ctx, "reduce", pipelineName, 0, storeManager,
 		window.Aligned, pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -1430,7 +1457,7 @@ func TestReduceDataForward_SumMultiPartitions(t *testing.T) {
 	op := pnf.NewPnFManager(ctx, keyedVertex, SumReduceTest{}, toBuffer, pbqManager, &myForwardTestRoundRobin{}, publishersMap, idleManager, windower)
 
 	var reduceDataForward *DataForward
-	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, &myForwardTestRoundRobin{}, f, publishersMap,
+	reduceDataForward, err = NewDataForward(ctx, keyedVertex, fromBuffer, toBuffer, pbqManager, storeManager, &myForwardTestRoundRobin{}, f, publishersMap,
 		windower, idleManager, op, WithReadBatchSize(10))
 
 	assert.NoError(t, err)
diff --git a/pkg/reduce/pbq/pbq_test.go b/pkg/reduce/pbq/pbq_test.go
index e25ba588..f7d790c9 100644
--- a/pkg/reduce/pbq/pbq_test.go
+++ b/pkg/reduce/pbq/pbq_test.go
@@ -81,7 +81,7 @@ func TestPBQ_ReadWrite(t *testing.T) {
 	}()
 
 	for _, msg := range writeRequests {
-		err := pq.Write(ctx, &msg)
+		err := pq.Write(ctx, &msg, true)
 		assert.NoError(t, err)
 	}
 	pq.CloseOfBook()
@@ -148,7 +148,7 @@ func Test_PBQReadWithCanceledContext(t *testing.T) {
 	}()
 
 	for _, msg := range windowRequests {
-		err := pq.Write(ctx, &msg)
+		err := pq.Write(ctx, &msg, true)
 		assert.NoError(t, err)
 	}
 
@@ -189,7 +189,7 @@ func TestPBQ_WriteWithStoreFull(t *testing.T) {
 	assert.NoError(t, err)
 
 	for _, req := range windowRequests {
-		err = pq.Write(ctx, &req)
+		err = pq.Write(ctx, &req, true)
 	}
 	pq.CloseOfBook()
 
diff --git a/pkg/reduce/pbq/pbqmanager_test.go b/pkg/reduce/pbq/pbqmanager_test.go
index 530d337c..b1fab18d 100644
--- a/pkg/reduce/pbq/pbqmanager_test.go
+++ b/pkg/reduce/pbq/pbqmanager_test.go
@@ -139,7 +139,7 @@ func TestPBQFlow(t *testing.T) {
 
 	readRequests := testutils.BuildTestWindowRequests(int64(msgsCount), time.Now(), window.Append)
 	for _, msg := range readRequests {
-		err := pq.Write(ctx, &msg)
+		err := pq.Write(ctx, &msg, true)
 		assert.NoError(t, err)
 	}
 
@@ -196,10 +196,9 @@ func TestPBQFlowWithNoOpStore(t *testing.T) {
 
 	requests := testutils.BuildTestWindowRequests(int64(msgsCount), time.Now(), window.Append)
 	for _, msg := range requests {
-		err := pq.Write(ctx, &msg)
+		err := pq.Write(ctx, &msg, true)
 		assert.NoError(t, err)
 	}
-	pbqManager.Replay(ctx)
 
 	pq.CloseOfBook()
 	assert.NoError(t, err)
@@ -209,87 +208,3 @@ func TestPBQFlowWithNoOpStore(t *testing.T) {
 	// no of windowRequests will be equal to no of produced messages
 	assert.Len(t, windowRequests, msgsCount)
 }
-
-func TestManager_Replay(t *testing.T) {
-	size := int64(100)
-
-	ctx := context.Background()
-	pbqManager, err := NewManager(ctx, "reduce", "test-pipeline", 0, memory.NewMemoryStores(memory.WithStoreSize(size)), window.Aligned, 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),
-		Slot:  "slot-1",
-	}
-
-	var pq ReadWriteCloser
-	pq, err = pbqManager.CreateNewPBQ(ctx, testPartition)
-	assert.NoError(t, err)
-	var wg sync.WaitGroup
-	wg.Add(1)
-
-	var readRequests []*window.TimedWindowRequest
-	// go routine which reads the requests from pbq
-	go func() {
-	readLoop:
-		for {
-			select {
-			case msg, ok := <-pq.ReadCh():
-				if msg != nil {
-					readRequests = append(readRequests, msg)
-				}
-				if !ok {
-					break readLoop
-				}
-			case <-ctx.Done():
-				break readLoop
-			}
-		}
-		wg.Done()
-	}()
-
-	// write 50 requests to pbq
-	msgsCount := 50
-	writeRequests := testutils.BuildTestWindowRequests(int64(msgsCount), time.Now(), window.Append)
-
-	for _, msg := range writeRequests {
-		err := pq.Write(ctx, &msg)
-		assert.NoError(t, err)
-	}
-
-	// after writing, replay records from the store using Manager
-	pbqManager.Replay(ctx)
-
-	pq.CloseOfBook()
-	wg.Wait()
-	// number of read requests should be twice compared to number of requests produced
-	// since we replayed the messages which are persisted in store
-	assert.Len(t, readRequests, 2*msgsCount)
-}
-
-func TestManager_StartUp(t *testing.T) {
-	size := int64(100)
-
-	pID1 := partition.ID{
-		Start: time.Now(),
-		End:   time.Now(),
-		Slot:  "test-partition-1",
-	}
-
-	pID2 := partition.ID{
-		Start: time.Now(),
-		End:   time.Now(),
-		Slot:  "test-partition-2",
-	}
-	dp := func(context.Context) ([]partition.ID, error) {
-		return []partition.ID{pID1,
-			pID2}, nil
-	}
-	stores := memory.NewMemoryStores(memory.WithStoreSize(size), memory.WithDiscoverer(dp))
-	ctx := context.Background()
-	pbqManager, err := NewManager(ctx, "reduce", "test-pipeline", 0, stores, window.Aligned, WithReadTimeout(1*time.Second), WithChannelBufferSize(10), WithReadBatchSize(10))
-	assert.NoError(t, err)
-	ids, err := pbqManager.GetExistingPartitions(ctx)
-	assert.NoError(t, err)
-	assert.Len(t, ids, 2)
-}
diff --git a/pkg/reduce/pbq/store/aligned/wal/stores_test.go b/pkg/reduce/pbq/store/aligned/fs/manager_test.go
similarity index 83%
rename from pkg/reduce/pbq/store/aligned/wal/stores_test.go
rename to pkg/reduce/pbq/store/aligned/fs/manager_test.go
index 41d69291..918d8472 100644
--- a/pkg/reduce/pbq/store/aligned/wal/stores_test.go
+++ b/pkg/reduce/pbq/store/aligned/fs/manager_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package wal
+package fs
 
 import (
 	"context"
@@ -60,26 +60,25 @@ func TestWalStores(t *testing.T) {
 	}
 
 	tmp := t.TempDir()
-	storeProvider := NewWALStores(vi, WithStorePath(tmp))
+	storeProvider := NewFSManager(vi, WithStorePath(tmp))
 
 	for _, partitionID := range partitionIds {
 		_, err = storeProvider.CreateStore(ctx, partitionID)
 		assert.NoError(t, err)
 	}
 
-	var discoveredPartitions []partition.ID
-	discoveredPartitions, err = storeProvider.DiscoverPartitions(ctx)
+	discoverStores, err := storeProvider.DiscoverStores(ctx)
 	assert.NoError(t, err)
 
-	assert.Len(t, discoveredPartitions, len(partitionIds))
+	assert.Len(t, discoverStores, len(partitionIds))
 
 	for _, partitionID := range partitionIds {
 		err = storeProvider.DeleteStore(partitionID)
 		assert.NoError(t, err)
 	}
 
-	discoveredPartitions, err = storeProvider.DiscoverPartitions(ctx)
+	discoverStores, err = storeProvider.DiscoverStores(ctx)
 	assert.NoError(t, err)
 
-	assert.Len(t, discoveredPartitions, 0)
+	assert.Len(t, discoverStores, 0)
 }
diff --git a/pkg/reduce/pbq/store/aligned/wal/segment_test.go b/pkg/reduce/pbq/store/aligned/fs/wal_segment_test.go
similarity index 80%
rename from pkg/reduce/pbq/store/aligned/wal/segment_test.go
rename to pkg/reduce/pbq/store/aligned/fs/wal_segment_test.go
index 495f99a4..6b963bed 100644
--- a/pkg/reduce/pbq/store/aligned/wal/segment_test.go
+++ b/pkg/reduce/pbq/store/aligned/fs/wal_segment_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package wal
+package fs
 
 import (
 	"bytes"
@@ -29,6 +29,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/store"
 )
 
 var vi = &dfv1.VertexInstance{
@@ -50,7 +51,7 @@ func Test_writeReadHeader(t *testing.T) {
 	}
 
 	tmp := t.TempDir()
-	stores := NewWALStores(vi, WithStorePath(tmp))
+	stores := NewFSManager(vi, WithStorePath(tmp))
 	store, err := stores.CreateStore(context.Background(), id)
 	assert.NoError(t, err)
 	wal := store.(*WAL)
@@ -63,14 +64,14 @@ func Test_writeReadHeader(t *testing.T) {
 	fmt.Println(fName)
 	assert.NoError(t, err)
 
-	openWAL, err := wal.walStores.openWAL(fName)
+	openWAL, err := NewWriteOnlyWAL(&id, fName, dfv1.DefaultStoreMaxBufferSize, dfv1.DefaultStoreSyncDuration, "testPipeline", "testVertex", 0)
 	assert.NoError(t, err)
 	// we have already read the header in OpenWAL
-	_, err = openWAL.readWALHeader()
+	_, err = openWAL.(*WAL).readWALHeader()
 	assert.Error(t, err)
 
 	// compare the original ID with read ID
-	assert.Equal(t, id, *openWAL.partitionID)
+	assert.Equal(t, id, *openWAL.(*WAL).partitionID)
 
 	err = openWAL.Close()
 	assert.NoError(t, err)
@@ -105,7 +106,7 @@ func Test_encodeDecodeHeader(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			tmp := t.TempDir()
-			stores := NewWALStores(vi, WithStorePath(tmp))
+			stores := NewFSManager(vi, WithStorePath(tmp))
 			wal, err := stores.CreateStore(context.Background(), *tt.id)
 			assert.NoError(t, err)
 			newWal := wal.(*WAL)
@@ -130,7 +131,7 @@ func Test_writeReadEntry(t *testing.T) {
 	}
 
 	tmp := t.TempDir()
-	stores := NewWALStores(vi, WithStorePath(tmp))
+	stores := NewFSManager(vi, WithStorePath(tmp))
 	wal, err := stores.CreateStore(context.Background(), id)
 	assert.NoError(t, err)
 
@@ -144,17 +145,36 @@ func Test_writeReadEntry(t *testing.T) {
 	assert.NoError(t, err)
 
 	// Reopen the WAL for read and write.
-	store, err := stores.CreateStore(context.Background(), id)
+	discoveredStores, err := stores.DiscoverStores(context.Background())
+	assert.Len(t, discoveredStores, 1)
 	assert.NoError(t, err)
+	store := discoveredStores[0]
 	newWal := store.(*WAL)
 	// we have already read the header in OpenWAL
 	_, err = newWal.readWALHeader()
 	assert.Error(t, err)
 
-	actualMessages, finished, err := newWal.Read(10000)
-	assert.NoError(t, err)
+	msgCh, errCh := newWal.Replay()
+	actualMessages := make([]*isb.ReadMessage, 0)
+outerLoop:
+	for {
+		select {
+		case msg, ok := <-msgCh:
+			if msg != nil {
+				actualMessages = append(actualMessages, msg)
+			}
+			if !ok {
+				break outerLoop
+			}
+		case err, ok := <-errCh:
+			if !ok {
+				break outerLoop
+			}
+			assert.NoError(t, err)
+			break outerLoop
+		}
+	}
 	// Check we reach the end of file
-	assert.Equal(t, true, finished)
 	assert.Equal(t, newWal.readUpTo, newWal.rOffset)
 
 	assert.Len(t, actualMessages, 1)
@@ -207,7 +227,7 @@ func Test_encodeDecodeEntry(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			tmp := t.TempDir()
-			stores := NewWALStores(vi, WithStorePath(tmp))
+			stores := NewFSManager(vi, WithStorePath(tmp))
 			wal, err := stores.CreateStore(context.Background(), partition.ID{})
 			assert.NoError(t, err)
 			newWal := wal.(*WAL)
@@ -233,6 +253,7 @@ func Test_encodeDecodeEntry(t *testing.T) {
 }
 
 func Test_batchSyncWithMaxBatchSize(t *testing.T) {
+	ctx := context.Background()
 	id := partition.ID{
 		Start: time.Unix(1665109020, 0).In(location),
 		End:   time.Unix(1665109020, 0).Add(time.Minute).In(location),
@@ -240,7 +261,7 @@ func Test_batchSyncWithMaxBatchSize(t *testing.T) {
 	}
 
 	tmp := t.TempDir()
-	stores := NewWALStores(vi, WithStorePath(tmp))
+	stores := NewFSManager(vi, WithStorePath(tmp))
 	wal, err := stores.CreateStore(context.Background(), id)
 	assert.NoError(t, err)
 
@@ -257,7 +278,7 @@ func Test_batchSyncWithMaxBatchSize(t *testing.T) {
 	assert.Equal(t, int64(0), tempWAL.prevSyncedWOffset)
 	assert.NoError(t, err)
 
-	tempWAL.walStores.maxBatchSize = 10
+	tempWAL.maxBatchSize = 10
 	assert.NoError(t, err)
 	err = wal.Write(&message)
 	assert.NoError(t, err)
@@ -267,17 +288,36 @@ func Test_batchSyncWithMaxBatchSize(t *testing.T) {
 	assert.NoError(t, err)
 
 	// Reopen the WAL for read and write.
-	store, err := stores.CreateStore(context.Background(), id)
+	discoveredStores, err := stores.DiscoverStores(ctx)
+	assert.Len(t, discoveredStores, 1)
 	assert.NoError(t, err)
-	newWal := store.(*WAL)
+	newWal := discoveredStores[0].(*WAL)
 	// we have already read the header in OpenWAL
 	_, err = newWal.readWALHeader()
 	assert.Error(t, err)
 
-	actualMessages, finished, err := newWal.Read(10000)
-	assert.NoError(t, err)
+	msgCh, errCh := newWal.Replay()
+	actualMessages := make([]*isb.ReadMessage, 0)
+outerLoop:
+	for {
+		select {
+		case msg, ok := <-msgCh:
+			if msg != nil {
+				actualMessages = append(actualMessages, msg)
+			}
+			if !ok {
+				break outerLoop
+			}
+		case err, ok := <-errCh:
+			if !ok {
+				break outerLoop
+			}
+			assert.NoError(t, err)
+			break outerLoop
+		}
+	}
+
 	// Check we reach the end of file
-	assert.Equal(t, true, finished)
 	assert.Equal(t, newWal.readUpTo, newWal.rOffset)
 
 	assert.Len(t, actualMessages, 2)
@@ -305,12 +345,20 @@ func Test_batchSyncWithSyncDuration(t *testing.T) {
 	}
 
 	tmp := t.TempDir()
-	stores := NewWALStores(vi, WithStorePath(tmp))
+	stores := &fsWAL{
+		storePath:    tmp,
+		maxBatchSize: dfv1.DefaultStoreMaxBufferSize,
+		syncDuration: 0,
+		pipelineName: vi.Vertex.Spec.PipelineName,
+		vertexName:   vi.Vertex.Spec.AbstractVertex.Name,
+		replicaIndex: vi.Replica,
+		activeStores: make(map[string]store.Store),
+	}
+
 	wal, err := stores.CreateStore(context.Background(), id)
 	assert.NoError(t, err)
 
 	tempWAL := wal.(*WAL)
-	tempWAL.walStores.syncDuration = 0
 
 	startTime := time.Unix(1665109020, 0).In(location)
 	msgCount := 2
@@ -323,7 +371,7 @@ func Test_batchSyncWithSyncDuration(t *testing.T) {
 	assert.NoError(t, err)
 
 	storePrevSyncedTime = tempWAL.prevSyncedTime
-	tempWAL.walStores.syncDuration = 10 * time.Second
+	tempWAL.syncDuration = 10 * time.Second
 	assert.NoError(t, err)
 	err = wal.Write(&message)
 	assert.NoError(t, err)
@@ -333,17 +381,35 @@ func Test_batchSyncWithSyncDuration(t *testing.T) {
 	assert.NoError(t, err)
 
 	// Reopen the WAL for read and write.
-	store, err := stores.CreateStore(context.Background(), id)
+	discoverStores, err := stores.DiscoverStores(context.Background())
+	assert.Len(t, discoverStores, 1)
 	assert.NoError(t, err)
-	newWal := store.(*WAL)
+	newWal := discoverStores[0].(*WAL)
 	// we have already read the header in OpenWAL
 	_, err = newWal.readWALHeader()
 	assert.Error(t, err)
 
-	actualMessages, finished, err := newWal.Read(10000)
-	assert.NoError(t, err)
+	msgCh, errCh := newWal.Replay()
+	actualMessages := make([]*isb.ReadMessage, 0)
+outerLoop:
+	for {
+		select {
+		case msg, ok := <-msgCh:
+			if msg != nil {
+				actualMessages = append(actualMessages, msg)
+			}
+			if !ok {
+				break outerLoop
+			}
+		case err, ok := <-errCh:
+			if !ok {
+				break outerLoop
+			}
+			assert.NoError(t, err)
+			break outerLoop
+		}
+	}
 	// Check we reach the end of file
-	assert.Equal(t, true, finished)
 	assert.Equal(t, newWal.readUpTo, newWal.rOffset)
 
 	assert.Len(t, actualMessages, 2)
diff --git a/pkg/reduce/pbq/store/aligned/memory/stores_test.go b/pkg/reduce/pbq/store/aligned/memory/manager_test.go
similarity index 83%
rename from pkg/reduce/pbq/store/aligned/memory/stores_test.go
rename to pkg/reduce/pbq/store/aligned/memory/manager_test.go
index fecc9cc0..89463abd 100644
--- a/pkg/reduce/pbq/store/aligned/memory/stores_test.go
+++ b/pkg/reduce/pbq/store/aligned/memory/manager_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/reduce/pbq/store"
 )
 
 func TestMemoryStores(t *testing.T) {
@@ -55,19 +56,19 @@ func TestMemoryStores(t *testing.T) {
 		assert.NoError(t, err)
 	}
 
-	var discoveredPartitions []partition.ID
-	discoveredPartitions, err = storeProvider.DiscoverPartitions(ctx)
+	var discoveredStores []store.Store
+	discoveredStores, err = storeProvider.DiscoverStores(ctx)
 	assert.NoError(t, err)
 
-	assert.Len(t, discoveredPartitions, len(partitionIds))
+	assert.Len(t, discoveredStores, len(partitionIds))
 
 	for _, partitionID := range partitionIds {
 		err = storeProvider.DeleteStore(partitionID)
 		assert.NoError(t, err)
 	}
 
-	discoveredPartitions, err = storeProvider.DiscoverPartitions(ctx)
+	discoveredStores, err = storeProvider.DiscoverStores(ctx)
 	assert.NoError(t, err)
 
-	assert.Len(t, discoveredPartitions, 0)
+	assert.Len(t, discoveredStores, 0)
 }
diff --git a/pkg/reduce/pbq/store/aligned/memory/store_test.go b/pkg/reduce/pbq/store/aligned/memory/wal_test.go
similarity index 80%
rename from pkg/reduce/pbq/store/aligned/memory/store_test.go
rename to pkg/reduce/pbq/store/aligned/memory/wal_test.go
index ba55a11c..07e424ca 100644
--- a/pkg/reduce/pbq/store/aligned/memory/store_test.go
+++ b/pkg/reduce/pbq/store/aligned/memory/wal_test.go
@@ -73,9 +73,26 @@ func TestMemoryStore_ReadFromStore(t *testing.T) {
 		err := memStore.Write(&msg)
 		assert.NoError(t, err)
 	}
-	var readMessages []*isb.ReadMessage
-	readMessages, _, err = memStore.Read(int64(msgCount))
-	assert.NoError(t, err)
+	msgCh, errCh := memStore.Replay()
+	readMessages := make([]*isb.ReadMessage, 0)
+outerLoop:
+	for {
+		select {
+		case msg, ok := <-msgCh:
+			if msg != nil {
+				readMessages = append(readMessages, msg)
+			}
+			if !ok {
+				break outerLoop
+			}
+		case err, ok := <-errCh:
+			if !ok {
+				break outerLoop
+			}
+			assert.NoError(t, err)
+			break outerLoop
+		}
+	}
 	// number of read messages should be equal to msgCount
 	assert.Len(t, readMessages, msgCount)
 }
@@ -93,11 +110,28 @@ func TestEmptyStore_Read(t *testing.T) {
 
 	memStore, err := NewMemoryStores(WithStoreSize(storeSize)).CreateStore(ctx, partitionID)
 	assert.NoError(t, err)
-	var eof bool
-	_, eof, err = memStore.Read(storeSize)
-	assert.NoError(t, err)
-	// since store is empty, eof will be true
-	assert.Equal(t, eof, true)
+	msgCh, errCh := memStore.Replay()
+	readMessages := make([]*isb.ReadMessage, 0)
+outerLoop:
+	for {
+		select {
+		case msg, ok := <-msgCh:
+			if msg != nil {
+				readMessages = append(readMessages, msg)
+			}
+			if !ok {
+				break outerLoop
+			}
+		case err, ok := <-errCh:
+			if !ok {
+				break outerLoop
+			}
+			assert.NoError(t, err)
+			break outerLoop
+		}
+	}
+	// since store is empty, there should not be any messages to read
+	assert.Len(t, readMessages, 0)
 
 }