Repository: numaproj/numaflow
Pull request #216 (2022-10-13 22:52:18)

feat: in-memory watermark store for better testing
watermark unit tests with in mem store


Patch:
diff --git a/pkg/shared/clients/jetstream/nats_conn.go b/pkg/shared/clients/jetstream/nats_conn.go
index f7b65f49..18549a8d 100644
--- a/pkg/shared/clients/jetstream/nats_conn.go
+++ b/pkg/shared/clients/jetstream/nats_conn.go
@@ -33,7 +33,9 @@ func NewNatsConn(conn *nats.Conn) *NatsConn {
 
 // Close function closes the underlying Nats connection.
 func (nc *NatsConn) Close() {
-	nc.Conn.Close()
+	if !nc.Conn.IsClosed() {
+		nc.Conn.Close()
+	}
 }
 
 // JetStream function invokes same function of underlying Nats connection for returning,
diff --git a/pkg/watermark/fetch/processor_manager.go b/pkg/watermark/fetch/processor_manager.go
index e52dbe45..f0f2e75d 100644
--- a/pkg/watermark/fetch/processor_manager.go
+++ b/pkg/watermark/fetch/processor_manager.go
@@ -77,6 +77,7 @@ func (v *ProcessorManager) GetProcessor(processor string) *ProcessorToFetch {
 }
 
 // DeleteProcessor deletes a processor.
+// Note: This operation is not used anywhere ATM.
 func (v *ProcessorManager) DeleteProcessor(processor string) {
 	v.lock.Lock()
 	defer v.lock.Unlock()
@@ -150,8 +151,9 @@ func (v *ProcessorManager) startHeatBeatWatcher() {
 				var entity = processor.NewProcessorEntity(value.Key(), processor.WithSeparateOTBuckets(v.opts.separateOTBucket))
 				// do we have such a processor
 				p := v.GetProcessor(value.Key())
-				if p == nil { // if no, create a new processor
-					// A fromProcessor need to be added to v.processors
+				if p == nil {
+					// if p is nil, create a new processor
+					// A fromProcessor needs to be added to v.processors
 					// The fromProcessor may have been deleted
 					// TODO: make capacity configurable
 					var fromProcessor = NewProcessorToFetch(v.ctx, entity, 10, v.otWatcher)
diff --git a/pkg/watermark/store/inmem/kv_store.go b/pkg/watermark/store/inmem/kv_store.go
new file mode 100644
index 00000000..4247fcd3
--- /dev/null
+++ b/pkg/watermark/store/inmem/kv_store.go
@@ -0,0 +1,140 @@
+/*
+Package inmem package implements the watermark progression using in mem store as the KV store.
+*/
+package inmem
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"sync"
+
+	"go.uber.org/zap"
+
+	"github.com/numaproj/numaflow/pkg/shared/logging"
+	"github.com/numaproj/numaflow/pkg/watermark/store"
+)
+
+var (
+	buckets     = make(map[string]*inMemStore)
+	bucketsLock sync.RWMutex
+)
+
+// kvEntry is each key-value entry in the store and the operation associated with the kv pair.
+type kvEntry struct {
+	key   string
+	value []byte
+	op    store.KVWatchOp
+}
+
+// Key returns the key
+func (k kvEntry) Key() string {
+	return k.key
+}
+
+// Value returns the value.
+func (k kvEntry) Value() []byte {
+	return k.value
+}
+
+// Operation returns the operation on that key-value pair.
+func (k kvEntry) Operation() store.KVWatchOp {
+	return k.op
+}
+
+// inMemStore implements the watermark's KV store backed up by in mem store.
+type inMemStore struct {
+	pipelineName string
+	bucketName   string
+	kv           map[string][]byte
+	kvLock       sync.RWMutex
+	kvEntryCh    chan store.WatermarkKVEntry
+	log          *zap.SugaredLogger
+}
+
+var _ store.WatermarkKVStorer = (*inMemStore)(nil)
+
+// NewKVInMemKVStore returns inMemStore.
+func NewKVInMemKVStore(ctx context.Context, pipelineName string, bucketName string) (store.WatermarkKVStorer, chan store.WatermarkKVEntry, error) {
+	s := &inMemStore{
+		pipelineName: pipelineName,
+		bucketName:   bucketName,
+		kv:           make(map[string][]byte),
+		kvEntryCh:    make(chan store.WatermarkKVEntry, 10),
+		log:          logging.FromContext(ctx).With("pipeline", pipelineName).With("bucketName", bucketName),
+	}
+	bucketsLock.Lock()
+	buckets[bucketName] = s
+	bucketsLock.Unlock()
+	return s, s.kvEntryCh, nil
+}
+
+// GetAllKeys returns all the keys in the key-value store.
+func (kv *inMemStore) GetAllKeys(_ context.Context) ([]string, error) {
+	kv.kvLock.Lock()
+	defer kv.kvLock.Unlock()
+	var keys []string
+	for key := range kv.kv {
+		keys = append(keys, key)
+	}
+	sort.Slice(keys, func(i int, j int) bool {
+		return keys[i] < keys[j]
+	})
+	return keys, nil
+}
+
+// GetValue returns the value for a given key.
+func (kv *inMemStore) GetValue(_ context.Context, k string) ([]byte, error) {
+	kv.kvLock.RLock()
+	defer kv.kvLock.RUnlock()
+	if val, ok := kv.kv[k]; ok {
+		return val, nil
+	} else {
+		return []byte(""), fmt.Errorf("key %s not found", k)
+	}
+}
+
+// GetStoreName returns the store name.
+func (kv *inMemStore) GetStoreName() string {
+	return kv.bucketName
+}
+
+// DeleteKey deletes the key from the in mem key-value store.
+func (kv *inMemStore) DeleteKey(_ context.Context, k string) error {
+	kv.kvLock.Lock()
+	defer kv.kvLock.Unlock()
+	if val, ok := kv.kv[k]; ok {
+		delete(kv.kv, k)
+		kv.kvEntryCh <- kvEntry{
+			key:   k,
+			value: val,
+			op:    store.KVDelete,
+		}
+		return nil
+	} else {
+		return fmt.Errorf("key %s not found", k)
+	}
+}
+
+// PutKV puts an element to the in mem key-value store.
+func (kv *inMemStore) PutKV(_ context.Context, k string, v []byte) error {
+	kv.kvLock.Lock()
+	defer kv.kvLock.Unlock()
+	var val = make([]byte, len(v))
+	copy(val, v)
+	kv.kv[k] = val
+	kv.kvEntryCh <- kvEntry{
+		key:   k,
+		value: val,
+		op:    store.KVPut,
+	}
+	return nil
+}
+
+// Close closes the channel connection and clean up the bucket.
+func (kv *inMemStore) Close() {
+	close(kv.kvEntryCh)
+	bucketsLock.Lock()
+	delete(buckets, kv.bucketName)
+	bucketsLock.Unlock()
+}
diff --git a/pkg/watermark/store/inmem/kv_watch.go b/pkg/watermark/store/inmem/kv_watch.go
new file mode 100644
index 00000000..97f21fc3
--- /dev/null
+++ b/pkg/watermark/store/inmem/kv_watch.go
@@ -0,0 +1,88 @@
+package inmem
+
+import (
+	"context"
+	"sync"
+
+	"github.com/numaproj/numaflow/pkg/shared/logging"
+	"github.com/numaproj/numaflow/pkg/shared/util"
+	"github.com/numaproj/numaflow/pkg/watermark/store"
+	"go.uber.org/zap"
+)
+
+// inMemWatch implements the watermark's KV store backed up by in memory store.
+type inMemWatch struct {
+	pipelineName string
+	bucketName   string
+	kvEntryCh    <-chan store.WatermarkKVEntry
+	kvHistory    []store.WatermarkKVEntry
+	updatesChMap map[string]chan store.WatermarkKVEntry
+	lock         sync.Mutex // the lock for kvHistory and updatesChMap
+	log          *zap.SugaredLogger
+}
+
+var _ store.WatermarkKVWatcher = (*inMemWatch)(nil)
+
+// NewInMemWatch returns inMemWatch which implements the WatermarkKVWatcher interface.
+func NewInMemWatch(ctx context.Context, pipelineName string, bucketName string, kvEntryCh <-chan store.WatermarkKVEntry) (store.WatermarkKVWatcher, error) {
+	k := &inMemWatch{
+		pipelineName: pipelineName,
+		bucketName:   bucketName,
+		kvEntryCh:    kvEntryCh,
+		kvHistory:    []store.WatermarkKVEntry{},
+		updatesChMap: make(map[string]chan store.WatermarkKVEntry),
+		log:          logging.FromContext(ctx).With("pipeline", pipelineName).With("bucketName", bucketName),
+	}
+	return k, nil
+}
+
+// Watch watches the key-value store.
+func (k *inMemWatch) Watch(ctx context.Context) <-chan store.WatermarkKVEntry {
+	// create a new updates channel and fill in the history
+	var id = util.RandomString(10)
+	var updates = make(chan store.WatermarkKVEntry)
+
+	// for new updates channel initialization
+	go func() {
+		k.lock.Lock()
+		for _, value := range k.kvHistory {
+			updates <- value
+		}
+		// add the new updates channel to the map before unlock
+		// so the new updates channel won't miss new kv operations
+		k.updatesChMap[id] = updates
+		k.lock.Unlock()
+	}()
+
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				k.log.Infow("stopping watching", zap.String("watcher", k.GetKVName()))
+				k.lock.Lock()
+				delete(k.updatesChMap, id)
+				close(updates)
+				k.lock.Unlock()
+				return
+			case value := <-k.kvEntryCh:
+				k.log.Debug(value.Key(), value.Value(), value.Operation())
+				k.lock.Lock()
+				k.kvHistory = append(k.kvHistory, value)
+				for _, ch := range k.updatesChMap {
+					ch <- value
+				}
+				k.lock.Unlock()
+			}
+		}
+	}()
+	return updates
+}
+
+// GetKVName returns the KV store (bucket) name.
+func (k *inMemWatch) GetKVName() string {
+	return k.bucketName
+}
+
+// Close does nothing here because the udpates channel has already been closed when exiting the Watch.
+func (k *inMemWatch) Close() {
+}
diff --git a/pkg/watermark/store/interface.go b/pkg/watermark/store/interface.go
index 5c55ecb8..6d0968d0 100644
--- a/pkg/watermark/store/interface.go
+++ b/pkg/watermark/store/interface.go
@@ -35,6 +35,7 @@ const (
 	// KVDelete represents a delete.
 	KVDelete
 	// KVPurge is when the kv bucket is purged.
+	// This value is only for JetStream.
 	KVPurge
 )
 
diff --git a/pkg/watermark/store/jetstream/kv_store.go b/pkg/watermark/store/jetstream/kv_store.go
index 21689534..c36c82ad 100644
--- a/pkg/watermark/store/jetstream/kv_store.go
+++ b/pkg/watermark/store/jetstream/kv_store.go
@@ -37,7 +37,9 @@ func NewKVJetStreamKVStore(ctx context.Context, pipelineName string, bucketName
 	// do we need to specify any opts? if yes, send it via options.
 	js, err := conn.JetStream(nats.PublishAsyncMaxPending(256))
 	if err != nil {
-		conn.Close()
+		if !conn.IsClosed() {
+			conn.Close()
+		}
 		return nil, fmt.Errorf("failed to get JetStream context for writer")
 	}
 
@@ -48,13 +50,20 @@ func NewKVJetStreamKVStore(ctx context.Context, pipelineName string, bucketName
 		log:          logging.FromContext(ctx).With("pipeline", pipelineName).With("bucketName", bucketName),
 	}
 
+	// for JetStream KeyValue store, the bucket should have been created in advance
 	j.kv, err = j.js.KeyValue(bucketName)
 	if err != nil {
+		if !conn.IsClosed() {
+			conn.Close()
+		}
 		return nil, err
 	}
 	// options if any
 	for _, o := range opts {
 		if err := o(j); err != nil {
+			if !conn.IsClosed() {
+				conn.Close()
+			}
 			return nil, err
 		}
 	}
@@ -91,11 +100,13 @@ func (kv *jetStreamStore) GetStoreName() string {
 
 // DeleteKey deletes the key from the JS key-value store.
 func (kv *jetStreamStore) DeleteKey(_ context.Context, k string) error {
+	// will return error if nats connection is closed
 	return kv.kv.Delete(k)
 }
 
 // PutKV puts an element to the JS key-value store.
 func (kv *jetStreamStore) PutKV(_ context.Context, k string, v []byte) error {
+	// will return error if nats connection is closed
 	_, err := kv.kv.Put(k, v)
 	return err
 }
diff --git a/pkg/watermark/store/jetstream/kv_watch.go b/pkg/watermark/store/jetstream/kv_watch.go
index 771872f7..a6d96f13 100644
--- a/pkg/watermark/store/jetstream/kv_watch.go
+++ b/pkg/watermark/store/jetstream/kv_watch.go
@@ -35,7 +35,9 @@ func NewKVJetStreamKVWatch(ctx context.Context, pipelineName string, kvBucketNam
 	// do we need to specify any opts? if yes, send it via options.
 	js, err := conn.JetStream(nats.PublishAsyncMaxPending(256))
 	if err != nil {
-		conn.Close()
+		if !conn.IsClosed() {
+			conn.Close()
+		}
 		return nil, fmt.Errorf("failed to get JetStream context for writer")
 	}
 
@@ -48,6 +50,9 @@ func NewKVJetStreamKVWatch(ctx context.Context, pipelineName string, kvBucketNam
 
 	j.kv, err = j.js.KeyValue(kvBucketName)
 	if err != nil {
+		if !conn.IsClosed() {
+			conn.Close()
+		}
 		return nil, err
 	}
 
@@ -56,6 +61,9 @@ func NewKVJetStreamKVWatch(ctx context.Context, pipelineName string, kvBucketNam
 	// options if any
 	for _, o := range opts {
 		if err := o(j); err != nil {
+			if !conn.IsClosed() {
+				conn.Close()
+			}
 			return nil, err
 		}
 	}
@@ -101,9 +109,15 @@ func (k *jetStreamWatch) Watch(ctx context.Context) <-chan store.WatermarkKVEntr
 		for {
 			select {
 			case <-ctx.Done():
-				k.log.Errorw("stopping WatchAll", zap.String("watcher", k.GetKVName()))
+				k.log.Infow("stopping WatchAll", zap.String("watcher", k.GetKVName()))
 				// call jetstream watch stop
-				_ = kvWatcher.Stop()
+				err = kvWatcher.Stop()
+				if err != nil {
+					k.log.Errorw("failed to stop", zap.String("watcher", k.GetKVName()), zap.Error(err))
+				} else {
+					k.log.Infow("WatchAll successfully stopped", zap.String("watcher", k.GetKVName()))
+				}
+				close(updates)
 				return
 			case value := <-kvWatcher.Updates():
 				// if channel is closed, nil could come in
@@ -140,6 +154,8 @@ func (k *jetStreamWatch) GetKVName() string {
 
 // Close closes the connection.
 func (k *jetStreamWatch) Close() {
+	// need to cancel the `Watch` ctx before calling Close()
+	// otherwise `kvWatcher.Stop()` will raise the nats connection is closed error
 	if !k.conn.IsClosed() {
 		k.conn.Close()
 	}


Test patch:
diff --git a/pkg/watermark/fetch/edge_fetcher_test.go b/pkg/watermark/fetch/edge_fetcher_test.go
index 4cea1155..ab032edc 100644
--- a/pkg/watermark/fetch/edge_fetcher_test.go
+++ b/pkg/watermark/fetch/edge_fetcher_test.go
@@ -1,5 +1,3 @@
-//go:build isb_jetstream
-
 package fetch
 
 import (
@@ -8,52 +6,24 @@ import (
 	"testing"
 	"time"
 
-	"github.com/nats-io/nats.go"
+	"github.com/numaproj/numaflow/pkg/watermark/store/noop"
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap/zaptest"
 
 	"github.com/numaproj/numaflow/pkg/isb"
-	jsclient "github.com/numaproj/numaflow/pkg/shared/clients/jetstream"
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/store"
-	"github.com/numaproj/numaflow/pkg/watermark/store/jetstream"
 )
 
-func createAndLaterDeleteBucket(js *jsclient.JetStreamContext, kvConfig *nats.KeyValueConfig) (func(), error) {
-	kv, err := js.CreateKeyValue(kvConfig)
-	if err != nil {
-		return nil, err
-	}
-	return func() {
-		// ignore error as this will be executed only via defer
-		_ = js.DeleteKeyValue(kv.Bucket())
-		return
-	}, nil
-}
-
 func TestBuffer_GetWatermark(t *testing.T) {
 	var ctx = context.Background()
-	defaultJetStreamClient := jsclient.NewDefaultJetStreamClient(nats.DefaultURL)
-	conn, err := defaultJetStreamClient.Connect(ctx)
-	assert.NoError(t, err)
-	js, err := conn.JetStream()
-	assert.NoError(t, err)
 
-	var publisherHBBucketName = "fetchTest_PROCESSORS"
-	deleteFn, err := createAndLaterDeleteBucket(js, &nats.KeyValueConfig{Bucket: publisherHBBucketName})
-	assert.NoError(t, err)
-	defer deleteFn()
-
-	var publisherOTBucketName = "fetchTest_OT"
-	deleteFn, err = createAndLaterDeleteBucket(js, &nats.KeyValueConfig{Bucket: publisherOTBucketName})
-	assert.NoError(t, err)
-	defer deleteFn()
-
-	hbWatcher, err := jetstream.NewKVJetStreamKVWatch(ctx, "testFetch", publisherHBBucketName, defaultJetStreamClient)
-	otWatcher, err := jetstream.NewKVJetStreamKVWatch(ctx, "testFetch", publisherOTBucketName, defaultJetStreamClient)
+	// We don't really need watcher because we manually call the `Put` function and the `addProcessor` function
+	// so use no op watcher for testing
+	hbWatcher := noop.NewKVOpWatch()
+	otWatcher := noop.NewKVOpWatch()
 	processorManager := NewProcessorManager(ctx, store.BuildWatermarkStoreWatcher(hbWatcher, otWatcher))
 	var (
-		// TODO: watcher should not be nil
 		testPod0     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod1"), 5, otWatcher)
 		testPod1     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod2"), 5, otWatcher)
 		testPod2     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod3"), 5, otWatcher)
@@ -125,7 +95,7 @@ func TestBuffer_GetWatermark(t *testing.T) {
 			want:             10,
 		},
 		{
-			name:             "offset_22",
+			name:             "offset_23",
 			processorManager: processorManager,
 			args:             args{23},
 			want:             10,
@@ -137,7 +107,7 @@ func TestBuffer_GetWatermark(t *testing.T) {
 			want:             14,
 		},
 		{
-			name:             "offset_28",
+			name:             "offset_29",
 			processorManager: processorManager,
 			args:             args{29},
 			want:             17,
diff --git a/pkg/watermark/fetch/processor_manager_inmem_test.go b/pkg/watermark/fetch/processor_manager_inmem_test.go
new file mode 100644
index 00000000..a5679748
--- /dev/null
+++ b/pkg/watermark/fetch/processor_manager_inmem_test.go
@@ -0,0 +1,216 @@
+package fetch
+
+import (
+	"context"
+	"encoding/binary"
+	"fmt"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/numaproj/numaflow/pkg/watermark/store/inmem"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/watermark/store"
+)
+
+func TestFetcherWithSameOTBucket_InMem(t *testing.T) {
+	var (
+		err          error
+		pipelineName       = "testFetch"
+		keyspace           = "fetcherTest"
+		hbBucketName       = keyspace + "_PROCESSORS"
+		otBucketName       = keyspace + "_OT"
+		epoch        int64 = 1651161600
+		testOffset   int64 = 100
+		wg           sync.WaitGroup
+	)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+	hb, hbWatcherCh, err := inmem.NewKVInMemKVStore(ctx, pipelineName, hbBucketName)
+	assert.NoError(t, err)
+	defer hb.Close()
+	ot, otWatcherCh, err := inmem.NewKVInMemKVStore(ctx, pipelineName, otBucketName)
+	assert.NoError(t, err)
+	defer ot.Close()
+
+	b := make([]byte, 8)
+	binary.LittleEndian.PutUint64(b, uint64(testOffset))
+	// this key format is meant for non-separate OT watcher
+	err = ot.PutKV(ctx, fmt.Sprintf("%s%s%d", "p1", "_", epoch), b)
+	assert.NoError(t, err)
+
+	epoch += 60
+	binary.LittleEndian.PutUint64(b, uint64(testOffset+5))
+	// this key format is meant for non-separate OT watcher
+	err = ot.PutKV(ctx, fmt.Sprintf("%s%s%d", "p2", "_", epoch), b)
+	assert.NoError(t, err)
+
+	hbWatcher, err := inmem.NewInMemWatch(ctx, "testFetch", keyspace+"_PROCESSORS", hbWatcherCh)
+	assert.NoError(t, err)
+	defer hbWatcher.Close()
+	otWatcher, err := inmem.NewInMemWatch(ctx, "testFetch", keyspace+"_OT", otWatcherCh)
+	assert.NoError(t, err)
+	defer otWatcher.Close()
+	var testVertex = NewProcessorManager(ctx, store.BuildWatermarkStoreWatcher(hbWatcher, otWatcher), WithPodHeartbeatRate(1), WithRefreshingProcessorsRate(1), WithSeparateOTBuckets(false))
+	var testBuffer = NewEdgeFetcher(ctx, "testBuffer", testVertex).(*edgeFetcher)
+
+	// start p1 heartbeat for 3 loops
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		var err error
+		for i := 0; i < 3; i++ {
+			err = hb.PutKV(ctx, "p1", []byte(fmt.Sprintf("%d", time.Now().Unix())))
+			assert.NoError(t, err)
+			time.Sleep(time.Duration(testVertex.opts.podHeartbeatRate) * time.Second)
+		}
+		err = hb.DeleteKey(ctx, "p1")
+		assert.NoError(t, err)
+	}()
+
+	// start p2 heartbeat for 20 loops (20 seconds)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		var err error
+		for i := 0; i < 20; i++ {
+			err = hb.PutKV(ctx, "p2", []byte(fmt.Sprintf("%d", time.Now().Unix())))
+			assert.NoError(t, err)
+			time.Sleep(time.Duration(testVertex.opts.podHeartbeatRate) * time.Second)
+		}
+	}()
+
+	allProcessors := testBuffer.processorManager.GetAllProcessors()
+	for len(allProcessors) != 2 {
+		select {
+		case <-ctx.Done():
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatalf("expected 2 processors, got %d: %s", len(allProcessors), ctx.Err())
+			}
+		default:
+			time.Sleep(1 * time.Millisecond)
+			allProcessors = testBuffer.processorManager.GetAllProcessors()
+		}
+	}
+
+	assert.True(t, allProcessors["p1"].IsActive())
+	assert.True(t, allProcessors["p2"].IsActive())
+
+	// "p1" status becomes deleted after 3 loops
+	for !allProcessors["p1"].IsDeleted() {
+		select {
+		case <-ctx.Done():
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatalf("expected p1 to be deleted: %s", ctx.Err())
+			}
+		default:
+			time.Sleep(1 * time.Millisecond)
+			allProcessors = testBuffer.processorManager.GetAllProcessors()
+		}
+	}
+
+	allProcessors = testBuffer.processorManager.GetAllProcessors()
+	assert.Equal(t, 2, len(allProcessors))
+	assert.True(t, allProcessors["p1"].IsDeleted())
+	assert.True(t, allProcessors["p2"].IsActive())
+
+	_ = testBuffer.GetWatermark(isb.SimpleOffset(func() string { return strconv.FormatInt(testOffset, 10) }))
+	allProcessors = testBuffer.processorManager.GetAllProcessors()
+	assert.Equal(t, 2, len(allProcessors))
+	assert.True(t, allProcessors["p1"].IsDeleted())
+	assert.True(t, allProcessors["p2"].IsActive())
+	_ = testBuffer.GetWatermark(isb.SimpleOffset(func() string { return strconv.FormatInt(testOffset+3, 10) }))
+	allProcessors = testBuffer.processorManager.GetAllProcessors()
+	// we don't delete inactive processors from processor manager, so the length should still be 2
+	assert.Equal(t, 2, len(allProcessors))
+	assert.True(t, allProcessors["p1"].IsDeleted())
+	assert.True(t, allProcessors["p2"].IsActive())
+
+	time.Sleep(time.Second)
+	// resume p1 after one second
+	// run for 5 loops
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		var err error
+		for i := 0; i < 5; i++ {
+			err = hb.PutKV(ctx, "p1", []byte(fmt.Sprintf("%d", time.Now().Unix())))
+			assert.NoError(t, err)
+			time.Sleep(time.Duration(testVertex.opts.podHeartbeatRate) * time.Second)
+		}
+	}()
+
+	// wait until p1 becomes active
+	time.Sleep(time.Duration(testVertex.opts.podHeartbeatRate) * time.Second)
+	allProcessors = testBuffer.processorManager.GetAllProcessors()
+	for !allProcessors["p1"].IsActive() {
+		select {
+		case <-ctx.Done():
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatalf("expected p1 to be active: %s", ctx.Err())
+			}
+		default:
+			time.Sleep(1 * time.Millisecond)
+			allProcessors = testBuffer.processorManager.GetAllProcessors()
+		}
+	}
+	assert.True(t, allProcessors["p1"].IsActive())
+	assert.True(t, allProcessors["p2"].IsActive())
+	// "p1" should still have the same offsetTimeline
+	_ = testBuffer.GetWatermark(isb.SimpleOffset(func() string { return strconv.FormatInt(testOffset+1, 10) }))
+	p1 := testBuffer.processorManager.GetProcessor("p1")
+	assert.Equal(t, int64(100), p1.offsetTimeline.GetHeadOffset())
+
+	// publish a new watermark 101
+	binary.LittleEndian.PutUint64(b, uint64(testOffset+1))
+	err = ot.PutKV(ctx, fmt.Sprintf("%s%s%d", "p1", "_", epoch), b)
+	assert.NoError(t, err)
+
+	// "p1" becomes inactive after 5 loops
+	for !allProcessors["p1"].IsInactive() {
+		select {
+		case <-ctx.Done():
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatalf("expected p1 to be inactive: %s", ctx.Err())
+			}
+		default:
+			time.Sleep(1 * time.Millisecond)
+			allProcessors = testBuffer.processorManager.GetAllProcessors()
+		}
+	}
+
+	time.Sleep(time.Second)
+
+	// resume after one second
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		var err error
+		for i := 0; i < 3; i++ {
+			err = hb.PutKV(ctx, "p1", []byte(fmt.Sprintf("%d", time.Now().Unix())))
+			assert.NoError(t, err)
+			time.Sleep(time.Duration(testVertex.opts.podHeartbeatRate) * time.Second)
+		}
+	}()
+
+	allProcessors = testBuffer.processorManager.GetAllProcessors()
+	for len(allProcessors) != 2 {
+		select {
+		case <-ctx.Done():
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatalf("expected 2 processors, got %d: %s", len(allProcessors), ctx.Err())
+			}
+		default:
+			time.Sleep(1 * time.Millisecond)
+			allProcessors = testBuffer.processorManager.GetAllProcessors()
+		}
+	}
+
+	// added 101 in the previous steps for p1, so the head should be 101 after resume
+	assert.Equal(t, int64(101), p1.offsetTimeline.GetHeadOffset())
+	wg.Wait()
+	cancel()
+}
diff --git a/pkg/watermark/fetch/processor_manager_test.go b/pkg/watermark/fetch/processor_manager_test.go
index e4e94e60..0e842951 100644
--- a/pkg/watermark/fetch/processor_manager_test.go
+++ b/pkg/watermark/fetch/processor_manager_test.go
@@ -1,4 +1,4 @@
-//go:build isb_jetstream
+// //go:build isb_jetstream
 
 package fetch
 
@@ -7,6 +7,7 @@ import (
 	"encoding/binary"
 	"fmt"
 	"strconv"
+	"sync"
 	"testing"
 	"time"
 
@@ -20,20 +21,26 @@ import (
 )
 
 func TestFetcherWithSameOTBucket(t *testing.T) {
-	var ctx = context.Background()
+	var (
+		keyspace         = "fetcherTest"
+		epoch      int64 = 1651161600
+		testOffset int64 = 100
+		wg         sync.WaitGroup
+	)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
 
-	// Connect to NATS
+	// connect to NATS
 	nc, err := jsclient.NewDefaultJetStreamClient(nats.DefaultURL).Connect(context.TODO())
-	assert.Nil(t, err)
+	assert.NoError(t, err)
+	defer nc.Close()
 
-	// Create JetStream Context
+	// create JetStream Context
 	js, err := nc.JetStream(nats.PublishAsyncMaxPending(256))
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 
 	// create heartbeat bucket
-	var keyspace = "fetcherTest"
-
-	heartbeatBucket, err := js.CreateKeyValue(&nats.KeyValueConfig{
+	_, err = js.CreateKeyValue(&nats.KeyValueConfig{
 		Bucket:       keyspace + "_PROCESSORS",
 		Description:  fmt.Sprintf("[%s] heartbeat bucket", keyspace),
 		MaxValueSize: 0,
@@ -45,10 +52,10 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 		Placement:    nil,
 	})
 	defer func() { _ = js.DeleteKeyValue(keyspace + "_PROCESSORS") }()
-	assert.Nil(t, err)
-	var epoch int64 = 1651161600
-	var testOffset int64 = 100
-	ot, _ := js.CreateKeyValue(&nats.KeyValueConfig{
+	assert.NoError(t, err)
+
+	// create offset timeline bucket
+	_, err = js.CreateKeyValue(&nats.KeyValueConfig{
 		Bucket:       keyspace + "_OT",
 		Description:  "",
 		MaxValueSize: 0,
@@ -60,51 +67,73 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 		Placement:    nil,
 	})
 	defer func() { _ = js.DeleteKeyValue(keyspace + "_OT") }()
+	assert.NoError(t, err)
+
+	// create default JetStream client
+	defaultJetStreamClient := jsclient.NewDefaultJetStreamClient(nats.DefaultURL)
+
+	// create hbStore
+	hbStore, err := jetstream.NewKVJetStreamKVStore(ctx, "testFetch", keyspace+"_PROCESSORS", defaultJetStreamClient)
+	assert.NoError(t, err)
+	defer hbStore.Close()
+
+	// create osStore
+	otStore, err := jetstream.NewKVJetStreamKVStore(ctx, "testFetch", keyspace+"_OT", defaultJetStreamClient)
+	assert.NoError(t, err)
+	defer otStore.Close()
+
+	// put values into osStore
 	b := make([]byte, 8)
 	binary.LittleEndian.PutUint64(b, uint64(testOffset))
-	_, err = ot.Put(fmt.Sprintf("%s%s%d", "p1", "_", epoch), b)
+	err = otStore.PutKV(ctx, fmt.Sprintf("%s%s%d", "p1", "_", epoch), b)
 	assert.NoError(t, err)
-
 	epoch += 60
 	binary.LittleEndian.PutUint64(b, uint64(testOffset+5))
-	_, err = ot.Put(fmt.Sprintf("%s%s%d", "p2", "_", epoch), b)
+	err = otStore.PutKV(ctx, fmt.Sprintf("%s%s%d", "p2", "_", epoch), b)
 	assert.NoError(t, err)
 
-	defaultJetStreamClient := jsclient.NewDefaultJetStreamClient(nats.DefaultURL)
-
+	// create watchers for heartbeat and offset timeline
 	hbWatcher, err := jetstream.NewKVJetStreamKVWatch(ctx, "testFetch", keyspace+"_PROCESSORS", defaultJetStreamClient)
+	assert.NoError(t, err)
+	defer hbWatcher.Close()
 	otWatcher, err := jetstream.NewKVJetStreamKVWatch(ctx, "testFetch", keyspace+"_OT", defaultJetStreamClient)
+	assert.NoError(t, err)
+	defer otWatcher.Close()
 	var testVertex = NewProcessorManager(ctx, store.BuildWatermarkStoreWatcher(hbWatcher, otWatcher), WithPodHeartbeatRate(1), WithRefreshingProcessorsRate(1), WithSeparateOTBuckets(false))
 	var testBuffer = NewEdgeFetcher(ctx, "testBuffer", testVertex).(*edgeFetcher)
 
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		var err error
 		for i := 0; i < 3; i++ {
-			_, err = heartbeatBucket.Put("p1", []byte(fmt.Sprintf("%d", time.Now().Unix())))
+			err = hbStore.PutKV(ctx, "p1", []byte(fmt.Sprintf("%d", time.Now().Unix())))
 			assert.NoError(t, err)
 			time.Sleep(time.Duration(testVertex.opts.podHeartbeatRate) * time.Second)
 		}
-		err = heartbeatBucket.Delete("p1")
+		err = hbStore.DeleteKey(ctx, "p1")
 		assert.NoError(t, err)
 	}()
 
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		var err error
-		for i := 0; i < 100; i++ {
-			_, err = heartbeatBucket.Put("p2", []byte(fmt.Sprintf("%d", time.Now().Unix())))
+		// run p2 for 20 seconds
+		for i := 0; i < 20; i++ {
+			err = hbStore.PutKV(ctx, "p2", []byte(fmt.Sprintf("%d", time.Now().Unix())))
 			assert.NoError(t, err)
 			time.Sleep(time.Duration(testVertex.opts.podHeartbeatRate) * time.Second)
 		}
 	}()
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
-	defer cancel()
-
 	allProcessors := testBuffer.processorManager.GetAllProcessors()
 	for len(allProcessors) != 2 {
 		select {
 		case <-ctx.Done():
-			t.Fatalf("expected 2 processors, got %d: %s", len(allProcessors), ctx.Err())
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatalf("expected 2 processors, got %d: %s", len(allProcessors), ctx.Err())
+			}
 		default:
 			time.Sleep(1 * time.Millisecond)
 			allProcessors = testBuffer.processorManager.GetAllProcessors()
@@ -118,7 +147,9 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 	for !allProcessors["p1"].IsDeleted() {
 		select {
 		case <-ctx.Done():
-			t.Fatalf("expected p1 to be deleted: %s", ctx.Err())
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatalf("expected p1 to be deleted: %s", ctx.Err())
+			}
 		default:
 			time.Sleep(1 * time.Millisecond)
 			allProcessors = testBuffer.processorManager.GetAllProcessors()
@@ -135,20 +166,21 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 	assert.Equal(t, 2, len(allProcessors))
 	assert.True(t, allProcessors["p1"].IsDeleted())
 	assert.True(t, allProcessors["p2"].IsActive())
-	// "p1" should be deleted after this GetWatermark offset=101
-	// because "p1" offsetTimeline's head offset=100, which is < inputOffset 103
 	_ = testBuffer.GetWatermark(isb.SimpleOffset(func() string { return strconv.FormatInt(testOffset+3, 10) }))
 	allProcessors = testBuffer.processorManager.GetAllProcessors()
+	// we don't delete inactive processors, so the length should still be 2
 	assert.Equal(t, 2, len(allProcessors))
 	assert.True(t, allProcessors["p2"].IsActive())
 	assert.False(t, allProcessors["p1"].IsActive())
 
 	time.Sleep(time.Second)
 	// resume after one second
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		var err error
 		for i := 0; i < 5; i++ {
-			_, err = heartbeatBucket.Put("p1", []byte(fmt.Sprintf("%d", time.Now().Unix())))
+			err = hbStore.PutKV(ctx, "p1", []byte(fmt.Sprintf("%d", time.Now().Unix())))
 			assert.NoError(t, err)
 			time.Sleep(time.Duration(testVertex.opts.podHeartbeatRate) * time.Second)
 		}
@@ -157,10 +189,12 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 	// wait until p1 becomes active
 	time.Sleep(time.Duration(testVertex.opts.podHeartbeatRate) * time.Second)
 	allProcessors = testBuffer.processorManager.GetAllProcessors()
-	for len(allProcessors) != 2 {
+	for !allProcessors["p1"].IsActive() {
 		select {
 		case <-ctx.Done():
-			t.Fatalf("expected 2 processors, got %d: %s", len(allProcessors), ctx.Err())
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatalf("expected p1 to be active: %s", ctx.Err())
+			}
 		default:
 			time.Sleep(1 * time.Millisecond)
 			allProcessors = testBuffer.processorManager.GetAllProcessors()
@@ -169,38 +203,26 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 
 	assert.True(t, allProcessors["p1"].IsActive())
 	assert.True(t, allProcessors["p2"].IsActive())
-	// "p1" has been deleted from vertex.Processors
-	// so "p1" will be considered as a new processors and a new offsetTimeline watcher for "p1" will be created
+	// "p1" status is back to active, the offset timeline should be preserved
 	_ = testBuffer.GetWatermark(isb.SimpleOffset(func() string { return strconv.FormatInt(testOffset+1, 10) }))
-	newP1 := testBuffer.processorManager.GetProcessor("p1")
-	assert.NotNil(t, newP1)
-	assert.True(t, newP1.IsActive())
-	assert.NotNil(t, newP1.offsetTimeline)
-
-	//Wait till the offsetTimeline has been populated
-	newP1head := newP1.offsetTimeline.GetHeadOffset()
-	for newP1head == -1 {
-		select {
-		case <-ctx.Done():
-			t.Fatalf("expected head offset to not be equal to -1, %s", ctx.Err())
-		default:
-			time.Sleep(1 * time.Millisecond)
-			newP1head = newP1.offsetTimeline.GetHeadOffset()
-		}
-	}
-	// because the bucket hasn't been cleaned up, the new watcher will read all the history data to create this new offsetTimeline
-	assert.Equal(t, int64(100), newP1.offsetTimeline.GetHeadOffset())
+	p1 := testBuffer.processorManager.GetProcessor("p1")
+	assert.NotNil(t, p1)
+	assert.True(t, p1.IsActive())
+	assert.NotNil(t, p1.offsetTimeline)
+	assert.Equal(t, int64(100), p1.offsetTimeline.GetHeadOffset())
 
 	// publish a new watermark 101
 	binary.LittleEndian.PutUint64(b, uint64(testOffset+1))
-	_, err = ot.Put(fmt.Sprintf("%s%s%d", "p1", "_", epoch), b)
+	err = otStore.PutKV(ctx, fmt.Sprintf("%s%s%d", "p1", "_", epoch), b)
 	assert.NoError(t, err)
 
 	// "p1" becomes inactive after 5 loops
 	for !allProcessors["p1"].IsInactive() {
 		select {
 		case <-ctx.Done():
-			t.Fatalf("expected p1 to be inactive: %s", ctx.Err())
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatalf("expected p1 to be inactive: %s", ctx.Err())
+			}
 		default:
 			time.Sleep(1 * time.Millisecond)
 			allProcessors = testBuffer.processorManager.GetAllProcessors()
@@ -210,10 +232,12 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 	time.Sleep(time.Second)
 
 	// resume after one second
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		var err error
-		for i := 0; i < 5; i++ {
-			_, err = heartbeatBucket.Put("p1", []byte(fmt.Sprintf("%d", time.Now().Unix())))
+		for i := 0; i < 3; i++ {
+			err = hbStore.PutKV(ctx, "p1", []byte(fmt.Sprintf("%d", time.Now().Unix())))
 			assert.NoError(t, err)
 			time.Sleep(time.Duration(testVertex.opts.podHeartbeatRate) * time.Second)
 		}
@@ -230,9 +254,17 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 		}
 	}
 
-	// added 101 in the previous steps for newP1, so the head should be 101 after resume
-	assert.Equal(t, int64(101), newP1.offsetTimeline.GetHeadOffset())
+	// added 101 in the previous steps for p1, so the head should be 101 after resume
+	assert.Equal(t, int64(101), p1.offsetTimeline.GetHeadOffset())
+
+	wg.Wait()
+	cancel()
 
+	// wait for all the ctx.Done() is completed
+	// we don't have wg on the Watch functions go routine..
+	// TODO: opened an issue https://github.com/numaproj/numaflow/issues/224
+	//   fix tests after we revisit the watermark store and watcher closing logic
+	time.Sleep(3 * time.Second)
 }
 
 func TestFetcherWithSeparateOTBucket(t *testing.T) {
@@ -240,7 +272,7 @@ func TestFetcherWithSeparateOTBucket(t *testing.T) {
 	//   Maybe we should not support seperate OT because controller does not support it
 	t.Skip()
 	// uncomment to debug
-	//os.Setenv("NUMAFLOW_DEBUG", "true")
+	// os.Setenv("NUMAFLOW_DEBUG", "true")
 
 	var ctx = context.Background()
 
@@ -308,11 +340,13 @@ func TestFetcherWithSeparateOTBucket(t *testing.T) {
 	defaultJetStreamClient := jsclient.NewDefaultJetStreamClient(nats.DefaultURL)
 
 	hbWatcher, err := jetstream.NewKVJetStreamKVWatch(ctx, "testFetch", keyspace+"_PROCESSORS", defaultJetStreamClient)
+	assert.NoError(t, err)
 	otWatcher, err := jetstream.NewKVJetStreamKVWatch(ctx, "testFetch", keyspace+"_OT", defaultJetStreamClient)
+	assert.NoError(t, err)
 	var testVertex = NewProcessorManager(ctx, store.BuildWatermarkStoreWatcher(hbWatcher, otWatcher), WithPodHeartbeatRate(1), WithRefreshingProcessorsRate(1), WithSeparateOTBuckets(true))
 	var testBuffer = NewEdgeFetcher(ctx, "testBuffer", testVertex).(*edgeFetcher)
 
-	//var location, _ = time.LoadLocation("UTC")
+	// var location, _ = time.LoadLocation("UTC")
 	go func() {
 		var err error
 		for i := 0; i < 3; i++ {
diff --git a/pkg/watermark/publish/publisher_inmem_test.go b/pkg/watermark/publish/publisher_inmem_test.go
new file mode 100644
index 00000000..c9978959
--- /dev/null
+++ b/pkg/watermark/publish/publisher_inmem_test.go
@@ -0,0 +1,97 @@
+package publish
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/numaproj/numaflow/pkg/watermark/store/inmem"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/watermark/processor"
+	"github.com/numaproj/numaflow/pkg/watermark/store"
+)
+
+func TestPublisherWithSeparateOTBuckets_InMem(t *testing.T) {
+	var ctx = context.Background()
+
+	var publisherHBKeyspace = "publisherTest_PROCESSORS"
+
+	// this test uses separate OT buckets, so it is an OT bucket per processor
+	var publisherOTKeyspace = "publisherTest_OT_publisherTestPod1"
+
+	heartbeatKV, _, err := inmem.NewKVInMemKVStore(ctx, "testPublisher", publisherHBKeyspace)
+	assert.NoError(t, err)
+	otKV, _, err := inmem.NewKVInMemKVStore(ctx, "testPublisher", publisherOTKeyspace)
+	assert.NoError(t, err)
+
+	publishEntity := processor.NewProcessorEntity("publisherTestPod1")
+
+	p := NewPublish(ctx, publishEntity, store.BuildWatermarkStore(heartbeatKV, otKV), WithAutoRefreshHeartbeatDisabled(), WithPodHeartbeatRate(1)).(*publish)
+
+	var epoch int64 = 1651161600
+	var location, _ = time.LoadLocation("UTC")
+	for i := 0; i < 3; i++ {
+		p.PublishWatermark(processor.Watermark(time.Unix(epoch, 0).In(location)), isb.SimpleOffset(func() string { return strconv.Itoa(i) }))
+		epoch += 60
+		time.Sleep(time.Millisecond)
+	}
+	// publish a stale watermark (offset doesn't matter)
+	p.PublishWatermark(processor.Watermark(time.Unix(epoch-120, 0).In(location)), isb.SimpleOffset(func() string { return strconv.Itoa(0) }))
+
+	keys := p.getAllOTKeysFromBucket()
+	assert.Equal(t, []string{"publisherTestPod1_1651161600", "publisherTestPod1_1651161660", "publisherTestPod1_1651161720"}, keys)
+
+	wm := p.loadLatestFromStore()
+	assert.Equal(t, processor.Watermark(time.Unix(epoch-60, 0).In(location)).String(), wm.String())
+
+	head := p.GetLatestWatermark()
+	assert.Equal(t, processor.Watermark(time.Unix(epoch-60, 0).In(location)).String(), head.String())
+
+	p.StopPublisher()
+
+	_, err = p.heartbeatStore.GetValue(ctx, publishEntity.GetID())
+	assert.Equal(t, fmt.Errorf("key publisherTestPod1 not found"), err)
+
+}
+
+func TestPublisherWithSharedOTBucket_InMem(t *testing.T) {
+	var ctx = context.Background()
+
+	var keyspace = "publisherTest"
+
+	publishEntity := processor.NewProcessorEntity("publisherTestPod1", processor.WithSeparateOTBuckets(true))
+	heartbeatKV, _, err := inmem.NewKVInMemKVStore(ctx, "testPublisher", keyspace+"_PROCESSORS")
+	assert.NoError(t, err)
+	otKV, _, err := inmem.NewKVInMemKVStore(ctx, "testPublisher", keyspace+"_OT")
+	assert.NoError(t, err)
+
+	p := NewPublish(ctx, publishEntity, store.BuildWatermarkStore(heartbeatKV, otKV), WithAutoRefreshHeartbeatDisabled(), WithPodHeartbeatRate(1)).(*publish)
+
+	var epoch int64 = 1651161600
+	var location, _ = time.LoadLocation("UTC")
+	for i := 0; i < 3; i++ {
+		p.PublishWatermark(processor.Watermark(time.Unix(epoch, 0).In(location)), isb.SimpleOffset(func() string { return strconv.Itoa(i) }))
+		epoch += 60
+		time.Sleep(time.Millisecond)
+	}
+	// publish a stale watermark (offset doesn't matter)
+	p.PublishWatermark(processor.Watermark(time.Unix(epoch-120, 0).In(location)), isb.SimpleOffset(func() string { return strconv.Itoa(0) }))
+
+	keys := p.getAllOTKeysFromBucket()
+	assert.Equal(t, []string{"1651161600", "1651161660", "1651161720"}, keys)
+
+	wm := p.loadLatestFromStore()
+	assert.Equal(t, processor.Watermark(time.Unix(epoch-60, 0).In(location)).String(), wm.String())
+
+	head := p.GetLatestWatermark()
+	assert.Equal(t, processor.Watermark(time.Unix(epoch-60, 0).In(location)).String(), head.String())
+
+	p.StopPublisher()
+
+	_, err = p.heartbeatStore.GetValue(ctx, publishEntity.GetID())
+	assert.ErrorContains(t, err, "key publisherTestPod1 not found")
+}