fix(streaming): unstable lifetime compiling bug in join state (#2704)

fix(streaming): potential lifetime bug in join state

Signed-off-by: Bugen Zhao <i@bugenzhao.com>

src/stream/src/executor/managed_state/join/mod.rs

@@ -177,7 +177,7 @@ impl<K: HashKey, S: StateStore> JoinHashMap<K, S> {
     /// Returns a mutable reference to the value of the key in the memory, if does not exist, look
     /// up in remote storage and return, if still not exist, return None.
     #[allow(dead_code)]
-    pub async fn get(&mut self, key: &K) -> Option<&HashValueType<S>> {
+    pub async fn get<'a>(&'a mut self, key: &K) -> Option<&'a HashValueType<S>> {
         let state = self.inner.get(key);
         // TODO: we should probably implement a entry function for `LruCache`
         match state {
@@ -212,10 +212,10 @@ impl<K: HashKey, S: StateStore> JoinHashMap<K, S> {
     /// Returns a mutable reference to the value of the key in the memory, if does not exist, look
     /// up in remote storage and return the [`JoinEntryState`] without cached state, if still not
     /// exist, return None.
-    pub async fn get_mut_without_cached(
-        &mut self,
+    pub async fn get_mut_without_cached<'a>(
+        &'a mut self,
         key: &K,
-    ) -> RwResult<Option<&mut HashValueType<S>>> {
+    ) -> RwResult<Option<&'a mut HashValueType<S>>> {
         let state = self.inner.get(key);
         // TODO: we should probably implement a entry function for `LruCache`
         match state {