streaming: report error to local barrier manager explicitly when error occurs

[BugenZhao]

If an error (not panic) occurs during the execution of an actor, the actor will exit with Err, and without reporting to the local barrier manager. The exiting of the actor will cause upstream/downstream to exit, and finally cause a failure when injecting the next barrier, which will be realized by the meta service and trigger the fail-over procedure.

However, if the in-flight concurrent checkpoint has reached the maximum (or is simply set to 1), the compute nodes will hang forever, while the heartbeat acts normally. We'll fail to recover the cluster.

We may need to explicitly report the error to the local barrier manager, so that the meta service will realize the failure ASAP.

cc @StrikeW @yezizp2012

[yezizp2012]

And the hang situation will only occur if the actor exit right after the barrier has been sent successfully and before the collect.