[Issue 1223] Support ZeroQueueConsumer

[crossoverJie]

Fixes #1223

Motivation

Support ZeroQueueConsumer, refer to Java ZeroQueueConsumerImpl

Modifications

• The consumer add a new optional parameter EnableZeroQueueConsumer
• Add a new zeroQueueConsumer

Verifying this change

• Make sure that the change passes the CI checks.

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (yes)
• The schema: (no)
• The default values of configurations: (yes)
• The wire protocol: (no)

Documentation

• Does this pull request introduce a new feature? (yes)
• If yes, how is the feature documented? (GoDocs)
• If a feature is not applicable for documentation, explain why?
• If a feature is not documented yet in this PR, please create a followup issue for adding the documentation

[crossoverJie]

@RobertIndie PTAL, thanks.

[RobertIndie]

Could we abstract this logic so that we can reuse the code?

pulsar-client-go/pulsar/consumer_impl.go

Lines 390 to 420 in fc15995

	opts := &partitionConsumerOpts{
	topic: pt,
	consumerName: c.consumerName,
	subscription: c.options.SubscriptionName,
	subscriptionType: c.options.Type,
	subscriptionInitPos: c.options.SubscriptionInitialPosition,
	partitionIdx: idx,
	receiverQueueSize: receiverQueueSize,
	nackRedeliveryDelay: nackRedeliveryDelay,
	nackBackoffPolicy: c.options.NackBackoffPolicy,
	metadata: metadata,
	subProperties: subProperties,
	replicateSubscriptionState: c.options.ReplicateSubscriptionState,
	startMessageID: c.options.startMessageID,
	startMessageIDInclusive: c.options.StartMessageIDInclusive,
	subscriptionMode: c.options.SubscriptionMode,
	readCompacted: c.options.ReadCompacted,
	interceptors: c.options.Interceptors,
	maxReconnectToBroker: c.options.MaxReconnectToBroker,
	backoffPolicy: c.options.BackoffPolicy,
	keySharedPolicy: c.options.KeySharedPolicy,
	schema: c.options.Schema,
	decryption: c.options.Decryption,
	ackWithResponse: c.options.AckWithResponse,
	maxPendingChunkedMessage: c.options.MaxPendingChunkedMessage,
	expireTimeOfIncompleteChunk: c.options.ExpireTimeOfIncompleteChunk,
	autoAckIncompleteChunk: c.options.AutoAckIncompleteChunk,
	consumerEventListener: c.options.EventListener,
	enableBatchIndexAck: c.options.EnableBatchIndexAcknowledgment,
	ackGroupingOptions: c.options.AckGroupingOptions,
	autoReceiverQueueSize: c.options.EnableAutoScaledReceiverQueueSize,

[crossoverJie]

Thanks, done.