diff --git a/pkg/ctl/topic/stop/teminate.go b/pkg/ctl/topic/stop/teminate.go new file mode 100644 index 000000000..ccc593b22 --- /dev/null +++ b/pkg/ctl/topic/stop/teminate.go @@ -0,0 +1,110 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package stop + +import ( + "github.com/streamnative/pulsarctl/pkg/cmdutils" + e "github.com/streamnative/pulsarctl/pkg/ctl/topic/errors" + "github.com/streamnative/pulsarctl/pkg/pulsar" + + "github.com/pkg/errors" + "github.com/spf13/pflag" +) + +func TopicTerminateCmd(vc *cmdutils.VerbCmd) { + var desc pulsar.LongDescription + desc.CommandUsedFor = "This command is used for terminating a non-partitioned topic or a partition of " + + "a partitioned topic. Upon termination, no more messages are allowed to published to it." + desc.CommandPermission = "This command requires tenant admin permissions." + + var examples []pulsar.Example + terminate := pulsar.Example{ + Desc: "Terminate a non-partitioned topic (topic-name)", + Command: "pulsarctl topic terminate (topic-name)", + } + + terminateWithPartition := pulsar.Example{ + Desc: "Terminate a partition of a partitioned topic", + Command: "pulsarctl topic terminate --partition (partition) (topic-name)", + } + examples = append(examples, terminate, terminateWithPartition) + desc.CommandExamples = examples + + var out []pulsar.Output + successOut := pulsar.Output{ + Desc: "normal output", + Out: "Topic (topic-name) is successfully terminated at (message-id)", + } + + partitionError := pulsar.Output{ + Desc: "the specified topic is a partitioned topic", + Out: "[✖] code: 405 reason: Termination of a partitioned topic is not allowed", + } + out = append(out, successOut, e.ArgError, e.TopicNotFoundError, partitionError) + out = append(out, e.TopicNameErrors...) + out = append(out, e.NamespaceErrors...) + desc.CommandOutput = out + + vc.SetDescription( + "terminate", + "Terminate a non-partitioned topic", + desc.ToString(), + desc.ExampleToString()) + + var partition int + + vc.SetRunFuncWithNameArg(func() error { + return doTerminate(vc, partition) + }) + + vc.FlagSetGroup.InFlagSet("Terminate", func(set *pflag.FlagSet) { + set.IntVarP(&partition, "partition", "p", -1, + "The partitioned topic index value") + }) +} + +func doTerminate(vc *cmdutils.VerbCmd, partition int) error { + // for testing + if vc.NameError != nil { + return vc.NameError + } + + topic, err := pulsar.GetTopicName(vc.NameArg) + if err != nil { + return err + } + + if !topic.IsPersistent() { + return errors.New("only support terminating a persistent topic") + } + + if partition >= 0 { + topic, err = topic.GetPartition(partition) + if err != nil { + return err + } + } + + admin := cmdutils.NewPulsarClient() + messageID, err := admin.Topics().Terminate(*topic) + if err == nil { + vc.Command.Printf("Topic %s is successfully terminated at %+v", topic.String(), messageID) + } + + return err +} diff --git a/pkg/ctl/topic/stop/teminate_test.go b/pkg/ctl/topic/stop/teminate_test.go new file mode 100644 index 000000000..e9ea17170 --- /dev/null +++ b/pkg/ctl/topic/stop/teminate_test.go @@ -0,0 +1,65 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package stop + +import ( + "strings" + "testing" + + "github.com/streamnative/pulsarctl/pkg/ctl/topic/crud" + "github.com/streamnative/pulsarctl/pkg/ctl/topic/test" + + "github.com/stretchr/testify/assert" +) + +func TestTerminateCmd(t *testing.T) { + args := []string{"create", "test-terminate-topic", "0"} + _, execErr, _, _ := test.TestTopicCommands(crud.CreateTopicCmd, args) + assert.Nil(t, execErr) + + args = []string{"terminate", "test-terminate-topic"} + out, execErr, _, _ := test.TestTopicCommands(TopicTerminateCmd, args) + assert.Nil(t, execErr) + assert.True(t, strings.HasPrefix(out.String(), + "Topic persistent://public/default/test-terminate-topic is successfully terminated at")) +} + +func TestTerminatePartitionedTopicError(t *testing.T) { + args := []string{"create", "test-terminate-partitioned-topic", "2"} + _, execErr, _, _ := test.TestTopicCommands(crud.CreateTopicCmd, args) + assert.Nil(t, execErr) + + args = []string{"terminate", "test-terminate-partitioned-topic"} + _, execErr, _, _ = test.TestTopicCommands(TopicTerminateCmd, args) + assert.NotNil(t, execErr) + assert.Equal(t, "code: 405 reason: Termination of a partitioned topic is not allowed", execErr.Error()) +} + +func TestTerminateArgError(t *testing.T) { + args := []string{"terminate"} + _, _, nameErr, _ := test.TestTopicCommands(TopicTerminateCmd, args) + assert.NotNil(t, nameErr) + assert.Equal(t, "only one argument is allowed to be used as a name", nameErr.Error()) +} + +func TestTerminateNonExistingTopic(t *testing.T) { + args := []string{"terminate", "non-existing-topic"} + _, execErr, _, _ := test.TestTopicCommands(TopicTerminateCmd, args) + assert.NotNil(t, execErr) + assert.Equal(t, "code: 404 reason: Topic not found", execErr.Error()) +} diff --git a/pkg/ctl/topic/topic.go b/pkg/ctl/topic/topic.go index 650b090cf..ac8da0062 100644 --- a/pkg/ctl/topic/topic.go +++ b/pkg/ctl/topic/topic.go @@ -26,6 +26,7 @@ import ( "github.com/streamnative/pulsarctl/pkg/ctl/topic/offload" "github.com/streamnative/pulsarctl/pkg/ctl/topic/permission" "github.com/streamnative/pulsarctl/pkg/ctl/topic/stats" + "github.com/streamnative/pulsarctl/pkg/ctl/topic/stop" "github.com/streamnative/pulsarctl/pkg/ctl/topic/unload" "github.com/spf13/cobra" @@ -39,6 +40,7 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { "topic") commands := []func(*cmdutils.VerbCmd){ + stop.TopicTerminateCmd, offload.TopicOffloadCmd, offload.TopicOffloadStatusCmd, unload.TopicUnloadCmd, diff --git a/pkg/pulsar/admin.go b/pkg/pulsar/admin.go index f36cbb3d8..4fdc22c3d 100644 --- a/pkg/pulsar/admin.go +++ b/pkg/pulsar/admin.go @@ -274,6 +274,10 @@ func (c *client) deleteWithQueryParams(endpoint string, obj interface{}, params } func (c *client) post(endpoint string, in interface{}) error { + return c.postWithObj(endpoint, in, nil) +} + +func (c *client) postWithObj(endpoint string, in, obj interface{}) error { req, err := c.newRequest(http.MethodPost, endpoint) if err != nil { return err @@ -286,6 +290,11 @@ func (c *client) post(endpoint string, in interface{}) error { return err } defer safeRespClose(resp) + if obj != nil { + if err := decodeJSONBody(resp, &obj); err != nil { + return err + } + } return nil } diff --git a/pkg/pulsar/topic.go b/pkg/pulsar/topic.go index 05ee0be15..ec00a4353 100644 --- a/pkg/pulsar/topic.go +++ b/pkg/pulsar/topic.go @@ -38,6 +38,7 @@ type Topics interface { GetStats(TopicName) (TopicStats, error) GetInternalStats(TopicName) (PersistentTopicInternalStats, error) GetPartitionedStats(TopicName, bool) (PartitionedTopicStats, error) + Terminate(TopicName) (MessageID, error) Offload(TopicName, MessageID) error OffloadStatus(TopicName) (OffloadProcessStatus, error) Unload(TopicName) error @@ -210,6 +211,13 @@ func (t *topics) GetPartitionedStats(topic TopicName, perPartition bool) (Partit return stats, err } +func (t *topics) Terminate(topic TopicName) (MessageID, error) { + endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "terminate") + var messageID MessageID + err := t.client.postWithObj(endpoint, "", &messageID) + return messageID, err +} + func (t *topics) Offload(topic TopicName, messageID MessageID) error { endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "offload") return t.client.put(endpoint, messageID)