Repository: apache/pulsar-client-go (go)

Issue:
[Question]Zero value of ReceiverQueueSize
For php client was discussed [#26](https://github.com/ikilobyte/pulsar-client-php/issues/26)

According to implementation
```go
if options.ReceiverQueueSize <= 0 {
	options.ReceiverQueueSize = 1000
}
```
According to [recommendations](https://pulsar.apache.org/docs/next/cookbooks-message-queue/#client-configuration-changes):
"...set the consumers' receiver queue size very low (potentially even to **_0_** if necessary)..."

So according to docs, I need set this value to 0, but go client sets it to 1000.

Please clarify




Pull request description:
[Issue 1223] Support ZeroQueueConsumer



Fixes #1223


### Motivation

Support ZeroQueueConsumer, refer to Java [ZeroQueueConsumerImpl](https://github.com/apache/pulsar/blob/8c50a6c2e91c81dbf187ce5e66cb39e2758a741e/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java#L42)

### Modifications

- The consumer add a new optional parameter `EnableZeroQueueConsumer`
- Add a new `zeroQueueConsumer`

### Verifying this change

- [x] 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


Patch:
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index 0b70a16..31f89a5 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -160,6 +160,11 @@ type ConsumerOptions struct {
 	// Default value is `1000` messages and should be good for most use cases.
 	ReceiverQueueSize int
 
+	// EnableZeroQueueConsumer, if enabled, the ReceiverQueueSize will be 0.
+	// Notice: only non-partitioned topic is supported.
+	// Default is false.
+	EnableZeroQueueConsumer bool
+
 	// EnableAutoScaledReceiverQueueSize, if enabled, the consumer receive queue will be auto-scaled
 	// by the consumer actual throughput. The ReceiverQueueSize will be the maximum size which consumer
 	// receive queue can be scaled.
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index 71ac323..291471c 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -22,9 +22,12 @@ import (
 	"fmt"
 	"math/rand"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
+	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
+
 	"github.com/apache/pulsar-client-go/pulsar/crypto"
 	"github.com/apache/pulsar-client-go/pulsar/internal"
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
@@ -81,6 +84,10 @@ func newConsumer(client *client, options ConsumerOptions) (Consumer, error) {
 		options.ReceiverQueueSize = defaultReceiverQueueSize
 	}
 
+	if options.EnableZeroQueueConsumer {
+		options.ReceiverQueueSize = 0
+	}
+
 	if options.Interceptors == nil {
 		options.Interceptors = defaultConsumerInterceptors
 	}
@@ -236,7 +243,24 @@ func newConsumer(client *client, options ConsumerOptions) (Consumer, error) {
 }
 
 func newInternalConsumer(client *client, options ConsumerOptions, topic string,
-	messageCh chan ConsumerMessage, dlq *dlqRouter, rlq *retryRouter, disableForceTopicCreation bool) (*consumer, error) {
+	messageCh chan ConsumerMessage, dlq *dlqRouter, rlq *retryRouter, disableForceTopicCreation bool) (Consumer, error) {
+	partitions, err := client.TopicPartitions(topic)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(partitions) > 1 && options.EnableZeroQueueConsumer {
+		return nil, pkgerrors.New("ZeroQueueConsumer is not supported for partitioned topics")
+	}
+
+	if len(partitions) == 1 && options.EnableZeroQueueConsumer &&
+		strings.Contains(partitions[0], utils.PARTITIONEDTOPICSUFFIX) {
+		return nil, pkgerrors.New("ZeroQueueConsumer is not supported for partitioned topics")
+	}
+
+	if len(partitions) == 1 && options.EnableZeroQueueConsumer {
+		return newZeroConsumer(client, options, topic, messageCh, dlq, rlq, disableForceTopicCreation)
+	}
 
 	consumer := &consumer{
 		topic:                     topic,
@@ -253,7 +277,7 @@ func newInternalConsumer(client *client, options ConsumerOptions, topic string,
 		metrics:                   client.metrics.GetLeveledMetrics(topic),
 	}
 
-	err := consumer.internalTopicSubscribeToPartitions()
+	err = consumer.internalTopicSubscribeToPartitions()
 	if err != nil {
 		return nil, err
 	}
@@ -343,10 +367,6 @@ func (c *consumer) internalTopicSubscribeToPartitions() error {
 		consumer  *partitionConsumer
 	}
 
-	receiverQueueSize := c.options.ReceiverQueueSize
-	metadata := c.options.Properties
-	subProperties := c.options.SubscriptionProperties
-
 	startPartition := oldNumPartitions
 	partitionsToAdd := newNumPartitions - oldNumPartitions
 
@@ -364,45 +384,7 @@ func (c *consumer) internalTopicSubscribeToPartitions() error {
 
 		go func(idx int, pt string) {
 			defer wg.Done()
-
-			var nackRedeliveryDelay time.Duration
-			if c.options.NackRedeliveryDelay == 0 {
-				nackRedeliveryDelay = defaultNackRedeliveryDelay
-			} else {
-				nackRedeliveryDelay = c.options.NackRedeliveryDelay
-			}
-			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,
-			}
+			opts := newPartitionConsumerOpts(pt, c.consumerName, idx, c.options)
 			cons, err := newPartitionConsumer(c, c.client, opts, c.messageCh, c.dlq, c.metrics)
 			ch <- ConsumerError{
 				err:       err,
@@ -444,6 +426,48 @@ func (c *consumer) internalTopicSubscribeToPartitions() error {
 	return nil
 }
 
+func newPartitionConsumerOpts(topic, consumerName string, idx int, options ConsumerOptions) *partitionConsumerOpts {
+
+	var nackRedeliveryDelay time.Duration
+	if options.NackRedeliveryDelay == 0 {
+		nackRedeliveryDelay = defaultNackRedeliveryDelay
+	} else {
+		nackRedeliveryDelay = options.NackRedeliveryDelay
+	}
+	return &partitionConsumerOpts{
+		topic:                       topic,
+		consumerName:                consumerName,
+		subscription:                options.SubscriptionName,
+		subscriptionType:            options.Type,
+		subscriptionInitPos:         options.SubscriptionInitialPosition,
+		partitionIdx:                idx,
+		receiverQueueSize:           options.ReceiverQueueSize,
+		nackRedeliveryDelay:         nackRedeliveryDelay,
+		nackBackoffPolicy:           options.NackBackoffPolicy,
+		metadata:                    options.Properties,
+		subProperties:               options.SubscriptionProperties,
+		replicateSubscriptionState:  options.ReplicateSubscriptionState,
+		startMessageID:              options.startMessageID,
+		startMessageIDInclusive:     options.StartMessageIDInclusive,
+		subscriptionMode:            options.SubscriptionMode,
+		readCompacted:               options.ReadCompacted,
+		interceptors:                options.Interceptors,
+		maxReconnectToBroker:        options.MaxReconnectToBroker,
+		backoffPolicy:               options.BackoffPolicy,
+		keySharedPolicy:             options.KeySharedPolicy,
+		schema:                      options.Schema,
+		decryption:                  options.Decryption,
+		ackWithResponse:             options.AckWithResponse,
+		maxPendingChunkedMessage:    options.MaxPendingChunkedMessage,
+		expireTimeOfIncompleteChunk: options.ExpireTimeOfIncompleteChunk,
+		autoAckIncompleteChunk:      options.AutoAckIncompleteChunk,
+		consumerEventListener:       options.EventListener,
+		enableBatchIndexAck:         options.EnableBatchIndexAcknowledgment,
+		ackGroupingOptions:          options.AckGroupingOptions,
+		autoReceiverQueueSize:       options.EnableAutoScaledReceiverQueueSize,
+	}
+}
+
 func (c *consumer) Subscription() string {
 	return c.options.SubscriptionName
 }
diff --git a/pulsar/consumer_zero_queue.go b/pulsar/consumer_zero_queue.go
new file mode 100644
index 0000000..8117127
--- /dev/null
+++ b/pulsar/consumer_zero_queue.go
@@ -0,0 +1,289 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pulsar
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/apache/pulsar-client-go/pulsar/internal"
+	"github.com/apache/pulsar-client-go/pulsar/log"
+	"github.com/pkg/errors"
+)
+
+type zeroQueueConsumer struct {
+	sync.Mutex
+	topic                     string
+	client                    *client
+	options                   ConsumerOptions
+	pc                        *partitionConsumer
+	consumerName              string
+	disableForceTopicCreation bool
+
+	messageCh chan ConsumerMessage
+
+	dlq       *dlqRouter
+	rlq       *retryRouter
+	closeOnce sync.Once
+	closeCh   chan struct{}
+	errorCh   chan error
+
+	log     log.Logger
+	metrics *internal.LeveledMetrics
+}
+
+func newZeroConsumer(client *client, options ConsumerOptions, topic string,
+	messageCh chan ConsumerMessage, dlq *dlqRouter,
+	rlq *retryRouter, disableForceTopicCreation bool) (*zeroQueueConsumer, error) {
+	zc := &zeroQueueConsumer{
+		topic:                     topic,
+		client:                    client,
+		options:                   options,
+		disableForceTopicCreation: disableForceTopicCreation,
+		messageCh:                 messageCh,
+		closeCh:                   make(chan struct{}),
+		errorCh:                   make(chan error),
+		dlq:                       dlq,
+		rlq:                       rlq,
+		log:                       client.log.SubLogger(log.Fields{"topic": topic}),
+		consumerName:              options.Name,
+		metrics:                   client.metrics.GetLeveledMetrics(topic),
+	}
+	opts := newPartitionConsumerOpts(zc.topic, zc.consumerName, 0, zc.options)
+	conn, err := newPartitionConsumer(zc, zc.client, opts, zc.messageCh, zc.dlq, zc.metrics)
+	if err != nil {
+		return nil, err
+	}
+	zc.pc = conn
+
+	return zc, nil
+}
+
+func (z *zeroQueueConsumer) Subscription() string {
+	return z.options.SubscriptionName
+}
+
+func (z *zeroQueueConsumer) Unsubscribe() error {
+	return z.unsubscribe(false)
+}
+
+func (z *zeroQueueConsumer) UnsubscribeForce() error {
+	return z.unsubscribe(true)
+}
+
+func (z *zeroQueueConsumer) unsubscribe(force bool) error {
+	z.Lock()
+	defer z.Unlock()
+
+	if err := z.pc.unsubscribe(force); err != nil {
+		return errors.Errorf("topic %s, subscription %s: %v", z.topic, z.Subscription(), err)
+	}
+
+	return nil
+}
+
+func (z *zeroQueueConsumer) GetLastMessageIDs() ([]TopicMessageID, error) {
+	id, err := z.pc.getLastMessageID()
+	if err != nil {
+		return nil, err
+	}
+	tm := &topicMessageID{topic: z.pc.topic, track: id}
+	return []TopicMessageID{tm}, nil
+}
+
+func (z *zeroQueueConsumer) Receive(ctx context.Context) (Message, error) {
+	if state := z.pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
+		z.log.WithField("state", state).Error("Failed to ack by closing or closed consumer")
+		return nil, errors.New("consumer state is closed")
+	}
+	z.Lock()
+	defer z.Unlock()
+	z.pc.availablePermits.inc()
+	for {
+		select {
+		case <-z.closeCh:
+			return nil, newError(ConsumerClosed, "consumer closed")
+		case cm, ok := <-z.messageCh:
+			if !ok {
+				return nil, newError(ConsumerClosed, "consumer closed")
+			}
+			return cm.Message, nil
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}
+
+}
+
+func (z *zeroQueueConsumer) Chan() <-chan ConsumerMessage {
+	panic("zeroQueueConsumer cannot support Chan method")
+}
+
+func (z *zeroQueueConsumer) Ack(m Message) error {
+	return z.AckID(m.ID())
+}
+
+func (z *zeroQueueConsumer) checkMsgIDPartition(msgID MessageID) error {
+	partition := msgID.PartitionIdx()
+	if partition != 0 {
+		z.log.Errorf("invalid partition index %d expected a partition equal to 0",
+			partition)
+		return fmt.Errorf("invalid partition index %d expected a partition equal to 0",
+			partition)
+	}
+	return nil
+}
+
+func (z *zeroQueueConsumer) messageID(msgID MessageID) *trackingMessageID {
+	if err := z.checkMsgIDPartition(msgID); err != nil {
+		return nil
+	}
+	mid := toTrackingMessageID(msgID)
+	return mid
+}
+
+func (z *zeroQueueConsumer) AckID(msgID MessageID) error {
+	if err := z.checkMsgIDPartition(msgID); err != nil {
+		return err
+	}
+
+	if z.options.AckWithResponse {
+		return z.pc.AckIDWithResponse(msgID)
+	}
+
+	return z.pc.AckID(msgID)
+}
+
+func (z *zeroQueueConsumer) AckWithTxn(msg Message, txn Transaction) error {
+	msgID := msg.ID()
+	if err := z.checkMsgIDPartition(msgID); err != nil {
+		return err
+	}
+
+	return z.pc.AckIDWithTxn(msgID, txn)
+}
+
+func (z *zeroQueueConsumer) AckCumulative(msg Message) error {
+	return z.AckIDCumulative(msg.ID())
+}
+
+func (z *zeroQueueConsumer) AckIDCumulative(msgID MessageID) error {
+	if err := z.checkMsgIDPartition(msgID); err != nil {
+		return err
+	}
+	if z.options.AckWithResponse {
+		return z.pc.AckIDWithResponseCumulative(msgID)
+	}
+	return z.pc.AckIDCumulative(msgID)
+}
+
+func (z *zeroQueueConsumer) ReconsumeLater(_ Message, _ time.Duration) {
+	z.log.Warnf("zeroQueueConsumer not support ReconsumeLater yet.")
+}
+
+func (z *zeroQueueConsumer) ReconsumeLaterWithCustomProperties(_ Message, _ map[string]string, _ time.Duration) {
+	z.log.Warnf("zeroQueueConsumer not support ReconsumeLaterWithCustomProperties yet.")
+}
+
+func (z *zeroQueueConsumer) Nack(msg Message) {
+	if !checkMessageIDType(msg.ID()) {
+		z.log.Warnf("invalid message id type %T", msg.ID())
+		return
+	}
+	if z.options.EnableDefaultNackBackoffPolicy || z.options.NackBackoffPolicy != nil {
+		mid := z.messageID(msg.ID())
+		if mid == nil {
+			return
+		}
+
+		if mid.consumer != nil {
+			mid.NackByMsg(msg)
+			return
+		}
+		z.pc.NackMsg(msg)
+		return
+	}
+
+	z.NackID(msg.ID())
+}
+
+func (z *zeroQueueConsumer) NackID(msgID MessageID) {
+	if err := z.checkMsgIDPartition(msgID); err != nil {
+		return
+	}
+	z.pc.NackID(msgID)
+}
+
+func (z *zeroQueueConsumer) Close() {
+	z.closeOnce.Do(func() {
+		z.Lock()
+		defer z.Unlock()
+
+		z.pc.Close()
+		close(z.closeCh)
+		z.client.handlers.Del(z)
+		z.dlq.close()
+		z.rlq.close()
+		z.metrics.ConsumersClosed.Inc()
+		z.metrics.ConsumersPartitions.Sub(float64(1))
+	})
+}
+
+func (z *zeroQueueConsumer) Seek(msgID MessageID) error {
+	z.Lock()
+	defer z.Unlock()
+
+	if err := z.checkMsgIDPartition(msgID); err != nil {
+		return err
+	}
+
+	if err := z.pc.Seek(msgID); err != nil {
+		return err
+	}
+
+	// clear messageCh
+	for len(z.messageCh) > 0 {
+		<-z.messageCh
+	}
+
+	return nil
+}
+
+func (z *zeroQueueConsumer) SeekByTime(time time.Time) error {
+	z.Lock()
+	defer z.Unlock()
+	var errs error
+
+	if err := z.pc.SeekByTime(time); err != nil {
+		msg := fmt.Sprintf("unable to SeekByTime for topic=%s subscription=%s", z.topic, z.Subscription())
+		errs = errors.Wrap(newError(SeekFailed, err.Error()), msg)
+	}
+
+	// clear messageCh
+	for len(z.messageCh) > 0 {
+		<-z.messageCh
+	}
+
+	return errs
+}
+
+func (z *zeroQueueConsumer) Name() string {
+	return z.consumerName
+}
diff --git a/pulsar/reader_impl.go b/pulsar/reader_impl.go
index 4f4e2aa..7cbae05 100644
--- a/pulsar/reader_impl.go
+++ b/pulsar/reader_impl.go
@@ -23,6 +23,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/pkg/errors"
+
 	"github.com/apache/pulsar-client-go/pulsar/crypto"
 	"github.com/apache/pulsar-client-go/pulsar/internal"
 	"github.com/apache/pulsar-client-go/pulsar/log"
@@ -141,7 +143,11 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
 		close(reader.messageCh)
 		return nil, err
 	}
-	reader.c = c
+	cs, ok := c.(*consumer)
+	if !ok {
+		return nil, errors.New("invalid consumer type")
+	}
+	reader.c = cs
 
 	reader.metrics.ReadersOpened.Inc()
 	return reader, nil
diff --git a/pulsaradmin/pkg/admin/admin.go b/pulsaradmin/pkg/admin/admin.go
index a22b22c..0b1b962 100644
--- a/pulsaradmin/pkg/admin/admin.go
+++ b/pulsaradmin/pkg/admin/admin.go
@@ -112,3 +112,8 @@ func (c *pulsarClient) endpoint(componentPath string, parts ...string) string {
 		path.Join(escapedParts...),
 	)
 }
+
+// this function won't do any escaping
+func (c *pulsarClient) endpointWithFullPath(componentPath string, fullPath string) string {
+	return path.Join(utils.MakeHTTPPath(c.APIVersion.String(), componentPath), fullPath)
+}
diff --git a/pulsaradmin/pkg/admin/brokers.go b/pulsaradmin/pkg/admin/brokers.go
index 7dcea80..8ac400f 100644
--- a/pulsaradmin/pkg/admin/brokers.go
+++ b/pulsaradmin/pkg/admin/brokers.go
@@ -19,7 +19,6 @@ package admin
 
 import (
 	"fmt"
-	"net/url"
 	"strings"
 
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
@@ -120,8 +119,8 @@ func (b *broker) GetOwnedNamespaces(cluster, brokerURL string) (map[string]utils
 }
 
 func (b *broker) UpdateDynamicConfiguration(configName, configValue string) error {
-	value := url.QueryEscape(configValue)
-	endpoint := b.pulsar.endpoint(b.basePath, "/configuration/", configName, value)
+	value := fmt.Sprintf("/configuration/%s/%s", configName, configValue)
+	endpoint := b.pulsar.endpointWithFullPath(b.basePath, value)
 	return b.pulsar.Client.Post(endpoint, nil)
 }
 


Test patch:
diff --git a/pulsar/consumer_zero_queue_test.go b/pulsar/consumer_zero_queue_test.go
new file mode 100644
index 0000000..d8c8f7c
--- /dev/null
+++ b/pulsar/consumer_zero_queue_test.go
@@ -0,0 +1,490 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pulsar
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"testing"
+	"time"
+
+	"github.com/apache/pulsar-client-go/pulsar/internal"
+	"github.com/apache/pulsar-client-go/pulsaradmin"
+	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/admin/config"
+	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNormalZeroQueueConsumer(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	ctx := context.Background()
+
+	// create consumer
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                   topic,
+		SubscriptionName:        "my-sub",
+		EnableZeroQueueConsumer: true,
+	})
+	assert.Nil(t, err)
+	_, ok := consumer.(*zeroQueueConsumer)
+	assert.True(t, ok)
+	defer consumer.Close()
+
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: false,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	// send 10 messages
+	for i := 0; i < 10; i++ {
+		msg, err := producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+			Key:     "pulsar",
+			Properties: map[string]string{
+				"key-1": "pulsar-1",
+			},
+		})
+		assert.Nil(t, err)
+		log.Printf("send message: %s", msg.String())
+	}
+
+	// receive 10 messages
+	for i := 0; i < 10; i++ {
+		msg, err := consumer.Receive(context.Background())
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		expectMsg := fmt.Sprintf("hello-%d", i)
+		expectProperties := map[string]string{
+			"key-1": "pulsar-1",
+		}
+		assert.Equal(t, []byte(expectMsg), msg.Payload())
+		assert.Equal(t, "pulsar", msg.Key())
+		assert.Equal(t, expectProperties, msg.Properties())
+		// ack message
+		consumer.Ack(msg)
+		log.Printf("receive message: %s", msg.ID().String())
+	}
+	err = consumer.Unsubscribe()
+	assert.Nil(t, err)
+}
+func TestPartitionZeroQueueConsumer(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	err = createPartitionedTopic(topic, 2)
+	assert.Nil(t, err)
+
+	// create consumer
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                   topic,
+		SubscriptionName:        "my-sub",
+		EnableZeroQueueConsumer: true,
+	})
+	assert.Nil(t, consumer)
+	assert.Error(t, err, "ZeroQueueConsumer is not supported for partitioned topics")
+}
+func TestOnePartitionZeroQueueConsumer(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	err = createPartitionedTopic(topic, 1)
+	assert.Nil(t, err)
+
+	// create consumer
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                   topic,
+		SubscriptionName:        "my-sub",
+		EnableZeroQueueConsumer: true,
+	})
+	assert.Nil(t, consumer)
+	assert.Error(t, err, "ZeroQueueConsumer is not supported for partitioned topics")
+}
+
+func TestZeroQueueConsumerGetLastMessageIDs(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	partition := 1
+	topic := newTopicName()
+	// create consumer
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                   topic,
+		SubscriptionName:        "my-sub",
+		Type:                    Shared,
+		EnableZeroQueueConsumer: true,
+	})
+	_, ok := consumer.(*zeroQueueConsumer)
+	assert.True(t, ok)
+	assert.Nil(t, err)
+	defer consumer.Close()
+
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: true,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	ctx := context.Background()
+	// send messages
+	totalMessage := 10
+	for i := 0; i < totalMessage; i++ {
+		if _, err := producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+		}); err != nil {
+			assert.Nil(t, err)
+		}
+	}
+
+	// create admin
+	admin, err := pulsaradmin.NewClient(&config.Config{})
+	assert.Nil(t, err)
+
+	messageIDs, err := consumer.GetLastMessageIDs()
+	assert.Nil(t, err)
+	assert.Equal(t, partition, len(messageIDs))
+
+	id := messageIDs[0]
+	topicName, err := utils.GetTopicName(id.Topic())
+	assert.Nil(t, err)
+	messages, err := admin.Subscriptions().GetMessagesByID(*topicName, id.LedgerID(), id.EntryID())
+	assert.Nil(t, err)
+	assert.Equal(t, 1, len(messages))
+	err = consumer.UnsubscribeForce()
+	assert.Nil(t, err)
+
+}
+
+func TestZeroQueueConsumer_Chan(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	ctx := context.Background()
+
+	// create consumer
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                   topic,
+		SubscriptionName:        "my-sub",
+		EnableZeroQueueConsumer: true,
+	})
+	assert.Nil(t, err)
+	_, ok := consumer.(*zeroQueueConsumer)
+	assert.True(t, ok)
+	defer consumer.Close()
+
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: false,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	// send 10 messages
+	for i := 0; i < 10; i++ {
+		msg, err := producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+		})
+		assert.Nil(t, err)
+		log.Printf("send message: %s", msg.String())
+	}
+	assertPanic(t, "zeroQueueConsumer cannot support Chan method", func() {
+		consumer.Chan()
+	})
+}
+
+func assertPanic(t *testing.T, panicValue interface{}, f func()) {
+	defer func() {
+		if r := recover(); r != panicValue {
+			t.Errorf("Expected panic %v, but got %v", panicValue, r)
+		}
+	}()
+	f()
+}
+
+func TestZeroQueueConsumer_AckCumulativeConsumer(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	ctx := context.Background()
+
+	// create consumer
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                   topic,
+		SubscriptionName:        "my-sub",
+		EnableZeroQueueConsumer: true,
+	})
+	assert.Nil(t, err)
+	_, ok := consumer.(*zeroQueueConsumer)
+	assert.True(t, ok)
+	defer consumer.Close()
+
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: false,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	// send 10 messages
+	for i := 0; i < 10; i++ {
+		msg, err := producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+		})
+		assert.Nil(t, err)
+		log.Printf("send message: %s", msg.String())
+	}
+
+	var lastID MessageID
+	// receive 10 messages
+	for i := 0; i < 10; i++ {
+		msg, err := consumer.Receive(context.Background())
+		lastID = msg.ID()
+		if err != nil {
+			log.Fatal(err)
+		}
+		expectMsg := fmt.Sprintf("hello-%d", i)
+		assert.Equal(t, []byte(expectMsg), msg.Payload())
+	}
+	err = consumer.AckIDCumulative(lastID)
+	assert.Nil(t, err)
+
+}
+
+func TestZeroQueueConsumer_Nack(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topicName := newTopicName()
+	ctx := context.Background()
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: topicName,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                   topicName,
+		SubscriptionName:        "sub-1",
+		Type:                    Shared,
+		NackRedeliveryDelay:     1 * time.Second,
+		EnableZeroQueueConsumer: true,
+	})
+	assert.Nil(t, err)
+	_, ok := consumer.(*zeroQueueConsumer)
+	assert.True(t, ok)
+	defer consumer.Close()
+
+	const N = 100
+
+	for i := 0; i < N; i++ {
+		if _, err := producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("msg-content-%d", i)),
+		}); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	for i := 0; i < N; i++ {
+		msg, err := consumer.Receive(ctx)
+		assert.Nil(t, err)
+		assert.Equal(t, fmt.Sprintf("msg-content-%d", i), string(msg.Payload()))
+
+		if i%2 == 0 {
+			// Only acks even messages
+			consumer.Ack(msg)
+		} else {
+			// Fails to process odd messages
+			consumer.Nack(msg)
+		}
+	}
+
+	// Failed messages should be resent
+
+	// We should only receive the odd messages
+	for i := 1; i < N; i += 2 {
+		msg, err := consumer.Receive(ctx)
+		assert.Nil(t, err)
+		assert.Equal(t, fmt.Sprintf("msg-content-%d", i), string(msg.Payload()))
+
+		consumer.Ack(msg)
+	}
+}
+
+func TestZeroQueueConsumer_Seek(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topicName := newTopicName()
+	ctx := context.Background()
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           topicName,
+		DisableBatching: false,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                   topicName,
+		EnableZeroQueueConsumer: true,
+		SubscriptionName:        "sub-1",
+	})
+	assert.Nil(t, err)
+	_, ok := consumer.(*zeroQueueConsumer)
+	assert.True(t, ok)
+	defer consumer.Close()
+
+	const N = 10
+	var seekID MessageID
+	for i := 0; i < N; i++ {
+		id, err := producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+		})
+		assert.Nil(t, err)
+
+		if i == N-5 {
+			seekID = id
+		}
+	}
+
+	// Don't consume all messages so some stay in queues
+	for i := 0; i < N; i++ {
+		msg, err := consumer.Receive(ctx)
+		assert.Nil(t, err)
+		assert.Equal(t, fmt.Sprintf("hello-%d", i), string(msg.Payload()))
+		consumer.Ack(msg)
+	}
+
+	err = consumer.Seek(seekID)
+	assert.Nil(t, err)
+
+	time.Sleep(time.Second * 3)
+	msg, err := consumer.Receive(ctx)
+	assert.Nil(t, err)
+	log.Printf("msg: %s", string(msg.Payload()))
+	assert.Equal(t, fmt.Sprintf("hello-%d", N-5), string(msg.Payload()))
+}
+
+func TestZeroQueueConsumer_SeekByTime(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topicName := newTopicName()
+	ctx := context.Background()
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           topicName,
+		DisableBatching: false,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                   topicName,
+		EnableZeroQueueConsumer: true,
+		SubscriptionName:        "my-sub",
+	})
+	assert.Nil(t, err)
+	_, ok := consumer.(*zeroQueueConsumer)
+	assert.True(t, ok)
+	defer consumer.Close()
+
+	const N = 10
+	resetTimeStr := "100s"
+	retentionTimeInSecond, err := internal.ParseRelativeTimeInSeconds(resetTimeStr)
+	assert.Nil(t, err)
+
+	for i := 0; i < N; i++ {
+		_, err := producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+		})
+		assert.Nil(t, err)
+	}
+
+	// Don't consume all messages so some stay in queues
+	for i := 0; i < N-2; i++ {
+		msg, err := consumer.Receive(ctx)
+		assert.Nil(t, err)
+		assert.Equal(t, fmt.Sprintf("hello-%d", i), string(msg.Payload()))
+		consumer.Ack(msg)
+	}
+
+	currentTimestamp := time.Now()
+	err = consumer.SeekByTime(currentTimestamp.Add(-retentionTimeInSecond))
+	assert.Nil(t, err)
+
+	time.Sleep(time.Second * 3)
+
+	for i := 0; i < N; i++ {
+		msg, err := consumer.Receive(ctx)
+		assert.Nil(t, err)
+		assert.Equal(t, fmt.Sprintf("hello-%d", i), string(msg.Payload()))
+		consumer.Ack(msg)
+	}
+}
diff --git a/pulsaradmin/pkg/admin/brokers_test.go b/pulsaradmin/pkg/admin/brokers_test.go
index 3ae9e4a..5be90f6 100644
--- a/pulsaradmin/pkg/admin/brokers_test.go
+++ b/pulsaradmin/pkg/admin/brokers_test.go
@@ -73,3 +73,21 @@ func TestGetAllActiveBrokers(t *testing.T) {
 	assert.NoError(t, err)
 	assert.NotEmpty(t, brokers)
 }
+
+func TestUpdateDynamicConfiguration(t *testing.T) {
+	readFile, err := os.ReadFile("../../../integration-tests/tokens/admin-token")
+	assert.NoError(t, err)
+	cfg := &config.Config{
+		Token: string(readFile),
+	}
+	admin, err := New(cfg)
+	assert.NoError(t, err)
+	assert.NotNil(t, admin)
+
+	err = admin.Brokers().UpdateDynamicConfiguration("allowAutoSubscriptionCreation", "true")
+	assert.NoError(t, err)
+
+	configurations, err := admin.Brokers().GetDynamicConfigurationNames()
+	assert.NoError(t, err)
+	assert.NotEmpty(t, configurations)
+}