fix(ci): fix executor metrics in release mode (#3087)

Fix executor metrics in release mode

src/stream/src/executor/debug.rs

@@ -94,7 +94,7 @@ impl DebugExecutor {
         stream: impl MessageStream + 'static,
     ) -> impl MessageStream + 'static {
         // Metrics
-        let stream = trace::metrics(extra.actor_id, extra.metrics, stream);
+        let stream = trace::metrics(extra.actor_id, extra.executor_id, extra.metrics, stream);
 
         // Epoch check
         let stream = epoch_check::epoch_check(info, stream);

src/stream/src/executor/debug/trace.rs

@@ -66,17 +66,22 @@ pub async fn trace(
 
 /// Streams wrapped by `metrics` will update actor metrics.
 #[try_stream(ok = Message, error = StreamExecutorError)]
-pub async fn metrics(actor_id: ActorId, metrics: Arc<StreamingMetrics>, input: impl MessageStream) {
+pub async fn metrics(
+    actor_id: ActorId,
+    executor_id: u64,
+    metrics: Arc<StreamingMetrics>,
+    input: impl MessageStream,
+) {
     let actor_id_string = actor_id.to_string();
-
+    let executor_id_string = executor_id.to_string();
     pin_mut!(input);
 
     while let Some(message) = input.next().await.transpose()? {
         if let Message::Chunk(chunk) = &message {
             if chunk.cardinality() > 0 {
                 metrics
                     .executor_row_count
-                    .with_label_values(&[&actor_id_string])
+                    .with_label_values(&[&actor_id_string, &executor_id_string])
                     .inc_by(chunk.cardinality() as u64);
             }
         }