Repository: risingwavelabs/risingwave
File: src/stream/src/executor/dispatch.rs (Rust)

Code before:
                    .for_each_concurrent(limit, |dispatcher| async {
                        let metrics = &dispatcher.actor_output_buffer_blocking_duration_ns;
                        let dispatcher_output = &mut dispatcher.dispatcher;
                        let fut = dispatcher_output.dispatch_watermark(watermark.clone());
                        await_with_metrics(fut, metrics).await;
                    })
                    .await;
            }
            MessageBatch::Chunk(chunk) => {
                futures::stream::iter(self.dispatchers.iter_mut())
                    .for_each_concurrent(limit, |dispatcher| async {
                        let metrics = &dispatcher.actor_output_buffer_blocking_duration_ns;
                        let dispatcher_output = &mut dispatcher.dispatcher;
                        let fut = dispatcher_output.dispatch_data(chunk.clone());
                        await_with_metrics(fut, metrics).await;
                    })
                    .await;

                self.metrics
                    .actor_out_record_cnt
                    .inc_by(chunk.cardinality() as _);
            }
        }
        self.dispatchers.retain(|dispatcher| match &**dispatcher {
            DispatcherImpl::Failed(dispatcher_id, e) => {
                warn!(%dispatcher_id, e = %e.as_report(), actor_id = %self.actor_id, "dispatch fail");
                false
            }
            _ => true,
        });
        Ok(())
    }

    /// Add new dispatchers to the executor. Will check whether their ids are unique.
    async fn add_dispatchers<'a>(
        &mut self,
        new_dispatchers: impl IntoIterator<Item = &'a PbDispatcher>,
    ) -> StreamResult<()> {
        for dispatcher in new_dispatchers {
            let outputs = self
                .collect_outputs(&dispatcher.downstream_actor_id)
                .await?;
            let dispatcher = DispatcherImpl::new(outputs, dispatcher)?;
            let dispatcher = self.metrics.monitor_dispatcher(dispatcher);
            self.dispatchers.push(dispatcher);
        }

        assert!(
            self.dispatchers
                .iter()
                .map(|d| d.dispatcher_id())

Review comment (bug):
```suggestion
                warn!(%dispatcher_id, error = %e.as_report(), actor_id = %self.actor_id, "dispatch fail");
```

Code after:
                        await_with_metrics(fut, metrics).await;
                    })
                    .await;
                self.post_mutate_dispatchers(&mutation)?;
            }
            MessageBatch::Watermark(watermark) => {
                futures::stream::iter(self.dispatchers.iter_mut())
                    .for_each_concurrent(limit, |dispatcher| async {
                        let metrics = &dispatcher.actor_output_buffer_blocking_duration_ns;
                        let dispatcher_output = &mut dispatcher.dispatcher;
                        let fut = dispatcher_output.dispatch_watermark(watermark.clone());
                        await_with_metrics(fut, metrics).await;
                    })
                    .await;
            }
            MessageBatch::Chunk(chunk) => {
                futures::stream::iter(self.dispatchers.iter_mut())
                    .for_each_concurrent(limit, |dispatcher| async {
                        let metrics = &dispatcher.actor_output_buffer_blocking_duration_ns;
                        let dispatcher_output = &mut dispatcher.dispatcher;
                        let fut = dispatcher_output.dispatch_data(chunk.clone());
                        await_with_metrics(fut, metrics).await;
                    })
                    .await;

                self.metrics
                    .actor_out_record_cnt
                    .inc_by(chunk.cardinality() as _);
            }
        }
        self.dispatchers.retain(|dispatcher| match &**dispatcher {
            DispatcherImpl::Failed(dispatcher_id, e) => {
                warn!(%dispatcher_id, err = %e.as_report(), actor_id = %self.actor_id, "dispatch fail");
                false
            }
            _ => true,
        });
        Ok(())
    }

    /// Add new dispatchers to the executor. Will check whether their ids are unique.
    async fn add_dispatchers<'a>(
        &mut self,
        new_dispatchers: impl IntoIterator<Item = &'a PbDispatcher>,
    ) -> StreamResult<()> {
        for dispatcher in new_dispatchers {
            let outputs = self
                .collect_outputs(&dispatcher.downstream_actor_id)
                .await?;
            let dispatcher = DispatcherImpl::new(outputs, dispatcher)?;
            let dispatcher = self.metrics.monitor_dispatcher(dispatcher);