feat: set expire when create flow

GreptimeTeam · May 16, 2024 · 1f4a901 · 1f4a901
1 parent f48c831
commit 1f4a901
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 1 deletion.
diff --git a/src/flow/src/adapter/worker.rs b/src/flow/src/adapter/worker.rs
@@ -208,7 +208,6 @@ impl<'s> Worker<'s> {
         create_if_not_exist: bool,
         err_collector: ErrCollector,
     ) -> Result<Option<FlowId>, Error> {
-        let _ = expire_when;
         let already_exist = self.task_states.contains_key(&flow_id);
         match (already_exist, create_if_not_exist) {
             (true, true) => return Ok(None),
@@ -220,6 +219,7 @@ impl<'s> Worker<'s> {
             err_collector,
             ..Default::default()
         };
+        cur_task_state.state.set_expire_after(expire_when);
 
         {
             let mut ctx = cur_task_state.new_ctx(sink_id);

diff --git a/src/flow/src/compute/state.rs b/src/flow/src/compute/state.rs
@@ -102,6 +102,10 @@ impl DataflowState {
         self.err_collector.clone()
     }
 
+    pub fn set_expire_after(&mut self, after: Option<repr::Duration>) {
+        self.expire_after = after;
+    }
+
     pub fn expire_after(&self) -> Option<Timestamp> {
         self.expire_after
     }