Add command topic max unacked messages per consumer. (streamnative/pu…

…lsarctl#246) (apache#358)

Add command topic max unacked messages per consumer:

* pulsarctl topics get-max-unacked-messages-per-consumer [topic]
* pulsarctl topics set-max-unacked-messages-per-consumer [topic] -m [max]
* pulsarctl topics remove-max-unacked-messages-per-consumer [topic]

pulsaradmin/pkg/pulsar/topic.go

@@ -125,6 +125,15 @@ type Topics interface {
 
 	// RemoveMaxConsumers Remove max number of consumers for a topic
 	RemoveMaxConsumers(utils.TopicName) error
+
+	// GetMaxUnackMessagesPerConsumer Get max unacked messages policy on consumer for a topic
+	GetMaxUnackMessagesPerConsumer(utils.TopicName) (int, error)
+
+	// SetMaxUnackMessagesPerConsumer Set max unacked messages policy on consumer for a topic
+	SetMaxUnackMessagesPerConsumer(utils.TopicName, int) error
+
+	// RemoveMaxUnackMessagesPerConsumer Remove max unacked messages policy on consumer for a topic
+	RemoveMaxUnackMessagesPerConsumer(utils.TopicName) error
 }
 
 type topics struct {
@@ -389,3 +398,20 @@ func (t *topics) RemoveMaxConsumers(topic utils.TopicName) error {
 	err := t.pulsar.Client.Delete(endpoint)
 	return err
 }
+
+func (t *topics) GetMaxUnackMessagesPerConsumer(topic utils.TopicName) (int, error) {
+	var maxNum int
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "maxUnackedMessagesOnConsumer")
+	err := t.pulsar.Client.Get(endpoint, &maxNum)
+	return maxNum, err
+}
+
+func (t *topics) SetMaxUnackMessagesPerConsumer(topic utils.TopicName, maxUnackedNum int) error {
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "maxUnackedMessagesOnConsumer")
+	return t.pulsar.Client.Post(endpoint, &maxUnackedNum)
+}
+
+func (t *topics) RemoveMaxUnackMessagesPerConsumer(topic utils.TopicName) error {
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "maxUnackedMessagesOnConsumer")
+	return t.pulsar.Client.Delete(endpoint)
+}