涉及文件： src/sage/stream/operators.py、src/sage/stream/datastream.py
问题描述：
JoinOperator 的状态缓冲完全由用户传入的 JoinFunction 实例变量管理，没有任何框架层的窗口约束。这在无限流场景中会导致：
- 如果 join 的两个流中有一侧数据速率较慢或某些 key 的数据稀少，对应 key 的等待缓冲会永久保留在内存中，直到进程 OOM 或重启。
- 没有 watermark 机制，无法区分"数据尚未到达"和"数据已经迟到"，也无法安全地触发窗口关闭和清理。
handle_stop_signal 的实现依赖 "Source" 字符串匹配来判断 stop signal 来源，这是脆弱的字符串耦合，当 source 名称不包含 "Source" 时会永远等待第二个 stop signal。
此外，JoinOperator 目前只支持 inner join 语义（双侧均有数据才输出），没有 left outer join 或窗口超时输出（如果某 key 在 N 秒内没有来自另一侧的数据则输出 null join 结果）。
需要明确并实现的内容：
- 框架层是否提供 tumbling/sliding/session window 支持？如果是，
JoinOperator 的窗口边界应由框架维护，而不是由 user function 自行管理缓冲。
- 如果框架层暂不提供窗口支持，至少应在
JoinOperator 上增加 max_buffer_size_per_key（超出后驱逐最旧条目）和 join_timeout_sec（超时触发 null join 输出或丢弃）两个保护参数，防止无界积累。
handle_stop_signal 的两侧 stop signal 检测逻辑应基于注册的输入流数量，而不是字符串匹配。
任务目标： 让 JoinOperator 在无限流场景中具备可控的内存占用，框架层承担窗口约束或至少提供 buffer bound 保护，使得生产环境中 Join pipeline 不会因数据倾斜或流速不均而 OOM。
完成标准： JoinOperator 在无限流下不会无界积累缓冲，具备 max_buffer_size_per_key 或窗口超时保护，handle_stop_signal 不依赖字符串匹配，而是基于注册的输入流数量判断终止条件。
父 Issue: #1484
涉及文件：
src/sage/stream/operators.py、src/sage/stream/datastream.py问题描述：
JoinOperator的状态缓冲完全由用户传入的JoinFunction实例变量管理，没有任何框架层的窗口约束。这在无限流场景中会导致：handle_stop_signal的实现依赖"Source"字符串匹配来判断 stop signal 来源，这是脆弱的字符串耦合，当 source 名称不包含 "Source" 时会永远等待第二个 stop signal。此外，
JoinOperator目前只支持 inner join 语义（双侧均有数据才输出），没有 left outer join 或窗口超时输出（如果某 key 在 N 秒内没有来自另一侧的数据则输出 null join 结果）。需要明确并实现的内容：
JoinOperator的窗口边界应由框架维护，而不是由 user function 自行管理缓冲。JoinOperator上增加max_buffer_size_per_key（超出后驱逐最旧条目）和join_timeout_sec（超时触发 null join 输出或丢弃）两个保护参数，防止无界积累。handle_stop_signal的两侧 stop signal 检测逻辑应基于注册的输入流数量，而不是字符串匹配。任务目标： 让
JoinOperator在无限流场景中具备可控的内存占用，框架层承担窗口约束或至少提供 buffer bound 保护，使得生产环境中 Join pipeline 不会因数据倾斜或流速不均而 OOM。完成标准：
JoinOperator在无限流下不会无界积累缓冲，具备max_buffer_size_per_key或窗口超时保护，handle_stop_signal不依赖字符串匹配，而是基于注册的输入流数量判断终止条件。父 Issue: #1484