diff --git a/pkg/ctl/topic/errors/errors_topic.go b/pkg/ctl/topic/errors/errors_topic.go index 410fdd5b6..982c9d6b2 100644 --- a/pkg/ctl/topic/errors/errors_topic.go +++ b/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", } diff --git a/pkg/ctl/topic/topic.go b/pkg/ctl/topic/topic.go index afc7e8035..780951e0c 100644 --- a/pkg/ctl/topic/topic.go +++ b/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, diff --git a/pkg/ctl/topic/unload/unload.go b/pkg/ctl/topic/unload/unload.go new file mode 100644 index 000000000..cd5ac70fc --- /dev/null +++ b/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 +} diff --git a/pkg/ctl/topic/unload/unload_test.go b/pkg/ctl/topic/unload/unload_test.go new file mode 100644 index 000000000..823a06d14 --- /dev/null +++ b/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()) +} diff --git a/pkg/pulsar/topic.go b/pkg/pulsar/topic.go index faac080a4..6b9c10e92 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) + 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, "")