Repository: numaproj/numaflow
Pull request #792 (2023-06-14 14:57:58)

chore: metrics changes for multi partition edges

vigith: look like we have a flaky unit test
```
=== RUN   TestFetcherWithSameOTBucketWithSinglePartition
{"level":"info","ts":"2023-06-14T16:10:58.171016269Z","logger":"numaflow","caller":"processor/processor_manager.go:219","msg":"Deleting","key":"p2","p2":"p2 status:active, timelines:  [1651161660000:105] -> [-1:-1] -> [-1:-1] -> [-1:-1] -> [-1:-1] -> [-1:-1] -> [-1:-1] -> [-1:-1] -> [-1:-1] -> [-1:-1]\n "}
panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x18 pc=0x183344b]

goroutine 82 [running]:
github.com/numaproj/numaflow/pkg/watermark/store/inmem.(*inMemWatch).Watch.func2()
	/home/runner/work/numaflow/numaflow/pkg/watermark/store/inmem/kv_watch.go:87 +0x18b
created by github.com/numaproj/numaflow/pkg/watermark/store/inmem.(*inMemWatch).Watch
	/home/runner/work/numaflow/numaflow/pkg/watermark/store/inmem/kv_watch.go:75 +0x2e6
FAIL	github.com/numaproj/numaflow/pkg/watermark/fetch	1.724s
?   	github.com/numaproj/numaflow/pkg/watermark/generic	[no test files]
?   	github.com/numaproj/numaflow/pkg/watermark/generic/jetstream	[no test files]
=== RUN   TestEntity```

Patch:
diff --git a/docs/operations/example-dashboard-template.json b/docs/operations/example-dashboard-template.json
index c7d7d69b..b1cca72b 100644
--- a/docs/operations/example-dashboard-template.json
+++ b/docs/operations/example-dashboard-template.json
@@ -1864,14 +1864,14 @@
           "type": "prometheus",
           "uid": "eF7wTnc4k"
         },
-        "definition": "label_values({namespace=~\"$namespace\", pipeline=~\"$pipeline\", vertex=~\"$vertex\"}, buffer)",
+        "definition": "label_values({namespace=~\"$namespace\", pipeline=~\"$pipeline\", vertex=~\"$vertex\"}, partition_name)",
         "hide": 0,
         "includeAll": false,
         "multi": false,
         "name": "buffer",
         "options": [],
         "query": {
-          "query": "label_values({namespace=~\"$namespace\", pipeline=~\"$pipeline\", vertex=~\"$vertex\"}, buffer)",
+          "query": "label_values({namespace=~\"$namespace\", pipeline=~\"$pipeline\", vertex=~\"$vertex\"}, partition_name)",
           "refId": "StandardVariableQuery"
         },
         "refresh": 1,
diff --git a/docs/operations/metrics/metrics.md b/docs/operations/metrics/metrics.md
index b869c6f4..83e617f9 100644
--- a/docs/operations/metrics/metrics.md
+++ b/docs/operations/metrics/metrics.md
@@ -12,21 +12,21 @@ These metrics can be used to determine throughput of your pipeline.
 
 #### Data-forward
 
-| Metric name                           | Metric type | Labels                                                                                                            | Description                                                                                            |
-|---------------------------------------|-------------|-------------------------------------------------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------|
-| `forwarder_read_total`                | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `buffer=<buffer-name>`                                | Provides the total number of messages read by a given Vertex from an Inter-Step Buffer                 |
-| `forwarder_read_bytes_total`          | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `buffer=<buffer-name>`                                | Provides the total number of bytes read by a given Vertex from an Inter-Step Buffer                    |
-| `forwarder_write_total`               | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `buffer=<buffer-name>`                                | Provides the total number of messages written to Inter-Step Buffer by a given Vertex                   |
-| `forwarder_write_bytes_total`         | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `buffer=<buffer-name>`                                | Provides the total number of bytes written to Inter-Step Buffer by a given Vertex                      |
-| `forwarder_ack_total`                 | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `buffer=<buffer-name>`                                | Provides the total number of messages acknowledged by a given Vertex from an Inter-Step Buffer         |
-| `forwarder_drop_total`                | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `buffer=<buffer-name>`                                | Provides the total number of messages dropped by a given Vertex due to a full Inter-Step Buffer        |
-| `forwarder_drop_bytes_total`          | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `buffer=<buffer-name>`                                | Provides the total number of bytes dropped by a given Vertex due to a full Inter-Step Buffer           |
-| `reduce_isb_reader_read_total`        | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `buffer=<buffer-name>` | Provides the total number of messages read by a given Reduce Vertex from an Inter-Step Buffer          |
-| `reduce_isb_reader_read_bytes_total`  | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `buffer=<buffer-name>` | Provides the total number of bytes read by a given Reduce Vertex from an Inter-Step Buffer             |
-| `reduce_isb_writer_write_total`       | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `buffer=<buffer-name>` | Provides the total number of messages written to Inter-Step Buffer by a given Reduce Vertex            |
-| `reduce_isb_writer_write_bytes_total` | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `buffer=<buffer-name>` | Provides the total number of bytes written to Inter-Step Buffer by a given Reduce Vertex               |
-| `reduce_isb_writer_drop_total`        | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `buffer=<buffer-name>` | Provides the total number of messages dropped by a given Reduce Vertex due to a full Inter-Step Buffer |
-| `reduce_isb_writer_drop_bytes_total`  | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `buffer=<buffer-name>` | Provides the total number of bytes dropped by a given Reduce Vertex due to a full Inter-Step Buffer    |
+| Metric name                           | Metric type | Labels                                                                                                                       | Description                                                                                                      |
+|---------------------------------------|-------------|------------------------------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------|
+| `forwarder_read_total`                | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of messages read by a given Vertex from an Inter-Step Buffer Partition                 |
+| `forwarder_read_bytes_total`          | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of bytes read by a given Vertex from an Inter-Step Buffer Partition                    |
+| `forwarder_write_total`               | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of messages written to Inter-Step Buffer by a given Vertex                             |
+| `forwarder_write_bytes_total`         | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of bytes written to Inter-Step Buffer by a given Vertex                                |
+| `forwarder_ack_total`                 | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of messages acknowledged by a given Vertex from an Inter-Step Buffer Partition         |
+| `forwarder_drop_total`                | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of messages dropped by a given Vertex due to a full Inter-Step Buffer Partition        |
+| `forwarder_drop_bytes_total`          | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of bytes dropped by a given Vertex due to a full Inter-Step Buffer Partition           |
+| `reduce_isb_reader_read_total`        | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `partition_name=<partition-name>` | Provides the total number of messages read by a given Reduce Vertex from an Inter-Step Buffer Partition          |
+| `reduce_isb_reader_read_bytes_total`  | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `partition_name=<partition-name>` | Provides the total number of bytes read by a given Reduce Vertex from an Inter-Step Buffer Partition             |
+| `reduce_isb_writer_write_total`       | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `partition_name=<partition-name>` | Provides the total number of messages written to Inter-Step Buffer by a given Reduce Vertex                      |
+| `reduce_isb_writer_write_bytes_total` | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `partition_name=<partition-name>` | Provides the total number of bytes written to Inter-Step Buffer by a given Reduce Vertex                         |
+| `reduce_isb_writer_drop_total`        | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `partition_name=<partition-name>` | Provides the total number of messages dropped by a given Reduce Vertex due to a full Inter-Step Buffer Partition |
+| `reduce_isb_writer_drop_bytes_total`  | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `partition_name=<partition-name>` | Provides the total number of bytes dropped by a given Reduce Vertex due to a full Inter-Step Buffer Partition    |
 
 #### Kafka Source
 
@@ -37,12 +37,12 @@ These metrics can be used to determine throughput of your pipeline.
 
 #### Redis Streams Source
 
-| Metric name                           | Metric type | Labels                                                 | Description                                                                                           |
-| ------------------------------------- | ----------- | ------------------------------------------------------ | ----------------------------------------------------------------------------------------------------- |
-| `redis_streams_source_read_total`     | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` | Provides the number of messages read by the Redis Streams Source Vertex/Processor                     |
-| `redis_streams_source_read_err_total` | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` | Provides the number of read failures for the Redis Streams Source Vertex/Processor                    |
-| `redis_streams_source_ack_total`      | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` | Provides the number of messages acked by the Redis Streams Source Vertex/Processor                    |
-| `redis_streams_source_ack_err_total`  | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` | Provides the number of errors attempting to ack by the Redis Streams Source Vertex/Processor          |
+| Metric name                           | Metric type | Labels                                                 | Description                                                                                  |
+|---------------------------------------|-------------|--------------------------------------------------------|----------------------------------------------------------------------------------------------|
+| `redis_streams_source_read_total`     | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` | Provides the number of messages read by the Redis Streams Source Vertex/Processor            |
+| `redis_streams_source_read_err_total` | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` | Provides the number of read failures for the Redis Streams Source Vertex/Processor           |
+| `redis_streams_source_ack_total`      | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` | Provides the number of messages acked by the Redis Streams Source Vertex/Processor           |
+| `redis_streams_source_ack_err_total`  | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` | Provides the number of errors attempting to ack by the Redis Streams Source Vertex/Processor |
 
 #### Generator Source
 
@@ -72,33 +72,33 @@ These metrics can be used to determine throughput of your pipeline.
 
 These metrics can be used to determine the latency of your pipeline.
 
-| Metric name                               | Metric type | Labels                                                                                | Description                                                                                    |
-|-------------------------------------------|-------------|---------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------|
-| `forwarder_udf_processing_time`           | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `buffer=<buffer-name>`    | Provides a histogram distribution of the processing times of User Defined Functions. (UDF's)   |
-| `forwarder_forward_chunk_processing_time` | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `buffer=<buffer-name>`    | Provides a histogram distribution of the processing times of the forwarder function as a whole |
-| `reduce_pnf_process_time`                 | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` | Provides a histogram distribution of the processing times of the reducer                       |
-| `reduce_pnf_forward_time`                 | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` | Provides a histogram distribution of the forwarding times of the reducer                       |
+| Metric name                               | Metric type | Labels                                                                                        | Description                                                                                    |
+|-------------------------------------------|-------------|-----------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------|
+| `forwarder_udf_processing_time`           | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>` | Provides a histogram distribution of the processing times of User Defined Functions. (UDF's)   |
+| `forwarder_forward_chunk_processing_time` | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>` | Provides a histogram distribution of the processing times of the forwarder function as a whole |
+| `reduce_pnf_process_time`                 | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>`         | Provides a histogram distribution of the processing times of the reducer                       |
+| `reduce_pnf_forward_time`                 | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>`         | Provides a histogram distribution of the forwarding times of the reducer                       |
 
 ### Errors
 
 These metrics can be used to determine if there are any errors in the pipeline
 
-| Metric name                           | Metric type | Labels                                                                                                            | Description                                                                   |
-|---------------------------------------|-------------|-------------------------------------------------------------------------------------------------------------------|-------------------------------------------------------------------------------|
-| `forwarder_platform_error`            | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `buffer=<buffer-name>`                                | Indicates any internal errors which could stop pipeline processing            |
-| `forwarder_read_error`                | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `buffer=<buffer-name>`                                | Indicates any errors while reading messages by the forwarder                  |
-| `forwarder_write_error`               | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `buffer=<buffer-name>`                                | Indicates any errors while writing messages by the forwarder                  |
-| `forwarder_ack_error`                 | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `buffer=<buffer-name>`                                | Indicates any errors while acknowledging messages by the forwarder            |
-| `kafka_source_offset_ack_errors`      | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>`                                                            | Indicates any kafka acknowledgement errors                                    |
-| `kafka_sink_write_error_total`        | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>`                                                            | Provides the number of errors while writing to the Kafka sink                 |
-| `kafka_sink_write_timeout_total`      | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>`                                                            | Provides the write timeouts while writing to the Kafka sink                   |
-| `isb_jetstream_read_error_total`      | Counter     | `buffer=<buffer-name>`                                                                                            | Indicates any read errors with NATS Jetstream ISB                             |
-| `isb_jetstream_write_error_total`     | Counter     | `buffer=<buffer-name>`                                                                                            | Indicates any write errors with NATS Jetstream ISB                            |
-| `isb_redis_read_error_total`          | Counter     | `buffer=<buffer-name>`                                                                                            | Indicates any read errors with Redis ISB                                      |
-| `isb_redis_write_error_total`         | Counter     | `buffer=<buffer-name>`                                                                                            | Indicates any write errors with Redis ISB                                     |
-| `reduce_isb_reader_read_error_total`  | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `buffer=<buffer-name>` | Indicates any read errors with Reducer ISB                                    |
-| `reduce_isb_writer_write_error_total` | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `buffer=<buffer-name>` | Indicates any write errors with Reducer ISB                                   |
-| `reduce_pnf_platform_error_total`     | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>`                             | Indicates any internal errors while processing and forwarding data by reducer |
+| Metric name                           | Metric type | Labels                                                                                                                       | Description                                                                   |
+|---------------------------------------|-------------|------------------------------------------------------------------------------------------------------------------------------|-------------------------------------------------------------------------------|
+| `forwarder_platform_error`            | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Indicates any internal errors which could stop pipeline processing            |
+| `forwarder_read_error`                | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Indicates any errors while reading messages by the forwarder                  |
+| `forwarder_write_error`               | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Indicates any errors while writing messages by the forwarder                  |
+| `forwarder_ack_error`                 | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Indicates any errors while acknowledging messages by the forwarder            |
+| `kafka_source_offset_ack_errors`      | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>`                                                                       | Indicates any kafka acknowledgement errors                                    |
+| `kafka_sink_write_error_total`        | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>`                                                                       | Provides the number of errors while writing to the Kafka sink                 |
+| `kafka_sink_write_timeout_total`      | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>`                                                                       | Provides the write timeouts while writing to the Kafka sink                   |
+| `isb_jetstream_read_error_total`      | Counter     | `partition_name=<partition-name>`                                                                                            | Indicates any read errors with NATS Jetstream ISB                             |
+| `isb_jetstream_write_error_total`     | Counter     | `partition_name=<partition-name>`                                                                                            | Indicates any write errors with NATS Jetstream ISB                            |
+| `isb_redis_read_error_total`          | Counter     | `partition_name=<partition-name>`                                                                                            | Indicates any read errors with Redis ISB                                      |
+| `isb_redis_write_error_total`         | Counter     | `partition_name=<partition-name>`                                                                                            | Indicates any write errors with Redis ISB                                     |
+| `reduce_isb_reader_read_error_total`  | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `partition_name=<partition-name>` | Indicates any read errors with Reducer ISB                                    |
+| `reduce_isb_writer_write_error_total` | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>` <br> `partition_name=<partition-name>` | Indicates any write errors with Reducer ISB                                   |
+| `reduce_pnf_platform_error_total`     | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>`                                        | Indicates any internal errors while processing and forwarding data by reducer |
 
 ### Saturation
 
diff --git a/pkg/forward/forward.go b/pkg/forward/forward.go
index f323cdfc..d455e123 100644
--- a/pkg/forward/forward.go
+++ b/pkg/forward/forward.go
@@ -15,7 +15,7 @@ limitations under the License.
 */
 
 /*
-Package forward does the Read (fromBuffer) -> Process (UDF) -> Forward (toBuffers) -> Ack (fromBuffer) loop.
+Package forward does the Read (fromBufferPartition) -> Process (UDF) -> Forward (toBuffers) -> Ack (fromBufferPartition) loop.
 */
 package forward
 
@@ -49,8 +49,8 @@ type InterStepDataForward struct {
 	ctx context.Context
 	// cancelFn cancels our new context, our cancellation is little more complex and needs to be well orchestrated, hence
 	// we need something more than a cancel().
-	cancelFn   context.CancelFunc
-	fromBuffer isb.BufferReader
+	cancelFn            context.CancelFunc
+	fromBufferPartition isb.BufferReader
 	// toBuffers is a map of toVertex name to the toVertex's owned buffers.
 	toBuffers map[string][]isb.BufferWriter
 	FSD       ToWhichStepDecider
@@ -88,14 +88,14 @@ func NewInterStepDataForward(vertex *dfv1.Vertex,
 	ctx, cancel := context.WithCancel(context.Background())
 
 	var isdf = InterStepDataForward{
-		ctx:          ctx,
-		cancelFn:     cancel,
-		fromBuffer:   fromStep,
-		toBuffers:    toSteps,
-		FSD:          fsd,
-		UDF:          applyUDF,
-		wmFetcher:    fetchWatermark,
-		wmPublishers: publishWatermark,
+		ctx:                 ctx,
+		cancelFn:            cancel,
+		fromBufferPartition: fromStep,
+		toBuffers:           toSteps,
+		FSD:                 fsd,
+		UDF:                 applyUDF,
+		wmFetcher:           fetchWatermark,
+		wmPublishers:        publishWatermark,
 		// should we do a check here for the values not being null?
 		vertexName:   vertex.Spec.Name,
 		pipelineName: vertex.Spec.PipelineName,
@@ -150,7 +150,7 @@ func (isdf *InterStepDataForward) Start() <-chan struct{} {
 				}
 			default:
 				// once context.Done() is called, we still have to try to forwardAChunk because in graceful
-				// shutdown the fromBuffer should be empty.
+				// shutdown the fromBufferPartition should be empty.
 			}
 			// keep doing what you are good at
 			isdf.forwardAChunk(isdf.ctx)
@@ -160,10 +160,10 @@ func (isdf *InterStepDataForward) Start() <-chan struct{} {
 	go func() {
 		wg.Wait()
 		// Clean up resources for buffer reader and all the writers if any.
-		if err := isdf.fromBuffer.Close(); err != nil {
+		if err := isdf.fromBufferPartition.Close(); err != nil {
 			log.Errorw("Failed to close buffer reader, shutdown anyways...", zap.Error(err))
 		} else {
-			log.Infow("Closed buffer reader", zap.String("bufferFrom", isdf.fromBuffer.GetName()))
+			log.Infow("Closed buffer reader", zap.String("bufferFrom", isdf.fromBufferPartition.GetName()))
 		}
 		for _, buffer := range isdf.toBuffers {
 			for _, partition := range buffer {
@@ -199,7 +199,7 @@ type readWriteMessagePair struct {
 	udfError      error
 }
 
-// forwardAChunk forwards a chunk of message from the fromBuffer to the toBuffers. It does the Read -> Process -> Forward -> Ack chain
+// forwardAChunk forwards a chunk of message from the fromBufferPartition to the toBuffers. It does the Read -> Process -> Forward -> Ack chain
 // for a chunk of messages returned by the first Read call. It will return only if only we are successfully able to ack
 // the message after forwarding, barring any platform errors. The platform errors include buffer-full,
 // buffer-not-reachable, etc., but does not include errors due to user code UDFs, WhereTo, etc.
@@ -208,12 +208,12 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 	// There is a chance that we have read the message and the container got forcefully terminated before processing. To provide
 	// at-least-once semantics for reading, during restart we will have to reprocess all unacknowledged messages. It is the
 	// responsibility of the Read function to do that.
-	readMessages, err := isdf.fromBuffer.Read(ctx, isdf.opts.readBatchSize)
+	readMessages, err := isdf.fromBufferPartition.Read(ctx, isdf.opts.readBatchSize)
 	if err != nil {
-		isdf.opts.logger.Warnw("failed to read fromBuffer", zap.Error(err))
-		readMessagesError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Inc()
+		isdf.opts.logger.Warnw("failed to read fromBufferPartition", zap.Error(err))
+		readMessagesError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Inc()
 	}
-	readMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Add(float64(len(readMessages)))
+	readMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(readMessages)))
 
 	// process only if we have any read messages. There is a natural looping here if there is an internal error while
 	// reading, and we are not able to proceed.
@@ -225,7 +225,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		// We also publish a control message if this is the first time we get this idle situation.
 		// we use the HeadWMB as the watermark for the idle
 		// we get the HeadWMB for the partition from which we read the messages
-		var processorWMB = isdf.wmFetcher.GetHeadWMB(isdf.fromBuffer.GetPartitionIdx())
+		var processorWMB = isdf.wmFetcher.GetHeadWMB(isdf.fromBufferPartition.GetPartitionIdx())
 		if !isdf.wmbChecker.ValidateHeadWMB(processorWMB) {
 			// validation failed, skip publishing
 			isdf.opts.logger.Debugw("skip publishing idle watermark",
@@ -270,7 +270,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		// let's track only the first element's watermark. This is important because we reassign the watermark we fetch
 		// to all the elements in the batch. If we were to assign last element's watermark, we will wrongly mark on-time data as late.
 		// we fetch the watermark for the partition from which we read the message.
-		processorWM = isdf.wmFetcher.GetWatermark(readMessages[0].ReadOffset, isdf.fromBuffer.GetPartitionIdx())
+		processorWM = isdf.wmFetcher.GetWatermark(readMessages[0].ReadOffset, isdf.fromBufferPartition.GetPartitionIdx())
 	}
 
 	var writeOffsets map[string][][]isb.Offset
@@ -303,7 +303,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		// send to UDF only the data messages
 		for idx, m := range dataMessages {
 			// emit message size metric
-			readBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Add(float64(len(m.Payload)))
+			readBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(m.Payload)))
 			// assign watermark to the message. assign time.UnixMilli(-1) as watermark when we are at source vertex.
 			m.Watermark = time.Time(processorWM)
 			// send UDF processing work to the channel
@@ -316,7 +316,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		// context.Done() is closed.
 		wg.Wait()
 		isdf.opts.logger.Debugw("concurrent applyUDF completed", zap.Int("concurrency", isdf.opts.udfConcurrency), zap.Duration("took", time.Since(concurrentUDFProcessingStart)))
-		concurrentUDFProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Observe(float64(time.Since(concurrentUDFProcessingStart).Microseconds()))
+		concurrentUDFProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Observe(float64(time.Since(concurrentUDFProcessingStart).Microseconds()))
 		// UDF processing is done.
 
 		// if vertex type is source, it means we have finished the source data transformation.
@@ -338,7 +338,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 			isdf.opts.srcWatermarkPublisher.PublishSourceWatermarks(transformedReadMessages)
 			// fetch the source watermark again, we might not get the latest watermark because of publishing delay,
 			// but ideally we should use the latest to determine the IsLate attribute.
-			processorWM = isdf.wmFetcher.GetWatermark(readMessages[0].ReadOffset, isdf.fromBuffer.GetPartitionIdx())
+			processorWM = isdf.wmFetcher.GetWatermark(readMessages[0].ReadOffset, isdf.fromBufferPartition.GetPartitionIdx())
 			// assign isLate
 			for _, m := range writeMessages {
 				if processorWM.After(m.EventTime) { // Set late data at source level
@@ -353,17 +353,17 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 			// look for errors in udf processing, if we see even 1 error NoAck all messages
 			// then return. Handling partial retrying is not worth ATM.
 			if m.udfError != nil {
-				udfError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Inc()
+				udfError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Inc()
 				isdf.opts.logger.Errorw("failed to applyUDF", zap.Error(m.udfError))
 				// As there's no partial failure, non-ack all the readOffsets
-				isdf.fromBuffer.NoAck(ctx, readOffsets)
+				isdf.fromBufferPartition.NoAck(ctx, readOffsets)
 				return
 			}
 			// update toBuffers
 			for _, message := range m.writeMessages {
 				if err := isdf.whereToStep(message, messageToStep, m.readMessage); err != nil {
 					isdf.opts.logger.Errorw("failed in whereToStep", zap.Error(err))
-					isdf.fromBuffer.NoAck(ctx, readOffsets)
+					isdf.fromBufferPartition.NoAck(ctx, readOffsets)
 					return
 				}
 			}
@@ -373,7 +373,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		writeOffsets, err = isdf.writeToBuffers(ctx, messageToStep)
 		if err != nil {
 			isdf.opts.logger.Errorw("failed to write to toBuffers", zap.Error(err))
-			isdf.fromBuffer.NoAck(ctx, readOffsets)
+			isdf.fromBufferPartition.NoAck(ctx, readOffsets)
 			return
 		}
 	} else {
@@ -381,7 +381,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		if err != nil {
 			isdf.opts.logger.Errorw("failed to streamMessage", zap.Error(err))
 			// As there's no partial failure, non-ack all the readOffsets
-			isdf.fromBuffer.NoAck(ctx, readOffsets)
+			isdf.fromBufferPartition.NoAck(ctx, readOffsets)
 			return
 		}
 	}
@@ -444,13 +444,13 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 	// implicit return for posterity :-)
 	if err != nil {
 		isdf.opts.logger.Errorw("failed to ack from buffer", zap.Error(err))
-		ackMessageError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Add(float64(len(readOffsets)))
+		ackMessageError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(readOffsets)))
 		return
 	}
-	ackMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Add(float64(len(readOffsets)))
+	ackMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(readOffsets)))
 
 	// ProcessingTimes of the entire forwardAChunk
-	forwardAChunkProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Observe(float64(time.Since(start).Microseconds()))
+	forwardAChunkProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Observe(float64(time.Since(start).Microseconds()))
 }
 
 // streamMessage streams the data messages to the next step.
@@ -478,14 +478,14 @@ func (isdf *InterStepDataForward) streamMessage(
 		// send to UDF only the data messages
 
 		// emit message size metric
-		readBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).
+		readBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).
 			Add(float64(len(dataMessages[0].Payload)))
 		// assign watermark to the message. assign time.UnixMilli(-1) as watermark when we are at source vertex.
 		dataMessages[0].Watermark = time.Time(processorWM)
 
 		// process the UDF and get the result
 		start := time.Now()
-		udfReadMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Inc()
+		udfReadMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Inc()
 
 		writeMessageCh := make(chan isb.WriteMessage)
 		errs, ctx := errgroup.WithContext(ctx)
@@ -498,9 +498,9 @@ func (isdf *InterStepDataForward) streamMessage(
 		msgIndex := 0
 		for writeMessage := range writeMessageCh {
 			// add partition to the ID, this is to make sure that the ID is unique across partitions
-			writeMessage.ID = fmt.Sprintf("%s-%d-%d", dataMessages[0].ReadOffset.String(), isdf.fromBuffer.GetPartitionIdx(), msgIndex)
+			writeMessage.ID = fmt.Sprintf("%s-%d-%d", dataMessages[0].ReadOffset.String(), isdf.fromBufferPartition.GetPartitionIdx(), msgIndex)
 			msgIndex += 1
-			udfWriteMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Add(float64(1))
+			udfWriteMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(1))
 
 			// update toBuffers
 			if err := isdf.whereToStep(&writeMessage, messageToStep, dataMessages[0]); err != nil {
@@ -524,7 +524,7 @@ func (isdf *InterStepDataForward) streamMessage(
 		// then return. Handling partial retrying is not worth ATM.
 		if err := errs.Wait(); err != nil {
 			udfError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName,
-				"buffer": isdf.fromBuffer.GetName()}).Inc()
+				metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Inc()
 			// We do not retry as we are streaming
 			if ok, _ := isdf.IsShuttingDown(); ok {
 				isdf.opts.logger.Errorw("UDF.Apply, Stop called while stuck on an internal error", zap.Error(err))
@@ -534,7 +534,7 @@ func (isdf *InterStepDataForward) streamMessage(
 		}
 
 		udfProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName,
-			"buffer": isdf.fromBuffer.GetName()}).Observe(float64(time.Since(start).Microseconds()))
+			metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Observe(float64(time.Since(start).Microseconds()))
 	} else {
 		// Even not data messages, forward the message to the edge buffer (could be multiple edges)
 		var err error
@@ -547,7 +547,7 @@ func (isdf *InterStepDataForward) streamMessage(
 	return writeOffsets, nil
 }
 
-// ackFromBuffer acknowledges an array of offsets back to fromBuffer and is a blocking call or until shutdown has been initiated.
+// ackFromBuffer acknowledges an array of offsets back to fromBufferPartition and is a blocking call or until shutdown has been initiated.
 func (isdf *InterStepDataForward) ackFromBuffer(ctx context.Context, offsets []isb.Offset) error {
 	var ackRetryBackOff = wait.Backoff{
 		Factor:   1,
@@ -559,7 +559,7 @@ func (isdf *InterStepDataForward) ackFromBuffer(ctx context.Context, offsets []i
 	attempt := 0
 
 	ctxClosedErr := wait.ExponentialBackoff(ackRetryBackOff, func() (done bool, err error) {
-		errs := isdf.fromBuffer.Ack(ctx, ackOffsets)
+		errs := isdf.fromBufferPartition.Ack(ctx, ackOffsets)
 		attempt += 1
 		summarizedErr := errorArrayToMap(errs)
 		var failedOffsets []isb.Offset
@@ -619,7 +619,7 @@ func (isdf *InterStepDataForward) writeToBuffers(
 }
 
 // writeToBuffer forwards an array of messages to a single buffer and is a blocking call or until shutdown has been initiated.
-func (isdf *InterStepDataForward) writeToBuffer(ctx context.Context, toBuffer isb.BufferWriter, messages []isb.Message) (writeOffsets []isb.Offset, err error) {
+func (isdf *InterStepDataForward) writeToBuffer(ctx context.Context, toBufferPartition isb.BufferWriter, messages []isb.Message) (writeOffsets []isb.Offset, err error) {
 	var (
 		totalCount int
 		writeCount int
@@ -630,7 +630,7 @@ func (isdf *InterStepDataForward) writeToBuffer(ctx context.Context, toBuffer is
 	writeOffsets = make([]isb.Offset, 0, totalCount)
 
 	for {
-		_writeOffsets, errs := toBuffer.Write(ctx, messages)
+		_writeOffsets, errs := toBufferPartition.Write(ctx, messages)
 		// Note: this is an unwanted memory allocation during a happy path. We want only minimal allocation since using failedMessages is an unlikely path.
 		var failedMessages []isb.Message
 		needRetry := false
@@ -638,13 +638,13 @@ func (isdf *InterStepDataForward) writeToBuffer(ctx context.Context, toBuffer is
 			if err := errs[idx]; err != nil {
 				// ATM there are no user defined errors during write, all are InternalErrors.
 				if errors.As(err, &isb.NoRetryableBufferWriteErr{}) {
-					// If toBuffer returns us a NoRetryableBufferWriteErr, we drop the message.
+					// If toBufferPartition returns us a NoRetryableBufferWriteErr, we drop the message.
 					dropBytes += float64(len(msg.Payload))
 				} else {
 					needRetry = true
 					// we retry only failed messages
 					failedMessages = append(failedMessages, msg)
-					writeMessagesError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": toBuffer.GetName()}).Inc()
+					writeMessagesError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Inc()
 					// a shutdown can break the blocking loop caused due to InternalErr
 					if ok, _ := isdf.IsShuttingDown(); ok {
 						platformError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName}).Inc()
@@ -664,9 +664,9 @@ func (isdf *InterStepDataForward) writeToBuffer(ctx context.Context, toBuffer is
 		if needRetry {
 			isdf.opts.logger.Errorw("Retrying failed messages",
 				zap.Any("errors", errorArrayToMap(errs)),
-				zap.String("pipeline", isdf.pipelineName),
-				zap.String("vertex", isdf.vertexName),
-				zap.String("buffer", toBuffer.GetName()),
+				zap.String(metrics.LabelPipeline, isdf.pipelineName),
+				zap.String(metrics.LabelVertex, isdf.vertexName),
+				zap.String(metrics.LabelPartitionName, toBufferPartition.GetName()),
 			)
 			// set messages to failed for the retry
 			messages = failedMessages
@@ -677,10 +677,10 @@ func (isdf *InterStepDataForward) writeToBuffer(ctx context.Context, toBuffer is
 		}
 	}
 
-	dropMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": toBuffer.GetName()}).Add(float64(totalCount - writeCount))
-	dropBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": toBuffer.GetName()}).Add(dropBytes)
-	writeMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": toBuffer.GetName()}).Add(float64(writeCount))
-	writeBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": toBuffer.GetName()}).Add(writeBytes)
+	dropMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(float64(totalCount - writeCount))
+	dropBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(dropBytes)
+	writeMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(float64(writeCount))
+	writeBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(writeBytes)
 	return writeOffsets, nil
 }
 
@@ -688,12 +688,12 @@ func (isdf *InterStepDataForward) writeToBuffer(ctx context.Context, toBuffer is
 func (isdf *InterStepDataForward) concurrentApplyUDF(ctx context.Context, readMessagePair <-chan *readWriteMessagePair) {
 	for message := range readMessagePair {
 		start := time.Now()
-		udfReadMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Inc()
+		udfReadMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Inc()
 		writeMessages, err := isdf.applyUDF(ctx, message.readMessage)
-		udfWriteMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Add(float64(len(writeMessages)))
+		udfWriteMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(writeMessages)))
 		message.writeMessages = append(message.writeMessages, writeMessages...)
 		message.udfError = err
-		udfProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Observe(float64(time.Since(start).Microseconds()))
+		udfProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Observe(float64(time.Since(start).Microseconds()))
 	}
 }
 
@@ -720,7 +720,7 @@ func (isdf *InterStepDataForward) applyUDF(ctx context.Context, readMessage *isb
 			// if we do not get a time from UDF, we set it to the time from (N-1)th vertex
 			for index, m := range writeMessages {
 				// add partition to the ID, this is to make sure that the ID is unique across partitions
-				m.ID = fmt.Sprintf("%s-%d-%d", readMessage.ReadOffset.String(), isdf.fromBuffer.GetPartitionIdx(), index)
+				m.ID = fmt.Sprintf("%s-%d-%d", readMessage.ReadOffset.String(), isdf.fromBufferPartition.GetPartitionIdx(), index)
 				if m.EventTime.IsZero() {
 					m.EventTime = readMessage.EventTime
 				}
@@ -735,7 +735,7 @@ func (isdf *InterStepDataForward) whereToStep(writeMessage *isb.WriteMessage, me
 	// call WhereTo and drop it on errors
 	to, err := isdf.FSD.WhereTo(writeMessage.Keys, writeMessage.Tags)
 	if err != nil {
-		isdf.opts.logger.Errorw("failed in whereToStep", zap.Error(isb.MessageWriteErr{Name: isdf.fromBuffer.GetName(), Header: readMessage.Header, Body: readMessage.Body, Message: fmt.Sprintf("WhereTo failed, %s", err)}))
+		isdf.opts.logger.Errorw("failed in whereToStep", zap.Error(isb.MessageWriteErr{Name: isdf.fromBufferPartition.GetName(), Header: readMessage.Header, Body: readMessage.Body, Message: fmt.Sprintf("WhereTo failed, %s", err)}))
 		// a shutdown can break the blocking loop caused due to InternalErr
 		if ok, _ := isdf.IsShuttingDown(); ok {
 			err := fmt.Errorf("whereToStep, Stop called while stuck on an internal error, %v", err)
@@ -747,7 +747,7 @@ func (isdf *InterStepDataForward) whereToStep(writeMessage *isb.WriteMessage, me
 
 	for _, t := range to {
 		if _, ok := messageToStep[t.ToVertexName]; !ok {
-			isdf.opts.logger.Errorw("failed in whereToStep", zap.Error(isb.MessageWriteErr{Name: isdf.fromBuffer.GetName(), Header: readMessage.Header, Body: readMessage.Body, Message: fmt.Sprintf("no such destination (%s)", t.ToVertexName)}))
+			isdf.opts.logger.Errorw("failed in whereToStep", zap.Error(isb.MessageWriteErr{Name: isdf.fromBufferPartition.GetName(), Header: readMessage.Header, Body: readMessage.Body, Message: fmt.Sprintf("no such destination (%s)", t.ToVertexName)}))
 		}
 		messageToStep[t.ToVertexName][t.ToVertexPartitionIdx] = append(messageToStep[t.ToVertexName][t.ToVertexPartitionIdx], writeMessage.Message)
 	}
diff --git a/pkg/forward/metrics.go b/pkg/forward/metrics.go
index 15f06519..09f42227 100644
--- a/pkg/forward/metrics.go
+++ b/pkg/forward/metrics.go
@@ -28,77 +28,77 @@ var readMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "read_total",
 	Help:      "Total number of Messages Read",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
 // readBytesCount is to indicate the number of bytes read
 var readBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "read_bytes_total",
 	Help:      "Total number of bytes read",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
 // readMessagesError is used to indicate the number of errors messages read
 var readMessagesError = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "read_error_total",
 	Help:      "Total number of Read Errors",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
 // writeMessagesCount is used to indicate the number of messages written
 var writeMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "write_total",
 	Help:      "Total number of Messages Written",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
 // writeBytesCount is to indicate the number of bytes written
 var writeBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "write_bytes_total",
 	Help:      "Total number of bytes written",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
 // writeMessagesError is used to indicate the number of errors messages written
 var writeMessagesError = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "write_error_total",
 	Help:      "Total number of Write Errors",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
 // dropMessagesCount is used to indicate the number of messages dropped
 var dropMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "drop_total",
 	Help:      "Total number of Messages Dropped",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
 // dropBytesCount is to indicate the number of bytes dropped
 var dropBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "drop_bytes_total",
 	Help:      "Total number of Bytes Dropped",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
 // ackMessagesCount is used to indicate the number of  messages acknowledged
 var ackMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "ack_total",
 	Help:      "Total number of Messages Acknowledged",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
 // ackMessageError is used to indicate the errors in the number of  messages acknowledged
 var ackMessageError = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "ack_error_total",
 	Help:      "Total number of Acknowledged Errors",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
 // udfError is used to indicate the number of UDF errors
 var udfError = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "udf_error_total",
 	Help:      "Total number of UDF Errors",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
 // platformError is used to indicate the number of Internal/Platform errors
 var platformError = promauto.NewCounterVec(prometheus.CounterOpts{
@@ -113,7 +113,7 @@ var forwardAChunkProcessingTime = promauto.NewHistogramVec(prometheus.HistogramO
 	Name:      "forward_chunk_processing_time",
 	Help:      "Processing times of the entire forward a chunk (100 microseconds to 20 minutes)",
 	Buckets:   prometheus.ExponentialBucketsRange(100, 60000000*20, 60),
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
 // udfProcessingTime is a histogram to Observe UDF Processing times as a whole
 var udfProcessingTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
@@ -121,7 +121,7 @@ var udfProcessingTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
 	Name:      "udf_processing_time",
 	Help:      "Processing times of UDF (100 microseconds to 15 minutes)",
 	Buckets:   prometheus.ExponentialBucketsRange(100, 60000000*15, 60),
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
 // concurrentUDFProcessingTime is a histogram to Observe UDF Processing times as a whole
 var concurrentUDFProcessingTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
@@ -129,18 +129,18 @@ var concurrentUDFProcessingTime = promauto.NewHistogramVec(prometheus.HistogramO
 	Name:      "concurrent_udf_processing_time",
 	Help:      "Processing times of Concurrent UDF (100 microseconds to 20 minutes)",
 	Buckets:   prometheus.ExponentialBucketsRange(100, 60000000*20, 60),
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
 // udfReadMessagesCount is used to indicate the number of messages read by UDF
 var udfReadMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "udf_read_total",
 	Help:      "Total number of Messages Read by UDF",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
 // udfWriteMessagesCount is used to indicate the number of messages written by UDF
 var udfWriteMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "udf_write_total",
 	Help:      "Total number of Messages Written by UDF",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go
index 04024331..3afcb8a3 100644
--- a/pkg/metrics/metrics.go
+++ b/pkg/metrics/metrics.go
@@ -43,6 +43,7 @@ const (
 	LabelVertex             = "vertex"
 	LabelPeriod             = "period"
 	LabelVertexReplicaIndex = "replica"
+	LabelPartitionName      = "partition_name"
 
 	VertexPendingMessages = "vertex_pending_messages"
 )
@@ -168,15 +169,18 @@ func (ms *metricsServer) buildupPendingInfo(ctx context.Context) {
 			return
 		case <-ticker.C:
 			totalPending := int64(0)
-			for _, pending := range ms.lagReaders {
-				if p, err := pending.Pending(ctx); err != nil {
+			skipPending := false
+			for _, pendingLag := range ms.lagReaders {
+				if p, err := pendingLag.Pending(ctx); err != nil || p == isb.PendingNotAvailable {
 					log.Errorw("Failed to get pending messages", zap.Error(err))
+					skipPending = true
+					break
 				} else {
 					totalPending += p
 				}
 			}
-
-			if totalPending != isb.PendingNotAvailable {
+			// Skip pending information if any pending is not available
+			if !skipPending {
 				ts := timestampedPending{pending: totalPending, timestamp: time.Now().Unix()}
 				ms.pendingInfo.Append(ts)
 			}
diff --git a/pkg/reduce/data_forward.go b/pkg/reduce/data_forward.go
index 6890c687..d5b06b47 100644
--- a/pkg/reduce/data_forward.go
+++ b/pkg/reduce/data_forward.go
@@ -59,7 +59,7 @@ type DataForward struct {
 	vertexName            string
 	pipelineName          string
 	vertexReplica         int32
-	fromBuffer            isb.BufferReader
+	fromBufferPartition   isb.BufferReader
 	toBuffers             map[string][]isb.BufferWriter
 	wmFetcher             fetch.Fetcher
 	wmPublishers          map[string]publish.Publisher
@@ -102,7 +102,7 @@ func NewDataForward(ctx context.Context,
 		vertexName:            vertexInstance.Vertex.Spec.Name,
 		pipelineName:          vertexInstance.Vertex.Spec.PipelineName,
 		vertexReplica:         vertexInstance.Replica,
-		fromBuffer:            fromBuffer,
+		fromBufferPartition:   fromBuffer,
 		toBuffers:             toBuffers,
 		wmFetcher:             fw,
 		wmPublishers:          watermarkPublishers,
@@ -179,7 +179,7 @@ func (df *DataForward) ReplayPersistedMessages(ctx context.Context) error {
 // forwardAChunk reads a chunk of messages from isb and assigns watermark to messages
 // and writes the messages to pbq
 func (df *DataForward) forwardAChunk(ctx context.Context) {
-	readMessages, err := df.fromBuffer.Read(ctx, df.opts.readBatchSize)
+	readMessages, err := df.fromBufferPartition.Read(ctx, df.opts.readBatchSize)
 	totalBytes := 0
 	if err != nil {
 		df.log.Errorw("Failed to read from isb", zap.Error(err))
@@ -187,13 +187,13 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 			metrics.LabelVertex:             df.vertexName,
 			metrics.LabelPipeline:           df.pipelineName,
 			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
-			"buffer":                        df.fromBuffer.GetName()}).Inc()
+			metrics.LabelPartitionName:      df.fromBufferPartition.GetName()}).Inc()
 	}
 
 	if len(readMessages) == 0 {
 		// we use the HeadWMB as the watermark for the idle
 		// we get the HeadWMB for the partition from which we read the messages
-		var processorWMB = df.wmFetcher.GetHeadWMB(df.fromBuffer.GetPartitionIdx())
+		var processorWMB = df.wmFetcher.GetHeadWMB(df.fromBufferPartition.GetPartitionIdx())
 		if !df.wmbChecker.ValidateHeadWMB(processorWMB) {
 			// validation failed, skip publishing
 			df.log.Debugw("skip publishing idle watermark",
@@ -244,12 +244,12 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 		metrics.LabelVertex:             df.vertexName,
 		metrics.LabelPipeline:           df.pipelineName,
 		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
-		"buffer":                        df.fromBuffer.GetName(),
+		metrics.LabelPartitionName:      df.fromBufferPartition.GetName(),
 	}).Add(float64(len(readMessages)))
 	// fetch watermark using the first element's watermark, because we assign the watermark to all other
 	// elements in the batch based on the watermark we fetch from 0th offset.
 	// get the watermark for the partition from which we read the messages
-	processorWM := df.wmFetcher.GetWatermark(readMessages[0].ReadOffset, df.fromBuffer.GetPartitionIdx())
+	processorWM := df.wmFetcher.GetWatermark(readMessages[0].ReadOffset, df.fromBufferPartition.GetPartitionIdx())
 	for _, m := range readMessages {
 		if !df.keyed {
 			m.Keys = []string{dfv1.DefaultKeyForNonKeyedData}
@@ -262,7 +262,7 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 		metrics.LabelVertex:             df.vertexName,
 		metrics.LabelPipeline:           df.pipelineName,
 		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
-		"buffer":                        df.fromBuffer.GetName(),
+		metrics.LabelPartitionName:      df.fromBufferPartition.GetName(),
 	}).Add(float64(totalBytes))
 
 	// readMessages has to be written to PBQ, acked, etc.
@@ -526,6 +526,7 @@ func (df *DataForward) ackMessages(ctx context.Context, messages []*isb.ReadMess
 						metrics.LabelVertex:             df.vertexName,
 						metrics.LabelPipeline:           df.pipelineName,
 						metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+						metrics.LabelPartitionName:      df.fromBufferPartition.GetName(),
 					}).Inc()
 
 					df.log.Errorw("Failed to ack message, retrying", zap.String("msgOffSet", o.String()), zap.Error(rErr), zap.Int("attempt", attempt))
@@ -544,6 +545,7 @@ func (df *DataForward) ackMessages(ctx context.Context, messages []*isb.ReadMess
 					metrics.LabelVertex:             df.vertexName,
 					metrics.LabelPipeline:           df.pipelineName,
 					metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+					metrics.LabelPartitionName:      df.fromBufferPartition.GetName(),
 				}).Inc()
 				return true, nil
 			})
@@ -563,10 +565,10 @@ func (df *DataForward) ShutDown(ctx context.Context) {
 
 	df.log.Infow("Stopping reduce data forwarder...")
 
-	if err := df.fromBuffer.Close(); err != nil {
+	if err := df.fromBufferPartition.Close(); err != nil {
 		df.log.Errorw("Failed to close buffer reader, shutdown anyways...", zap.Error(err))
 	} else {
-		df.log.Infow("Closed buffer reader", zap.String("bufferFrom", df.fromBuffer.GetName()))
+		df.log.Infow("Closed buffer reader", zap.String("bufferFrom", df.fromBufferPartition.GetName()))
 	}
 
 	// flush pending messages to persistent storage
diff --git a/pkg/reduce/metrics.go b/pkg/reduce/metrics.go
index 9a73282a..a0d1ce0b 100644
--- a/pkg/reduce/metrics.go
+++ b/pkg/reduce/metrics.go
@@ -60,32 +60,32 @@ var ackMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "reduce_isb_reader",
 	Name:      "ack_total",
 	Help:      "Total number of Messages Acknowledged",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, metrics.LabelPartitionName})
 
 // ackMessageError is used to indicate the errors in the number of  messages acknowledged
 var ackMessageError = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "reduce_isb_reader",
 	Name:      "ack_error_total",
 	Help:      "Total number of Acknowledged Errors",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, metrics.LabelPartitionName})
 
 // readMessagesCount is used to indicate the number of messages read
 var readMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "reduce_isb_reader",
 	Name:      "read_total",
 	Help:      "Total number of Messages Read",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, metrics.LabelPartitionName})
 
 // readBytesCount is to indicate the number of bytes read
 var readBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "reduce_isb_reader",
 	Name:      "read_bytes_total",
 	Help:      "Total number of bytes read",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, metrics.LabelPartitionName})
 
 // readMessagesError is used to indicate the number of read errors
 var readMessagesError = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "reduce_isb_reader",
 	Name:      "read_error_total",
 	Help:      "Total number of Read Errors",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, metrics.LabelPartitionName})
diff --git a/pkg/reduce/pnf/metrics.go b/pkg/reduce/pnf/metrics.go
index 1a18ec36..faa2b9c0 100644
--- a/pkg/reduce/pnf/metrics.go
+++ b/pkg/reduce/pnf/metrics.go
@@ -28,35 +28,35 @@ var writeMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "reduce_isb_writer",
 	Name:      "write_total",
 	Help:      "Total number of Messages Written",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, metrics.LabelPartitionName})
 
 // writeBytesCount is to indicate the number of bytes written
 var writeBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "reduce_isb_writer",
 	Name:      "write_bytes_total",
 	Help:      "Total number of bytes written",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, metrics.LabelPartitionName})
 
 // writeMessagesError is used to indicate the number of errors messages written
 var writeMessagesError = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "reduce_isb_writer",
 	Name:      "write_error_total",
 	Help:      "Total number of Write Errors",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, metrics.LabelPartitionName})
 
 // dropMessagesCount is used to indicate the number of messages dropped
 var dropMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "reduce_isb_writer",
 	Name:      "drop_total",
 	Help:      "Total number of Messages Dropped",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, metrics.LabelPartitionName})
 
 // dropBytesCount is used to indicate the number of bytes dropped
 var dropBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "reduce_isb_writer",
 	Name:      "drop_bytes_total",
 	Help:      "Total number of Bytes Dropped",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, "buffer"})
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, metrics.LabelPartitionName})
 
 // platformError is used to indicate the number of Internal/Platform errors
 var platformError = promauto.NewCounterVec(prometheus.CounterOpts{
diff --git a/pkg/reduce/pnf/processandforward.go b/pkg/reduce/pnf/processandforward.go
index 03236882..4b855888 100644
--- a/pkg/reduce/pnf/processandforward.go
+++ b/pkg/reduce/pnf/processandforward.go
@@ -246,7 +246,7 @@ func (p *processAndForward) writeToBuffer(ctx context.Context, edgeName string,
 				metrics.LabelVertex:             p.vertexName,
 				metrics.LabelPipeline:           p.pipelineName,
 				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(p.vertexReplica)),
-				"buffer":                        p.toBuffers[edgeName][partition].GetName()}).Add(float64(len(failedMessages)))
+				metrics.LabelPartitionName:      p.toBuffers[edgeName][partition].GetName()}).Add(float64(len(failedMessages)))
 			return false, nil
 		}
 		return true, nil
@@ -256,25 +256,25 @@ func (p *processAndForward) writeToBuffer(ctx context.Context, edgeName string,
 		metrics.LabelVertex:             p.vertexName,
 		metrics.LabelPipeline:           p.pipelineName,
 		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(p.vertexReplica)),
-		"buffer":                        p.toBuffers[edgeName][partition].GetName()}).Add(float64(len(resultMessages) - writeCount))
+		metrics.LabelPartitionName:      p.toBuffers[edgeName][partition].GetName()}).Add(float64(len(resultMessages) - writeCount))
 
 	dropBytesCount.With(map[string]string{
 		metrics.LabelVertex:             p.vertexName,
 		metrics.LabelPipeline:           p.pipelineName,
 		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(p.vertexReplica)),
-		"buffer":                        p.toBuffers[edgeName][partition].GetName()}).Add(dropBytes)
+		metrics.LabelPartitionName:      p.toBuffers[edgeName][partition].GetName()}).Add(dropBytes)
 
 	writeMessagesCount.With(map[string]string{
 		metrics.LabelVertex:             p.vertexName,
 		metrics.LabelPipeline:           p.pipelineName,
 		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(p.vertexReplica)),
-		"buffer":                        p.toBuffers[edgeName][partition].GetName()}).Add(float64(writeCount))
+		metrics.LabelPartitionName:      p.toBuffers[edgeName][partition].GetName()}).Add(float64(writeCount))
 
 	writeBytesCount.With(map[string]string{
 		metrics.LabelVertex:             p.vertexName,
 		metrics.LabelPipeline:           p.pipelineName,
 		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(p.vertexReplica)),
-		"buffer":                        p.toBuffers[edgeName][partition].GetName()}).Add(writeBytes)
+		metrics.LabelPartitionName:      p.toBuffers[edgeName][partition].GetName()}).Add(writeBytes)
 	return offsets, ctxClosedErr
 }
 


Test patch:
diff --git a/pkg/forward/forward_test.go b/pkg/forward/forward_test.go
index ef0f1d87..58a68eb6 100644
--- a/pkg/forward/forward_test.go
+++ b/pkg/forward/forward_test.go
@@ -1470,7 +1470,7 @@ func validateMetrics(t *testing.T, batchSize int64) {
 		# TYPE forwarder_read_total counter
 		`
 	expected := `
-		forwarder_read_total{buffer="from",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)) + `
+		forwarder_read_total{partition_name="from",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)) + `
 	`
 
 	err := testutil.CollectAndCompare(readMessagesCount, strings.NewReader(metadata+expected), "forwarder_read_total")
@@ -1485,13 +1485,13 @@ func validateMetrics(t *testing.T, batchSize int64) {
 	var writeExpected string
 	if batchSize > 1 {
 		writeExpected = `
-		forwarder_write_total{buffer="to1-1",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)/2) + `
-		forwarder_write_total{buffer="to1-2",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)/2) + `
+		forwarder_write_total{partition_name="to1-1",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)/2) + `
+		forwarder_write_total{partition_name="to1-2",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)/2) + `
 	`
 	} else {
 		writeExpected = `
-		forwarder_write_total{buffer="to1-1",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)) + `
-		forwarder_write_total{buffer="to1-2",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(0)) + `
+		forwarder_write_total{partition_name="to1-1",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)) + `
+		forwarder_write_total{partition_name="to1-2",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(0)) + `
 	`
 	}
 
@@ -1505,7 +1505,7 @@ func validateMetrics(t *testing.T, batchSize int64) {
 		# TYPE forwarder_ack_total counter
 		`
 	ackExpected := `
-		forwarder_ack_total{buffer="from",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%d", batchSize) + `
+		forwarder_ack_total{partition_name="from",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%d", batchSize) + `
 	`
 
 	err = testutil.CollectAndCompare(ackMessagesCount, strings.NewReader(ackMetadata+ackExpected), "forwarder_ack_total")
diff --git a/pkg/forward/shutdown_test.go b/pkg/forward/shutdown_test.go
index 70a310e1..4f4eb653 100644
--- a/pkg/forward/shutdown_test.go
+++ b/pkg/forward/shutdown_test.go
@@ -120,7 +120,7 @@ func TestInterStepDataForward(t *testing.T) {
 			f, err := NewInterStepDataForward(vertex, fromStep, toSteps, myShutdownTest{}, myShutdownTest{}, fetchWatermark, publishWatermark, WithReadBatchSize(batchSize), WithUDFStreaming(tt.streamEnabled))
 			assert.NoError(t, err)
 			stopped := f.Start()
-			// write some data such that the fromBuffer can be empty, that is toBuffer gets full
+			// write some data such that the fromBufferPartition can be empty, that is toBuffer gets full
 			_, errs := fromStep.Write(ctx, writeMessages[0:4*batchSize])
 			assert.Equal(t, make([]error, 4*batchSize), errs)
 
diff --git a/pkg/reduce/data_forward_test.go b/pkg/reduce/data_forward_test.go
index 83e6bc3c..23d88fdd 100644
--- a/pkg/reduce/data_forward_test.go
+++ b/pkg/reduce/data_forward_test.go
@@ -431,7 +431,7 @@ func TestReduceDataForward_IdleWM(t *testing.T) {
 		Watermark: 1679961600000,
 	}, otDecode)
 
-	// check the fromBuffer to see if we've acked the message
+	// check the fromBufferPartition to see if we've acked the message
 	for !fromBuffer.IsEmpty() {
 		select {
 		case <-ctx.Done():
@@ -1309,7 +1309,7 @@ func writeMessages(ctx context.Context, count int, key string, fromBuffer *simpl
 		messages := buildMessagesForReduce(count, key+strconv.Itoa(i), publishTime)
 		i++
 
-		// write the messages to fromBuffer, so that it will be available for consuming
+		// write the messages to fromBufferPartition, so that it will be available for consuming
 		offsets, _ := fromBuffer.Write(ctx, messages)
 		for _, offset := range offsets {
 			publish.PublishWatermark(wmb.Watermark(publishTime), offset, 0)
diff --git a/pkg/watermark/fetch/edge_fetcher_test.go b/pkg/watermark/fetch/edge_fetcher_test.go
index 4e072492..33a788f8 100644
--- a/pkg/watermark/fetch/edge_fetcher_test.go
+++ b/pkg/watermark/fetch/edge_fetcher_test.go
@@ -887,8 +887,8 @@ func TestFetcherWithSameOTBucket_InMem(t *testing.T) {
 	var fetcher = NewEdgeFetcher(ctx, "testBuffer", storeWatcher, processorManager, 1)
 
 	var heartBeatManagerMap = make(map[string]*heartBeatManager)
-	heartBeatManagerMap["p1"] = manageHeartbeat(ctx, "p1", hbStore)
-	heartBeatManagerMap["p2"] = manageHeartbeat(ctx, "p2", hbStore)
+	heartBeatManagerMap["p1"] = manageHeartbeat(ctx, "p1", hbStore, &wg)
+	heartBeatManagerMap["p2"] = manageHeartbeat(ctx, "p2", hbStore, &wg)
 
 	heartBeatManagerMap["p1"].start()
 	heartBeatManagerMap["p2"].start()
@@ -1037,14 +1037,14 @@ func TestFetcherWithSameOTBucket_InMem(t *testing.T) {
 	assert.Equal(t, int64(101), p1.GetOffsetTimelines()[0].GetHeadOffset())
 	heartBeatManagerMap["p1"].stop()
 	heartBeatManagerMap["p2"].stop()
-	wg.Wait()
 	cancel()
+	wg.Wait()
 }
 
 // end to end test for fetcher with same ot bucket
 func TestFetcherWithSameOTBucketWithSinglePartition(t *testing.T) {
 	var (
-		keyspace         = "fetcherTest"
+		keyspace         = "fetcherTestSinglePartition"
 		epoch      int64 = 1651161600000
 		testOffset int64 = 100
 		wg         sync.WaitGroup
@@ -1141,8 +1141,8 @@ func TestFetcherWithSameOTBucketWithSinglePartition(t *testing.T) {
 	fetcher := NewEdgeFetcher(ctx, "testBuffer", storeWatcher, processorManager, 1)
 
 	var heartBeatManagerMap = make(map[string]*heartBeatManager)
-	heartBeatManagerMap["p1"] = manageHeartbeat(ctx, "p1", hbStore)
-	heartBeatManagerMap["p2"] = manageHeartbeat(ctx, "p2", hbStore)
+	heartBeatManagerMap["p1"] = manageHeartbeat(ctx, "p1", hbStore, &wg)
+	heartBeatManagerMap["p2"] = manageHeartbeat(ctx, "p2", hbStore, &wg)
 
 	// start the heartbeats for p1 and p2
 	heartBeatManagerMap["p1"].start()
@@ -1188,7 +1188,6 @@ func TestFetcherWithSameOTBucketWithSinglePartition(t *testing.T) {
 		}
 	}
 
-	allProcessors = processorManager.GetAllProcessors()
 	assert.Equal(t, 2, len(allProcessors))
 	assert.True(t, allProcessors["p1"].IsDeleted())
 	assert.True(t, allProcessors["p2"].IsActive())
@@ -1242,14 +1241,14 @@ func TestFetcherWithSameOTBucketWithSinglePartition(t *testing.T) {
 	heartBeatManagerMap["p1"].resume()
 
 	// "p1" becomes inactive after stopping the heartbeat
-	for !allProcessors["p1"].IsInactive() {
+	for allProcessors["p1"].IsActive() {
 		select {
 		case <-ctx.Done():
 			if ctx.Err() == context.DeadlineExceeded {
 				t.Fatalf("expected p1 to be inactive: %s", ctx.Err())
 			}
 		default:
-			time.Sleep(100 * time.Millisecond)
+			time.Sleep(1 * time.Millisecond)
 			allProcessors = processorManager.GetAllProcessors()
 		}
 	}
@@ -1295,7 +1294,7 @@ func TestFetcherWithSameOTBucketWithSinglePartition(t *testing.T) {
 				t.Fatalf("expected p1 has the offset timeline [IDLE 1651161660600:106] -> [1651161660500:103] -> [-1:-1] -> [-1:-1]..., got %s: %s", allProcessors["p1"].GetOffsetTimelines()[0].Dump(), ctx.Err())
 			}
 		default:
-			time.Sleep(100 * time.Millisecond)
+			time.Sleep(1 * time.Millisecond)
 			allProcessors = processorManager.GetAllProcessors()
 		}
 	}
@@ -1320,8 +1319,8 @@ func TestFetcherWithSameOTBucketWithSinglePartition(t *testing.T) {
 	}
 	heartBeatManagerMap["p1"].stop()
 	heartBeatManagerMap["p2"].stop()
-	wg.Wait()
 	cancel()
+	wg.Wait()
 }
 
 // end to end test for fetcher with same ot bucket
@@ -1457,8 +1456,8 @@ func TestFetcherWithSameOTBucketWithMultiplePartition(t *testing.T) {
 	fetcher := NewEdgeFetcher(ctx, "testBuffer", storeWatcher, processorManager, 3)
 
 	var heartBeatManagerMap = make(map[string]*heartBeatManager)
-	heartBeatManagerMap["p1"] = manageHeartbeat(ctx, "p1", hbStore)
-	heartBeatManagerMap["p2"] = manageHeartbeat(ctx, "p2", hbStore)
+	heartBeatManagerMap["p1"] = manageHeartbeat(ctx, "p1", hbStore, &wg)
+	heartBeatManagerMap["p2"] = manageHeartbeat(ctx, "p2", hbStore, &wg)
 
 	// start the heartbeats for p1 and p2
 	heartBeatManagerMap["p1"].start()
@@ -1572,7 +1571,7 @@ func TestFetcherWithSameOTBucketWithMultiplePartition(t *testing.T) {
 
 	heartBeatManagerMap["p1"].resume()
 	// "p1" is inactive since we resume the heartbeat
-	for !allProcessors["p1"].IsInactive() {
+	for allProcessors["p1"].IsActive() {
 		select {
 		case <-ctx.Done():
 			if ctx.Err() == context.DeadlineExceeded {
@@ -1678,8 +1677,8 @@ func TestFetcherWithSameOTBucketWithMultiplePartition(t *testing.T) {
 	}
 	heartBeatManagerMap["p1"].stop()
 	heartBeatManagerMap["p2"].stop()
-	wg.Wait()
 	cancel()
+	wg.Wait()
 }
 
 type heartBeatManager struct {
@@ -1698,11 +1697,13 @@ func (h *heartBeatManager) resume() {
 	h.heartBeatCh <- 2
 }
 
-func manageHeartbeat(ctx context.Context, entityName string, hbStore store.WatermarkKVStorer) *heartBeatManager {
+func manageHeartbeat(ctx context.Context, entityName string, hbStore store.WatermarkKVStorer, wg *sync.WaitGroup) *heartBeatManager {
 	hbManager := &heartBeatManager{
 		heartBeatCh: make(chan int),
 	}
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		var start bool
 		for {
 			select {
diff --git a/test/e2e-suite-1/functional_test.go b/test/e2e-suite-1/functional_test.go
index d2a2eb41..cf1c13b3 100644
--- a/test/e2e-suite-1/functional_test.go
+++ b/test/e2e-suite-1/functional_test.go
@@ -333,7 +333,7 @@ func (s *FunctionalSuite) TestWatermarkEnabled() {
 	}()
 	buffers, err := client.ListPipelineBuffers(context.Background(), pipelineName)
 	assert.NoError(s.T(), err)
-	assert.Equal(s.T(), 5, len(buffers))
+	assert.Equal(s.T(), 8, len(buffers))
 	bufferInfo, err := client.GetPipelineBuffer(context.Background(), pipelineName, dfv1.GenerateBufferName(Namespace, pipelineName, "cat1", 0))
 	assert.NoError(s.T(), err)
 	assert.Equal(s.T(), pipelineName, *bufferInfo.Pipeline)
diff --git a/test/e2e-suite-1/testdata/even-odd.yaml b/test/e2e-suite-1/testdata/even-odd.yaml
index b676495d..83c4aee0 100644
--- a/test/e2e-suite-1/testdata/even-odd.yaml
+++ b/test/e2e-suite-1/testdata/even-odd.yaml
@@ -8,11 +8,13 @@ spec:
       source:
         http: {}
     - name: even-or-odd
+      partitions: 2
       udf:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/even_odd
           image: quay.io/numaio/numaflow-go/map-even-odd
     - name: even-sink
+      partitions: 2
       sink:
         udsink:
           container:
diff --git a/test/e2e-suite-1/testdata/time-extraction-filter.yaml b/test/e2e-suite-1/testdata/time-extraction-filter.yaml
index f8f29f08..d85a1698 100644
--- a/test/e2e-suite-1/testdata/time-extraction-filter.yaml
+++ b/test/e2e-suite-1/testdata/time-extraction-filter.yaml
@@ -14,6 +14,7 @@ spec:
               filterExpr: int(json(payload).id) < 100
               eventTimeExpr: json(payload).time
     - name: out
+      partitions: 2
       scale:
         min: 1
       sink:
diff --git a/test/e2e-suite-1/testdata/watermark.yaml b/test/e2e-suite-1/testdata/watermark.yaml
index 77769904..5ca0e7d1 100644
--- a/test/e2e-suite-1/testdata/watermark.yaml
+++ b/test/e2e-suite-1/testdata/watermark.yaml
@@ -17,6 +17,7 @@ spec:
           - name: NUMAFLOW_DEBUG
             value: "true" # DO NOT forget the double quotes!!!
     - name: cat1
+      partitions: 2
       udf:
         builtin:
           name: cat # A builtin UDF which simply cats the message
@@ -33,6 +34,7 @@ spec:
           - name: NUMAFLOW_DEBUG
             value: "true" # DO NOT forget the double quotes!!!
     - name: cat3
+      partitions: 3
       udf:
         builtin:
           name: cat # A builtin UDF which simply cats the message
diff --git a/test/e2e-suite-2/functional_test.go b/test/e2e-suite-2/functional_test.go
index be82b63a..17ee32d1 100644
--- a/test/e2e-suite-2/functional_test.go
+++ b/test/e2e-suite-2/functional_test.go
@@ -56,7 +56,7 @@ func (s *FunctionalSuite) TestDropOnFull() {
 	time.Sleep(time.Second * 5)
 	w.SendMessageTo(pipelineName, "in", NewHttpPostRequest().WithBody([]byte("2")))
 
-	expectedDropMetric := `forwarder_drop_total{buffer="numaflow-system-drop-on-full-sink-0",pipeline="drop-on-full",vertex="in"} 1`
+	expectedDropMetric := `forwarder_drop_total{partition_name="numaflow-system-drop-on-full-sink-0",pipeline="drop-on-full",vertex="in"} 1`
 	// wait for the drop metric to be updated, time out after 10s.
 	timeoutChan := time.After(time.Second * 10)
 	ticker := time.NewTicker(time.Second * 2)
diff --git a/test/e2e-suite-2/testdata/drop-on-full.yaml b/test/e2e-suite-2/testdata/drop-on-full.yaml
index 9c68037c..9ba7df57 100644
--- a/test/e2e-suite-2/testdata/drop-on-full.yaml
+++ b/test/e2e-suite-2/testdata/drop-on-full.yaml
@@ -15,6 +15,7 @@ spec:
       source:
         http: {}
     - name: sink
+      partitions: 2
       sink:
         log: {}
   edges:
diff --git a/test/reduce-e2e/testdata/complex-sliding-window-pipeline.yaml b/test/reduce-e2e/testdata/complex-sliding-window-pipeline.yaml
index 1118ccbe..78bcd5c6 100644
--- a/test/reduce-e2e/testdata/complex-sliding-window-pipeline.yaml
+++ b/test/reduce-e2e/testdata/complex-sliding-window-pipeline.yaml
@@ -19,7 +19,7 @@ spec:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/even_odd
           image: quay.io/numaio/numaflow-go/map-even-odd
     - name: keyed-fixed-sum
-      partitions: 1
+      partitions: 2
       udf:
         container:
           # Compute the sum, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/sum
@@ -57,6 +57,7 @@ spec:
           storage:
             emptyDir: {}
     - name: sink
+      partitions: 2
       scale:
         min: 1
       sink:
diff --git a/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml b/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml
index 414c3e9e..b826ae03 100644
--- a/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml
+++ b/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml
@@ -17,7 +17,7 @@ spec:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/even_odd
           image: quay.io/numaio/numaflow-go/map-even-odd
     - name: compute-sum
-      partitions: 1
+      partitions: 2
       udf:
         container:
           # Compute the sum, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/sum
diff --git a/test/sdks-e2e/testdata/flatmap-stream.yaml b/test/sdks-e2e/testdata/flatmap-stream.yaml
index 39a59f18..47162b3d 100644
--- a/test/sdks-e2e/testdata/flatmap-stream.yaml
+++ b/test/sdks-e2e/testdata/flatmap-stream.yaml
@@ -8,6 +8,7 @@ spec:
       source:
         http: {}
     - name: go-split
+      partitions: 3
       metadata:
         annotations:
           numaflow.numaproj.io/map-stream: "true"
@@ -36,6 +37,7 @@ spec:
             # https://github.com/numaproj/numaflow-go/tree/main/pkg/sink/examples/log
             image: quay.io/numaio/numaflow-go/sink-log:latest
     - name: python-split
+      partitions: 3
       metadata:
         annotations:
           numaflow.numaproj.io/map-stream: "true"
@@ -62,6 +64,7 @@ spec:
             # https://github.com/numaproj/numaflow-python/tree/main/examples/sink/log
             image: quay.io/numaio/numaflow-python/sink-log:latest
     - name: java-split
+      partitions: 3
       metadata:
         annotations:
           numaflow.numaproj.io/map-stream: "true"
diff --git a/test/sdks-e2e/testdata/transformer/event-time-filter-go.yaml b/test/sdks-e2e/testdata/transformer/event-time-filter-go.yaml
index 11ccaa79..4f3385d3 100644
--- a/test/sdks-e2e/testdata/transformer/event-time-filter-go.yaml
+++ b/test/sdks-e2e/testdata/transformer/event-time-filter-go.yaml
@@ -12,6 +12,7 @@ spec:
             # Filter messages based on event time, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/event_time_filter
             image: quay.io/numaio/numaflow-go/mapt-event-time-filter:latest
     - name: sink-within-2022
+      partitions: 2
       scale:
         min: 1
       sink:
diff --git a/test/sdks-e2e/testdata/transformer/event-time-filter-java.yaml b/test/sdks-e2e/testdata/transformer/event-time-filter-java.yaml
index 760574af..be0f1d25 100644
--- a/test/sdks-e2e/testdata/transformer/event-time-filter-java.yaml
+++ b/test/sdks-e2e/testdata/transformer/event-time-filter-java.yaml
@@ -22,6 +22,7 @@ spec:
       sink:
         log: {}
     - name: sink-all
+      partitions: 2
       scale:
         min: 1
       sink:
diff --git a/test/sdks-e2e/testdata/transformer/event-time-filter-python.yaml b/test/sdks-e2e/testdata/transformer/event-time-filter-python.yaml
index 70999d3c..42b30ff6 100644
--- a/test/sdks-e2e/testdata/transformer/event-time-filter-python.yaml
+++ b/test/sdks-e2e/testdata/transformer/event-time-filter-python.yaml
@@ -19,6 +19,7 @@ spec:
       sink:
         log: {}
     - name: sink-after-2022
+      partitions: 2
       scale:
         min: 1
       sink: