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

Code before:
        self.merge_barrier_align_duration.clone()
    }

    pub fn flush_buffered_watermarks(&mut self) {
        self.buffered_watermarks
            .values_mut()
            .for_each(|buffers| buffers.clear());
    }

    pub fn upstream_input_ids(&self) -> impl Iterator<Item = InputId> + '_ {
        self.blocked
            .iter()
            .map(|s| s.id())
            .chain(self.active.iter().map(|s| s.get_ref().unwrap().id()))
    }

    pub fn is_empty(&self) -> bool {
        self.blocked.is_empty() && self.active.is_empty()
    }
}

// Explanation of why we need `DispatchBarrierBuffer`:
//
// When we need to create or replace an upstream fragment for the current fragment, the `Merge` operator must
// add some new upstream actor inputs. However, the `Merge` operator may still have old upstreams. We must wait
// for these old upstreams to completely process their barriers and align before we can safely update the
// `upstream-input-set`.
//
// Meanwhile, the creation of a new upstream actor can only succeed after the channel to the downstream `Merge`
// operator has been established. This creates a potential dependency chain: [new_actor_creation ->
// downstream_merge_update -> old_actor_processing]
//
// To address this, we split the application of a barrier's `Mutation` into two steps:
// 1. Parse the `Mutation`. If there is an addition on the upstream-set, establish a channel with the upstream
//    and cache it.
// 2. When the upstream barrier actually arrives, apply the cached upstream changes to the upstream-set
//
// Additionally, since receiving a barrier from current upstream input and from the `barrier_rx` are
// asynchronous, we cannot determine which will arrive first. Therefore, when a barrier is received from an
// upstream: if a cached mutation is present, we apply it. Otherwise, we must fetch a new barrier from
// `barrier_rx`.
pub(crate) struct DispatchBarrierBuffer {
    buffer: VecDeque<(Barrier, Option<Vec<BoxedActorInput>>)>,
    barrier_rx: mpsc::UnboundedReceiver<Barrier>,
    recv_state: BarrierReceiverState,
    curr_upstream_fragment_id: FragmentId,
    actor_id: ActorId,
    // read-only context for building new inputs
    build_input_ctx: Arc<BuildInputContext>,
}


Review comment (suggestion):
make it a streaming developer config? I feel we may need to tune the value if needs better sensitivity. 

Code after:
        local_barrier_manager: LocalBarrierManager,
        metrics: Arc<StreamingMetrics>,
        fragment_id: FragmentId,
    ) -> Self {
        Self {
            buffer: VecDeque::new(),
            barrier_rx,
            recv_state: BarrierReceiverState::ReceivingBarrier,
            curr_upstream_fragment_id,
            actor_id,
            build_input_ctx: Arc::new(BuildInputContext {
                actor_id,
                local_barrier_manager,
                metrics,
                fragment_id,
            }),
        }
    }

    pub async fn await_next_message(
        &mut self,
        stream: &mut (impl Stream<Item = StreamExecutorResult<DispatcherMessage>> + Unpin),
        metrics: &ActorInputMetrics,
    ) -> StreamExecutorResult<DispatcherMessage> {
        let mut start_time = Instant::now();
        let interval_duration = Duration::from_secs(15);
        let mut interval =
            tokio::time::interval_at(start_time + interval_duration, interval_duration);

        loop {
            tokio::select! {
                biased;
                msg = stream.try_next() => {
                    metrics
                        .actor_input_buffer_blocking_duration_ns
                        .inc_by(start_time.elapsed().as_nanos() as u64);
                    return msg?.ok_or_else(
                        || StreamExecutorError::channel_closed("upstream executor closed unexpectedly")
                    );
                }

                e = self.continuously_fetch_barrier_rx() => {
                    return Err(e);
                }

                _ = interval.tick() => {
                    start_time = Instant::now();
                    metrics.actor_input_buffer_blocking_duration_ns.inc_by(interval_duration.as_nanos() as u64);
                    continue;
                }
            }