feat: add pulsar admin namespace properties methods for PUT/GET/DELETE (#1390)

Author: thomas-bousquet

pulsaradmin/pkg/admin/namespace.go

@@ -305,6 +305,15 @@ type Namespaces interface {
 	// RemoveSubscriptionExpirationTime removes subscription expiration time from a namespace,
 	// defaulting to broker settings
 	RemoveSubscriptionExpirationTime(namespace utils.NameSpaceName) error
+
+	// UpdateProperties updates the properties of a namespace
+	UpdateProperties(namespace utils.NameSpaceName, properties map[string]string) error
+
+	// GetProperties returns the properties of a namespace
+	GetProperties(namespace utils.NameSpaceName) (map[string]string, error)
+
+	// RemoveProperties clears the properties of a namespace
+	RemoveProperties(namespace utils.NameSpaceName) error
 }
 
 type namespaces struct {
@@ -970,3 +979,20 @@ func (n *namespaces) RemoveSubscriptionExpirationTime(namespace utils.NameSpaceN
 	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "subscriptionExpirationTime")
 	return n.pulsar.Client.Delete(endpoint)
 }
+
+func (n *namespaces) UpdateProperties(namespace utils.NameSpaceName, properties map[string]string) error {
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "properties")
+	return n.pulsar.Client.Put(endpoint, properties)
+}
+
+func (n *namespaces) GetProperties(namespace utils.NameSpaceName) (map[string]string, error) {
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "properties")
+	properties := make(map[string]string)
+	err := n.pulsar.Client.Get(endpoint, &properties)
+	return properties, err
+}
+
+func (n *namespaces) RemoveProperties(namespace utils.NameSpaceName) error {
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "properties")
+	return n.pulsar.Client.Delete(endpoint)
+}

pulsaradmin/pkg/admin/namespace_test.go

@@ -495,3 +495,31 @@ func TestNamespaces_GetSchemaCompatibilityStrategy(t *testing.T) {
 	}
 	assert.Equal(t, utils.SchemaCompatibilityStrategyUndefined, defaultStrategy)
 }
+
+func TestNamespaces_Properties(t *testing.T) {
+	config := &config.Config{}
+	admin, err := New(config)
+	require.NoError(t, err)
+	require.NotNil(t, admin)
+
+	namespace, err := utils.GetNamespaceName("public/default")
+	assert.Equal(t, err, nil)
+
+	// Namespace properties are expected to be set and retrieved successfully
+	properties := map[string]string{
+		"key-1": "value-1",
+	}
+	err = admin.Namespaces().UpdateProperties(*namespace, properties)
+	assert.Equal(t, err, nil)
+
+	actualProperties, err := admin.Namespaces().GetProperties(*namespace)
+	assert.Equal(t, err, nil)
+	assert.Equal(t, actualProperties, properties)
+
+	// All namespace properties are expected to be deleted successfully
+	err = admin.Namespaces().RemoveProperties(*namespace)
+	assert.Equal(t, err, nil)
+	actualPropertiesAfterRemoveCall, err := admin.Namespaces().GetProperties(*namespace)
+	assert.Equal(t, err, nil)
+	assert.Equal(t, actualPropertiesAfterRemoveCall, map[string]string{})
+}