Support update or remove topic properties

pulsaradmin/pkg/admin/topic.go

@@ -51,6 +51,12 @@ type Topics interface {
 	// GetProperties returns the properties of a topic
 	GetProperties(topic utils.TopicName) (map[string]string, error)
 
+	// UpdateProperties updates the properties of a topic
+	UpdateProperties(topic utils.TopicName, properties map[string]string) error
+
+	// RemoveProperty removes a property with the given key of a topic
+	RemoveProperty(topic utils.TopicName, key string) error
+
 	// Delete a topic, this function can delete both partitioned or non-partitioned topic
 	//
 	// @param topic
@@ -433,6 +439,16 @@ func (t *topics) GetProperties(topic utils.TopicName) (map[string]string, error)
 	return properties, err
 }
 
+func (t *topics) UpdateProperties(topic utils.TopicName, properties map[string]string) error {
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "properties")
+	return t.pulsar.Client.Put(endpoint, properties)
+}
+
+func (t *topics) RemoveProperty(topic utils.TopicName, key string) error {
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "properties")
+	return t.pulsar.Client.DeleteWithQueryParams(endpoint, map[string]string{"key": key})
+}
+
 func (t *topics) Delete(topic utils.TopicName, force bool, nonPartitioned bool) error {
 	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "partitions")
 	if nonPartitioned {

pulsaradmin/pkg/admin/topic_test.go

@@ -65,37 +65,63 @@ func TestCreateTopic(t *testing.T) {
 	t.Error("Couldn't find topic: " + topic)
 }
 
-func TestTopics_CreateWithProperties(t *testing.T) {
+func TestTopics_Properties(t *testing.T) {
+	t.Run("NonPartitioned", func(t *testing.T) {
+		internalTestTopicsProperties(t, 0)
+	})
+	t.Run("Partitioned", func(t *testing.T) {
+		internalTestTopicsProperties(t, 4)
+	})
+}
+
+func verifyTopicProperties(t *testing.T, admin Client, topic *utils.TopicName,
+	expected map[string]string) {
+	properties, err := admin.Topics().GetProperties(*topic)
+	assert.NoError(t, err)
+	assert.Equal(t, expected, properties)
+}
+
+func internalTestTopicsProperties(t *testing.T, partitions int) {
 	topic := newTopicName()
 	cfg := &config.Config{}
 	admin, err := New(cfg)
 	assert.NoError(t, err)
 	assert.NotNil(t, admin)
 
-	// Create non-partition topic
 	topicName, err := utils.GetTopicName(topic)
 	assert.NoError(t, err)
-	err = admin.Topics().CreateWithProperties(*topicName, 0, map[string]string{
+	err = admin.Topics().CreateWithProperties(*topicName, partitions, map[string]string{
 		"key1": "value1",
 	})
 	assert.NoError(t, err)
+	verifyTopicProperties(t, admin, topicName, map[string]string{"key1": "value1"})
 
 	properties, err := admin.Topics().GetProperties(*topicName)
 	assert.NoError(t, err)
 	assert.Equal(t, properties["key1"], "value1")
 
-	// Create partition topic
-	topic = newTopicName()
-	topicName, err = utils.GetTopicName(topic)
+	newProperties := map[string]string{
+		"key1": "value1-updated",
+		"key2": "value2",
+	}
+	err = admin.Topics().UpdateProperties(*topicName, newProperties)
+	assert.NoError(t, err)
+	verifyTopicProperties(t, admin, topicName, newProperties)
+
+	err = admin.Topics().UpdateProperties(*topicName, map[string]string{"key3": "value3"})
 	assert.NoError(t, err)
-	err = admin.Topics().CreateWithProperties(*topicName, 4, map[string]string{
+	verifyTopicProperties(t, admin, topicName, map[string]string{
+		"key1": "value1-updated",
 		"key2": "value2",
+		"key3": "value3",
 	})
-	assert.NoError(t, err)
 
-	properties, err = admin.Topics().GetProperties(*topicName)
+	err = admin.Topics().RemoveProperty(*topicName, "key1")
 	assert.NoError(t, err)
-	assert.Equal(t, properties["key2"], "value2")
+	verifyTopicProperties(t, admin, topicName, map[string]string{
+		"key2": "value2",
+		"key3": "value3",
+	})
 }
 
 func TestPartitionState(t *testing.T) {