You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2026/07/23 08:59:10 UTC
(pulsar-client-go) branch master updated: feat: add RemoveBacklogQuotaByType to namespace (#1525)
This is an automated email from the ASF dual-hosted git repository.
RobertIndie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git
The following commit(s) were added to refs/heads/master by this push:
new 8213b20c feat: add RemoveBacklogQuotaByType to namespace (#1525)
8213b20c is described below
commit 8213b20c03b79d899eca91bfc8fc7c104f983c01
Author: Rui Fu <fr...@users.noreply.github.com>
AuthorDate: Thu Jul 23 16:59:05 2026 +0800
feat: add RemoveBacklogQuotaByType to namespace (#1525)
### Motivation
support following API:
```http
DELETE /admin/v2/namespaces/{tenant}/{namespace}/backlogQuota
?backlogQuotaType=message_age|destination_storage
```
---
pulsaradmin/pkg/admin/namespace.go | 30 ++++++--
pulsaradmin/pkg/admin/namespace_test.go | 117 ++++++++++++++++++++++++++++++++
2 files changed, 143 insertions(+), 4 deletions(-)
diff --git a/pulsaradmin/pkg/admin/namespace.go b/pulsaradmin/pkg/admin/namespace.go
index 1bd80566..7f73750f 100644
--- a/pulsaradmin/pkg/admin/namespace.go
+++ b/pulsaradmin/pkg/admin/namespace.go
@@ -140,12 +140,22 @@ type Namespaces interface {
backlogQuotaType utils.BacklogQuotaType,
) error
- // RemoveBacklogQuota removes a backlog quota policy from a namespace
+ // RemoveBacklogQuota removes the destination storage backlog quota policy from a namespace
RemoveBacklogQuota(namespace string) error
- // RemoveBacklogQuotaWithContext removes a backlog quota policy from a namespace
+ // RemoveBacklogQuotaWithContext removes the destination storage backlog quota policy from a namespace
RemoveBacklogQuotaWithContext(ctx context.Context, namespace string) error
+ // RemoveBacklogQuotaByType removes a backlog quota policy of the specified type from a namespace
+ RemoveBacklogQuotaByType(namespace string, backlogQuotaType utils.BacklogQuotaType) error
+
+ // RemoveBacklogQuotaByTypeWithContext removes a backlog quota policy of the specified type from a namespace
+ RemoveBacklogQuotaByTypeWithContext(
+ ctx context.Context,
+ namespace string,
+ backlogQuotaType utils.BacklogQuotaType,
+ ) error
+
// GetTopicAutoCreation returns the topic auto-creation config for a namespace.
// Returns nil if the topic auto-creation config is not configured at the namespace level.
GetTopicAutoCreation(namespace utils.NameSpaceName) (*utils.TopicAutoCreationConfig, error)
@@ -1048,17 +1058,29 @@ func (n *namespaces) SetBacklogQuotaWithContext(ctx context.Context, namespace s
}
func (n *namespaces) RemoveBacklogQuota(namespace string) error {
- return n.RemoveBacklogQuotaWithContext(context.Background(), namespace)
+ return n.RemoveBacklogQuotaByType(namespace, utils.DestinationStorage)
}
func (n *namespaces) RemoveBacklogQuotaWithContext(ctx context.Context, namespace string) error {
+ return n.RemoveBacklogQuotaByTypeWithContext(ctx, namespace, utils.DestinationStorage)
+}
+
+func (n *namespaces) RemoveBacklogQuotaByType(namespace string, backlogQuotaType utils.BacklogQuotaType) error {
+ return n.RemoveBacklogQuotaByTypeWithContext(context.Background(), namespace, backlogQuotaType)
+}
+
+func (n *namespaces) RemoveBacklogQuotaByTypeWithContext(
+ ctx context.Context,
+ namespace string,
+ backlogQuotaType utils.BacklogQuotaType,
+) error {
nsName, err := utils.GetNamespaceName(namespace)
if err != nil {
return err
}
endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "backlogQuota")
params := map[string]string{
- "backlogQuotaType": string(utils.DestinationStorage),
+ "backlogQuotaType": string(backlogQuotaType),
}
return n.pulsar.Client.DeleteWithQueryParamsWithContext(ctx, endpoint, params)
}
diff --git a/pulsaradmin/pkg/admin/namespace_test.go b/pulsaradmin/pkg/admin/namespace_test.go
index f14027dd..8f3131b9 100644
--- a/pulsaradmin/pkg/admin/namespace_test.go
+++ b/pulsaradmin/pkg/admin/namespace_test.go
@@ -18,6 +18,10 @@
package admin
import (
+ "context"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
"os"
"testing"
"time"
@@ -33,6 +37,119 @@ func ptr(n int) *int {
return &n
}
+func TestNamespaces_RemoveBacklogQuotaByType(t *testing.T) {
+ tests := []struct {
+ name string
+ quotaType utils.BacklogQuotaType
+ remove func(Namespaces) error
+ }{
+ {
+ name: "legacy",
+ quotaType: utils.DestinationStorage,
+ remove: func(namespaces Namespaces) error {
+ return namespaces.RemoveBacklogQuota("public/default")
+ },
+ },
+ {
+ name: "legacy with context",
+ quotaType: utils.DestinationStorage,
+ remove: func(namespaces Namespaces) error {
+ return namespaces.RemoveBacklogQuotaWithContext(context.Background(), "public/default")
+ },
+ },
+ {
+ name: "message age",
+ quotaType: utils.MessageAge,
+ remove: func(namespaces Namespaces) error {
+ return namespaces.RemoveBacklogQuotaByType("public/default", utils.MessageAge)
+ },
+ },
+ {
+ name: "message age with context",
+ quotaType: utils.MessageAge,
+ remove: func(namespaces Namespaces) error {
+ return namespaces.RemoveBacklogQuotaByTypeWithContext(
+ context.Background(),
+ "public/default",
+ utils.MessageAge,
+ )
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ type request struct {
+ method string
+ path string
+ quotaType string
+ querySize int
+ }
+ requests := make(chan request, 1)
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ query := r.URL.Query()
+ requests <- request{
+ method: r.Method,
+ path: r.URL.EscapedPath(),
+ quotaType: query.Get("backlogQuotaType"),
+ querySize: len(query),
+ }
+ w.WriteHeader(http.StatusNoContent)
+ }))
+ t.Cleanup(server.Close)
+
+ client, err := New(&config.Config{WebServiceURL: server.URL})
+ require.NoError(t, err)
+
+ require.NoError(t, tt.remove(client.Namespaces()))
+ actual := <-requests
+ assert.Equal(t, http.MethodDelete, actual.method)
+ assert.Equal(t, "/admin/v2/namespaces/public/default/backlogQuota", actual.path)
+ assert.Equal(t, tt.quotaType.String(), actual.quotaType)
+ assert.Equal(t, 1, actual.querySize)
+ })
+ }
+}
+
+func TestNamespaces_RemoveBacklogQuotaByTypeIntegration(t *testing.T) {
+ client, err := New(&config.Config{})
+ require.NoError(t, err)
+
+ namespaces := client.Namespaces()
+ namespace := fmt.Sprintf("public/remove-backlog-quota-%d", time.Now().UnixNano())
+ require.NoError(t, namespaces.CreateNamespace(namespace))
+ t.Cleanup(func() {
+ _ = namespaces.DeleteNamespace(namespace)
+ })
+
+ require.NoError(t, namespaces.SetBacklogQuota(
+ namespace,
+ utils.NewBacklogQuota(1024, -1, utils.ProducerException),
+ utils.DestinationStorage,
+ ))
+ require.NoError(t, namespaces.SetBacklogQuota(
+ namespace,
+ utils.NewBacklogQuota(-1, 60, utils.ConsumerBacklogEviction),
+ utils.MessageAge,
+ ))
+
+ quotas, err := namespaces.GetBacklogQuotaMap(namespace)
+ require.NoError(t, err)
+ require.Contains(t, quotas, utils.DestinationStorage)
+ require.Contains(t, quotas, utils.MessageAge)
+
+ require.NoError(t, namespaces.RemoveBacklogQuotaByType(namespace, utils.MessageAge))
+ quotas, err = namespaces.GetBacklogQuotaMap(namespace)
+ require.NoError(t, err)
+ require.Contains(t, quotas, utils.DestinationStorage)
+ require.NotContains(t, quotas, utils.MessageAge)
+
+ require.NoError(t, namespaces.RemoveBacklogQuota(namespace))
+ quotas, err = namespaces.GetBacklogQuotaMap(namespace)
+ require.NoError(t, err)
+ require.Empty(t, quotas)
+}
+
func TestSetTopicAutoCreation(t *testing.T) {
config := &config.Config{}
admin, err := New(config)