[fix] Close consumer resources if creation fails (#1070)

### Motivation

When a consumer fails to get created, we should close any resources that it created to prevent leaks of internal resources and leaks of the consumer on the broker side. The broker leak could happen if the connection was left open. These fixes are similar to #1061.

### Modifications

* Close `ackGroupingTracker` and `chunkedMsgCtxMap` if `grabConn` fails. We cannot call `Close` on the consumer because the state is not `Ready`. If we re-design the consumer, it could be nice to be able to call `Close` in this scenario.
* Call `Close` on the consumer in cases where we move it to `Ready` but determine it is not able to be created.
* Fix typo in comment

(cherry picked from commit a3fcc9a)

Author: michaeljmarshall

pulsar/consumer_partition.go

@@ -372,6 +372,8 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 	if err != nil {
 		pc.log.WithError(err).Error("Failed to create consumer")
 		pc.nackTracker.Close()
+		pc.ackGroupingTracker.close()
+		pc.chunkedMsgCtxMap.Close()
 		return nil, err
 	}
 	pc.log.Info("Created consumer")
@@ -381,7 +383,7 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 	if pc.options.startMessageIDInclusive && startingMessageID != nil && startingMessageID.equal(latestMessageID) {
 		msgID, err := pc.requestGetLastMessageID()
 		if err != nil {
-			pc.nackTracker.Close()
+			pc.Close()
 			return nil, err
 		}
 		if msgID.entryID != noMessageEntry {
@@ -390,7 +392,7 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 			// use the WithoutClear version because the dispatcher is not started yet
 			err = pc.requestSeekWithoutClear(msgID.messageID)
 			if err != nil {
-				pc.nackTracker.Close()
+				pc.Close()
 				return nil, err
 			}
 		}

pulsar/producer_partition.go

@@ -1405,7 +1405,7 @@ func (p *partitionProducer) setProducerState(state producerState) {
 	p.state.Swap(int32(state))
 }
 
-// set a new consumerState and return the last state
+// set a new producerState and return the last state
 // returns bool if the new state has been set or not
 func (p *partitionProducer) casProducerState(oldState, newState producerState) bool {
 	return p.state.CAS(int32(oldState), int32(newState))