updated with distributed stateful processing