Repository: numaproj/numaflow
Pull request #105 (2022-07-08 20:32:22)

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



Patch:
diff --git a/README.md b/README.md
index a9ede6ed..94bc6ddb 100644
--- a/README.md
+++ b/README.md
@@ -27,7 +27,7 @@ Numaflow installs in less than a minute and is easier and cheaper for simple dat
 
 ## Resources
 
-- [QUICK_START](docs/QUICK_START.md)
+- [QUICK_START](docs/quick-start.md)
 - [EXAMPLES](examples)
-- [DEVELOPMENT](docs/DEVELOPMENT.md)
+- [DEVELOPMENT](docs/development.md)
 - [CONTRIBUTING](https://github.com/numaproj/numaproj/blob/main/CONTRIBUTING.md)
diff --git a/config/base/controller-manager/controller-manager-deployment.yaml b/config/base/controller-manager/controller-manager-deployment.yaml
index 33bd2ced..cb24e26a 100644
--- a/config/base/controller-manager/controller-manager-deployment.yaml
+++ b/config/base/controller-manager/controller-manager-deployment.yaml
@@ -1,7 +1,7 @@
 apiVersion: apps/v1
 kind: Deployment
 metadata:
-  name: controller-manager
+  name: numaflow-controller
 spec:
   replicas: 1
   selector:
diff --git a/config/base/kustomization.yaml b/config/base/kustomization.yaml
index 9cbb9e89..d6073eb4 100644
--- a/config/base/kustomization.yaml
+++ b/config/base/kustomization.yaml
@@ -17,7 +17,7 @@ patchesStrategicMerge:
     apiVersion: apps/v1
     kind: Deployment
     metadata:
-      name: controller-manager
+      name: numaflow-controller
     spec:
       template:
         spec:
diff --git a/config/install.yaml b/config/install.yaml
index 5dcd18bd..daf9c0c7 100644
--- a/config/install.yaml
+++ b/config/install.yaml
@@ -12833,7 +12833,7 @@ spec:
 apiVersion: apps/v1
 kind: Deployment
 metadata:
-  name: controller-manager
+  name: numaflow-controller
   namespace: numaflow-system
 spec:
   replicas: 1
diff --git a/docs/README.md b/docs/README.md
index b7620273..f9ea853e 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -10,4 +10,4 @@ Numaflow is a Kubernetes-native platform for running massive parallel data proce
 
 ## Getting Started
 
-For set-up information and running your first Numaflow pipeline, please see our [getting started guide](QUICK_START.md).
+For set-up information and running your first Numaflow pipeline, please see our [getting started guide](./quick-start.md).
diff --git a/docs/architecture/ARCHITECTURE.md b/docs/architecture/architecture.md
similarity index 100%
rename from docs/architecture/ARCHITECTURE.md
rename to docs/architecture/architecture.md
diff --git a/docs/AUTOSCALING.md b/docs/autoscaling.md
similarity index 96%
rename from docs/AUTOSCALING.md
rename to docs/autoscaling.md
index e9a683e4..e8e226aa 100644
--- a/docs/AUTOSCALING.md
+++ b/docs/autoscaling.md
@@ -41,7 +41,7 @@ spec:
 
 With the configuration above, Kubernetes HPA controller will keep the target utilization of the pods of the Vertex at 50%.
 
-Kubernetes HPA autoscaling is usful for those Source vertice not able to count pending messages, such as [HTTP](sources/HTTP.md).
+Kubernetes HPA autoscaling is usful for those Source vertice not able to count pending messages, such as [HTTP](sources/http.md).
 
 ### Third Party Autoscaling
 
diff --git a/docs/builtin-functions/README.md b/docs/builtin-functions/README.md
index 165ee071..cc075a08 100644
--- a/docs/builtin-functions/README.md
+++ b/docs/builtin-functions/README.md
@@ -18,7 +18,7 @@ spec:
 **Filter**
 
 A `filter` builtin UDF does filter the message based on expression. `payload` keyword represents message object.
-see documentation for expression [here](FILTER.md#expression)
+see documentation for expression [here](filter.md#expression)
 
 ```yaml
 spec:
diff --git a/docs/builtin-functions/CAT.md b/docs/builtin-functions/cat.md
similarity index 100%
rename from docs/builtin-functions/CAT.md
rename to docs/builtin-functions/cat.md
diff --git a/docs/builtin-functions/FILTER.md b/docs/builtin-functions/filter.md
similarity index 100%
rename from docs/builtin-functions/FILTER.md
rename to docs/builtin-functions/filter.md
diff --git a/docs/CONDITIONAL_FORWARDING.md b/docs/conditional-forwarding.md
similarity index 100%
rename from docs/CONDITIONAL_FORWARDING.md
rename to docs/conditional-forwarding.md
diff --git a/docs/container-resources.md b/docs/container-resources.md
new file mode 100644
index 00000000..3caf1754
--- /dev/null
+++ b/docs/container-resources.md
@@ -0,0 +1,72 @@
+# Container Resources
+
+[Container Resources](https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/) can be customized for all the types of vertices.
+
+## Main Container
+
+To specify `resources` for the `main` container of vertex pods:
+
+```yaml
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: my-pipeline
+spec:
+  vertices:
+    - name: my-vertex
+      containerTemplate:
+        resources:
+          limits:
+            cpu: "3"
+            memory: 6Gi
+          requests:
+            cpu: "1"
+            memory: 4Gi
+```
+
+## UDF Container
+
+To specify `resources` for `udf` container of vertex pods:
+
+```yaml
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: my-pipeline
+spec:
+  vertices:
+    - name: my-vertex
+      udf:
+        container:
+          resources:
+            limits:
+              cpu: "3"
+              memory: 6Gi
+            requests:
+              cpu: "1"
+              memory: 4Gi
+```
+
+## UDSink Container
+
+To specify `resources` for `udsink` container of vertex pods:
+
+```yaml
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: my-pipeline
+spec:
+  vertices:
+    - name: my-vertex
+      sink:
+        udsink:
+          container:
+            resources:
+              limits:
+                cpu: "3"
+                memory: 6Gi
+              requests:
+                cpu: "1"
+                memory: 4Gi
+```
diff --git a/docs/CONTROLLER_CONFIGMAP.md b/docs/controller-configmap.md
similarity index 96%
rename from docs/CONTROLLER_CONFIGMAP.md
rename to docs/controller-configmap.md
index ea6d63c0..c15d3dfe 100644
--- a/docs/CONTROLLER_CONFIGMAP.md
+++ b/docs/controller-configmap.md
@@ -27,7 +27,7 @@ data:
 
 ### ISB Service Configuration
 
-One of the important configuration items in the ConfigMap is about [ISB Service](./INTER_STEP_BUFFER_SERVICE.md). We currently use 3rd party technologies such as `JetStream` to implmement ISB Services, if those applications have new releases, to make them available in Numaflow, the new versions need to be added in the ConfigMap.
+One of the important configuration items in the ConfigMap is about [ISB Service](./inter-step-buffer-service.md). We currently use 3rd party technologies such as `JetStream` to implmement ISB Services, if those applications have new releases, to make them available in Numaflow, the new versions need to be added in the ConfigMap.
 
 For example, there's a new `Nats JetStream` version `x.y.x` available, a new version configuration like below needs to be added before it can be referenced in the `InterStepBufferService` spec.
 
diff --git a/docs/DEBUGGING.md b/docs/debugging.md
similarity index 93%
rename from docs/DEBUGGING.md
rename to docs/debugging.md
index 83c66d99..0f6e3d4e 100644
--- a/docs/DEBUGGING.md
+++ b/docs/debugging.md
@@ -9,7 +9,7 @@ metadata:
   name: simple-pipeline
 spec:
   vertices:
-    - name: input
+    - name: in
       source:
         generator:
           rpu: 100
@@ -22,14 +22,14 @@ spec:
         env:
           - name: NUMAFLOW_DEBUG
             value: "true" # DO NOT forget the double quotes!!!
-    - name: output
+    - name: out
       sink:
         log: {}
   edges:
-    - from: input
+    - from: in
       to: p1
     - from: p1
-      to: output
+      to: out
 ```
 
 ## Profiling
diff --git a/docs/DEVELOPMENT.md b/docs/development.md
similarity index 100%
rename from docs/DEVELOPMENT.md
rename to docs/development.md
diff --git a/docs/environment-variables.md b/docs/environment-variables.md
new file mode 100644
index 00000000..245b00da
--- /dev/null
+++ b/docs/environment-variables.md
@@ -0,0 +1,44 @@
+# Environment Variables
+
+For the `main` container of vertex pods, environment variable `NUMAFLOW_DEBUG` can be set to `true` for [debugging](./debugging.md).
+
+In [`udf`](./user-defined-functions.md) and [`udsink`](./sinks/user-defined-sinks.md) containers, there are some preset environment variables that can be used directly.
+
+- `NUMAFLOW_NAMESPACE` - Namespace.
+- `NUMAFLOW_POD` - Pod name.
+- `NUMAFLOW_REPLICA` - Replica index.
+- `NUMAFLOW_PIPELINE_NAME` - Name of the pipeline.
+- `NUMAFLOW_VERTEX_NAME` - Name of the vertex.
+
+## Your Own Environment Variables
+
+To add your own environment variables to `udf` or `udsink` containers, check the example below.
+
+```yaml
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: my-pipeline
+spec:
+  vertices:
+    - name: my-udf
+      udf:
+        container:
+          image: my-function:latest
+          env:
+            - name: env01
+              value: value01
+            - name: env02
+              valueFrom:
+                secretKeyRef:
+                  name: my-secret
+                  key: my-key
+    - name: my-sink
+      sink:
+        udsink:
+          container:
+            image: my-sink:latest
+            env:
+              - name: env03
+                value: value03
+```
diff --git a/docs/INTER_STEP_BUFFER_SERVICE.md b/docs/inter-step-buffer-service.md
similarity index 99%
rename from docs/INTER_STEP_BUFFER_SERVICE.md
rename to docs/inter-step-buffer-service.md
index 03379451..739fccbc 100644
--- a/docs/INTER_STEP_BUFFER_SERVICE.md
+++ b/docs/inter-step-buffer-service.md
@@ -1,6 +1,6 @@
 # Inter-Step Buffer Service
 
-Inter-Step Buffer Service is the service to provide [Inter-Step Buffers](INTER_STEP_BUFFER.md).
+Inter-Step Buffer Service is the service to provide [Inter-Step Buffers](./inter-step-buffer.md).
 
 An Inter-Step Buffer Service is describe by a [Custom Resource](https://kubernetes.io/docs/concepts/extend-kubernetes/api-extension/custom-resources/), it is required to be existing in a namespace before Pipeline objects are created. A sample `InterStepBufferService` with JetStream implementation looks like below.
 
diff --git a/docs/INTER_STEP_BUFFER.md b/docs/inter-step-buffer.md
similarity index 100%
rename from docs/INTER_STEP_BUFFER.md
rename to docs/inter-step-buffer.md
diff --git a/docs/metrics/METRICS.md b/docs/metrics/metrics.md
similarity index 100%
rename from docs/metrics/METRICS.md
rename to docs/metrics/metrics.md
diff --git a/docs/PIPELINE_TUNING.md b/docs/pipeline-tuning.md
similarity index 100%
rename from docs/PIPELINE_TUNING.md
rename to docs/pipeline-tuning.md
diff --git a/docs/PIPELINE.md b/docs/pipeline.md
similarity index 93%
rename from docs/PIPELINE.md
rename to docs/pipeline.md
index 19593920..818b700b 100644
--- a/docs/PIPELINE.md
+++ b/docs/pipeline.md
@@ -2,7 +2,7 @@
 
 The `Pipeline` is the most important concept in Numaflow, it represents a data processing job, it defines:
 
-1. A list of [vertices](./VERTEX.md), which define the data processing tasks;
+1. A list of [vertices](./vertex.md), which define the data processing tasks;
 1. A list of `edges`, which are used to describe the relationship between the vertices.
 
 The `Pipeline` is abstracted as a [Kubernetes Custom Resource](https://kubernetes.io/docs/concepts/extend-kubernetes/api-extension/custom-resources/). A `Pipeline` spec looks like below.
diff --git a/docs/QUICK_START.md b/docs/quick-start.md
similarity index 90%
rename from docs/QUICK_START.md
rename to docs/quick-start.md
index 374ac62d..c6440b62 100644
--- a/docs/QUICK_START.md
+++ b/docs/quick-start.md
@@ -33,7 +33,7 @@ kubectl -n numaflow-system port-forward deployment/numaflow-server 8443:8443
 
 ## A Simple Pipeline
 
-[Inter-Step Buffers](./INTER_STEP_BUFFER.md) are essential to run Numaflow pipelines, to get Inter-Step Buffers in place, an [Inter-Step Buffer Service](./INTER_STEP_BUFFER_SERVICE.md) is required. We use [Nats JetStream](https://docs.nats.io/nats-concepts/jetstream) as an `Inter-Step Buffer Service` provider in this example.
+[Inter-Step Buffers](./inter-step-buffer.md) are essential to run Numaflow pipelines, to get Inter-Step Buffers in place, an [Inter-Step Buffer Service](./inter-step-buffer-service.md) is required. We use [Nats JetStream](https://docs.nats.io/nats-concepts/jetstream) as an `Inter-Step Buffer Service` provider in this example.
 
 ```shell
 kubectl apply -f https://raw.githubusercontent.com/numaproj/numaflow/stable/examples/0-isbsvc-jetstream.yaml
@@ -73,4 +73,4 @@ kubectl delete -f https://raw.githubusercontent.com/numaproj/numaflow/stable/exa
 
 ## What's Next
 
-After exploring how a Numaflow pipeline runs, you can check what data [Sources](./sources/GENERATOR.md) and [Sinks](./sinks/KAFKA.md) Numaflow supports out of the box, or learn how to write [User Defined Functions](./USER_DEFINED_FUNCTIONS.md).
+After exploring how a Numaflow pipeline runs, you can check what data [Sources](./sources/generator.md) and [Sinks](./sinks/kafka.md) Numaflow supports out of the box, or learn how to write [User Defined Functions](./user-defined-functions.md).
diff --git a/docs/RELEASES.md b/docs/releases.md
similarity index 100%
rename from docs/RELEASES.md
rename to docs/releases.md
diff --git a/docs/RELEASING.md b/docs/releasing.md
similarity index 100%
rename from docs/RELEASING.md
rename to docs/releasing.md
diff --git a/docs/sinks/KAFKA.md b/docs/sinks/kafka.md
similarity index 100%
rename from docs/sinks/KAFKA.md
rename to docs/sinks/kafka.md
diff --git a/docs/sinks/LOG.md b/docs/sinks/log.md
similarity index 100%
rename from docs/sinks/LOG.md
rename to docs/sinks/log.md
diff --git a/docs/USER_DEFINED_SINKS.md b/docs/sinks/user-defined-sinks.md
similarity index 100%
rename from docs/USER_DEFINED_SINKS.md
rename to docs/sinks/user-defined-sinks.md
diff --git a/docs/sources/GENERATOR.md b/docs/sources/generator.md
similarity index 100%
rename from docs/sources/GENERATOR.md
rename to docs/sources/generator.md
diff --git a/docs/sources/HTTP.md b/docs/sources/http.md
similarity index 100%
rename from docs/sources/HTTP.md
rename to docs/sources/http.md
diff --git a/docs/sources/KAFKA.md b/docs/sources/kafka.md
similarity index 100%
rename from docs/sources/KAFKA.md
rename to docs/sources/kafka.md
diff --git a/docs/STATIC_CODE_ANALYSIS.md b/docs/static-code-analysis.md
similarity index 100%
rename from docs/STATIC_CODE_ANALYSIS.md
rename to docs/static-code-analysis.md
diff --git a/docs/USER_DEFINED_FUNCTIONS.md b/docs/user-defined-functions.md
similarity index 100%
rename from docs/USER_DEFINED_FUNCTIONS.md
rename to docs/user-defined-functions.md
diff --git a/docs/VERTEX.md b/docs/vertex.md
similarity index 88%
rename from docs/VERTEX.md
rename to docs/vertex.md
index 86bfe223..d59a3e25 100644
--- a/docs/VERTEX.md
+++ b/docs/vertex.md
@@ -1,6 +1,6 @@
 # Vertex
 
-The `Vertex` is also a key component of Numaflow `Pipeline` where the data processing happens. `Vertex` is defined as a list in the [pipeline](./PIPELINE.md) spec, each representing a data processing task.
+The `Vertex` is also a key component of Numaflow `Pipeline` where the data processing happens. `Vertex` is defined as a list in the [pipeline](./pipeline.md) spec, each representing a data processing task.
 
 There are 3 types of `Vertex` in Numaflow today:
 
@@ -10,7 +10,7 @@ There are 3 types of `Vertex` in Numaflow today:
 
 We have defined a [Kubernetes Custom Resource](https://kubernetes.io/docs/concepts/extend-kubernetes/api-extension/custom-resources/) defined for `Vertex`. A `Pipeline` containing multiple vertices will automatically generate multiple `Vertex` objects by the controller. As a user, you should NOT create a `Vertex` object directly.
 
-In a `Pipeline`, the vertices are not connected directly, but through [Inter-Step Buffers](./INTER_STEP_BUFFER.md).
+In a `Pipeline`, the vertices are not connected directly, but through [Inter-Step Buffers](./inter-step-buffer.md).
 
 To query `Vertex` objects with `kubectl`:
 
diff --git a/docs/volumes.md b/docs/volumes.md
new file mode 100644
index 00000000..a6a2fd95
--- /dev/null
+++ b/docs/volumes.md
@@ -0,0 +1,37 @@
+# Volumes
+
+`Volumes` can be mounted to [`udf`](./user-defined-functions.md) or [`udsink`](./sinks/user-defined-sinks.md) containers.
+
+Following example shows how to mount a ConfigMap to an `udf` vertex and an `udsink` vertex.
+
+```yaml
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: my-pipeline
+spec:
+  vertices:
+    - name: my-udf
+      volumes:
+        - name: my-udf-config
+          configMap:
+            name: udf-config
+      udf:
+        container:
+          image: my-function:latest
+          volumeMounts:
+            - mountPath: /path/to/my-function-config
+              name: my-udf-config
+    - name: my-sink
+      volumes:
+        - name: my-udsink-config
+          configMap:
+            name: udsink-config
+      sink:
+        udsink:
+          container:
+            image: my-sink:latest
+            volumeMounts:
+              - mountPath: /path/to/my-sink-config
+                name: my-udsink-config
+```
diff --git a/mkdocs.yml b/mkdocs.yml
index 2001e4a3..4b585a98 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -37,39 +37,42 @@ markdown_extensions:
 nav:
   - Home: README.md
   - Getting Started:
-      - QUICK_START.md
+      - quick-start.md
   - User Guide:
       - Core Concepts:
-          - PIPELINE.md
-          - VERTEX.md
-          - INTER_STEP_BUFFER.md
-          - INTER_STEP_BUFFER_SERVICE.md
+          - pipeline.md
+          - vertex.md
+          - inter-step-buffer.md
+          - inter-step-buffer-service.md
       - Sources:
-          - sources/GENERATOR.md
-          - sources/HTTP.md
-          - sources/KAFKA.md
+          - sources/generator.md
+          - sources/http.md
+          - sources/kafka.md
       - Sinks:
-          - sinks/KAFKA.md
-          - sinks/LOG.md
-          - User Defined Sinks: "USER_DEFINED_SINKS.md"
+          - sinks/kafka.md
+          - sinks/log.md
+          - User Defined Sinks: "sinks/user-defined-sinks.md"
       - Functions:
           - Builtin Functions:
               - Overview: "builtin-functions/README.md"
-              - Cat: "builtin-functions/CAT.md"
-              - Filter: "builtin-functions/FILTER.md"
-          - User Defined Functions: "USER_DEFINED_FUNCTIONS.md"
-      - PIPELINE_TUNING.md
-      - CONDITIONAL_FORWARDING.md
-      - AUTOSCALING.md
+              - Cat: "builtin-functions/cat.md"
+              - Filter: "builtin-functions/filter.md"
+          - User Defined Functions: "user-defined-functions.md"
+      - pipeline-tuning.md
+      - conditional-forwarding.md
+      - autoscaling.md
+      - container-resources.md
+      - volumes.md
+      - environment-variables.md
       - APIs.md
   - Operator Manual:
-      - Releases ⧉: "RELEASES.md"
+      - Releases ⧉: "releases.md"
       - Configuration:
-          - Controller Configuration: "CONTROLLER_CONFIGMAP.md"
-          - metrics/METRICS.md
+          - Controller Configuration: "controller-configmap.md"
+          - metrics/metrics.md
   - Developer Guide:
-      - DEVELOPMENT.md
-      - Architecture: "architecture/ARCHITECTURE.md"
-      - DEBUGGING.md
-      - STATIC_CODE_ANALYSIS.md
-      - RELEASING.md
+      - development.md
+      - Architecture: "architecture/architecture.md"
+      - debugging.md
+      - static-code-analysis.md
+      - releasing.md
diff --git a/pkg/isb/forward/forward.go b/pkg/isb/forward/forward.go
index decf2843..e93293bc 100644
--- a/pkg/isb/forward/forward.go
+++ b/pkg/isb/forward/forward.go
@@ -372,7 +372,7 @@ func (isdf *InterStepDataForward) concurrentApplyUDF(ctx context.Context, readMe
 	}
 }
 
-// applyUDF applies the UDF and will block if there is any InternalErr. On the other hand, if this is an UserError
+// applyUDF applies the UDF and will block if there is any InternalErr. On the other hand, if this is a UserError
 // the skip flag is set. ShutDown flag will only if there is an InternalErr and ForceStop has been invoked.
 // The UserError retry will be done on the ApplyUDF.
 func (isdf *InterStepDataForward) applyUDF(ctx context.Context, readMessage *isb.ReadMessage) ([]*isb.Message, error) {
diff --git a/pkg/sources/generator/watermark.go b/pkg/sources/generator/watermark.go
index cc21aa11..16555852 100644
--- a/pkg/sources/generator/watermark.go
+++ b/pkg/sources/generator/watermark.go
@@ -11,6 +11,6 @@ import (
 func (mg *memgen) buildSourceWatermarkPublisher(publishWMStores *generic.PublishWMStores) publish.Publisher {
 	// for tickgen, it can be the name of the replica
 	entityName := fmt.Sprintf("%s-%d", mg.vertexInstance.Vertex.Name, mg.vertexInstance.Replica)
-	processorEntity := processor.NewProcessorEntity(entityName, mg.vertexInstance.Vertex.Name+"_SOURCE")
+	processorEntity := processor.NewProcessorEntity(entityName)
 	return publish.NewPublish(mg.lifecycleCtx, processorEntity, publishWMStores.HBStore, publishWMStores.OTStore)
 }
diff --git a/pkg/watermark/fetch/edge.go b/pkg/watermark/fetch/edge.go
index 472472f9..e931a047 100644
--- a/pkg/watermark/fetch/edge.go
+++ b/pkg/watermark/fetch/edge.go
@@ -49,7 +49,7 @@ func (e *Edge) GetWatermark(inputOffset isb.Offset) processor.Watermark {
 	var epoch int64 = math.MaxInt64
 	var podMap = e.fromVertex.GetAllProcessors()
 	for _, p := range podMap {
-		debugString.WriteString(fmt.Sprintf("[%s] %s\n", p.entity.GetBucketName(), p))
+		debugString.WriteString(fmt.Sprintf("[HB:%s OT:%s] %s\n", e.fromVertex.hbWatcher.GetKVName(), e.fromVertex.otWatcher.GetKVName(), p))
 		var t = p.offsetTimeline.GetEventTime(inputOffset)
 		if t != -1 && t < epoch {
 			epoch = t
diff --git a/pkg/watermark/fetch/processor_to_fetch.go b/pkg/watermark/fetch/processor_to_fetch.go
index ef3f8e61..0db8eac0 100644
--- a/pkg/watermark/fetch/processor_to_fetch.go
+++ b/pkg/watermark/fetch/processor_to_fetch.go
@@ -125,8 +125,7 @@ func (p *ProcessorToFetch) startTimeLineWatcher() {
 					watermark: epoch,
 					offset:    int64(uint64Value),
 				})
-				p.log.Debugf("[%s]timelineWatcher- Updates:  [%s] > %d: %d", p.entity.GetBucketName(), p.otWatcher.GetKVName(), epoch, int64(uint64Value))
-				p.log.Debugf("[%s]%s", p.entity.GetBucketName(), p)
+				p.log.Debugw("timelineWatcher- Updates", zap.String("bucket", p.otWatcher.GetKVName()), zap.Int64("epoch", epoch), zap.Uint64("value", uint64Value))
 			case store.KVDelete:
 				// we do not care about Delete events because the timeline bucket is meant to grow and the TTL will
 				// naturally trim the KV store.
diff --git a/pkg/watermark/fetch/vertex.go b/pkg/watermark/fetch/vertex.go
index 579d3396..20d9abac 100644
--- a/pkg/watermark/fetch/vertex.go
+++ b/pkg/watermark/fetch/vertex.go
@@ -26,9 +26,7 @@ type FromVertexer interface {
 // It has the mapping of all the processors which in turn has all the information about each processor
 // timelines.
 type FromVertex struct {
-	ctx context.Context
-	// keyspace is only used if we have a processorHBWatcher
-	keyspace   string
+	ctx        context.Context
 	hbWatcher  store.WatermarkKVWatcher
 	otWatcher  store.WatermarkKVWatcher
 	heartbeat  *ProcessorHeartbeat
@@ -41,7 +39,7 @@ type FromVertex struct {
 }
 
 // NewFromVertex returns `FromVertex`
-func NewFromVertex(ctx context.Context, keyspace string, hbWatcher store.WatermarkKVWatcher, otWatcher store.WatermarkKVWatcher, inputOpts ...VertexOption) *FromVertex {
+func NewFromVertex(ctx context.Context, hbWatcher store.WatermarkKVWatcher, otWatcher store.WatermarkKVWatcher, inputOpts ...VertexOption) *FromVertex {
 	opts := &vertexOptions{
 		podHeartbeatRate:         5,
 		refreshingProcessorsRate: 5,
@@ -53,7 +51,6 @@ func NewFromVertex(ctx context.Context, keyspace string, hbWatcher store.Waterma
 
 	v := &FromVertex{
 		ctx:        ctx,
-		keyspace:   keyspace,
 		hbWatcher:  hbWatcher,
 		otWatcher:  otWatcher,
 		heartbeat:  NewProcessorHeartbeat(),
@@ -140,7 +137,7 @@ func (v *FromVertex) refreshingProcessors() {
 			debugStr.WriteString(fmt.Sprintf("[%s] ", pName))
 		}
 	}
-	v.log.Debugf("[%s] active processors %+v", v.keyspace, debugStr.String())
+	v.log.Debugw("active processors", zap.String("HB", v.hbWatcher.GetKVName()), zap.String("OT", v.otWatcher.GetKVName()), zap.String("DebugStr", debugStr.String()))
 }
 
 // startHeatBeatWatcher starts the processor Heartbeat Watcher to listen to the processor bucket and update the processor
@@ -157,7 +154,7 @@ func (v *FromVertex) startHeatBeatWatcher() {
 			}
 			switch value.Operation() {
 			case store.KVPut:
-				var entity = processor.NewProcessorEntity(value.Key(), v.keyspace, processor.WithSeparateOTBuckets(v.opts.separateOTBucket))
+				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
@@ -191,7 +188,8 @@ func (v *FromVertex) startHeatBeatWatcher() {
 			case store.KVPurge:
 				v.log.Errorw("received nats.KeyValuePurge", zap.String("bucket", v.hbWatcher.GetKVName()))
 			}
-			v.log.Debugf("[%s] processorHeartbeatWatcher - Updates: [%s] %s > %s: %s", v.keyspace, value.Operation(), v.hbWatcher.GetKVName(), value.Key(), value.Value())
+			v.log.Debugw("processorHeartbeatWatcher - Updates:", zap.String("Operation", value.Operation().String()),
+				zap.String("HB", v.hbWatcher.GetKVName()), zap.String("key", value.Key()), zap.String("value", string(value.Value())))
 		}
 
 	}
diff --git a/pkg/watermark/generic/fetcher.go b/pkg/watermark/generic/fetcher.go
index f49748dc..a2562776 100644
--- a/pkg/watermark/generic/fetcher.go
+++ b/pkg/watermark/generic/fetcher.go
@@ -31,10 +31,10 @@ func BuildFetchWMWatchers(hbWatch store.WatermarkKVWatcher, otWatch store.Waterm
 }
 
 // NewGenericFetch returns GenericFetch. vertexName is the vertex currently processing.
-// fetchKeyspace is obsolete, and will be removed in subsequent iterations. fetchWM is a struct for retrieving both the heartbeat
+// fetchWM is a struct for retrieving both the heartbeat
 // and the offset watermark timeline (Vn-1 vertex).
-func NewGenericFetch(ctx context.Context, vertexName string, fetchKeyspace string, fetchWM FetchWMWatchers) *GenericFetch {
-	fromVertex := fetch.NewFromVertex(ctx, fetchKeyspace, fetchWM.HBWatch, fetchWM.OTWatch)
+func NewGenericFetch(ctx context.Context, vertexName string, fetchWM FetchWMWatchers) *GenericFetch {
+	fromVertex := fetch.NewFromVertex(ctx, fetchWM.HBWatch, fetchWM.OTWatch)
 	fromEdge := fetch.NewEdgeBuffer(ctx, vertexName, fromVertex)
 
 	gf := &GenericFetch{
diff --git a/pkg/watermark/generic/generic.go b/pkg/watermark/generic/generic.go
index c9adc392..1e7d2fe6 100644
--- a/pkg/watermark/generic/generic.go
+++ b/pkg/watermark/generic/generic.go
@@ -74,7 +74,7 @@ func BuildJetStreamWatermarkProgressors(ctx context.Context, vertexInstance *v1a
 	}
 
 	var fetchWmWatchers = BuildFetchWMWatchers(hbWatch, otWatch)
-	fetchWatermark = NewGenericFetch(ctx, vertexInstance.Vertex.Name, GetFetchKeyspace(vertexInstance.Vertex), fetchWmWatchers)
+	fetchWatermark = NewGenericFetch(ctx, vertexInstance.Vertex.Name, fetchWmWatchers)
 
 	// Publisher map creation, we need a publisher per edge.
 
@@ -95,7 +95,7 @@ func BuildJetStreamWatermarkProgressors(ctx context.Context, vertexInstance *v1a
 
 		var publishStores = BuildPublishWMStores(hbStore, otStore)
 		var processorName = fmt.Sprintf("%s-%d", vertexInstance.Vertex.Name, vertexInstance.Replica)
-		publishWatermark[buffer.Name] = NewGenericPublish(ctx, processorName, GetPublishKeySpace(vertexInstance.Vertex), publishStores)
+		publishWatermark[buffer.Name] = NewGenericPublish(ctx, processorName, publishStores)
 	}
 
 	return fetchWatermark, publishWatermark
diff --git a/pkg/watermark/generic/publisher.go b/pkg/watermark/generic/publisher.go
index 25068837..c3f381ad 100644
--- a/pkg/watermark/generic/publisher.go
+++ b/pkg/watermark/generic/publisher.go
@@ -33,8 +33,8 @@ var _ publish.Publisher = (*GenericPublish)(nil)
 // NewGenericPublish returns GenericPublish. processorName is the unique processor (pod) that is running on this vertex.
 // publishKeyspace is obsolete, and will be removed in subsequent iterations. publishWM is a struct for storing both the heartbeat
 // and the offset watermark timeline stores for the Vn vertex.
-func NewGenericPublish(ctx context.Context, processorName string, publishKeyspace string, publishWM PublishWMStores) *GenericPublish {
-	publishEntity := processor.NewProcessorEntity(processorName, publishKeyspace)
+func NewGenericPublish(ctx context.Context, processorName string, publishWM PublishWMStores) *GenericPublish {
+	publishEntity := processor.NewProcessorEntity(processorName)
 	udfPublish := publish.NewPublish(ctx, publishEntity, publishWM.HBStore, publishWM.OTStore)
 	gp := &GenericPublish{
 		toEdge: udfPublish,
diff --git a/pkg/watermark/processor/entity.go b/pkg/watermark/processor/entity.go
index 3c4eace2..734760e7 100644
--- a/pkg/watermark/processor/entity.go
+++ b/pkg/watermark/processor/entity.go
@@ -41,19 +41,16 @@ func WithSeparateOTBuckets(separate bool) EntityOption {
 // monotonically increase.
 type ProcessorEntitier interface {
 	GetID() string
-	GetBucketName() string
 	BuildOTWatcherKey(Watermark) string
 	ParseOTWatcherKey(string) (int64, bool, error)
 	IsOTBucketShared() bool
-	GetPublishKeyspace() string
 }
 
 // ProcessorEntity implements ProcessorEntitier.
 type ProcessorEntity struct {
 	// name is the name of the entity
-	name            string
-	publishKeyspace string
-	opts            *entityOptions
+	name string
+	opts *entityOptions
 }
 
 var _ ProcessorEntitier = (*ProcessorEntity)(nil)
@@ -64,8 +61,7 @@ var _ ProcessorEntitier = (*ProcessorEntity)(nil)
 const _defaultKeySeparator = "_"
 
 // NewProcessorEntity returns a new `ProcessorEntity`.
-// TODO: remove publishKeyspace
-func NewProcessorEntity(name string, publishKeyspace string, inputOpts ...EntityOption) *ProcessorEntity {
+func NewProcessorEntity(name string, inputOpts ...EntityOption) *ProcessorEntity {
 	opts := &entityOptions{
 		separateOTBucket: false,
 		keySeparator:     _defaultKeySeparator,
@@ -74,9 +70,8 @@ func NewProcessorEntity(name string, publishKeyspace string, inputOpts ...Entity
 		opt(opts)
 	}
 	return &ProcessorEntity{
-		name:            name,
-		publishKeyspace: publishKeyspace,
-		opts:            opts,
+		name: name,
+		opts: opts,
 	}
 }
 
@@ -85,20 +80,6 @@ func (p *ProcessorEntity) GetID() string {
 	return p.name
 }
 
-// GetPublishKeyspace returns the publishKeyspace of the entity
-func (p *ProcessorEntity) GetPublishKeyspace() string {
-	return p.publishKeyspace
-}
-
-// GetBucketName returns the offset-timeline for the entity.
-func (p *ProcessorEntity) GetBucketName() string {
-	if p.opts.separateOTBucket {
-		return p.publishKeyspace + "_OT_" + p.name
-	} else {
-		return p.publishKeyspace + "_OT"
-	}
-}
-
 // IsOTBucketShared returns true if the OT bucket is shared.
 func (p *ProcessorEntity) IsOTBucketShared() bool {
 	return p.opts.separateOTBucket
diff --git a/pkg/watermark/publish/publisher.go b/pkg/watermark/publish/publisher.go
index 10fe0dff..76e09b27 100644
--- a/pkg/watermark/publish/publisher.go
+++ b/pkg/watermark/publish/publisher.go
@@ -101,7 +101,7 @@ func (p *Publish) PublishWatermark(wm processor.Watermark, offset isb.Offset) {
 	for {
 		err := p.otStore.PutKV(p.ctx, key, value)
 		if err != nil {
-			p.log.Errorw("unable to publish watermark", zap.String("entity", p.entity.GetBucketName()), zap.String("key", key), zap.Error(err))
+			p.log.Errorw("unable to publish watermark", zap.String("HB", p.heartbeatStore.GetStoreName()), zap.String("OT", p.otStore.GetStoreName()), zap.String("key", key), zap.Error(err))
 			// TODO: better exponential backoff
 			time.Sleep(time.Millisecond * 250)
 		} else {


Test patch:
diff --git a/pkg/watermark/fetch/edge_test.go b/pkg/watermark/fetch/edge_test.go
index 05580291..9e736cf3 100644
--- a/pkg/watermark/fetch/edge_test.go
+++ b/pkg/watermark/fetch/edge_test.go
@@ -37,24 +37,24 @@ func TestBuffer_GetWatermark(t *testing.T) {
 	js, err := conn.JetStream()
 	assert.NoError(t, err)
 
-	var publisherHBKeyspace = "fetchTest_PROCESSORS"
-	deleteFn, err := createAndLaterDeleteBucket(js, &nats.KeyValueConfig{Bucket: publisherHBKeyspace})
+	var publisherHBBucketName = "fetchTest_PROCESSORS"
+	deleteFn, err := createAndLaterDeleteBucket(js, &nats.KeyValueConfig{Bucket: publisherHBBucketName})
 	assert.NoError(t, err)
 	defer deleteFn()
 
-	var publisherOTKeyspace = "fetchTest_OT"
-	deleteFn, err = createAndLaterDeleteBucket(js, &nats.KeyValueConfig{Bucket: publisherOTKeyspace})
+	var publisherOTBucketName = "fetchTest_OT"
+	deleteFn, err = createAndLaterDeleteBucket(js, &nats.KeyValueConfig{Bucket: publisherOTBucketName})
 	assert.NoError(t, err)
 	defer deleteFn()
 
-	hbWatcher, err := jetstream.NewKVJetStreamKVWatch(ctx, "testFetch", publisherHBKeyspace, defaultJetStreamClient)
-	otWatcher, err := jetstream.NewKVJetStreamKVWatch(ctx, "testFetch", publisherOTKeyspace, defaultJetStreamClient)
-	testVertex := NewFromVertex(ctx, "testVertex", hbWatcher, otWatcher)
+	hbWatcher, err := jetstream.NewKVJetStreamKVWatch(ctx, "testFetch", publisherHBBucketName, defaultJetStreamClient)
+	otWatcher, err := jetstream.NewKVJetStreamKVWatch(ctx, "testFetch", publisherOTBucketName, defaultJetStreamClient)
+	testVertex := NewFromVertex(ctx, hbWatcher, otWatcher)
 	var (
 		// TODO: watcher should not be nil
-		testPod0     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod1", "test"), 5, otWatcher)
-		testPod1     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod2", "test"), 5, otWatcher)
-		testPod2     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod3", "test"), 5, otWatcher)
+		testPod0     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod1"), 5, otWatcher)
+		testPod1     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod2"), 5, otWatcher)
+		testPod2     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod3"), 5, otWatcher)
 		pod0Timeline = []OffsetWatermark{
 			{watermark: 11, offset: 9},
 			{watermark: 12, offset: 20},
diff --git a/pkg/watermark/fetch/processor_to_fetch_test.go b/pkg/watermark/fetch/processor_to_fetch_test.go
index 550ffd46..507c84c3 100644
--- a/pkg/watermark/fetch/processor_to_fetch_test.go
+++ b/pkg/watermark/fetch/processor_to_fetch_test.go
@@ -12,7 +12,7 @@ import (
 
 func TestFromProcessor_setStatus(t *testing.T) {
 	var ctx = context.Background()
-	p := NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod1", "test"), 5, nil)
+	p := NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod1"), 5, nil)
 	p.setStatus(_inactive)
 	assert.Equal(t, _inactive, p.status)
 }
diff --git a/pkg/watermark/fetch/vertex_test.go b/pkg/watermark/fetch/vertex_test.go
index 89edbf8e..d75c86d0 100644
--- a/pkg/watermark/fetch/vertex_test.go
+++ b/pkg/watermark/fetch/vertex_test.go
@@ -72,7 +72,7 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 
 	hbWatcher, err := jetstream.NewKVJetStreamKVWatch(ctx, "testFetch", keyspace+"_PROCESSORS", defaultJetStreamClient)
 	otWatcher, err := jetstream.NewKVJetStreamKVWatch(ctx, "testFetch", keyspace+"_OT", defaultJetStreamClient)
-	var testVertex = NewFromVertex(ctx, keyspace, hbWatcher, otWatcher, WithPodHeartbeatRate(1), WithRefreshingProcessorsRate(1), WithSeparateOTBuckets(false))
+	var testVertex = NewFromVertex(ctx, hbWatcher, otWatcher, WithPodHeartbeatRate(1), WithRefreshingProcessorsRate(1), WithSeparateOTBuckets(false))
 	var testBuffer = NewEdgeBuffer(ctx, "testBuffer", testVertex)
 
 	go func() {
@@ -305,7 +305,7 @@ func TestFetcherWithSeparateOTBucket(t *testing.T) {
 
 	hbWatcher, err := jetstream.NewKVJetStreamKVWatch(ctx, "testFetch", keyspace+"_PROCESSORS", defaultJetStreamClient)
 	otWatcher, err := jetstream.NewKVJetStreamKVWatch(ctx, "testFetch", keyspace+"_OT", defaultJetStreamClient)
-	var testVertex = NewFromVertex(ctx, keyspace, hbWatcher, otWatcher, WithPodHeartbeatRate(1), WithRefreshingProcessorsRate(1), WithSeparateOTBuckets(true))
+	var testVertex = NewFromVertex(ctx, hbWatcher, otWatcher, WithPodHeartbeatRate(1), WithRefreshingProcessorsRate(1), WithSeparateOTBuckets(true))
 	var testBuffer = NewEdgeBuffer(ctx, "testBuffer", testVertex)
 
 	//var location, _ = time.LoadLocation("UTC")
diff --git a/pkg/watermark/processor/entity_test.go b/pkg/watermark/processor/entity_test.go
index 2e9aaf46..b5bd0eaa 100644
--- a/pkg/watermark/processor/entity_test.go
+++ b/pkg/watermark/processor/entity_test.go
@@ -9,17 +9,15 @@ import (
 )
 
 func TestEntity(t *testing.T) {
-	e := NewProcessorEntity("pod0", "v0")
+	e := NewProcessorEntity("pod0")
 	assert.False(t, e.opts.separateOTBucket)
 	assert.Equal(t, "pod0", e.GetID())
-	assert.Equal(t, "v0_OT", e.GetBucketName())
 }
 
 func TestEntityDifferentBuckets(t *testing.T) {
-	e := NewProcessorEntity("pod0", "v0", WithSeparateOTBuckets(true))
+	e := NewProcessorEntity("pod0", WithSeparateOTBuckets(true))
 	assert.True(t, e.IsOTBucketShared())
 	assert.Equal(t, "pod0", e.GetID())
-	assert.Equal(t, "v0_OT_pod0", e.GetBucketName())
 }
 
 func ExampleWatermark_String() {
@@ -41,7 +39,7 @@ func TestProcessorEntity_ParseOTWatcherKey(t *testing.T) {
 	}{
 		{
 			name:      "good_with_split",
-			p:         NewProcessorEntity("test1", "keyspace1", WithSeparateOTBuckets(true)),
+			p:         NewProcessorEntity("test1", WithSeparateOTBuckets(true)),
 			arg:       "1234",
 			wantEpoch: 1234,
 			wantSkip:  false,
@@ -49,7 +47,7 @@ func TestProcessorEntity_ParseOTWatcherKey(t *testing.T) {
 		},
 		{
 			name:      "bad_with_split",
-			p:         NewProcessorEntity("test1", "keyspace1", WithSeparateOTBuckets(true)),
+			p:         NewProcessorEntity("test1", WithSeparateOTBuckets(true)),
 			arg:       _defaultKeySeparator + "1234",
 			wantEpoch: 0,
 			wantSkip:  false,
@@ -57,7 +55,7 @@ func TestProcessorEntity_ParseOTWatcherKey(t *testing.T) {
 		},
 		{
 			name:      "bad_without_split_butSkip",
-			p:         NewProcessorEntity("test1", "keyspace1", WithSeparateOTBuckets(false)),
+			p:         NewProcessorEntity("test1", WithSeparateOTBuckets(false)),
 			arg:       _defaultKeySeparator + "1234", // name is missing
 			wantEpoch: 1234,
 			wantSkip:  true,
@@ -65,7 +63,7 @@ func TestProcessorEntity_ParseOTWatcherKey(t *testing.T) {
 		},
 		{
 			name:      "bad_without_split_missing_separator",
-			p:         NewProcessorEntity("test1", "keyspace1", WithSeparateOTBuckets(false)),
+			p:         NewProcessorEntity("test1", WithSeparateOTBuckets(false)),
 			arg:       "1234",
 			wantEpoch: 0,
 			wantSkip:  false,
@@ -73,7 +71,7 @@ func TestProcessorEntity_ParseOTWatcherKey(t *testing.T) {
 		},
 		{
 			name:      "good_without_split",
-			p:         NewProcessorEntity("test1", "keyspace1", WithSeparateOTBuckets(false)),
+			p:         NewProcessorEntity("test1", WithSeparateOTBuckets(false)),
 			arg:       "test1" + _defaultKeySeparator + "1234",
 			wantEpoch: 1234,
 			wantSkip:  false,
@@ -81,7 +79,7 @@ func TestProcessorEntity_ParseOTWatcherKey(t *testing.T) {
 		},
 		{
 			name:      "good_without_split_skip",
-			p:         NewProcessorEntity("test1", "keyspace1", WithSeparateOTBuckets(false)),
+			p:         NewProcessorEntity("test1", WithSeparateOTBuckets(false)),
 			arg:       "test-not-this" + _defaultKeySeparator + "1234",
 			wantEpoch: 1234,
 			wantSkip:  true,
@@ -112,7 +110,7 @@ func TestProcessorEntity_splitKey(t *testing.T) {
 	}{
 		{
 			name:    "good_separate_bucket",
-			p:       NewProcessorEntity("test1", "keyspace1", WithSeparateOTBuckets(true)),
+			p:       NewProcessorEntity("test1", WithSeparateOTBuckets(true)),
 			arg:     "1234",
 			want:    "",
 			want1:   "1234",
@@ -120,7 +118,7 @@ func TestProcessorEntity_splitKey(t *testing.T) {
 		},
 		{
 			name:    "bad_expected_name",
-			p:       NewProcessorEntity("test1", "keyspace1", WithSeparateOTBuckets(false)),
+			p:       NewProcessorEntity("test1", WithSeparateOTBuckets(false)),
 			arg:     "1234",
 			want:    "",
 			want1:   "",
@@ -128,7 +126,7 @@ func TestProcessorEntity_splitKey(t *testing.T) {
 		},
 		{
 			name:    "good_same_bucket",
-			p:       NewProcessorEntity("test1", "keyspace1", WithSeparateOTBuckets(false)),
+			p:       NewProcessorEntity("test1", WithSeparateOTBuckets(false)),
 			arg:     "p1" + _defaultKeySeparator + "1234",
 			want:    "p1",
 			want1:   "1234",
diff --git a/pkg/watermark/publish/publisher_test.go b/pkg/watermark/publish/publisher_test.go
index 4313b84e..38e101f5 100644
--- a/pkg/watermark/publish/publisher_test.go
+++ b/pkg/watermark/publish/publisher_test.go
@@ -53,7 +53,7 @@ func TestPublisherWithSeparateOTBuckets(t *testing.T) {
 	otKV, err := jetstream.NewKVJetStreamKVStore(ctx, "testPublisher", publisherOTKeyspace, defaultJetStreamClient)
 	assert.NoError(t, err)
 
-	publishEntity := processor.NewProcessorEntity("publisherTestPod1", publisherHBKeyspace)
+	publishEntity := processor.NewProcessorEntity("publisherTestPod1")
 
 	p := NewPublish(ctx, publishEntity, heartbeatKV, otKV, WithAutoRefreshHeartbeatDisabled(), WithPodHeartbeatRate(1))
 
@@ -78,9 +78,6 @@ func TestPublisherWithSeparateOTBuckets(t *testing.T) {
 
 	p.StopPublisher()
 
-	_, err = js.KeyValue(publishEntity.GetBucketName())
-	assert.Equal(t, nats.ErrBucketNotFound, err)
-
 	_, err = p.heartbeatStore.GetValue(ctx, publishEntity.GetID())
 	assert.Equal(t, nats.ErrKeyNotFound, err)
 
@@ -103,7 +100,7 @@ func TestPublisherWithSharedOTBucket(t *testing.T) {
 	deleteFn, err = createAndLaterDeleteBucket(js, &nats.KeyValueConfig{Bucket: keyspace + "_OT"})
 	defer deleteFn()
 
-	publishEntity := processor.NewProcessorEntity("publisherTestPod1", keyspace, processor.WithSeparateOTBuckets(true))
+	publishEntity := processor.NewProcessorEntity("publisherTestPod1", processor.WithSeparateOTBuckets(true))
 
 	heartbeatKV, err := jetstream.NewKVJetStreamKVStore(ctx, "testPublisher", keyspace+"_PROCESSORS", defaultJetStreamClient)
 	assert.NoError(t, err)
@@ -133,12 +130,6 @@ func TestPublisherWithSharedOTBucket(t *testing.T) {
 
 	p.StopPublisher()
 
-	// run this test only if we are not sharing the bucket
-	if !publishEntity.IsOTBucketShared() {
-		_, err = js.KeyValue(publishEntity.GetBucketName())
-		assert.Equal(t, nats.ErrBucketNotFound, err)
-	}
-
 	_, err = p.heartbeatStore.GetValue(ctx, publishEntity.GetID())
 	assert.Equal(t, nats.ErrKeyNotFound, err)
 }