Add topic command unload

pkg/ctl/topic/errors/errors_topic.go

@@ -35,17 +35,17 @@ var TopicAlreadyExistError = pulsar.Output{
 }
 
 var TopicNotFoundError = pulsar.Output{
-	Desc: "the specified topic does not found",
+	Desc: "the specified topic is not found",
 	Out:  "[✖]  code: 404 reason: Topic not found",
 }
 
 var TenantNotExistError = pulsar.Output{
-	Desc: "the tenant of the namespace is not exist",
+	Desc: "the tenant of the namespace does not exist",
 	Out:  "[✖]  code: 404 reason: Tenant does not exist",
 }
 
 var NamespaceNotExistError = pulsar.Output{
-	Desc: "the namespace is not exist",
+	Desc: "the namespace does not exist",
 	Out:  "[✖]  code: 404 reason: Namespace does not exist",
 }
 

pkg/ctl/topic/topic.go

@@ -25,6 +25,7 @@ import (
 	"github.com/streamnative/pulsarctl/pkg/ctl/topic/lookup"
 	"github.com/streamnative/pulsarctl/pkg/ctl/topic/permission"
 	"github.com/streamnative/pulsarctl/pkg/ctl/topic/stats"
+	"github.com/streamnative/pulsarctl/pkg/ctl/topic/unload"
 
 	"github.com/spf13/cobra"
 )
@@ -37,6 +38,7 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command {
 		"topic")
 
 	commands := []func(*cmdutils.VerbCmd){
+		unload.TopicUnloadCmd,
 		compact.StatusCmd,
 		crud.CreateTopicCmd,
 		crud.DeleteTopicCmd,

pkg/ctl/topic/unload/unload.go

@@ -0,0 +1,78 @@
+// 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 unload
+
+import (
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	e "github.com/streamnative/pulsarctl/pkg/ctl/topic/errors"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+)
+
+func TopicUnloadCmd(vc *cmdutils.VerbCmd) {
+	var desc pulsar.LongDescription
+	desc.CommandUsedFor = "This command is used for unloading a topic."
+	desc.CommandPermission = "This command requires super-user permissions."
+
+	var examples []pulsar.Example
+	unload := pulsar.Example{
+		Desc:    "Unload a topic (topic-name)",
+		Command: "pulsarctl topic unload (topic-name)",
+	}
+	examples = append(examples, unload)
+	desc.CommandExamples = examples
+
+	var out []pulsar.Output
+	successOut := pulsar.Output{
+		Desc: "normal output",
+		Out:  "Unload topic (topic-name) successfully",
+	}
+	out = append(out, successOut, e.ArgError, e.TopicNotFoundError)
+	out = append(out, e.TopicNameErrors...)
+	out = append(out, e.NamespaceErrors...)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"unload",
+		"Unloading a topic",
+		desc.ToString(),
+		desc.ExampleToString())
+
+	vc.SetRunFuncWithNameArg(func() error {
+		return doUnloadCmd(vc)
+	})
+}
+
+func doUnloadCmd(vc *cmdutils.VerbCmd) error {
+	// for testing
+	if vc.NameError != nil {
+		return vc.NameError
+	}
+
+	topic, err := pulsar.GetTopicName(vc.NameArg)
+	if err != nil {
+		return err
+	}
+
+	admin := cmdutils.NewPulsarClient()
+	err = admin.Topics().Unload(*topic)
+	if err == nil {
+		vc.Command.Printf("Unload topic %s successfully/n", topic.String())
+	}
+
+	return err
+}

pkg/ctl/topic/unload/unload_test.go

@@ -0,0 +1,52 @@
+// 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 unload
+
+import (
+	"testing"
+
+	"github.com/streamnative/pulsarctl/pkg/ctl/topic/crud"
+	"github.com/streamnative/pulsarctl/pkg/ctl/topic/test"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestUnloadCmd(t *testing.T) {
+	args := []string{"create", "test-unload-topic", "0"}
+	_, execErr, _, _ := test.TestTopicCommands(crud.CreateTopicCmd, args)
+	assert.Nil(t, execErr)
+
+	args = []string{"unload", "test-unload-topic"}
+	out, execErr, _, _ := test.TestTopicCommands(TopicUnloadCmd, args)
+	assert.Nil(t, execErr)
+	assert.Equal(t, "Unload topic persistent://public/default/test-unload-topic successfully/n", out.String())
+}
+
+func TestUnloadArgError(t *testing.T) {
+	args := []string{"unload"}
+	_, _, nameErr, _ := test.TestTopicCommands(TopicUnloadCmd, args)
+	assert.NotNil(t, nameErr)
+	assert.Equal(t, "only one argument is allowed to be used as a name", nameErr.Error())
+}
+
+func TestUnloadNonExistingTopic(t *testing.T) {
+	args := []string{"unload", "test-unload-non-existing-topic"}
+	_, execErr, _, _ := test.TestTopicCommands(TopicUnloadCmd, args)
+	assert.NotNil(t, execErr)
+	assert.Equal(t, "code: 404 reason: Topic not found", execErr.Error())
+}

pkg/pulsar/topic.go

@@ -38,6 +38,7 @@ type Topics interface {
 	GetStats(TopicName) (TopicStats, error)
 	GetInternalStats(TopicName) (PersistentTopicInternalStats, error)
 	GetPartitionedStats(TopicName, bool) (PartitionedTopicStats, error)
+	Unload(TopicName) error
 	Compact(TopicName) error
 	CompactStatus(TopicName) (LongRunningProcessStatus, error)
 }
@@ -207,6 +208,10 @@ func (t *topics) GetPartitionedStats(topic TopicName, perPartition bool) (Partit
 	return stats, err
 }
 
+func (t *topics) Unload(topic TopicName) error {
+	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "unload")
+	return t.client.put(endpoint, "")
+}
 func (t *topics) Compact(topic TopicName) error {
 	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "compaction")
 	return t.client.put(endpoint, "")