goroutine leak

#### Expected behavior

```
func (c *connection) start() {
// Each connection gets its own goroutine that will
go func() {
if c.connect() {
if c.doHandshake() {
c.run()
} else {
c.changeState(connectionClosed)
}
} else {
c.changeState(connectionClosed)
}
}()
}
```
this code, If the pulsar doesn't work properly, it will cause a goroutine leak.

The issue points to connection.start and its connect, doHandshake, run, and changeState calls; begin by tracing how failures move through this goroutine and how connection shutdown is represented. Reproduce the reported non-working-Pulsar case if possible, then verify that the goroutine terminates and the connection reaches connectionClosed.

Written by the indexing model from the issue text.