fix: Error handling in StreamingBatchWriter (#1921)

original PR: #1913

Reverts the revertion, plus more improvements:
- Client's handlers are now allowed to return an immediate error without draining the channel. The channel will be automatically drained for the error condition, and subsequent writes for that table won't get sent to the client.
- Some potential race conditions are fixed (`ensureOpened()` inline-func now got refactored into `s.send()` and handles the sending as well. The spawned goroutine doesn't refer to `inputCh` directly so that it can be replaced in `closeFlush()`.
- Shutdown logic is handled better: even if `client.Write()` returns an error after `msgs` is closed, it's still logged and returned.

Author: disq