Repository: numaproj/numaflow
Pull request #344 (2022-11-14 07:00:08)

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


vigith: @chromevoid, could you please do one dry-run to make sure we are seeing the expected behavior?
chromevoid: Tested successfully in k8s cluster using the sample reduce pipeline: didn't see data loss, and the events were processed in the expected order.

Patch:
diff --git a/docs/APIs.md b/docs/APIs.md
index d1f04eda..c9c34fc4 100644
--- a/docs/APIs.md
+++ b/docs/APIs.md
@@ -3454,7 +3454,7 @@ StoreType (<code>string</code> alias)
 </h3>
 <p>
 <p>
-PBQ store’s backend type.
+StoreType is the PBQ store’s backend type.
 </p>
 </p>
 <h3 id="numaflow.numaproj.io/v1alpha1.TLS">
diff --git a/pkg/apis/numaflow/v1alpha1/const.go b/pkg/apis/numaflow/v1alpha1/const.go
index 4b0aa632..d4a08194 100644
--- a/pkg/apis/numaflow/v1alpha1/const.go
+++ b/pkg/apis/numaflow/v1alpha1/const.go
@@ -131,10 +131,11 @@ const (
 	DefaultPBQReadBatchSize     = 100             // Default read batch size for pbq
 
 	// Default persistent store options
-	DefaultStoreSyncDuration  = 2 * time.Second // Default sync duration for pbq
-	DefaultStoreType          = NoOpType        // Default store type
-	DefaultStoreSize          = 1000000         // Default persistent store size
-	DefaultStoreMaxBufferSize = 100000          // Default buffer size for pbq in bytes
+	DefaultStoreSyncDuration  = 2 * time.Second        // Default sync duration for pbq
+	DefaultStoreType          = NoOpType               // Default store type
+	DefaultStoreSize          = 1000000                // Default persistent store size
+	DefaultStoreMaxBufferSize = 100000                 // Default buffer size for pbq in bytes
+	DefaultStorePath          = PathPBQMount + "/wals" // Default store path
 
 	// Default window options
 	DefaultWindowType     = FixedType
@@ -144,7 +145,7 @@ const (
 	PathPBQMount = "/var/numaflow/pbq"
 )
 
-// PBQ store's backend type.
+// StoreType is the PBQ store's backend type.
 type StoreType string
 
 const (
diff --git a/pkg/isb/stores/simplebuffer/buffer.go b/pkg/isb/stores/simplebuffer/buffer.go
index 5b77504d..b8999deb 100644
--- a/pkg/isb/stores/simplebuffer/buffer.go
+++ b/pkg/isb/stores/simplebuffer/buffer.go
@@ -128,9 +128,9 @@ func (b *InMemoryBuffer) Write(_ context.Context, messages []isb.Message) ([]isb
 			errs[idx] = nil
 			b.buffer[currentIdx].dirty = true
 			b.writeIdx = (currentIdx + 1) % b.size
-			writeOffsets = append(writeOffsets, isb.SimpleIntOffset(func() int64 {
+			writeOffsets[idx] = isb.SimpleIntOffset(func() int64 {
 				return currentIdx
-			}))
+			})
 			// access buffer via lock
 			b.rwlock.Unlock()
 		} else {
diff --git a/pkg/pbq/options.go b/pkg/pbq/options.go
index 4e735e09..8fe7dab1 100644
--- a/pkg/pbq/options.go
+++ b/pkg/pbq/options.go
@@ -17,9 +17,9 @@ limitations under the License.
 package pbq
 
 import (
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/pbq/store"
 	"time"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 )
 
 type options struct {
@@ -29,12 +29,6 @@ type options struct {
 	readTimeout time.Duration
 	// readBatchSize max size of batch to read from store
 	readBatchSize int64
-	// storeOptions options for pbq store
-	storeOptions *store.StoreOptions
-}
-
-func (o *options) StoreOptions() *store.StoreOptions {
-	return o.storeOptions
 }
 
 type PBQOption func(options *options) error
@@ -44,7 +38,6 @@ func DefaultOptions() *options {
 		channelBufferSize: dfv1.DefaultPBQChannelBufferSize,
 		readTimeout:       dfv1.DefaultPBQReadTimeout,
 		readBatchSize:     dfv1.DefaultPBQReadBatchSize,
-		storeOptions:      store.DefaultOptions(),
 	}
 }
 
@@ -71,16 +64,3 @@ func WithReadBatchSize(size int64) PBQOption {
 		return nil
 	}
 }
-
-// WithPBQStoreOptions sets different pbq store options
-func WithPBQStoreOptions(opts ...store.StoreOption) PBQOption {
-	return func(options *options) error {
-		for _, opt := range opts {
-			err := opt(options.storeOptions)
-			if err != nil {
-				return err
-			}
-		}
-		return nil
-	}
-}
diff --git a/pkg/pbq/pbq.go b/pkg/pbq/pbq.go
index 4f0f921d..94b65c29 100644
--- a/pkg/pbq/pbq.go
+++ b/pkg/pbq/pbq.go
@@ -20,10 +20,11 @@ import (
 	"context"
 	"errors"
 
+	"go.uber.org/zap"
+
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/pbq/store"
-	"go.uber.org/zap"
 )
 
 var ErrCOB = errors.New("error while writing to pbq, pbq is closed")
@@ -46,8 +47,8 @@ var _ ReadWriteCloser = (*PBQ)(nil)
 func (p *PBQ) Write(ctx context.Context, message *isb.ReadMessage) error {
 	// if cob we should return
 	if p.cob {
-		p.log.Errorw("Failed to write message to pbq, pbq is closed", zap.Any("ID", p.PartitionID), zap.Any("header", message.Header))
-		return ErrCOB
+		p.log.Errorw("Failed to write message to pbq, pbq is closed", zap.Any("ID", p.PartitionID), zap.Any("header", message.Header), zap.Any("message", message))
+		return nil
 	}
 	var writeErr error
 	// we need context to get out of blocking write
@@ -87,9 +88,12 @@ func (p *PBQ) ReadCh() <-chan *isb.ReadMessage {
 // finished forwarding the output to ISB.
 func (p *PBQ) GC() error {
 	err := p.store.GC()
+	if err != nil {
+		return err
+	}
 	p.store = nil
 	p.manager.deregister(p.PartitionID)
-	return err
+	return nil
 }
 
 // replayRecordsFromStore replays store messages when replay flag is set during start up time. It replays by reading from
diff --git a/pkg/pbq/pbqmanager.go b/pkg/pbq/pbqmanager.go
index dd7a2772..21d155ec 100644
--- a/pkg/pbq/pbqmanager.go
+++ b/pkg/pbq/pbqmanager.go
@@ -18,7 +18,6 @@ package pbq
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"math"
 	"sync"
@@ -27,21 +26,18 @@ import (
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/util/wait"
 
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/pbq/store"
-	"github.com/numaproj/numaflow/pkg/pbq/store/memory"
-	"github.com/numaproj/numaflow/pkg/pbq/store/noop"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 )
 
 // Manager helps in managing the lifecycle of PBQ instances
 type Manager struct {
-	storeOptions *store.StoreOptions
-	pbqOptions   *options
-	pbqMap       map[string]*PBQ
-	log          *zap.SugaredLogger
+	storeProvider store.StoreProvider
+	pbqOptions    *options
+	pbqMap        map[string]*PBQ
+	log           *zap.SugaredLogger
 	// we need lock to access pbqMap, since deregister will be called inside pbq
 	// and each pbq will be inside a go routine, and also entire PBQ could be managed
 	// through a go routine (depends on the orchestrator)
@@ -50,7 +46,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, opts ...PBQOption) (*Manager, error) {
+func NewManager(ctx context.Context, storeProvider store.StoreProvider, opts ...PBQOption) (*Manager, error) {
 	pbqOpts := DefaultOptions()
 	for _, opt := range opts {
 		if opt != nil {
@@ -61,10 +57,10 @@ func NewManager(ctx context.Context, opts ...PBQOption) (*Manager, error) {
 	}
 
 	pbqManager := &Manager{
-		pbqMap:       make(map[string]*PBQ),
-		pbqOptions:   pbqOpts,
-		storeOptions: pbqOpts.storeOptions,
-		log:          logging.FromContext(ctx),
+		storeProvider: storeProvider,
+		pbqMap:        make(map[string]*PBQ),
+		pbqOptions:    pbqOpts,
+		log:           logging.FromContext(ctx),
 	}
 
 	return pbqManager, nil
@@ -72,23 +68,9 @@ func NewManager(ctx context.Context, opts ...PBQOption) (*Manager, error) {
 
 // CreateNewPBQ creates new pbq for a partition
 func (m *Manager) CreateNewPBQ(ctx context.Context, partitionID partition.ID) (ReadWriteCloser, error) {
-
-	var persistentStore store.Store
-	var err error
-
-	switch m.storeOptions.PBQStoreType() {
-	case dfv1.NoOpType:
-		persistentStore, _ = noop.NewPBQNoOpStore()
-	case dfv1.InMemoryType:
-		persistentStore, err = memory.NewMemoryStore(ctx, partitionID, m.storeOptions)
-		if err != nil {
-			m.log.Errorw("Error while creating persistent store", zap.Any("ID", partitionID), zap.Any("storeType", m.storeOptions.PBQStoreType()), zap.Error(err))
-			return nil, err
-		}
-	case dfv1.FileSystemType:
-		return nil, fmt.Errorf("not implemented, %s", dfv1.FileSystemType)
-	default:
-		return nil, errors.New("not implemented (default)")
+	persistentStore, err := m.storeProvider.CreateStore(ctx, partitionID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create a PBQ store, %w", err)
 	}
 
 	// output channel is buffered to support bulk reads
@@ -106,20 +88,6 @@ func (m *Manager) CreateNewPBQ(ctx context.Context, partitionID partition.ID) (R
 	return p, nil
 }
 
-// discoverPartitions discovers partitions.
-func (m *Manager) discoverPartitions(ctx context.Context) ([]partition.ID, error) {
-	switch m.storeOptions.PBQStoreType() {
-	case dfv1.NoOpType:
-		return noop.DiscoverPartitions(ctx, m.storeOptions)
-	case dfv1.InMemoryType:
-		return memory.DiscoverPartitions(ctx, m.storeOptions)
-	case dfv1.FileSystemType:
-		return nil, fmt.Errorf("not implemented, %s", dfv1.FileSystemType)
-	default:
-		return nil, errors.New("not implemented (default)")
-	}
-}
-
 // ListPartitions returns all the pbq instances
 func (m *Manager) ListPartitions() []*PBQ {
 	m.RLock()
@@ -147,9 +115,9 @@ func (m *Manager) GetPBQ(partitionID partition.ID) ReadWriteCloser {
 	return nil
 }
 
-// StartUp restores the state of the pbqManager. It reads from the PBQs store to get the persisted partitions
+// 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) StartUp(ctx context.Context) {
+func (m *Manager) GetExistingPartitions(ctx context.Context) ([]partition.ID, error) {
 	var ctxClosedErr error
 	var partitionIDs []partition.ID
 
@@ -163,7 +131,7 @@ func (m *Manager) StartUp(ctx context.Context) {
 	ctxClosedErr = wait.ExponentialBackoffWithContext(ctx, discoverPartitionsBackoff, func() (done bool, err error) {
 		var attempt int
 
-		partitionIDs, err = m.discoverPartitions(ctx)
+		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))
@@ -173,32 +141,10 @@ func (m *Manager) StartUp(ctx context.Context) {
 	})
 	if ctxClosedErr != nil {
 		m.log.Errorw("Context closed while discovering partitions", zap.Error(ctxClosedErr))
-		return
-	}
-
-	var createPBQBackoff = wait.Backoff{
-		Steps:    math.MaxInt,
-		Duration: 100 * time.Millisecond,
-		Factor:   1,
-		Jitter:   0.1,
+		return partitionIDs, ctxClosedErr
 	}
 
-	for _, partitionID := range partitionIDs {
-		ctxClosedErr = wait.ExponentialBackoffWithContext(ctx, createPBQBackoff, func() (done bool, err error) {
-			var attempt int
-			_, err = m.CreateNewPBQ(ctx, partitionID)
-
-			if err != nil {
-				attempt += 1
-				m.log.Errorw("Failed to create pbq during startup, retrying", zap.Any("attempt", attempt), zap.Any("partitionID", partitionID.String()), zap.Error(err))
-				return false, nil
-			}
-			return true, nil
-		})
-		if ctxClosedErr != nil {
-			m.log.Errorw("Context closed while creating new pbq", zap.Any("partitionID", partitionID.String()), zap.Error(ctxClosedErr))
-		}
-	}
+	return partitionIDs, nil
 }
 
 // ShutDown for clean shut down, flushes pending messages to store and closes the store
@@ -260,7 +206,7 @@ func (m *Manager) deregister(partitionID partition.ID) {
 func (m *Manager) getPBQs() []*PBQ {
 	m.RLock()
 	defer m.RUnlock()
-	var pbqs = make([]*PBQ, 0)
+	var pbqs = make([]*PBQ, 0, len(m.pbqMap))
 	for _, pbq := range m.pbqMap {
 		pbqs = append(pbqs, pbq)
 	}
@@ -271,8 +217,10 @@ func (m *Manager) getPBQs() []*PBQ {
 // 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) {
@@ -282,4 +230,6 @@ func (m *Manager) Replay(ctx context.Context) {
 	}
 
 	wg.Wait()
+	m.log.Infow("Finished replaying records from store", zap.Duration("took", time.Since(tm)), zap.Any("partitions", partitionsIds))
+
 }
diff --git a/pkg/pbq/store/interfaces.go b/pkg/pbq/store/interfaces.go
index 95b67cfa..e51e874e 100644
--- a/pkg/pbq/store/interfaces.go
+++ b/pkg/pbq/store/interfaces.go
@@ -17,12 +17,16 @@ limitations under the License.
 package store
 
 import (
+	"context"
+
 	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
 )
 
 // Store provides methods to read, write and delete data from the store.
 type Store interface {
-	// Read returns upto N(size) messages from the persisted store
+	// Read 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)
 	// Write writes message to persistence store
 	Write(msg *isb.ReadMessage) error
@@ -31,3 +35,11 @@ type Store interface {
 	// GC does garbage collection and deletes all the messages that are persisted
 	GC() error
 }
+
+// StoreProvider defines the functions for store implementation
+type StoreProvider 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)
+}
diff --git a/pkg/pbq/store/memory/discover.go b/pkg/pbq/store/memory/options.go
similarity index 53%
rename from pkg/pbq/store/memory/discover.go
rename to pkg/pbq/store/memory/options.go
index e1f5312c..2696d469 100644
--- a/pkg/pbq/store/memory/discover.go
+++ b/pkg/pbq/store/memory/options.go
@@ -20,21 +20,20 @@ import (
 	"context"
 
 	"github.com/numaproj/numaflow/pkg/pbq/partition"
-	"github.com/numaproj/numaflow/pkg/pbq/store"
 )
 
-var discoverer func(*store.StoreOptions) ([]partition.ID, error)
+type Option func(stores *memoryStores)
 
-// DiscoverPartitions discovers the in-memory partitions. This is very useful during testing.
-func DiscoverPartitions(ctx context.Context, options *store.StoreOptions) ([]partition.ID, error) {
-	if discoverer == nil {
-		return []partition.ID{}, nil
+// WithDiscoverer sets the discover func of memorystores
+func WithDiscoverer(f func(ctx context.Context) ([]partition.ID, error)) Option {
+	return func(stores *memoryStores) {
+		stores.discoverFunc = f
 	}
-	return discoverer(options)
 }
 
-// SetDiscoverer sets the function that returns the initial list of partitions
-// Only for testing purposes
-func SetDiscoverer(d func(*store.StoreOptions) ([]partition.ID, error)) {
-	discoverer = d
+// WithStoreSize sets the store size
+func WithStoreSize(size int64) Option {
+	return func(stores *memoryStores) {
+		stores.storeSize = size
+	}
 }
diff --git a/pkg/pbq/store/memory/store.go b/pkg/pbq/store/memory/store.go
index afe87ff3..b28e4ca5 100644
--- a/pkg/pbq/store/memory/store.go
+++ b/pkg/pbq/store/memory/store.go
@@ -17,13 +17,11 @@ limitations under the License.
 package memory
 
 import (
-	"context"
 	"math"
 
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/pbq/store"
-	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"go.uber.org/zap"
 )
 
@@ -33,27 +31,11 @@ type memoryStore struct {
 	writePos    int64
 	readPos     int64
 	storage     []*isb.ReadMessage
-	options     *store.StoreOptions
+	storeSize   int64
 	log         *zap.SugaredLogger
 	partitionID partition.ID
 }
 
-// NewMemoryStore returns new memory store
-func NewMemoryStore(ctx context.Context, partitionID partition.ID, options *store.StoreOptions) (store.Store, error) {
-
-	memStore := &memoryStore{
-		writePos:    0,
-		readPos:     0,
-		closed:      false,
-		storage:     make([]*isb.ReadMessage, options.StoreSize()),
-		options:     options,
-		log:         logging.FromContext(ctx).With("PBQ store", "Memory store").With("Partition ID", partitionID),
-		partitionID: partitionID,
-	}
-
-	return memStore, nil
-}
-
 // ReadFromStore will return upto N messages persisted in store
 // this function will be invoked during bootstrap if there is a restart
 func (m *memoryStore) Read(size int64) ([]*isb.ReadMessage, bool, error) {
@@ -72,7 +54,7 @@ func (m *memoryStore) Read(size int64) ([]*isb.ReadMessage, bool, error) {
 
 // WriteToStore writes a message to store
 func (m *memoryStore) Write(msg *isb.ReadMessage) error {
-	if m.writePos >= m.options.StoreSize() {
+	if m.writePos >= m.storeSize {
 		m.log.Errorw(store.WriteStoreFullErr.Error(), zap.Any("msg header", msg.Header))
 		return store.WriteStoreFullErr
 	}
diff --git a/pkg/pbq/store/memory/stores.go b/pkg/pbq/store/memory/stores.go
new file mode 100644
index 00000000..084ecdb0
--- /dev/null
+++ b/pkg/pbq/store/memory/stores.go
@@ -0,0 +1,62 @@
+/*
+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 memory
+
+import (
+	"context"
+
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/pbq/store"
+	"github.com/numaproj/numaflow/pkg/shared/logging"
+)
+
+type memoryStores struct {
+	storeSize    int64
+	discoverFunc func(ctx context.Context) ([]partition.ID, error)
+}
+
+func NewMemoryStores(opts ...Option) store.StoreProvider {
+	s := &memoryStores{
+		storeSize: 100,
+	}
+	for _, o := range opts {
+		o(s)
+	}
+	return s
+}
+
+func (ms *memoryStores) CreateStore(ctx context.Context, partitionID partition.ID) (store.Store, error) {
+	memStore := &memoryStore{
+		writePos:    0,
+		readPos:     0,
+		closed:      false,
+		storage:     make([]*isb.ReadMessage, ms.storeSize),
+		storeSize:   ms.storeSize,
+		log:         logging.FromContext(ctx).With("pbqStore", "Memory").With("partitionID", partitionID),
+		partitionID: partitionID,
+	}
+
+	return memStore, nil
+}
+
+func (ms *memoryStores) DiscoverPartitions(ctx context.Context) ([]partition.ID, error) {
+	if ms.discoverFunc != nil {
+		return ms.discoverFunc(ctx)
+	}
+	return []partition.ID{}, nil
+}
diff --git a/pkg/pbq/store/noop/discover.go b/pkg/pbq/store/noop/stores.go
similarity index 69%
rename from pkg/pbq/store/noop/discover.go
rename to pkg/pbq/store/noop/stores.go
index 25b1f539..0f89c138 100644
--- a/pkg/pbq/store/noop/discover.go
+++ b/pkg/pbq/store/noop/stores.go
@@ -23,6 +23,17 @@ import (
 	"github.com/numaproj/numaflow/pkg/pbq/store"
 )
 
-func DiscoverPartitions(_ context.Context, _ *store.StoreOptions) ([]partition.ID, error) {
+type noopStores struct {
+}
+
+func NewNoopStores() store.StoreProvider {
+	return &noopStores{}
+}
+
+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
 }
diff --git a/pkg/pbq/store/options.go b/pkg/pbq/store/options.go
deleted file mode 100644
index bf1bfd2e..00000000
--- a/pkg/pbq/store/options.go
+++ /dev/null
@@ -1,85 +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 store
-
-import (
-	"time"
-
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-)
-
-type StoreOptions struct {
-	// maxBufferSize max size of batch before it's flushed to store
-	maxBatchSize int64
-	// syncDuration timeout to sync to store
-	syncDuration time.Duration
-	// pbqStoreType store type (memory or s3 or file system)
-	pbqStoreType dfv1.StoreType
-	// storeSize store array size
-	storeSize int64
-}
-
-func (o *StoreOptions) StoreSize() int64 {
-	return o.storeSize
-}
-
-func (o *StoreOptions) PBQStoreType() dfv1.StoreType {
-	return o.pbqStoreType
-}
-
-func DefaultOptions() *StoreOptions {
-	return &StoreOptions{
-		maxBatchSize: dfv1.DefaultStoreMaxBufferSize,
-		syncDuration: dfv1.DefaultStoreSyncDuration,
-		pbqStoreType: dfv1.DefaultStoreType,
-		storeSize:    dfv1.DefaultStoreSize,
-	}
-}
-
-type StoreOption func(options *StoreOptions) error
-
-// WithMaxBufferSize sets buffer max size option
-func WithMaxBufferSize(size int64) StoreOption {
-	return func(o *StoreOptions) error {
-		o.maxBatchSize = size
-		return nil
-	}
-}
-
-// WithSyncDuration sets sync duration option
-func WithSyncDuration(maxDuration time.Duration) StoreOption {
-	return func(o *StoreOptions) error {
-		o.syncDuration = maxDuration
-		return nil
-	}
-}
-
-// WithPbqStoreType sets store type option
-func WithPbqStoreType(storeType dfv1.StoreType) StoreOption {
-	return func(o *StoreOptions) error {
-		o.pbqStoreType = storeType
-		return nil
-	}
-}
-
-// WithStoreSize sets store size option
-func WithStoreSize(size int64) StoreOption {
-	return func(o *StoreOptions) error {
-		o.storeSize = size
-		return nil
-	}
-}
diff --git a/pkg/pbq/store/wal/bootstrap.go b/pkg/pbq/store/wal/bootstrap.go
new file mode 100644
index 00000000..7e6fc808
--- /dev/null
+++ b/pkg/pbq/store/wal/bootstrap.go
@@ -0,0 +1,221 @@
+/*
+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 (
+	"bytes"
+	"encoding/binary"
+	"encoding/gob"
+	"fmt"
+	"io"
+	"os"
+	"time"
+
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
+)
+
+var location *time.Location
+
+func init() {
+	var err error
+	location, err = time.LoadLocation("UTC")
+	if err != nil {
+		panic(fmt.Sprint("cannot load UTC", err))
+	}
+}
+
+// IsCorrupted checks whether the file is corrupt
+func (w *WAL) IsCorrupted() bool {
+	return w.corrupted
+}
+
+// OpenWAL returns a WAL if present
+func 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,
+	}
+
+	w.partitionID, err = w.readHeader()
+
+	return w, err
+}
+
+func (w *WAL) readHeader() (*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)
+	}
+
+	id, err := decodeHeader(w.fp)
+	if err != nil {
+		return nil, err
+	}
+
+	seek, err := w.fp.Seek(0, io.SeekCurrent)
+	if err != nil {
+		return nil, err
+	}
+
+	w.rOffset += seek
+
+	return id, err
+}
+
+func (w *WAL) isEnd() bool {
+	// TODO: If we are done reading, check that we got the expected
+	// number of entries and return EOF.
+	return w.rOffset >= w.readUpTo
+}
+
+// decodeHeader decodes the header which is encoded by encodeHeader.
+func decodeHeader(buf io.Reader) (*partition.ID, error) {
+	var err error
+	// read the fixed vals
+	var hp = new(headerPreamble)
+	err = binary.Read(buf, binary.LittleEndian, hp)
+	if err != nil {
+		return nil, err
+	}
+	// read the variadic key
+	var key = make([]rune, hp.KLen)
+	err = binary.Read(buf, binary.LittleEndian, key)
+	if err != nil {
+		return nil, err
+	}
+
+	return &partition.ID{
+		Start: time.UnixMilli(hp.S).In(location),
+		End:   time.UnixMilli(hp.E).In(location),
+		Key:   string(key),
+	}, 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 := decodeEntry(w.fp)
+		if err != nil {
+			if err == errChecksumMismatch {
+				w.corrupted = true
+			}
+			return nil, false, err
+		}
+
+		w.rOffset += sizeRead
+		messages = append(messages, message)
+	}
+	currentTime := time.Now()
+	if w.isEnd() {
+		w.wOffset = w.rOffset
+		w.prevSyncedWOffset = w.wOffset
+		w.prevSyncedTime = currentTime
+		w.numOfUnsyncedMsgs = 0
+		return messages, true, nil
+	}
+	return messages, false, nil
+}
+
+// decodeEntry decodes the header which is encoded by encodeEntry.
+func decodeEntry(buf io.Reader) (*isb.ReadMessage, int64, error) {
+	entryHeader, err := decodeEntryHeader(buf)
+	if err != nil {
+		return nil, 0, err
+	}
+
+	entryBody, err := decodeEntryBody(buf, entryHeader)
+	if err != nil {
+		return nil, 0, err
+	}
+	size := EntryHeaderSize + entryHeader.MessageLen
+
+	return &isb.ReadMessage{
+		Message:    *entryBody,
+		Watermark:  time.UnixMilli(entryHeader.WaterMark).In(location),
+		ReadOffset: isb.SimpleIntOffset(func() int64 { return entryHeader.Offset }),
+	}, size, nil
+}
+
+// decodeEntryHeader decodes the header which is encoded by encodeEntryHeader.
+func decodeEntryHeader(buf io.Reader) (*entryHeaderPreamble, error) {
+	// read the fixed vals
+	var entryHeader = new(entryHeaderPreamble)
+	err := binary.Read(buf, binary.LittleEndian, entryHeader)
+	if err != nil {
+		return nil, err
+	}
+	return entryHeader, nil
+}
+
+// decodeEntryBody decodes the header which is encoded by encodeEntryBody.
+// Returns errChecksumMismatch to indicate if corrupted entry is found.
+func decodeEntryBody(buf io.Reader, entryHeader *entryHeaderPreamble) (*isb.Message, error) {
+	var err error
+
+	body := make([]byte, entryHeader.MessageLen)
+	size, err := buf.Read(body)
+	if err != nil {
+		return nil, err
+	}
+	if int64(size) != entryHeader.MessageLen {
+		return nil, fmt.Errorf("expected to read length of %d, but wrote only %d", entryHeader.MessageLen, size)
+	}
+
+	// verify the checksum
+	checksum := calculateChecksum(body)
+	if checksum != entryHeader.Checksum {
+		return nil, errChecksumMismatch
+	}
+
+	var message = new(isb.Message)
+	dec := gob.NewDecoder(bytes.NewReader(body))
+	err = dec.Decode(&message)
+	if err != nil {
+		return nil, err
+	}
+	return message, nil
+}
diff --git a/pkg/pbq/store/wal/doc.go b/pkg/pbq/store/wal/doc.go
new file mode 100644
index 00000000..3b613302
--- /dev/null
+++ b/pkg/pbq/store/wal/doc.go
@@ -0,0 +1,18 @@
+/*
+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 implements write-ahead-log.
+package wal
diff --git a/pkg/pbq/store/wal/metrics.go b/pkg/pbq/store/wal/metrics.go
new file mode 100644
index 00000000..f5d0f7f2
--- /dev/null
+++ b/pkg/pbq/store/wal/metrics.go
@@ -0,0 +1,74 @@
+/*
+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 (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+const (
+	LabelPartitionKey = "partitionKey"
+)
+
+// TODO - Adjust metric bucket range after we get more map reduce use cases.
+
+var entriesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "pbq_wal",
+	Name:      "wal_entries_total",
+	Help:      "Total number of entries written across ALL wal files/partitions",
+}, []string{})
+
+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)",
+}, []string{})
+
+var activeFilesCount = promauto.NewGaugeVec(prometheus.GaugeOpts(prometheus.CounterOpts{
+	Subsystem: "pbq_wal",
+	Name:      "active_wal_files_total",
+	Help:      "Total number of active wal files/partitions",
+}), []string{})
+
+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)",
+	Buckets:   prometheus.ExponentialBucketsRange(100, 5000, 5),
+}, []string{LabelPartitionKey})
+
+var fileSyncWaitTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Subsystem: "pbq_wal",
+	Name:      "wal_file_sync_wait_time",
+	Help:      "File Sync wait time (1 to 60 milliseconds)",
+	Buckets:   prometheus.ExponentialBucketsRange(1, 60, 5),
+}, []string{LabelPartitionKey})
+
+var entryWriteTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Subsystem: "pbq_wal",
+	Name:      "wal_entry_write_time",
+	Help:      "Entry write time (1 to 60 milliseconds)",
+	Buckets:   prometheus.ExponentialBucketsRange(1, 60, 5),
+}, []string{LabelPartitionKey})
+
+var lifespan = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Subsystem: "pbq_wal",
+	Name:      "wal_lifespan",
+	Help:      "Lifespan of a pbq wal (1 to 20 minutes)",
+	Buckets:   prometheus.ExponentialBucketsRange(1, 20, 5),
+}, []string{LabelPartitionKey})
diff --git a/pkg/pbq/store/wal/options.go b/pkg/pbq/store/wal/options.go
new file mode 100644
index 00000000..4afd0a58
--- /dev/null
+++ b/pkg/pbq/store/wal/options.go
@@ -0,0 +1,42 @@
+/*
+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 "time"
+
+type Option func(stores *walStores)
+
+// WithStorePath sets the WAL store path
+func WithStorePath(path string) Option {
+	return func(stores *walStores) {
+		stores.storePath = path
+	}
+}
+
+// WithMaxBufferSize sets the WAL buffer max size option
+func WithMaxBufferSize(size int64) Option {
+	return func(stores *walStores) {
+		stores.maxBatchSize = size
+	}
+}
+
+// WithSyncDuration sets the WAL sync duration option
+func WithSyncDuration(maxDuration time.Duration) Option {
+	return func(stores *walStores) {
+		stores.syncDuration = maxDuration
+	}
+}
diff --git a/pkg/pbq/store/wal/segment.go b/pkg/pbq/store/wal/segment.go
new file mode 100644
index 00000000..2716ff60
--- /dev/null
+++ b/pkg/pbq/store/wal/segment.go
@@ -0,0 +1,291 @@
+/*
+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 (
+	"bytes"
+	"encoding/binary"
+	"encoding/gob"
+	"fmt"
+	"hash/crc32"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
+)
+
+const (
+	IEEE            = 0xedb88320
+	EntryHeaderSize = 28
+	SegmentPrefix   = "segment"
+)
+
+// Various errors contained in DNSError.
+var (
+	errChecksumMismatch = fmt.Errorf("data checksum not match")
+)
+
+// WAL implements a write-ahead-log. It represents both reader and writer. This WAL is write heavy and read is
+// 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
+	numOfUnsyncedMsgs int64
+}
+
+// writeHeader writes the header to the file
+func (w *WAL) writeHeader() error {
+	header, err := encodeHeader(w.partitionID)
+	if err != nil {
+		return err
+	}
+	wrote, err := w.fp.WriteAt(header.Bytes(), w.wOffset)
+	if wrote != header.Len() {
+		return fmt.Errorf("expected to write %d, but wrote only %d, %w", header.Len(), wrote, err)
+	}
+
+	// Only increase the write offset when we successfully write for atomicity.
+	w.wOffset += int64(wrote)
+	return err
+}
+
+// headerPreamble is the header preamble (excludes variadic key)
+type headerPreamble struct {
+	S    int64
+	E    int64
+	KLen int16
+}
+
+// entryHeaderPreamble is the header for each WAL entry
+type entryHeaderPreamble struct {
+	WaterMark  int64
+	Offset     int64
+	MessageLen int64
+	Checksum   uint32
+}
+
+// encodeHeader builds the header. The header is of the following format.
+//
+//	+--------------------+------------------+-----------------+------------+
+//	| start time (int64) | end time (int64) | key-len (int16) | key []rune |
+//	+--------------------+------------------+-----------------+------------+
+//
+// We require the key-len because key is variadic.
+func encodeHeader(id *partition.ID) (*bytes.Buffer, error) {
+	buf := new(bytes.Buffer)
+	hp := headerPreamble{
+		S:    id.Start.UnixMilli(),
+		E:    id.End.UnixMilli(),
+		KLen: int16(len(id.Key)),
+	}
+
+	// write the fixed values
+	err := binary.Write(buf, binary.LittleEndian, hp)
+	if err != nil {
+		return nil, err
+	}
+
+	// write the variadic values
+	err = binary.Write(buf, binary.LittleEndian, []rune(id.Key))
+
+	return buf, err
+}
+
+func encodeEntry(message *isb.ReadMessage) (*bytes.Buffer, error) {
+	buf := new(bytes.Buffer)
+	body, err := encodeEntryBody(message)
+	if err != nil {
+		return nil, err
+	}
+	checksum := calculateChecksum(body.Bytes())
+
+	// Writes the message header
+	header, err := encodeEntryHeader(message, int64(body.Len()), checksum)
+	if err != nil {
+		return nil, err
+	}
+	wrote, err := buf.Write(header.Bytes())
+	if err != nil {
+		return nil, err
+	}
+	if wrote != header.Len() {
+		return nil, fmt.Errorf("expected to write %d, but wrote only %d, %w", header.Len(), wrote, err)
+	}
+
+	// Writes the message body
+	wrote, err = buf.Write(body.Bytes())
+	if err != nil {
+		return nil, err
+	}
+	if wrote != body.Len() {
+		return nil, fmt.Errorf("expected to write %d, but wrote only %d, %w", header.Len(), wrote, err)
+	}
+	return buf, nil
+}
+
+func encodeEntryHeader(message *isb.ReadMessage, messageLen int64, checksum uint32) (*bytes.Buffer, error) {
+	watermark := message.Watermark.UnixMilli()
+
+	offset, err := message.ReadOffset.Sequence()
+	if err != nil {
+		return nil, err
+	}
+
+	entryHeader := entryHeaderPreamble{
+		WaterMark:  watermark,
+		Offset:     offset,
+		MessageLen: messageLen,
+		Checksum:   checksum,
+	}
+
+	// write the fixed values
+	buf := new(bytes.Buffer)
+	err = binary.Write(buf, binary.LittleEndian, entryHeader)
+	if err != nil {
+		return nil, err
+	}
+	return buf, nil
+}
+
+func encodeEntryBody(message *isb.ReadMessage) (*bytes.Buffer, error) {
+	m := new(bytes.Buffer)
+	enc := gob.NewEncoder(m)
+	err := enc.Encode(message.Message)
+	if err != nil {
+		return nil, fmt.Errorf("entry body encountered encode err: %w", err)
+	}
+	return m, nil
+}
+
+func calculateChecksum(data []byte) uint32 {
+	crc32q := crc32.MakeTable(IEEE)
+	return crc32.Checksum(data, crc32q)
+}
+
+// Write writes the message to the WAL. The format as follow is
+//
+//	+-------------------+----------------+-----------------+--------------+----------------+
+//	| watermark (int64) | offset (int64) | msg-len (int64) | CRC (unit32) | message []byte |
+//	+-------------------+----------------+-----------------+--------------+----------------+
+//
+// CRC will be used for detecting entry corruptions.
+func (w *WAL) Write(message *isb.ReadMessage) error {
+	writeStart := time.Now()
+	entry, err := encodeEntry(message)
+	if err != nil {
+		return err
+	}
+
+	wrote, err := w.fp.WriteAt(entry.Bytes(), w.wOffset)
+	if wrote != entry.Len() {
+		return fmt.Errorf("expected to write %d, but wrote only %d, %w", entry.Len(), wrote, err)
+	}
+	if err != nil {
+		return err
+	}
+
+	w.numOfUnsyncedMsgs = w.numOfUnsyncedMsgs + 1
+	// Only increase the write offset when we successfully write for atomicity.
+	w.wOffset += int64(wrote)
+	// TODO: add batch sync()
+
+	currentTime := time.Now()
+
+	if w.wOffset-w.prevSyncedWOffset > w.walStores.maxBatchSize || currentTime.Sub(w.prevSyncedTime) > w.walStores.syncDuration {
+		w.prevSyncedWOffset = w.wOffset
+		w.prevSyncedTime = currentTime
+		fSyncStart := time.Now()
+		err = w.fp.Sync()
+		fileSyncWaitTime.With(map[string]string{LabelPartitionKey: w.partitionID.Key}).Observe(float64(time.Since(fSyncStart).Milliseconds()))
+		if err == nil {
+			entryWriteTime.With(map[string]string{LabelPartitionKey: w.partitionID.Key}).Observe(float64(time.Since(writeStart).Milliseconds()) / float64(w.numOfUnsyncedMsgs))
+			entriesCount.With(map[string]string{}).Add(float64(w.numOfUnsyncedMsgs))
+		}
+		w.numOfUnsyncedMsgs = 0
+		return err
+	}
+	return err
+}
+
+// Close closes the WAL Segment.
+func (w *WAL) Close() error {
+	start := time.Now()
+	err := w.fp.Sync()
+	fileSyncWaitTime.With(map[string]string{LabelPartitionKey: w.partitionID.Key}).Observe(float64(time.Since(start).Milliseconds()))
+
+	if err != nil {
+		return err
+	}
+
+	err = w.fp.Close()
+	if err != nil {
+		return err
+	}
+
+	w.closed = true
+
+	return nil
+}
+
+// GC cleans up the WAL Segment.
+func (w *WAL) GC() error {
+	start := time.Now()
+	if !w.closed {
+		err := w.Close()
+		if err != nil {
+			return err
+		}
+	}
+	err := os.Remove(w.fp.Name())
+
+	if err == nil {
+		garbageCollectingTime.With(map[string]string{LabelPartitionKey: w.partitionID.Key}).Observe(float64(time.Since(start).Microseconds()))
+		lifespan.With(map[string]string{LabelPartitionKey: w.partitionID.Key}).Observe(time.Since(w.createTime).Minutes())
+		activeFilesCount.With(map[string]string{}).Dec()
+	}
+	return err
+}
+
+func getSegmentFilePath(id *partition.ID, dir string) string {
+	filename := fmt.Sprintf("%s_%d.%d.%s", SegmentPrefix, id.Start.Unix(), id.End.Unix(), id.Key)
+	return filepath.Join(dir, filename)
+}
diff --git a/pkg/pbq/store/wal/stores.go b/pkg/pbq/store/wal/stores.go
new file mode 100644
index 00000000..2436b91e
--- /dev/null
+++ b/pkg/pbq/store/wal/stores.go
@@ -0,0 +1,159 @@
+/*
+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"
+	"strings"
+	"time"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/pbq/store"
+)
+
+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
+}
+
+func NewWALStores(opts ...Option) store.StoreProvider {
+	s := &walStores{
+		storePath:    dfv1.DefaultStorePath,
+		maxBatchSize: dfv1.DefaultStoreMaxBufferSize,
+		syncDuration: dfv1.DefaultStoreSyncDuration,
+	}
+	for _, o := range opts {
+		o(s)
+	}
+	return s
+}
+
+func (ws *walStores) CreateStore(ctx context.Context, partitionID partition.ID) (store.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, 0644)
+		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
+
+	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(getSegmentFilePath(id, ws.storePath), os.O_WRONLY|os.O_CREATE, 0644)
+		if err != nil {
+			return nil, err
+		}
+		filesCount.With(map[string]string{}).Inc()
+		activeFilesCount.With(map[string]string{}).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.writeHeader()
+		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.readHeader()
+		if err != nil {
+			return nil, err
+		}
+		if id.Key != readPartition.Key {
+			return nil, fmt.Errorf("expected partition key %s, but got %s", id.Key, readPartition.Key)
+		}
+	}
+
+	return wal, err
+}
+
+func (ws *walStores) DiscoverPartitions(ctx 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 := OpenWAL(filePath)
+			if err != nil {
+				return nil, err
+			}
+			partitions = append(partitions, *wal.partitionID)
+		}
+	}
+
+	return partitions, nil
+}
diff --git a/pkg/reduce/pnf/processandforward.go b/pkg/reduce/pnf/processandforward.go
index 526cee6f..543aa163 100644
--- a/pkg/reduce/pnf/processandforward.go
+++ b/pkg/reduce/pnf/processandforward.go
@@ -27,14 +27,15 @@ import (
 	"sync"
 	"time"
 
+	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/util/wait"
+
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb/forward"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
-	"go.uber.org/zap"
-	"k8s.io/apimachinery/pkg/util/wait"
 
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/pbq"
@@ -187,6 +188,8 @@ func (p *ProcessAndForward) writeToBuffer(ctx context.Context, bufferID string,
 		for i, message := range writeMessages {
 			if writeErrs[i] != nil {
 				failedMessages = append(failedMessages, message)
+			} else {
+				p.log.Debugw("Forwarded message", zap.String("bufferID", bufferID), zap.Any("message", message))
 			}
 		}
 		// retry only the failed messages
diff --git a/pkg/reduce/readloop/ordered.go b/pkg/reduce/readloop/ordered.go
index 5afddeab..6ae7757f 100644
--- a/pkg/reduce/readloop/ordered.go
+++ b/pkg/reduce/readloop/ordered.go
@@ -103,24 +103,21 @@ func (of *orderedForwarder) reduceOp(ctx context.Context, t *task) {
 		if err == nil {
 			break
 		} else if err == ctx.Err() {
-			of.log.Infow("ReduceOp exiting", zap.Error(ctx.Err()))
+			of.log.Infow("ReduceOp exiting", zap.String("partitionID", t.pf.PartitionID.String()), zap.Error(ctx.Err()))
 			return
 		}
-		of.log.Errorw("Process failed", zap.Error(err))
+		of.log.Errorw("Process failed", zap.String("partitionID", t.pf.PartitionID.String()), zap.Error(err))
 		time.Sleep(retryDelay)
 	}
 	// after retrying indicate that we are done with processing the package. the processing can move on
 	close(t.doneCh)
-	of.log.Debugw("Process->Reduce call took ", zap.Int64("duration(ms)", time.Since(start).Milliseconds()))
+	of.log.Debugw("Process->Reduce call took ", zap.String("partitionID", t.pf.PartitionID.String()), zap.Int64("duration(ms)", time.Since(start).Milliseconds()))
 	// notify that some work has been completed
 	select {
 	case of.taskDone <- struct{}{}:
 	case <-ctx.Done():
 		return
 	}
-
-	// TODO: remove this?
-	of.log.Debugw("Post to task done chan took ", zap.Int64("duration(ms)", time.Since(start).Milliseconds()))
 }
 
 // forward monitors the task queue, as soon as the task at the head of the queue has been completed, the result is
@@ -144,6 +141,8 @@ func (of *orderedForwarder) forward(ctx context.Context) {
 		// for every signal, we try to empty out the task-queue.
 		of.RLock()
 		n := of.taskQueue.Len()
+		of.log.Debugw("Received a signal in task queue ", zap.Int("task count", n))
+
 		of.RUnlock()
 		// n could be 0 because we have emptied the queue
 		if n == 0 {
@@ -174,9 +173,10 @@ func (of *orderedForwarder) forward(ctx context.Context) {
 				rm := currElement
 				currElement = currElement.Next()
 				of.taskQueue.Remove(rm)
+				of.log.Debugw("Removing task post forward call", zap.String("partitionID", t.pf.PartitionID.String()))
 				of.Unlock()
 			case <-ctx.Done():
-				of.log.Infow("Forward exiting while waiting on the head of the queue task", zap.Error(ctx.Err()))
+				of.log.Infow("Forward exiting while waiting on the head of the queue task", zap.String("partitionID", t.pf.PartitionID.String()), zap.Error(ctx.Err()))
 				return
 			}
 		}
diff --git a/pkg/reduce/readloop/readloop.go b/pkg/reduce/readloop/readloop.go
index 86cd008b..78aedf65 100644
--- a/pkg/reduce/readloop/readloop.go
+++ b/pkg/reduce/readloop/readloop.go
@@ -87,36 +87,36 @@ func NewReadLoop(ctx context.Context,
 
 // Startup starts up the read-loop, because during boot up, it has to replay the data from the persistent store of
 // PBQ before it can start reading from ISB. Startup will return only after the replay has been completed.
-func (rl *ReadLoop) Startup(ctx context.Context) {
+func (rl *ReadLoop) Startup(ctx context.Context) error {
 	// start the PBQManager which discovers and builds the state from persistent store of the PBQ.
-	rl.pbqManager.StartUp(ctx)
-	// gets the partitions from the state
-	partitions := rl.pbqManager.ListPartitions()
+	partitions, err := rl.pbqManager.GetExistingPartitions(ctx)
+	if err != nil {
+		return err
+	}
+
 	rl.log.Infow("Partitions to be replayed ", zap.Int("count", len(partitions)), zap.Any("partitions", partitions))
 
 	for _, p := range partitions {
 		// Create keyed window for a given partition
 		// so that the window can be closed when the watermark
 		// crosses the window.
-		id := p.PartitionID
-		intervalWindow := &keyed.AlignedKeyedWindow{
-			Start: id.Start,
-			End:   id.End,
-		}
+
+		alignedKeyedWindow := keyed.NewKeyedWindow(p.Start, p.End)
+
 		// These windows have to be recreated as they are completely in-memory
-		rl.windower.CreateWindow(intervalWindow)
+		rl.windower.CreateWindow(alignedKeyedWindow)
 
 		// create and invoke process and forward for the partition
-		rl.associatePBQAndPnF(ctx, p.PartitionID)
+		rl.associatePBQAndPnF(ctx, p)
 	}
 
 	// replays the data (replay internally writes the data from persistent store on to the PBQ)
 	rl.pbqManager.Replay(ctx)
+	return nil
 }
 
 // Process is one iteration of the read loop.
 func (rl *ReadLoop) Process(ctx context.Context, messages []*isb.ReadMessage) {
-
 	// There is no Cap on backoff because setting a Cap will result in
 	// backoff stopped once the duration exceeds the Cap
 	var pbqWriteBackoff = wait.Backoff{
@@ -126,67 +126,76 @@ func (rl *ReadLoop) Process(ctx context.Context, messages []*isb.ReadMessage) {
 		Jitter:   0.1,
 	}
 
-	for _, m := range messages {
-		// identify and add window for the message
-		var ctxClosedErr error
-		windows := rl.upsertWindowsAndKeys(m)
-		// for each window we will have a PBQ. A message could belong to multiple windows (e.g., sliding).
-		for _, kw := range windows {
-			// identify partition for message
-			partitionID := partition.ID{
-				Start: kw.StartTime(),
-				End:   kw.EndTime(),
-				Key:   m.Key,
-			}
-
-			q := rl.associatePBQAndPnF(ctx, partitionID)
+	select {
+	case <-ctx.Done():
+		return
+
+	default:
+		for _, m := range messages {
+			// identify and add window for the message
+			windows := rl.upsertWindowsAndKeys(m)
+			// for each window we will have a PBQ. A message could belong to multiple windows (e.g., sliding).
+			for _, kw := range windows {
+				// identify partition for message
+				partitionID := partition.ID{
+					Start: kw.StartTime(),
+					End:   kw.EndTime(),
+					Key:   m.Key,
+				}
 
-			// write the message to PBQ
-			attempt := 0
-			ctxClosedErr = wait.ExponentialBackoffWithContext(ctx, pbqWriteBackoff, func() (done bool, err error) {
-				rErr := q.Write(ctx, m)
-				attempt += 1
-				if rErr != nil {
-					rl.log.Errorw("Failed to write message", zap.Any("msgOffSet", m.ReadOffset.String()), zap.String("partitionID", partitionID.String()), zap.Any("attempt", attempt), zap.Error(rErr))
-					return false, nil
+				q := rl.associatePBQAndPnF(ctx, partitionID)
+				// write the message to PBQ
+				attempt := 0
+				timeoutErr := wait.ExponentialBackoff(pbqWriteBackoff, func() (done bool, err error) {
+					// FIXME(p0): temporarily fixing happy path
+					rErr := q.Write(context.Background(), m)
+					attempt += 1
+					if rErr != nil {
+						rl.log.Errorw("Failed to write message", zap.Any("msgOffSet", m.ReadOffset.String()), zap.String("partitionID", partitionID.String()), zap.Any("attempt", attempt), zap.Error(rErr))
+						return false, nil
+					}
+					return true, nil
+				})
+
+				if timeoutErr != nil {
+					rl.log.Errorw("Not able to write the messages to all partitions. To avoid message loss, this message will not be acked", zap.String("msgOffSet", m.ReadOffset.String()))
+					return
 				}
-				return true, nil
-			})
 
-			if ctxClosedErr != nil {
-				rl.log.Errorw("Error while writing the message to PBQ", zap.Error(ctxClosedErr))
-				return
 			}
 
 			// Ack the message to ISB
-			attempt = 0
-			ctxClosedErr = wait.ExponentialBackoffWithContext(ctx, pbqWriteBackoff, func() (done bool, err error) {
+			attempt := 0
+			timeoutErr := wait.ExponentialBackoff(pbqWriteBackoff, func() (done bool, err error) {
 				rErr := m.ReadOffset.AckIt()
 				attempt += 1
 				if rErr != nil {
 					rl.log.Errorw("Failed to ack message", zap.String("msgOffSet", m.ReadOffset.String()), zap.Int("attempt", attempt), zap.Error(rErr))
 					return false, nil
 				}
+				rl.log.Debugw("Successfully acked message", zap.String("msgOffSet", m.ReadOffset.String()))
+
 				return true, nil
 			})
 
-			if ctxClosedErr != nil {
-				rl.log.Errorw("Error while acknowledging the message", zap.Error(ctxClosedErr))
+			if timeoutErr != nil {
+				rl.log.Errorw("Timed out while trying to ack a message.", zap.String("msgOffSet", m.ReadOffset.String()))
 				return
 			}
-		}
 
-		// close any windows that need to be closed.
-		wm := processor.Watermark(m.Watermark)
-		closedWindows := rl.windower.RemoveWindows(time.Time(wm))
-		rl.log.Debugw("closing windows", zap.Int("length", len(closedWindows)), zap.Time("watermark", time.Time(wm)))
+			// close any windows that need to be closed.
+			wm := processor.Watermark(m.Watermark)
+			closedWindows := rl.windower.RemoveWindows(time.Time(wm))
+			rl.log.Infow("closing windows", zap.Int("length", len(closedWindows)), zap.Time("watermark", time.Time(wm)))
 
-		for _, cw := range closedWindows {
-			partitions := cw.Partitions()
-			rl.closePartitions(partitions)
-			rl.log.Debugw("Closing Window", zap.Time("windowStart", cw.StartTime()), zap.Time("windowEnd", cw.EndTime()))
+			for _, cw := range closedWindows {
+				partitions := cw.Partitions()
+				rl.closePartitions(partitions)
+				rl.log.Debugw("Closing Window", zap.Time("windowStart", cw.StartTime()), zap.Time("windowEnd", cw.EndTime()))
+			}
 		}
 	}
+
 }
 
 // associatePBQAndPnF associates a PBQ with the partition if a PBQ exists, else creates a new one and then associates
@@ -218,7 +227,9 @@ func (rl *ReadLoop) associatePBQAndPnF(ctx context.Context, partitionID partitio
 		// we should create and attach the read side of the loop (PnF) to the partition and then
 		// start process-and-forward (pnf) loop
 		rl.op.schedulePnF(ctx, rl.UDF, q, partitionID, rl.toBuffers, rl.whereToDecider, rl.publishWatermark)
+		rl.log.Debugw("Successfully Created/Found pbq and started PnF", zap.String("partitionID", partitionID.String()))
 	}
+
 	return q
 }
 
@@ -242,11 +253,12 @@ func (rl *ReadLoop) upsertWindowsAndKeys(m *isb.ReadMessage) []window.AlignedKey
 		w := rl.windower.GetWindow(win)
 		if w == nil {
 			w = rl.windower.CreateWindow(win)
+			rl.log.Debugw("Creating new keyed window", zap.Any("key", w.Keys()), zap.String("msg.offset", m.ID), zap.Int64("startTime", w.StartTime().UnixMilli()), zap.Int64("endTime", w.EndTime().UnixMilli()))
+		} else {
+			rl.log.Debugw("Found an existing window", zap.Any("key", w.Keys()), zap.String("msg.offset", m.ID), zap.Int64("startTime", w.StartTime().UnixMilli()), zap.Int64("endTime", w.EndTime().UnixMilli()))
 		}
 		// track the key to window relationship
 		w.AddKey(m.Key)
-		rl.log.Debugw("Creating new keyed window", zap.Any("key", w.Keys()), zap.Int64("startTime", w.StartTime().UnixMilli()), zap.Int64("endTime", w.EndTime().UnixMilli()))
-
 		kWindows = append(kWindows, w)
 	}
 	return kWindows
@@ -256,6 +268,7 @@ func (rl *ReadLoop) upsertWindowsAndKeys(m *isb.ReadMessage) []window.AlignedKey
 func (rl *ReadLoop) closePartitions(partitions []partition.ID) {
 	for _, p := range partitions {
 		q := rl.pbqManager.GetPBQ(p)
+		rl.log.Infow("Close of book", zap.String("partitionID", p.String()))
 		q.CloseOfBook()
 	}
 }
diff --git a/pkg/reduce/reduce.go b/pkg/reduce/reduce.go
index 3c54981a..650fa963 100644
--- a/pkg/reduce/reduce.go
+++ b/pkg/reduce/reduce.go
@@ -81,20 +81,29 @@ func NewDataForward(ctx context.Context,
 
 // Start starts forwarding messages to readloop
 func (d *DataForward) Start(ctx context.Context) {
-	d.readloop.Startup(ctx)
+	err := d.readloop.Startup(ctx)
+	if err != nil {
+		d.log.Errorw("Failed to start the data forwarder in reduce vertex", zap.Error(err))
+	}
+	rCtx, rCancel := context.WithCancel(context.Background())
 	for {
 		select {
 		case <-ctx.Done():
 			d.log.Infow("Stopping reduce data forwarder...")
 			cctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 			defer cancel()
-			d.readloop.ShutDown(cctx)
 
 			if err := d.fromBuffer.Close(); err != nil {
 				d.log.Errorw("Failed to close buffer reader, shutdown anyways...", zap.Error(err))
 			} else {
 				d.log.Infow("Closed buffer reader", zap.String("bufferFrom", d.fromBuffer.GetName()))
 			}
+
+			// cancel the readloop context so that all the messages are persisted and acked.
+			rCancel()
+			// allow readloop to clean itself up.
+			d.readloop.ShutDown(cctx)
+
 			for _, v := range d.toBuffers {
 				if err := v.Close(); err != nil {
 					d.log.Errorw("Failed to close buffer writer, shutdown anyways...", zap.Error(err), zap.String("bufferTo", v.GetName()))
@@ -116,7 +125,10 @@ func (d *DataForward) Start(ctx context.Context) {
 			}
 			return
 		default:
-			d.forwardAChunk(ctx)
+			// pass the child context so that the reader can be closed before the readloop
+			// this way we can avoid the race condition and have all the read messages persisted
+			// and acked.
+			d.forwardAChunk(rCtx)
 		}
 	}
 }
diff --git a/pkg/udf/reduce_udf.go b/pkg/udf/reduce_udf.go
index 690ed51f..8fc2b671 100644
--- a/pkg/udf/reduce_udf.go
+++ b/pkg/udf/reduce_udf.go
@@ -31,6 +31,8 @@ import (
 	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/pbq"
 	"github.com/numaproj/numaflow/pkg/pbq/store"
+	"github.com/numaproj/numaflow/pkg/pbq/store/memory"
+	"github.com/numaproj/numaflow/pkg/pbq/store/wal"
 	"github.com/numaproj/numaflow/pkg/reduce"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
@@ -137,9 +139,15 @@ func (u *ReduceUDFProcessor) Start(ctx context.Context) error {
 	}()
 	log.Infow("Start processing reduce udf messages", zap.String("isbsvc", string(u.ISBSvcType)), zap.String("from", fromBuffer.Name))
 
-	var pbqManager *pbq.Manager
-	pbqManager, err = pbq.NewManager(ctx, pbq.WithPBQStoreOptions(store.WithPbqStoreType(dfv1.InMemoryType)))
+	var storeProvider store.StoreProvider
+	if storage := u.VertexInstance.Vertex.Spec.UDF.GroupBy.Storage; storage != nil && storage.PersistentVolumeClaim != nil {
+		storeProvider = wal.NewWALStores(wal.WithStorePath(dfv1.DefaultStorePath), wal.WithMaxBufferSize(dfv1.DefaultStoreMaxBufferSize), wal.WithSyncDuration(dfv1.DefaultStoreSyncDuration))
+	} else {
+		// use in memory type by default
+		storeProvider = memory.NewMemoryStores()
+	}
 
+	pbqManager, err := pbq.NewManager(ctx, storeProvider)
 	if err != nil {
 		log.Errorw("Failed to create pbq manager", zap.Error(err))
 		return fmt.Errorf("failed to create pbq manager, %w", err)


Test patch:
diff --git a/pkg/isb/testutils/rw.go b/pkg/isb/testutils/rw.go
index 011f02bb..3b10fe54 100644
--- a/pkg/isb/testutils/rw.go
+++ b/pkg/isb/testutils/rw.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strconv"
 	"time"
 
 	"github.com/go-redis/redis/v8"
@@ -73,6 +74,22 @@ func BuildTestReadMessages(count int64, startTime time.Time) []isb.ReadMessage {
 	return readMessages
 }
 
+// BuildTestReadMessages builds test isb.ReadMessage which can be used for testing.
+func BuildTestReadMessagesIntOffset(count int64, startTime time.Time) []isb.ReadMessage {
+	writeMessages := BuildTestWriteMessages(count, startTime)
+	var readMessages = make([]isb.ReadMessage, count)
+
+	for idx, writeMessage := range writeMessages {
+		offset, _ := strconv.Atoi(writeMessage.Header.ID)
+		readMessages[idx] = isb.ReadMessage{
+			Message:    writeMessage,
+			ReadOffset: isb.SimpleIntOffset(func() int64 { return int64(offset) }),
+		}
+	}
+
+	return readMessages
+}
+
 // ReadMessagesLen is used to test the length of the messages read as they arrive on the stream
 // If a stream already has 5 elements which have been read and then we add another set of elements of 10 the total number would be 15.
 func ReadMessagesLen(ctx context.Context, options *redis.UniversalOptions, streamName string, expectedValue int64) bool {
diff --git a/pkg/pbq/options_test.go b/pkg/pbq/options_test.go
index c04d077a..7d3f8cf2 100644
--- a/pkg/pbq/options_test.go
+++ b/pkg/pbq/options_test.go
@@ -17,11 +17,10 @@ limitations under the License.
 package pbq
 
 import (
-	"github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/pbq/store"
-	"github.com/stretchr/testify/assert"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestOptions(t *testing.T) {
@@ -29,14 +28,12 @@ func TestOptions(t *testing.T) {
 		WithReadBatchSize(100),
 		WithChannelBufferSize(10),
 		WithReadTimeout(2 * time.Second),
-		WithPBQStoreOptions(store.WithPbqStoreType(v1alpha1.NoOpType), store.WithStoreSize(1000)),
 	}
 
 	queueOption := &options{
 		channelBufferSize: 5,
 		readTimeout:       1,
 		readBatchSize:     5,
-		storeOptions:      &store.StoreOptions{},
 	}
 
 	for _, opt := range testOpts {
diff --git a/pkg/pbq/pbq_test.go b/pkg/pbq/pbq_test.go
index 1d88a1ed..8dc8b8b6 100644
--- a/pkg/pbq/pbq_test.go
+++ b/pkg/pbq/pbq_test.go
@@ -22,11 +22,11 @@ import (
 	"testing"
 	"time"
 
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
 	"github.com/numaproj/numaflow/pkg/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/pbq/store"
+	"github.com/numaproj/numaflow/pkg/pbq/store/memory"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -34,14 +34,13 @@ import (
 
 func TestPBQ_ReadWrite(t *testing.T) {
 	// create a store of size 100 (it can store max 100 messages)
-	storeSize := 100
+	storeSize := int64(100)
 	// create a pbq with buffer size 5
 	buffSize := 5
 
 	ctx := context.Background()
 
-	qManager, _ := NewManager(ctx, WithChannelBufferSize(int64(buffSize)), WithReadTimeout(1*time.Second),
-		WithPBQStoreOptions(store.WithPbqStoreType(dfv1.InMemoryType), store.WithStoreSize(int64(storeSize))))
+	qManager, _ := NewManager(ctx, memory.NewMemoryStores(memory.WithStoreSize(storeSize)), WithChannelBufferSize(int64(buffSize)), WithReadTimeout(1*time.Second))
 
 	// write 10 isb messages to persisted store
 	msgCount := 10
@@ -94,7 +93,7 @@ func TestPBQ_ReadWrite(t *testing.T) {
 
 func Test_PBQReadWithCanceledContext(t *testing.T) {
 	// create a store of size 100 (it can store max 100 messages)
-	storeSize := 100
+	storeSize := int64(100)
 	//create a pbq with buffer size 10
 	bufferSize := 10
 	var err error
@@ -102,8 +101,7 @@ func Test_PBQReadWithCanceledContext(t *testing.T) {
 
 	ctx := context.Background()
 
-	qManager, err = NewManager(ctx, WithChannelBufferSize(int64(bufferSize)), WithReadTimeout(1*time.Second),
-		WithPBQStoreOptions(store.WithPbqStoreType(dfv1.InMemoryType), store.WithStoreSize(int64(storeSize))))
+	qManager, err = NewManager(ctx, memory.NewMemoryStores(memory.WithStoreSize(storeSize)), WithChannelBufferSize(int64(bufferSize)), WithReadTimeout(1*time.Second))
 
 	assert.NoError(t, err)
 
@@ -162,15 +160,14 @@ func Test_PBQReadWithCanceledContext(t *testing.T) {
 func TestPBQ_WriteWithStoreFull(t *testing.T) {
 
 	// create a store of size 100 (it can store max 100 messages)
-	storeSize := 100
+	storeSize := int64(100)
 	// create a pbq with buffer size 101
 	buffSize := 101
 	var qManager *Manager
 	var err error
 	ctx := context.Background()
 
-	qManager, err = NewManager(ctx, WithChannelBufferSize(int64(buffSize)), WithReadTimeout(1*time.Second),
-		WithPBQStoreOptions(store.WithPbqStoreType(dfv1.InMemoryType), store.WithStoreSize(int64(storeSize))))
+	qManager, err = NewManager(ctx, memory.NewMemoryStores(memory.WithStoreSize(storeSize)), WithChannelBufferSize(int64(buffSize)), WithReadTimeout(1*time.Second))
 	assert.NoError(t, err)
 
 	// write 101 isb messages to pbq, but the store size is 100, we should get store is full error
diff --git a/pkg/pbq/pbqmanager_test.go b/pkg/pbq/pbqmanager_test.go
index e1a8937c..a453e340 100644
--- a/pkg/pbq/pbqmanager_test.go
+++ b/pkg/pbq/pbqmanager_test.go
@@ -22,22 +22,23 @@ import (
 	"testing"
 	"time"
 
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/stretchr/testify/assert"
+
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
 	"github.com/numaproj/numaflow/pkg/pbq/partition"
-	"github.com/numaproj/numaflow/pkg/pbq/store"
 	"github.com/numaproj/numaflow/pkg/pbq/store/memory"
-	"github.com/stretchr/testify/assert"
+	"github.com/numaproj/numaflow/pkg/pbq/store/noop"
 )
 
 // tests for pbqManager (store type - in-memory)
 
 func TestManager_ListPartitions(t *testing.T) {
-	size := 100
+	size := int64(100)
+	t.Skip()
 
 	ctx := context.Background()
-	pbqManager, err := NewManager(ctx, WithPBQStoreOptions(store.WithStoreSize(int64(size)), store.WithPbqStoreType(dfv1.InMemoryType)),
+	pbqManager, err := NewManager(ctx, memory.NewMemoryStores(memory.WithStoreSize(size)),
 		WithReadTimeout(1*time.Second), WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -72,10 +73,10 @@ func TestManager_ListPartitions(t *testing.T) {
 }
 
 func TestManager_GetPBQ(t *testing.T) {
-	size := 100
+	size := int64(100)
 	var pb1, pb2 ReadWriteCloser
 	ctx := context.Background()
-	pbqManager, err := NewManager(ctx, WithPBQStoreOptions(store.WithStoreSize(int64(size)), store.WithPbqStoreType(dfv1.InMemoryType)),
+	pbqManager, err := NewManager(ctx, memory.NewMemoryStores(memory.WithStoreSize(size)),
 		WithReadTimeout(1*time.Second), WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -96,10 +97,10 @@ func TestManager_GetPBQ(t *testing.T) {
 
 // manager -> pbq -> store
 func TestPBQFlow(t *testing.T) {
-	size := 100
+	size := int64(100)
 
 	ctx := context.Background()
-	pbqManager, err := NewManager(ctx, WithPBQStoreOptions(store.WithStoreSize(int64(size)), store.WithPbqStoreType(dfv1.InMemoryType)),
+	pbqManager, err := NewManager(ctx, memory.NewMemoryStores(memory.WithStoreSize(size)),
 		WithReadTimeout(1*time.Second), WithChannelBufferSize(10))
 	assert.NoError(t, err)
 	testPartition := partition.ID{
@@ -157,10 +158,9 @@ func TestPBQFlow(t *testing.T) {
 }
 
 func TestPBQFlowWithNoOpStore(t *testing.T) {
-	size := 100
 
 	ctx := context.Background()
-	pbqManager, err := NewManager(ctx, WithPBQStoreOptions(store.WithStoreSize(int64(size)), store.WithPbqStoreType(dfv1.NoOpType)),
+	pbqManager, err := NewManager(ctx, noop.NewNoopStores(),
 		WithReadTimeout(1*time.Second), WithChannelBufferSize(10))
 	assert.NoError(t, err)
 	testPartition := partition.ID{
@@ -215,10 +215,10 @@ func TestPBQFlowWithNoOpStore(t *testing.T) {
 }
 
 func TestManager_Replay(t *testing.T) {
-	size := 100
+	size := int64(100)
 
 	ctx := context.Background()
-	pbqManager, err := NewManager(ctx, WithPBQStoreOptions(store.WithStoreSize(int64(size)), store.WithPbqStoreType(dfv1.InMemoryType)),
+	pbqManager, err := NewManager(ctx, memory.NewMemoryStores(memory.WithStoreSize(size)),
 		WithReadTimeout(1*time.Second), WithChannelBufferSize(10), WithReadBatchSize(10))
 	assert.NoError(t, err)
 	testPartition := partition.ID{
@@ -273,12 +273,8 @@ func TestManager_Replay(t *testing.T) {
 }
 
 func TestManager_StartUp(t *testing.T) {
-	size := 100
+	size := int64(100)
 
-	ctx := context.Background()
-	pbqManager, err := NewManager(ctx, WithPBQStoreOptions(store.WithStoreSize(int64(size)), store.WithPbqStoreType(dfv1.InMemoryType)),
-		WithReadTimeout(1*time.Second), WithChannelBufferSize(10), WithReadBatchSize(10))
-	assert.NoError(t, err)
 	pID1 := partition.ID{
 		Start: time.Now(),
 		End:   time.Now(),
@@ -290,13 +286,16 @@ func TestManager_StartUp(t *testing.T) {
 		End:   time.Now(),
 		Key:   "test-partition-2",
 	}
-	dp := func(storeOptions *store.StoreOptions) ([]partition.ID, error) {
+	dp := func(context.Context) ([]partition.ID, error) {
 		return []partition.ID{pID1,
 			pID2}, nil
 	}
-	memory.SetDiscoverer(dp)
-	pbqManager.StartUp(ctx)
-	assert.Len(t, pbqManager.ListPartitions(), 2)
-	assert.NotNil(t, pbqManager.GetPBQ(pID1))
-	assert.NotNil(t, pbqManager.GetPBQ(pID2))
+	stores := memory.NewMemoryStores(memory.WithStoreSize(size), memory.WithDiscoverer(dp))
+	ctx := context.Background()
+	pbqManager, err := NewManager(ctx, stores,
+		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/pbq/store/memory/store_test.go b/pkg/pbq/store/memory/store_test.go
index 78e08397..8c537ff6 100644
--- a/pkg/pbq/store/memory/store_test.go
+++ b/pkg/pbq/store/memory/store_test.go
@@ -21,20 +21,13 @@ import (
 	"testing"
 	"time"
 
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
 	"github.com/numaproj/numaflow/pkg/pbq/partition"
-	"github.com/numaproj/numaflow/pkg/pbq/store"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestMemoryStore_WriteToStore(t *testing.T) {
-	// create a store of size 100 (it can store max 100 messages)
-	storeSize := 100
-	options := &store.StoreOptions{}
-	_ = store.WithPbqStoreType(dfv1.InMemoryType)(options)
-	_ = store.WithStoreSize(int64(storeSize))(options)
 	ctx := context.Background()
 
 	partitionID := partition.ID{
@@ -42,8 +35,8 @@ func TestMemoryStore_WriteToStore(t *testing.T) {
 		End:   time.Unix(120, 0),
 		Key:   "new-partition",
 	}
-
-	memStore, err := NewMemoryStore(ctx, partitionID, options)
+	// create a store of size 100 (it can store max 100 messages)
+	memStore, err := NewMemoryStores(WithStoreSize(100)).CreateStore(ctx, partitionID)
 	assert.NoError(t, err)
 
 	//write 10 isb messages to persisted store
@@ -58,11 +51,6 @@ func TestMemoryStore_WriteToStore(t *testing.T) {
 }
 
 func TestMemoryStore_ReadFromStore(t *testing.T) {
-	// create a store of size 100 (it can store max 100 messages)
-	storeSize := 100
-	options := &store.StoreOptions{}
-	_ = store.WithPbqStoreType(dfv1.InMemoryType)(options)
-	_ = store.WithStoreSize(int64(storeSize))(options)
 	ctx := context.Background()
 
 	partitionID := partition.ID{
@@ -71,7 +59,8 @@ func TestMemoryStore_ReadFromStore(t *testing.T) {
 		Key:   "new-partition",
 	}
 
-	memStore, err := NewMemoryStore(ctx, partitionID, options)
+	// create a store of size 100 (it can store max 100 messages)
+	memStore, err := NewMemoryStores(WithStoreSize(100)).CreateStore(ctx, partitionID)
 	assert.NoError(t, err)
 
 	//write 10 isb messages to persisted store
@@ -92,9 +81,7 @@ func TestMemoryStore_ReadFromStore(t *testing.T) {
 
 func TestEmptyStore_Read(t *testing.T) {
 	// create a store of size 100 (it can store max 100 messages)
-	storeSize := 100
-	options := &store.StoreOptions{}
-	_ = store.WithPbqStoreType(dfv1.InMemoryType)(options)
+	storeSize := int64(100)
 	ctx := context.Background()
 
 	partitionID := partition.ID{
@@ -103,10 +90,10 @@ func TestEmptyStore_Read(t *testing.T) {
 		Key:   "new-partition",
 	}
 
-	memStore, err := NewMemoryStore(ctx, partitionID, options)
+	memStore, err := NewMemoryStores(WithStoreSize(storeSize)).CreateStore(ctx, partitionID)
 	assert.NoError(t, err)
 	var eof bool
-	_, eof, err = memStore.Read(int64(storeSize))
+	_, eof, err = memStore.Read(storeSize)
 	assert.NoError(t, err)
 	// since store is empty, eof will be true
 	assert.Equal(t, eof, true)
@@ -115,10 +102,7 @@ func TestEmptyStore_Read(t *testing.T) {
 
 func TestFullStore_Write(t *testing.T) {
 	// create a store of size 100 (it can store max 100 messages)
-	storeSize := 100
-	options := &store.StoreOptions{}
-	_ = store.WithPbqStoreType(dfv1.InMemoryType)(options)
-	_ = store.WithStoreSize(int64(storeSize))(options)
+	storeSize := int64(100)
 	ctx := context.Background()
 
 	partitionID := partition.ID{
@@ -127,7 +111,7 @@ func TestFullStore_Write(t *testing.T) {
 		Key:   "new-partition",
 	}
 
-	memStore, err := NewMemoryStore(ctx, partitionID, options)
+	memStore, err := NewMemoryStores(WithStoreSize(storeSize)).CreateStore(ctx, partitionID)
 	assert.NoError(t, err)
 
 	//write 100 isb messages to persisted store
diff --git a/pkg/pbq/store/options_test.go b/pkg/pbq/store/options_test.go
deleted file mode 100644
index 45ef9b7e..00000000
--- a/pkg/pbq/store/options_test.go
+++ /dev/null
@@ -1,50 +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 store
-
-import (
-	"github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/stretchr/testify/assert"
-	"testing"
-	"time"
-)
-
-func TestOptions(t *testing.T) {
-	testOpts := []StoreOption{
-		WithStoreSize(100),
-		WithMaxBufferSize(10),
-		WithPbqStoreType(v1alpha1.InMemoryType),
-		WithSyncDuration(2 * time.Second),
-	}
-
-	storeOptions := &StoreOptions{
-		maxBatchSize: 0,
-		syncDuration: 1 * time.Second,
-		pbqStoreType: "in-memory",
-		storeSize:    0,
-	}
-
-	for _, opt := range testOpts {
-		err := opt(storeOptions)
-		assert.NoError(t, err)
-	}
-
-	assert.Equal(t, int64(100), storeOptions.storeSize)
-	assert.Equal(t, int64(10), storeOptions.maxBatchSize)
-	assert.Equal(t, 2*time.Second, storeOptions.syncDuration)
-	assert.Equal(t, v1alpha1.InMemoryType, storeOptions.pbqStoreType)
-}
diff --git a/pkg/pbq/store/wal/segment_test.go b/pkg/pbq/store/wal/segment_test.go
new file mode 100644
index 00000000..b78affc9
--- /dev/null
+++ b/pkg/pbq/store/wal/segment_test.go
@@ -0,0 +1,358 @@
+/*
+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 (
+	"bytes"
+	"context"
+	"encoding/binary"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/isb/testutils"
+	"github.com/numaproj/numaflow/pkg/pbq/partition"
+)
+
+func Test_writeReadHeader(t *testing.T) {
+	id := partition.ID{
+		Start: time.Unix(1665109020, 0).In(location),
+		End:   time.Unix(1665109020, 0).Add(time.Minute).In(location),
+		Key:   "test1",
+	}
+
+	tmp := t.TempDir()
+	stores := NewWALStores(WithStorePath(tmp))
+	store, err := stores.CreateStore(context.Background(), id)
+	assert.NoError(t, err)
+	wal := store.(*WAL)
+	fName := wal.fp.Name()
+	assert.NoError(t, err)
+	// read will fail because the file was opened only in write only mode
+	_, err = wal.readHeader()
+	assert.Error(t, err)
+	err = wal.Close()
+	fmt.Println(fName)
+	assert.NoError(t, err)
+
+	openWAL, err := OpenWAL(fName)
+	assert.NoError(t, err)
+	// we have already read the header in OpenWAL
+	_, err = openWAL.readHeader()
+	assert.Error(t, err)
+
+	// compare the original ID with read ID
+	assert.Equal(t, id, *openWAL.partitionID)
+}
+
+func Test_encodeDecodeHeader(t *testing.T) {
+	tests := []struct {
+		name    string
+		id      *partition.ID
+		want    *bytes.Buffer
+		wantErr assert.ErrorAssertionFunc
+	}{
+		{
+			name:    "enc_dec_good",
+			wantErr: assert.NoError,
+			id: &partition.ID{
+				Start: time.Unix(1665109020, 0).In(location),
+				End:   time.Unix(1665109020, 0).Add(time.Minute).In(location),
+				Key:   "test1,test2",
+			},
+		},
+		{
+			name:    "enc_dec_nodata",
+			wantErr: assert.NoError,
+			id: &partition.ID{
+				Start: time.Time{},
+				End:   time.Time{},
+				Key:   "",
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := encodeHeader(tt.id)
+			if !tt.wantErr(t, err, fmt.Sprintf("encodeHeader(%v)", tt.id)) {
+				return
+			}
+			result, err := decodeHeader(got)
+			assert.NoError(t, err)
+			assert.Equalf(t, tt.id, result, "encodeHeader(%v)", tt.id)
+		})
+	}
+}
+
+func Test_runeEncoding(t *testing.T) {
+	var inputs = [][]rune{
+		{'世', '界'},
+		{'1', '2'},
+		{'g', 'o', 'o', 'd', ' ', 'b', 'y', 'e'},
+	}
+	for _, input := range inputs {
+		var err error
+		// write
+		var buf = new(bytes.Buffer)
+		for _, runeValue := range input {
+			err := binary.Write(buf, binary.LittleEndian, runeValue)
+			assert.NoError(t, err)
+		}
+
+		// read
+		var output = make([]rune, len(input))
+		err = binary.Read(buf, binary.LittleEndian, &output)
+		assert.NoError(t, err)
+
+		assert.NotEmpty(t, output[0]) // shouldn't be 0 at any cost
+		assert.Equal(t, input, output)
+	}
+}
+
+func Test_writeReadEntry(t *testing.T) {
+	id := partition.ID{
+		Start: time.Unix(1665109020, 0).In(location),
+		End:   time.Unix(1665109020, 0).Add(time.Minute).In(location),
+		Key:   "test1",
+	}
+
+	tmp := t.TempDir()
+	stores := NewWALStores(WithStorePath(tmp))
+	wal, err := stores.CreateStore(context.Background(), id)
+	assert.NoError(t, err)
+
+	startTime := time.Unix(1665109020, 0).In(location)
+	msgCount := 2
+	writeMessages := testutils.BuildTestReadMessagesIntOffset(int64(msgCount), startTime)
+	message := writeMessages[0]
+	err = wal.Write(&message)
+	assert.NoError(t, err)
+	err = wal.Close()
+	assert.NoError(t, err)
+
+	// Reopen the WAL for read and write.
+	store, err := stores.CreateStore(context.Background(), id)
+	assert.NoError(t, err)
+	newWal := store.(*WAL)
+	// we have already read the header in OpenWAL
+	_, err = newWal.readHeader()
+	assert.Error(t, err)
+
+	actualMessages, finished, err := newWal.Read(10000)
+	assert.NoError(t, err)
+	// Check we reach the end of file
+	assert.Equal(t, true, finished)
+	assert.Equal(t, newWal.readUpTo, newWal.rOffset)
+
+	assert.Len(t, actualMessages, 1)
+	actualMessage := actualMessages[0]
+	assert.Equalf(t, message.Message, actualMessage.Message, "Read(%v)", message.Message)
+	expectedOffset, err := message.ReadOffset.Sequence()
+	assert.NoError(t, err)
+	actualOffset, err := actualMessage.ReadOffset.Sequence()
+	assert.NoError(t, err)
+	assert.Equalf(t, expectedOffset, actualOffset, "Read(%v)", message.ReadOffset)
+	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeEntry(%v)", message.Watermark)
+
+	// Start to write an entry again
+	err = newWal.Write(&message)
+	assert.NoError(t, err)
+	err = newWal.Close()
+	assert.NoError(t, err)
+}
+
+func Test_encodeDecodeEntry(t *testing.T) {
+	// write 1 isb messages to persisted store
+	startTime := time.Unix(1665109020, 0).In(location)
+	writeMessages := testutils.BuildTestReadMessagesIntOffset(1, startTime)
+	firstMessage := writeMessages[0]
+	tests := []struct {
+		name    string
+		message *isb.ReadMessage
+		want    *bytes.Buffer
+		wantErr assert.ErrorAssertionFunc
+	}{
+		{
+			name:    "enc_dec_good",
+			wantErr: assert.NoError,
+			message: &firstMessage,
+		},
+		{
+			name:    "enc_dec_nodata",
+			wantErr: assert.NoError,
+			message: &isb.ReadMessage{
+				Message: isb.Message{
+					Header: isb.Header{},
+					Body:   isb.Body{},
+				},
+				ReadOffset: isb.SimpleIntOffset(func() int64 { return int64(2) }),
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := encodeEntry(tt.message)
+			if !tt.wantErr(t, err, fmt.Sprintf("encodeEntry(%v)", tt.message)) {
+				return
+			}
+
+			result, _, err := decodeEntry(bytes.NewReader(got.Bytes()))
+			assert.NoError(t, err)
+			assert.Equalf(t, tt.message.Message, result.Message, "encodeEntry(%v)", tt.message.Message)
+			expectedOffset, err := tt.message.ReadOffset.Sequence()
+			assert.NoError(t, err)
+			actualOffset, err := result.ReadOffset.Sequence()
+			assert.NoError(t, err)
+			assert.Equalf(t, expectedOffset, actualOffset, "encodeEntry(%v)", tt.message.ReadOffset)
+			assert.Equalf(t, tt.message.Watermark, result.Watermark, "encodeEntry(%v)", tt.message.Watermark)
+		})
+	}
+}
+
+func Test_batchSyncWithMaxBatchSize(t *testing.T) {
+	id := partition.ID{
+		Start: time.Unix(1665109020, 0).In(location),
+		End:   time.Unix(1665109020, 0).Add(time.Minute).In(location),
+		Key:   "test1",
+	}
+
+	tmp := t.TempDir()
+	stores := NewWALStores(WithStorePath(tmp))
+	wal, err := stores.CreateStore(context.Background(), id)
+	assert.NoError(t, err)
+
+	tempWAL := wal.(*WAL)
+	tempWAL.prevSyncedTime = time.Now()
+
+	startTime := time.Unix(1665109020, 0).In(location)
+	msgCount := 2
+	writeMessages := testutils.BuildTestReadMessagesIntOffset(int64(msgCount), startTime)
+	message := writeMessages[0]
+	err = wal.Write(&message)
+	assert.NoError(t, err)
+
+	assert.Equal(t, tempWAL.prevSyncedWOffset, int64(0))
+	assert.NoError(t, err)
+
+	tempWAL.walStores.maxBatchSize = 10
+	assert.NoError(t, err)
+	err = wal.Write(&message)
+	assert.NoError(t, err)
+	assert.Equal(t, tempWAL.prevSyncedWOffset, int64(844))
+
+	err = wal.Close()
+	assert.NoError(t, err)
+
+	// Reopen the WAL for read and write.
+	store, err := stores.CreateStore(context.Background(), id)
+	assert.NoError(t, err)
+	newWal := store.(*WAL)
+	// we have already read the header in OpenWAL
+	_, err = newWal.readHeader()
+	assert.Error(t, err)
+
+	actualMessages, finished, err := newWal.Read(10000)
+	assert.NoError(t, err)
+	// Check we reach the end of file
+	assert.Equal(t, true, finished)
+	assert.Equal(t, newWal.readUpTo, newWal.rOffset)
+
+	assert.Len(t, actualMessages, 2)
+	actualMessage := actualMessages[0]
+	assert.Equalf(t, message.Message, actualMessage.Message, "Read(%v)", message.Message)
+	expectedOffset, err := message.ReadOffset.Sequence()
+	assert.NoError(t, err)
+	actualOffset, err := actualMessage.ReadOffset.Sequence()
+	assert.NoError(t, err)
+	assert.Equalf(t, expectedOffset, actualOffset, "Read(%v)", message.ReadOffset)
+	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeEntry(%v)", message.Watermark)
+
+	// Start to write an entry again
+	err = newWal.Write(&message)
+	assert.NoError(t, err)
+	err = newWal.Close()
+	assert.NoError(t, err)
+}
+
+func Test_batchSyncWithSyncDuration(t *testing.T) {
+	id := partition.ID{
+		Start: time.Unix(1665109020, 0).In(location),
+		End:   time.Unix(1665109020, 0).Add(time.Minute).In(location),
+		Key:   "test1",
+	}
+
+	tmp := t.TempDir()
+	stores := NewWALStores(WithStorePath(tmp))
+	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
+	writeMessages := testutils.BuildTestReadMessagesIntOffset(int64(msgCount), startTime)
+	message := writeMessages[0]
+	storePrevSyncedTime := tempWAL.prevSyncedTime
+	err = wal.Write(&message)
+	assert.Equal(t, tempWAL.prevSyncedWOffset, int64(441))
+	assert.NotEqual(t, storePrevSyncedTime, tempWAL.prevSyncedTime)
+	assert.NoError(t, err)
+
+	storePrevSyncedTime = tempWAL.prevSyncedTime
+	tempWAL.walStores.syncDuration = 10 * time.Second
+	assert.NoError(t, err)
+	err = wal.Write(&message)
+	assert.NoError(t, err)
+	assert.Equal(t, tempWAL.prevSyncedTime, storePrevSyncedTime)
+
+	err = wal.Close()
+	assert.NoError(t, err)
+
+	// Reopen the WAL for read and write.
+	store, err := stores.CreateStore(context.Background(), id)
+	assert.NoError(t, err)
+	newWal := store.(*WAL)
+	// we have already read the header in OpenWAL
+	_, err = newWal.readHeader()
+	assert.Error(t, err)
+
+	actualMessages, finished, err := newWal.Read(10000)
+	assert.NoError(t, err)
+	// Check we reach the end of file
+	assert.Equal(t, true, finished)
+	assert.Equal(t, newWal.readUpTo, newWal.rOffset)
+
+	assert.Len(t, actualMessages, 2)
+	actualMessage := actualMessages[0]
+	assert.Equalf(t, message.Message, actualMessage.Message, "Read(%v)", message.Message)
+	expectedOffset, err := message.ReadOffset.Sequence()
+	assert.NoError(t, err)
+	actualOffset, err := actualMessage.ReadOffset.Sequence()
+	assert.NoError(t, err)
+	assert.Equalf(t, expectedOffset, actualOffset, "Read(%v)", message.ReadOffset)
+	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeEntry(%v)", message.Watermark)
+
+	// Start to write an entry again
+	err = newWal.Write(&message)
+	assert.NoError(t, err)
+	err = newWal.Close()
+	assert.NoError(t, err)
+}
diff --git a/pkg/reduce/pnf/processandforward_test.go b/pkg/reduce/pnf/processandforward_test.go
index 07ad7910..ed7c52b7 100644
--- a/pkg/reduce/pnf/processandforward_test.go
+++ b/pkg/reduce/pnf/processandforward_test.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
+	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"github.com/numaproj/numaflow/pkg/watermark/generic"
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
@@ -34,14 +35,15 @@ import (
 	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
 	"github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1/funcmock"
 	"github.com/numaproj/numaflow-go/pkg/function/clienttest"
+	"github.com/stretchr/testify/assert"
+
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
 	"github.com/numaproj/numaflow/pkg/pbq"
 	"github.com/numaproj/numaflow/pkg/pbq/partition"
-	"github.com/numaproj/numaflow/pkg/pbq/store"
+	"github.com/numaproj/numaflow/pkg/pbq/store/memory"
 	udfcall "github.com/numaproj/numaflow/pkg/udf/function"
 	wmstore "github.com/numaproj/numaflow/pkg/watermark/store"
-	"github.com/stretchr/testify/assert"
 )
 
 type myForwardTest struct {
@@ -80,7 +82,6 @@ func TestProcessAndForward_Process(t *testing.T) {
 		}
 	}()
 
-	size := 100
 	testPartition := partition.ID{
 		Start: time.UnixMilli(60000),
 		End:   time.UnixMilli(120000),
@@ -89,7 +90,7 @@ func TestProcessAndForward_Process(t *testing.T) {
 	var err error
 	var pbqManager *pbq.Manager
 
-	pbqManager, err = pbq.NewManager(ctx, pbq.WithPBQStoreOptions(store.WithStoreSize(int64(size)), store.WithPbqStoreType(dfv1.InMemoryType)),
+	pbqManager, err = pbq.NewManager(ctx, memory.NewMemoryStores(memory.WithStoreSize(100)),
 		pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -146,7 +147,7 @@ func TestProcessAndForward_Forward(t *testing.T) {
 
 	var pbqManager *pbq.Manager
 
-	pbqManager, _ = pbq.NewManager(ctx)
+	pbqManager, _ = pbq.NewManager(ctx, memory.NewMemoryStores())
 
 	test1Buffer1 := simplebuffer.NewInMemoryBuffer("buffer1", 10)
 	test1Buffer2 := simplebuffer.NewInMemoryBuffer("buffer2", 10)
@@ -279,7 +280,7 @@ func createProcessAndForward(ctx context.Context, key string, pbqManager *pbq.Ma
 		UDF:              nil,
 		result:           result,
 		pbqReader:        simplePbq,
-		log:              nil,
+		log:              logging.FromContext(ctx),
 		toBuffers:        toBuffers,
 		whereToDecider:   myForwardTest{},
 		publishWatermark: pw,
diff --git a/pkg/reduce/readloop/ordered_test.go b/pkg/reduce/readloop/ordered_test.go
index 267d3576..63c0f1a7 100644
--- a/pkg/reduce/readloop/ordered_test.go
+++ b/pkg/reduce/readloop/ordered_test.go
@@ -30,7 +30,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/pbq"
 	"github.com/numaproj/numaflow/pkg/pbq/partition"
-	"github.com/numaproj/numaflow/pkg/pbq/store"
+	"github.com/numaproj/numaflow/pkg/pbq/store/memory"
 	"github.com/numaproj/numaflow/pkg/udf/applier"
 	"github.com/stretchr/testify/assert"
 )
@@ -107,7 +107,7 @@ func TestOrderedProcessing(t *testing.T) {
 			op.startUp(ctx)
 			// although this could be declared outside, since we are using common naming scheme for partitions,
 			// things will go haywire.
-			pbqManager, _ := pbq.NewManager(ctx, pbq.WithPBQStoreOptions(store.WithStoreSize(int64(100)), store.WithPbqStoreType(dfv1.InMemoryType)),
+			pbqManager, _ := pbq.NewManager(ctx, memory.NewMemoryStores(memory.WithStoreSize(100)),
 				pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 			cCtx, cancelFn := context.WithCancel(ctx)
 			defer cancelFn()
diff --git a/pkg/reduce/reduce_test.go b/pkg/reduce/reduce_test.go
index a4324803..cc719d46 100644
--- a/pkg/reduce/reduce_test.go
+++ b/pkg/reduce/reduce_test.go
@@ -22,43 +22,51 @@ import (
 	"fmt"
 	"math"
 	"strconv"
+	"sync"
 	"testing"
 	"time"
 
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/stretchr/testify/assert"
+
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/stores/simplebuffer"
 	"github.com/numaproj/numaflow/pkg/pbq"
 	"github.com/numaproj/numaflow/pkg/pbq/partition"
-	"github.com/numaproj/numaflow/pkg/pbq/store"
+	"github.com/numaproj/numaflow/pkg/pbq/store/memory"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	wmstore "github.com/numaproj/numaflow/pkg/watermark/store"
 	"github.com/numaproj/numaflow/pkg/watermark/store/inmem"
 	"github.com/numaproj/numaflow/pkg/window/strategy/fixed"
-	"github.com/stretchr/testify/assert"
 )
 
-type EventTypeWMProgressor struct{}
-
-func (e EventTypeWMProgressor) PublishWatermark(watermark processor.Watermark, offset isb.Offset) {
+type EventTypeWMProgressor struct {
+	watermarks map[string]processor.Watermark
+	m          sync.Mutex
+}
 
+func (e *EventTypeWMProgressor) PublishWatermark(watermark processor.Watermark, offset isb.Offset) {
+	e.m.Lock()
+	defer e.m.Unlock()
+	e.watermarks[offset.String()] = watermark
 }
 
-func (e EventTypeWMProgressor) GetLatestWatermark() processor.Watermark {
+func (e *EventTypeWMProgressor) GetLatestWatermark() processor.Watermark {
 	return processor.Watermark{}
 }
 
-func (e EventTypeWMProgressor) Close() error {
+func (e *EventTypeWMProgressor) Close() error {
 	return nil
 }
 
-func (e EventTypeWMProgressor) GetWatermark(_ isb.Offset) processor.Watermark {
-	return processor.Watermark(time.Now())
+func (e *EventTypeWMProgressor) GetWatermark(offset isb.Offset) processor.Watermark {
+	e.m.Lock()
+	defer e.m.Unlock()
+	return e.watermarks[offset.String()]
 }
 
-func (e EventTypeWMProgressor) GetHeadWatermark() processor.Watermark {
+func (e *EventTypeWMProgressor) GetHeadWatermark() processor.Watermark {
 	return processor.Watermark{}
 }
 
@@ -180,8 +188,12 @@ func TestDataForward_StartWithNoOpWM(t *testing.T) {
 	fromBuffer := simplebuffer.NewInMemoryBuffer(fromBufferName, fromBufferSize)
 	to := simplebuffer.NewInMemoryBuffer(toBufferName, toBufferSize)
 
+	wmpublisher := &EventTypeWMProgressor{
+		watermarks: make(map[string]processor.Watermark),
+	}
+
 	// keep on writing <count> messages every 1 second for the supplied key
-	go writeMessages(child, 10, "no-op-test", fromBuffer, EventTypeWMProgressor{}, time.Second*1)
+	go writeMessages(child, 10, "no-op-test", fromBuffer, wmpublisher, time.Second*1)
 
 	toBuffer := map[string]isb.BufferWriter{
 		toBufferName: to,
@@ -191,19 +203,19 @@ func TestDataForward_StartWithNoOpWM(t *testing.T) {
 	var pbqManager *pbq.Manager
 
 	// create pbqManager
-	pbqManager, err = pbq.NewManager(child, pbq.WithPBQStoreOptions(store.WithStoreSize(100), store.WithPbqStoreType(dfv1.InMemoryType)),
+	pbqManager, err = pbq.NewManager(child, memory.NewMemoryStores(memory.WithStoreSize(100)),
 		pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
 	publisher := map[string]publish.Publisher{
-		"to": EventTypeWMProgressor{},
+		"to": &EventTypeWMProgressor{},
 	}
 
 	// create new fixed window of (windowTime)
 	window := fixed.NewFixed(windowTime)
 
 	var reduceDataForwarder *DataForward
-	reduceDataForwarder, err = NewDataForward(child, CounterReduceTest{}, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, EventTypeWMProgressor{}, publisher, window, WithReadBatchSize(10))
+	reduceDataForwarder, err = NewDataForward(child, CounterReduceTest{}, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, wmpublisher, publisher, window, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
 	go reduceDataForwarder.Start(child)
@@ -229,7 +241,7 @@ func TestDataForward_StartWithNoOpWM(t *testing.T) {
 	var readMessagePayload PayloadForTest
 	_ = json.Unmarshal(msgs[0].Payload, &readMessagePayload)
 	// here the count will always be equal to the number of messages written per key
-	assert.Equal(t, 1, readMessagePayload.Value)
+	assert.Equal(t, 10, readMessagePayload.Value)
 	assert.Equal(t, "count", readMessagePayload.Key)
 
 }
@@ -260,7 +272,7 @@ func TestReduceDataForward_Count(t *testing.T) {
 
 	// create pbq manager
 	var pbqManager *pbq.Manager
-	pbqManager, err = pbq.NewManager(ctx, pbq.WithPBQStoreOptions(store.WithStoreSize(1000), store.WithPbqStoreType(dfv1.InMemoryType)),
+	pbqManager, err = pbq.NewManager(ctx, memory.NewMemoryStores(memory.WithStoreSize(1000)),
 		pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -333,7 +345,7 @@ func TestReduceDataForward_Sum(t *testing.T) {
 
 	// create pbq manager
 	var pbqManager *pbq.Manager
-	pbqManager, err = pbq.NewManager(ctx, pbq.WithPBQStoreOptions(store.WithStoreSize(1000), store.WithPbqStoreType(dfv1.InMemoryType)),
+	pbqManager, err = pbq.NewManager(ctx, memory.NewMemoryStores(memory.WithStoreSize(1000)),
 		pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -406,7 +418,7 @@ func TestReduceDataForward_Max(t *testing.T) {
 
 	// create pbq manager
 	var pbqManager *pbq.Manager
-	pbqManager, err = pbq.NewManager(ctx, pbq.WithPBQStoreOptions(store.WithStoreSize(1000), store.WithPbqStoreType(dfv1.InMemoryType)),
+	pbqManager, err = pbq.NewManager(ctx, memory.NewMemoryStores(memory.WithStoreSize(1000)),
 		pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -479,7 +491,7 @@ func TestReduceDataForward_SumWithDifferentKeys(t *testing.T) {
 
 	// create pbq manager
 	var pbqManager *pbq.Manager
-	pbqManager, err = pbq.NewManager(ctx, pbq.WithPBQStoreOptions(store.WithStoreSize(1000), store.WithPbqStoreType(dfv1.InMemoryType)),
+	pbqManager, err = pbq.NewManager(ctx, memory.NewMemoryStores(memory.WithStoreSize(1000)),
 		pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
 	assert.NoError(t, err)
 
@@ -585,24 +597,22 @@ func fetcherAndPublisher(ctx context.Context, toBuffers map[string]isb.BufferWri
 
 // buildMessagesForReduce builds test isb.Message which can be used for testing reduce.
 func buildMessagesForReduce(count int, key string, publishTime time.Time) []isb.Message {
-	var messages = make([]isb.Message, 0, count)
+	var messages = make([]isb.Message, count)
 	for i := 0; i < count; i++ {
 		result, _ := json.Marshal(PayloadForTest{
 			Key:   key,
 			Value: i,
 		})
-		messages = append(messages,
-			isb.Message{
-				Header: isb.Header{
-					PaneInfo: isb.PaneInfo{
-						EventTime: publishTime,
-					},
-					ID:  fmt.Sprintf("%d", i),
-					Key: key,
+		messages[i] = isb.Message{
+			Header: isb.Header{
+				PaneInfo: isb.PaneInfo{
+					EventTime: publishTime,
 				},
-				Body: isb.Body{Payload: result},
+				ID:  fmt.Sprintf("%d", i),
+				Key: key,
 			},
-		)
+			Body: isb.Body{Payload: result},
+		}
 	}
 
 	return messages
@@ -610,27 +620,20 @@ func buildMessagesForReduce(count int, key string, publishTime time.Time) []isb.
 
 // writeMessages writes message to simple buffer and publishes the watermark using write offsets
 func writeMessages(ctx context.Context, count int, key string, fromBuffer *simplebuffer.InMemoryBuffer, publish publish.Publisher, interval time.Duration) {
-	generateTime := 100 * time.Millisecond
-	ticker := time.NewTicker(generateTime)
 	intervalDuration := interval
 	publishTime := time.Unix(60, 0)
 	i := 1
-	for {
-		select {
-		case <-ticker.C:
-			// build  messages with eventTime time.Now()
-			publishTime = publishTime.Add(intervalDuration)
-			messages := buildMessagesForReduce(count, key+strconv.Itoa(i), publishTime)
-			i++
-
-			// write the messages to fromBuffer, so that it will be available for consuming
-			offsets, _ := fromBuffer.Write(ctx, messages)
-			if len(offsets) > 0 {
-				publish.PublishWatermark(processor.Watermark(publishTime), offsets[len(offsets)-1])
-			}
-		case <-ctx.Done():
-			ticker.Stop()
-			return
+
+	for k := 0; k < 3; k++ {
+		// build  messages with eventTime set to publish time
+		publishTime = publishTime.Add(intervalDuration)
+		messages := buildMessagesForReduce(count, key+strconv.Itoa(i), publishTime)
+		i++
+
+		// write the messages to fromBuffer, so that it will be available for consuming
+		offsets, _ := fromBuffer.Write(ctx, messages)
+		for _, offset := range offsets {
+			publish.PublishWatermark(processor.Watermark(publishTime), offset)
 		}
 	}
 }