add doc and gen proto

Signed-off-by: Bugen Zhao <i@bugenzhao.com>
risingwavelabs · Mar 16, 2023 · 262bf00 · 262bf00
1 parent 22cbb79
commit 262bf00
Show file tree

Hide file tree

Showing 3 changed files with 55 additions and 6 deletions.
diff --git a/dashboard/proto/gen/stream_plan.ts b/dashboard/proto/gen/stream_plan.ts
diff --git a/proto/stream_plan.proto b/proto/stream_plan.proto
@@ -131,9 +131,9 @@ message StreamSource {
   string source_name = 8;
 }
 
-message BarrierRecvNode {
-
-}
+// The executor only for receiving barrier from the meta service. It always resides in the leaves
+// of the streaming graph.
+message BarrierRecvNode {}
 
 message SourceNode {
   // The source node can contain either a stream source or nothing. So here we extract all

diff --git a/src/stream/src/executor/barrier_recv.rs b/src/stream/src/executor/barrier_recv.rs
@@ -22,11 +22,13 @@ use super::{
     StreamExecutorError,
 };
 
+/// The executor only for receiving barrier from the meta service. It always resides in the leaves
+/// of the streaming graph.
 pub struct BarrierRecvExecutor {
     _ctx: ActorContextRef,
-
     identity: String,
 
+    /// The barrier receiver registered in the local barrier manager.
     barrier_receiver: UnboundedReceiver<Barrier>,
 }
 
@@ -53,6 +55,8 @@ impl Executor for BarrierRecvExecutor {
         UnboundedReceiverStream::new(self.barrier_receiver)
             .map(|barrier| Ok(Message::Barrier(barrier)))
             .chain(futures::stream::once(async {
+                // We do not use the stream termination as the control message, and this line should
+                // never be reached in normal cases. So we just return an error here.
                 Err(StreamExecutorError::channel_closed("barrier receiver"))
             }))
             .boxed()