Add tests for stateful stream operators