Fix using closed connection in consumer

[hrsakai]

Motivation

Related to #761

This PR fixed the issue using closed connection at consumer side.

If same issue as #761 happens at consumer side, consumer will report the following errors and can't consume any messages.

ERRO[0180] [Connection was closed when request flow cmd]  consumerID=1 name=wzrtq subscription=sub1 topic="persistent://public/default/topic-go-1"
ERRO[0180] [unable to send initial permits to broker]    consumerID=1 error="connection closed" name=wzrtq subscription=sub1 topic="persistent://public/default/topic-go-1"

Modifications

• return an error when unable to add consumer handler.