Skip to content

Commit 08e3f3d

Browse files
feat: add set-replication-clusters command for topics (#1973)
1 parent da10b8f commit 08e3f3d

3 files changed

Lines changed: 156 additions & 0 deletions

File tree

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
package topic
19+
20+
import (
21+
"github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
22+
"github.com/pkg/errors"
23+
"github.com/spf13/cobra"
24+
"github.com/spf13/pflag"
25+
26+
"github.com/streamnative/pulsarctl/pkg/cmdutils"
27+
)
28+
29+
func SetReplicationClustersCmd(vc *cmdutils.VerbCmd) {
30+
desc := cmdutils.LongDescription{}
31+
desc.CommandUsedFor = "Set the replication clusters for a topic"
32+
desc.CommandPermission = "This command requires tenant admin permissions."
33+
34+
var examples []cmdutils.Example
35+
setReplication := cmdutils.Example{
36+
Desc: "Set the replication clusters for a topic",
37+
Command: "pulsarctl topics set-replication-clusters tenant/namespace/topic --clusters cluster1,cluster2",
38+
}
39+
examples = append(examples, setReplication)
40+
desc.CommandExamples = examples
41+
42+
var out []cmdutils.Output
43+
successOut := cmdutils.Output{
44+
Desc: "normal output",
45+
Out: "Set the replication clusters for [topic] successfully",
46+
}
47+
48+
noTopicName := cmdutils.Output{
49+
Desc: "you must specify a tenant/namespace/topic name, please check if the tenant/namespace/topic name is provided",
50+
Out: "[✖] the topic name is not specified or the topic name is specified more than one",
51+
}
52+
53+
tenantNotExistError := cmdutils.Output{
54+
Desc: "the tenant does not exist",
55+
Out: "[✖] code: 404 reason: Tenant does not exist",
56+
}
57+
58+
nsNotExistError := cmdutils.Output{
59+
Desc: "the namespace does not exist",
60+
Out: "[✖] code: 404 reason: Namespace (tenant/namespace) does not exist",
61+
}
62+
63+
out = append(out, successOut, noTopicName, tenantNotExistError, nsNotExistError)
64+
desc.CommandOutput = out
65+
66+
vc.SetDescription(
67+
"set-replication-clusters",
68+
"Set the replication clusters for a topic",
69+
desc.ToString(),
70+
desc.ExampleToString(),
71+
"set-replication-clusters",
72+
)
73+
74+
var clusters []string
75+
76+
vc.FlagSetGroup.InFlagSet("Set replication clusters", func(flagSet *pflag.FlagSet) {
77+
flagSet.StringSliceVarP(&clusters, "clusters", "c", nil,
78+
"Replication cluster ids.")
79+
_ = cobra.MarkFlagRequired(flagSet, "clusters")
80+
})
81+
vc.EnableOutputFlagSet()
82+
83+
vc.SetRunFuncWithNameArg(func() error {
84+
return doSetReplicationClusters(vc, clusters)
85+
}, "the topic name is not specified or the topic name is specified more than one")
86+
}
87+
88+
func doSetReplicationClusters(vc *cmdutils.VerbCmd, clusters []string) error {
89+
topic := vc.NameArg
90+
91+
if len(clusters) == 0 {
92+
return errors.New("clusters cannot be empty")
93+
}
94+
95+
admin := cmdutils.NewPulsarClient()
96+
topicName, err := utils.GetTopicName(topic)
97+
if err != nil {
98+
return err
99+
}
100+
101+
err = admin.Topics().SetReplicationClusters(*topicName, clusters)
102+
if err == nil {
103+
vc.Command.Printf("Set the replication clusters successfully on [%s]\n", topic)
104+
}
105+
106+
return err
107+
}
Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
package topic
19+
20+
import (
21+
"fmt"
22+
"testing"
23+
24+
"github.com/onsi/gomega"
25+
"github.com/streamnative/pulsarctl/pkg/test"
26+
)
27+
28+
func TestSetReplicationClustersCmd(t *testing.T) {
29+
g := gomega.NewWithT(t)
30+
31+
topic := fmt.Sprintf("test-replication-clusters-topic-%s", test.RandomSuffix())
32+
33+
args := []string{"create", topic, "0"}
34+
_, execErr, _, _ := TestTopicCommands(CreateTopicCmd, args)
35+
g.Expect(execErr).Should(gomega.BeNil())
36+
37+
args = []string{"set-replication-clusters", topic, "--clusters", "standalone"}
38+
out, execErr, nameErr, cmdErr := TestTopicCommands(SetReplicationClustersCmd, args)
39+
g.Expect(execErr).Should(gomega.BeNil())
40+
g.Expect(nameErr).Should(gomega.BeNil())
41+
g.Expect(cmdErr).Should(gomega.BeNil())
42+
g.Expect(out).ShouldNot(gomega.BeNil())
43+
g.Expect(out.String()).ShouldNot(gomega.BeEmpty())
44+
45+
// Since there is no get-replication-clusters command in this PR, we only test the set command success.
46+
// In a real scenario, we might want to verify using the client or adding a get command.
47+
// The set command output verification implies the call was successful.
48+
}

pkg/ctl/topic/topic.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,7 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command {
9696
GetInactiveTopicCmd,
9797
SetInactiveTopicCmd,
9898
RemoveInactiveTopicCmd,
99+
SetReplicationClustersCmd,
99100
}
100101

101102
cmdutils.AddVerbCmds(flagGrouping, resourceCmd, commands...)

0 commit comments

Comments
 (0)