Add topic command terminate

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
+}

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())
+}

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,

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
 }

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)