(pulsar-client-go) branch master updated: admin: add remove replication clusters topic API (#1470)

[rf...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

rfu 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 70d34c96 admin: add remove replication clusters topic API (#1470)
70d34c96 is described below

commit 70d34c965f3961cbd77f517af46d24b706612be6
Author: Rui Fu <fr...@users.noreply.github.com>
AuthorDate: Thu Mar 19 08:34:19 2026 +0800

    admin: add remove replication clusters topic API (#1470)
    
    * admin: add remove replication clusters topic API
    
    * admin: add topic replication clusters remove test
    
    * admin: strengthen replication clusters remove assertion
---
 pulsaradmin/pkg/admin/topic.go      | 15 ++++++++++++
 pulsaradmin/pkg/admin/topic_test.go | 47 +++++++++++++++++++++++++++++++++++++
 2 files changed, 62 insertions(+)

diff --git a/pulsaradmin/pkg/admin/topic.go b/pulsaradmin/pkg/admin/topic.go
index 65f8c747..14662ddc 100644
--- a/pulsaradmin/pkg/admin/topic.go
+++ b/pulsaradmin/pkg/admin/topic.go
@@ -832,6 +832,12 @@ type Topics interface {
 	//        list of replication cluster id
 	SetReplicationClustersWithContext(ctx context.Context, topic utils.TopicName, data []string) error
 
+	// RemoveReplicationClusters removes the replication clusters override on a topic
+	RemoveReplicationClusters(topic utils.TopicName) error
+
+	// RemoveReplicationClustersWithContext removes the replication clusters override on a topic
+	RemoveReplicationClustersWithContext(ctx context.Context, topic utils.TopicName) error
+
 	// GetSubscribeRate returns subscribe rate configuration for a topic.
 	// Returns nil if the subscribe rate is not configured at the topic level.
 	GetSubscribeRate(utils.TopicName) (*utils.SubscribeRate, error)
@@ -2032,6 +2038,15 @@ func (t *topics) GetReplicationClustersWithContext(ctx context.Context, topic ut
 	return data, err
 }
 
+func (t *topics) RemoveReplicationClusters(topic utils.TopicName) error {
+	return t.RemoveReplicationClustersWithContext(context.Background(), topic)
+}
+
+func (t *topics) RemoveReplicationClustersWithContext(ctx context.Context, topic utils.TopicName) error {
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "replication")
+	return t.pulsar.Client.DeleteWithContext(ctx, endpoint)
+}
+
 func (t *topics) GetSubscribeRate(topic utils.TopicName) (*utils.SubscribeRate, error) {
 	return t.GetSubscribeRateWithContext(context.Background(), topic)
 }
diff --git a/pulsaradmin/pkg/admin/topic_test.go b/pulsaradmin/pkg/admin/topic_test.go
index ae850105..d4f9494f 100644
--- a/pulsaradmin/pkg/admin/topic_test.go
+++ b/pulsaradmin/pkg/admin/topic_test.go
@@ -1251,6 +1251,53 @@ func TestSchemaCompatibilityStrategy(t *testing.T) {
 	)
 }
 
+func TestTopics_ReplicationClusters(t *testing.T) {
+	randomName := newTopicName()
+	topic := "persistent://public/default/" + randomName
+	cfg := &config.Config{}
+	admin, err := New(cfg)
+	assert.NoError(t, err)
+	assert.NotNil(t, admin)
+
+	topicName, err := utils.GetTopicName(topic)
+	assert.NoError(t, err)
+	err = admin.Topics().Create(*topicName, 4)
+	assert.NoError(t, err)
+
+	clusters, err := admin.Clusters().List()
+	assert.NoError(t, err)
+	assert.NotEmpty(t, clusters)
+
+	targetClusters := []string{clusters[0]}
+	err = admin.Topics().SetReplicationClusters(*topicName, targetClusters)
+	assert.NoError(t, err)
+
+	assert.Eventually(
+		t,
+		func() bool {
+			current, err := admin.Topics().GetReplicationClusters(*topicName)
+			return err == nil && len(current) == len(targetClusters) && current[0] == targetClusters[0]
+		},
+		10*time.Second,
+		100*time.Millisecond,
+	)
+
+	err = admin.Topics().RemoveReplicationClusters(*topicName)
+	assert.NoError(t, err)
+
+	// GetReplicationClusters reads topic-level explicit policy.
+	// After delete, it should no longer return the previous override.
+	assert.Eventually(
+		t,
+		func() bool {
+			current, err := admin.Topics().GetReplicationClusters(*topicName)
+			return err == nil && len(current) == 0
+		},
+		10*time.Second,
+		100*time.Millisecond,
+	)
+}
+
 func TestTopics_MaxProducers(t *testing.T) {
 	randomName := newTopicName()
 	topic := "persistent://public/default/" + randomName