Repository: numaproj/numaflow (go)

Issue:
Raptor generates wrong processing rate when multi-partition isb enabled
**Describe the bug**
Expecting source vertex and sink vertex to have close processing rate but not. The issue emerges when I specify partitions for the sink vertex.

![Screenshot 2023-09-08 at 1 03 53 PM](https://github.com/numaproj/numaflow/assets/13165977/adb1f5c3-ab78-40e9-bed6-c08ddb2f66b4)

**To Reproduce**
Use the following template to deploy the pipeline.

```
apiVersion: numaflow.numaproj.io/v1alpha1
kind: Pipeline
metadata:
  name: multi-partition-pipeline
spec:
  templates:
    daemon:
      containerTemplate:
        env:
          - name: NUMAFLOW_DEBUG
            value: "true" # DO NOT forget the double quotes!!!
  vertices:
    - name: in
      scale:
        min: 5
      source:
        # A self data generating source
        generator:
          msgSize: 1
          rpu: 1
          duration: 1s
    - name: out
      partitions: 5
      scale:
        min: 5
      sink:
        # A simple log printing sink
        log: {}
  edges:
    - from: in
      to: out

```

Pull request description:
fix: message count read in forwarder
Fixes #1026 

`readMessagesCount` in `forward.go` was adding both data messages as well as higher control messages(for low throughput pipelines with multiple partitions) leading to a higher processing rate calculated and shown for the sink vertex

count of messages read now is just based on data messages 
![image](https://github.com/numaproj/numaflow/assets/49195734/5dd3bb73-55f1-4b11-a11b-f20eb077df13)


Patch:
diff --git a/docs/operations/example-dashboard-template.json b/docs/operations/example-dashboard-template.json
index b1cca72b..5dead77e 100644
--- a/docs/operations/example-dashboard-template.json
+++ b/docs/operations/example-dashboard-template.json
@@ -125,7 +125,7 @@
             "uid": "eF7wTnc4k"
           },
           "editorMode": "builder",
-          "expr": "rate(forwarder_read_total{namespace=\"$namespace\", pipeline=\"$pipeline\", vertex=\"$vertex\"}[$__rate_interval])",
+          "expr": "rate(forwarder_data_read{namespace=\"$namespace\", pipeline=\"$pipeline\", vertex=\"$vertex\"}[$__rate_interval])",
           "legendFormat": "__auto",
           "range": true,
           "refId": "A"
diff --git a/docs/operations/metrics/metrics.md b/docs/operations/metrics/metrics.md
index 1afc2bf5..03e7f196 100644
--- a/docs/operations/metrics/metrics.md
+++ b/docs/operations/metrics/metrics.md
@@ -21,14 +21,14 @@ These metrics can be used to determine throughput of your pipeline.
 | `source_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 Source Vertex                                      |
 | `source_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 Source Vertex due to a full Inter-Step Buffer Partition |
 | `source_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 Source Vertex due to a full Inter-Step Buffer Partition    |
-| `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_data_read`                 | 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_data_read`         | 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                         |
diff --git a/pkg/daemon/server/service/rater/rater.go b/pkg/daemon/server/service/rater/rater.go
index 68d7ee81..d5da3f69 100644
--- a/pkg/daemon/server/service/rater/rater.go
+++ b/pkg/daemon/server/service/rater/rater.go
@@ -218,14 +218,14 @@ func sleep(ctx context.Context, duration time.Duration) {
 // since a pod can read from multiple partitions, we will return a map of partition to read count.
 func (r *Rater) getPodReadCounts(vertexName, vertexType, podName string) *PodReadCount {
 	metricNames := map[string]string{
-		keyVertexTypeReduce: "reduce_isb_reader_read_total",
+		keyVertexTypeReduce: "reduce_isb_reader_data_read",
 		keyVertexTypeSource: "source_forwarder_read_total",
-		keyVertexTypeSink:   "sink_forwarder_read_total",
+		keyVertexTypeSink:   "sink_forwarder_data_read",
 	}
 
 	readTotalMetricName, ok := metricNames[vertexType]
 	if !ok {
-		readTotalMetricName = "forwarder_read_total"
+		readTotalMetricName = "forwarder_data_read"
 	}
 
 	// scrape the read total metric from pod metric port
@@ -243,6 +243,7 @@ func (r *Rater) getPodReadCounts(vertexName, vertexType, podName string) *PodRea
 		r.log.Errorf("failed parsing to prometheus metric families, %v", err.Error())
 		return nil
 	}
+
 	if value, ok := result[readTotalMetricName]; ok && value != nil && len(value.GetMetric()) > 0 {
 		metricsList := value.GetMetric()
 		partitionReadCount := make(map[string]float64)
diff --git a/pkg/forward/forward.go b/pkg/forward/forward.go
index 8f192d04..ef2bc47d 100644
--- a/pkg/forward/forward.go
+++ b/pkg/forward/forward.go
@@ -199,7 +199,6 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		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, 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.
@@ -242,6 +241,8 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 			dataMessages = append(dataMessages, m)
 		}
 	}
+	readMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(dataMessages)))
+	totalMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(readMessages)))
 
 	// fetch watermark if available
 	// TODO: make it async (concurrent and wait later)
diff --git a/pkg/forward/metrics.go b/pkg/forward/metrics.go
index 09f42227..75d11be0 100644
--- a/pkg/forward/metrics.go
+++ b/pkg/forward/metrics.go
@@ -23,13 +23,20 @@ import (
 	"github.com/numaproj/numaflow/pkg/metrics"
 )
 
-// readMessagesCount is used to indicate the number of messages read
-var readMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+// totalMessagesCount is used to indicate the number of total messages read
+var totalMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
-	Name:      "read_total",
+	Name:      "total_read",
 	Help:      "Total number of Messages Read",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
+// readMessagesCount is used to indicate the number of data messages read
+var readMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "forwarder",
+	Name:      "data_read",
+	Help:      "Total number of Data Messages Read",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
 // readBytesCount is to indicate the number of bytes read
 var readBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
diff --git a/pkg/reduce/data_forward.go b/pkg/reduce/data_forward.go
index 267e2acd..0d5f5217 100644
--- a/pkg/reduce/data_forward.go
+++ b/pkg/reduce/data_forward.go
@@ -243,12 +243,7 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 		}
 		return
 	}
-	readMessagesCount.With(map[string]string{
-		metrics.LabelVertex:             df.vertexName,
-		metrics.LabelPipeline:           df.pipelineName,
-		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
-		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
@@ -322,6 +317,18 @@ func (df *DataForward) Process(ctx context.Context, messages []*isb.ReadMessage)
 			ctrlMessages = append(ctrlMessages, message)
 		}
 	}
+	readMessagesCount.With(map[string]string{
+		metrics.LabelVertex:             df.vertexName,
+		metrics.LabelPipeline:           df.pipelineName,
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+		metrics.LabelPartitionName:      df.fromBufferPartition.GetName(),
+	}).Add(float64(len(dataMessages)))
+	totalMessagesCount.With(map[string]string{
+		metrics.LabelVertex:             df.vertexName,
+		metrics.LabelPipeline:           df.pipelineName,
+		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(df.vertexReplica)),
+		metrics.LabelPartitionName:      df.fromBufferPartition.GetName(),
+	}).Add(float64(len(messages)))
 
 	// write messages to windows based by PBQs.
 	successfullyWrittenMessages, err := df.writeMessagesToWindows(ctx, dataMessages)
diff --git a/pkg/reduce/metrics.go b/pkg/reduce/metrics.go
index 4c21a20d..b566b503 100644
--- a/pkg/reduce/metrics.go
+++ b/pkg/reduce/metrics.go
@@ -70,13 +70,20 @@ var ackMessageError = promauto.NewCounterVec(prometheus.CounterOpts{
 	Help:      "Total number of Acknowledged Errors",
 }, []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{
+// totalMessagesCount is used to indicate the number of total messages read
+var totalMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "reduce_isb_reader",
-	Name:      "read_total",
+	Name:      "total_read",
 	Help:      "Total number of Messages Read",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelVertexReplicaIndex, metrics.LabelPartitionName})
 
+// readMessagesCount is used to indicate the number of data messages read
+var readMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "reduce_isb_reader",
+	Name:      "data_read",
+	Help:      "Total number of Data Messages Read",
+}, []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",
diff --git a/pkg/sinks/forward/forward.go b/pkg/sinks/forward/forward.go
index fec810f4..cd8af191 100644
--- a/pkg/sinks/forward/forward.go
+++ b/pkg/sinks/forward/forward.go
@@ -173,7 +173,6 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 		df.opts.logger.Warnw("failed to read fromBufferPartition", zap.Error(err))
 		readMessagesError.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelPartitionName: df.fromBufferPartition.GetName()}).Inc()
 	}
-	readMessagesCount.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelPartitionName: df.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.
@@ -210,6 +209,8 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 			dataMessages = append(dataMessages, m)
 		}
 	}
+	readMessagesCount.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelPartitionName: df.fromBufferPartition.GetName()}).Add(float64(len(dataMessages)))
+	totalMessagesCount.With(map[string]string{metrics.LabelVertex: df.vertexName, metrics.LabelPipeline: df.pipelineName, metrics.LabelPartitionName: df.fromBufferPartition.GetName()}).Add(float64(len(readMessages)))
 
 	// fetch watermark if available
 	// TODO: make it async (concurrent and wait later)
diff --git a/pkg/sinks/forward/metrics.go b/pkg/sinks/forward/metrics.go
index 1814306b..2a7925e5 100644
--- a/pkg/sinks/forward/metrics.go
+++ b/pkg/sinks/forward/metrics.go
@@ -23,13 +23,20 @@ import (
 	"github.com/numaproj/numaflow/pkg/metrics"
 )
 
-// readMessagesCount is used to indicate the number of messages read
-var readMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+// totalMessagesCount is used to indicate the number of total messages read
+var totalMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "sink_forwarder",
-	Name:      "read_total",
+	Name:      "total_read",
 	Help:      "Total number of Messages Read",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
+// readMessagesCount is used to indicate the number of data messages read
+var readMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "sink_forwarder",
+	Name:      "data_read",
+	Help:      "Total number of Data Messages Read",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
 // readBytesCount is to indicate the number of bytes read
 var readBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "sink_forwarder",
diff --git a/pkg/sources/forward/data_forward.go b/pkg/sources/forward/data_forward.go
index 3dfa0c59..885a5eeb 100644
--- a/pkg/sources/forward/data_forward.go
+++ b/pkg/sources/forward/data_forward.go
@@ -200,13 +200,13 @@ func (isdf *DataForward) forwardAChunk(ctx context.Context) {
 		isdf.opts.logger.Warnw("failed to read from source", zap.Error(err))
 		readMessagesError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Inc()
 	}
-	readMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.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.
 	if len(readMessages) == 0 {
 		return
 	}
+	readMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Add(float64(len(readMessages)))
 
 	// store the offsets of the messages we read from source
 	var readOffsets = make([]isb.Offset, len(readMessages))


Test patch:
diff --git a/pkg/daemon/server/service/rater/rater_test.go b/pkg/daemon/server/service/rater/rater_test.go
index fe115850..8520050c 100644
--- a/pkg/daemon/server/service/rater/rater_test.go
+++ b/pkg/daemon/server/service/rater/rater_test.go
@@ -46,11 +46,11 @@ func (m *raterMockHttpClient) Get(url string) (*http.Response, error) {
 		resp := &http.Response{
 			StatusCode: 200,
 			// the test uses an abstract vertex without specifying vertex type, meaning it's neither source nor reduce,
-			// hence the default forwarder metric name "forwarder_read_total" is used to retrieve the metric
+			// hence the default forwarder metric name "forwarder_data_read" is used to retrieve the metric
 			Body: io.NopCloser(bytes.NewReader([]byte(fmt.Sprintf(`
-# HELP forwarder_read_total Total number of Messages Read
-# TYPE forwarder_read_total counter
-forwarder_read_total{buffer="input",pipeline="simple-pipeline",vertex="input",partition_name="p-v-0"} %d
+# HELP forwarder_data_read Total number of Messages Read
+# TYPE forwarder_data_read counter
+forwarder_data_read{buffer="input",pipeline="simple-pipeline",vertex="input",partition_name="p-v-0"} %d
 `, m.podOneCount))))}
 		return resp, nil
 	} else if url == "https://p-v-1.p-v-headless.default.svc:2469/metrics" {
@@ -58,9 +58,9 @@ forwarder_read_total{buffer="input",pipeline="simple-pipeline",vertex="input",pa
 		resp := &http.Response{
 			StatusCode: 200,
 			Body: io.NopCloser(bytes.NewReader([]byte(fmt.Sprintf(`
-# HELP forwarder_read_total Total number of Messages Read
-# TYPE forwarder_read_total counter
-forwarder_read_total{buffer="input",pipeline="simple-pipeline",vertex="input", partition_name="p-v-1"} %d
+# HELP forwarder_data_read Total number of Messages Read
+# TYPE forwarder_data_read counter
+forwarder_data_read{buffer="input",pipeline="simple-pipeline",vertex="input", partition_name="p-v-1"} %d
 `, m.podTwoCount))))}
 		return resp, nil
 	} else {
diff --git a/pkg/forward/forward_test.go b/pkg/forward/forward_test.go
index 1cb977c2..e386c0d8 100644
--- a/pkg/forward/forward_test.go
+++ b/pkg/forward/forward_test.go
@@ -1585,14 +1585,14 @@ func (f myForwardApplyUDFErrTest) ApplyMapStream(_ context.Context, _ *isb.ReadM
 
 func validateMetrics(t *testing.T, batchSize int64) {
 	metadata := `
-		# HELP forwarder_read_total Total number of Messages Read
-		# TYPE forwarder_read_total counter
+		# HELP forwarder_data_read Total number of Data Messages Read
+		# TYPE forwarder_data_read counter
 		`
 	expected := `
-		forwarder_read_total{partition_name="from",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)) + `
+		forwarder_data_read{partition_name="from",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)) + `
 	`
 
-	err := testutil.CollectAndCompare(readMessagesCount, strings.NewReader(metadata+expected), "forwarder_read_total")
+	err := testutil.CollectAndCompare(readMessagesCount, strings.NewReader(metadata+expected), "forwarder_data_read")
 	if err != nil {
 		t.Errorf("unexpected collecting result:\n%s", err)
 	}
diff --git a/pkg/sinks/forward/forward_test.go b/pkg/sinks/forward/forward_test.go
index 2715546b..acb209f9 100644
--- a/pkg/sinks/forward/forward_test.go
+++ b/pkg/sinks/forward/forward_test.go
@@ -348,14 +348,14 @@ func TestWriteToBuffer(t *testing.T) {
 
 func validateMetrics(batchSize int64) (err error) {
 	metadata := `
-		# HELP sink_forwarder_read_total Total number of Messages Read
-		# TYPE sink_forwarder_read_total counter
+		# HELP sink_forwarder_data_read Total number of Data Messages Read
+		# TYPE sink_forwarder_data_read counter
 		`
 	expected := `
-		sink_forwarder_read_total{partition_name="from",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)) + `
+		sink_forwarder_data_read{partition_name="from",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)) + `
 	`
 
-	err = testutil.CollectAndCompare(readMessagesCount, strings.NewReader(metadata+expected), "sink_forwarder_read_total")
+	err = testutil.CollectAndCompare(readMessagesCount, strings.NewReader(metadata+expected), "sink_forwarder_data_read")
 	if err != nil {
 		return err
 	}