From c8f01e3d94b36ed80e00e6bfea82ba9b00674369 Mon Sep 17 00:00:00 2001 From: Yong Zhang Date: Tue, 24 Sep 2019 17:20:38 +0800 Subject: [PATCH 1/4] Add topic command `terminate` *Motivation* - Add command `terminate` --- pkg/ctl/topic/errors/errors_topic.go | 5 ++ pkg/ctl/topic/terminate/teminate.go | 87 ++++++++++++++++++++++++ pkg/ctl/topic/terminate/teminate_test.go | 64 +++++++++++++++++ pkg/pulsar/topic.go | 10 ++- pkg/pulsar/topic_name.go | 4 ++ 5 files changed, 169 insertions(+), 1 deletion(-) create mode 100644 pkg/ctl/topic/terminate/teminate.go create mode 100644 pkg/ctl/topic/terminate/teminate_test.go diff --git a/pkg/ctl/topic/errors/errors_topic.go b/pkg/ctl/topic/errors/errors_topic.go index b52225a04..cc74e53bd 100644 --- a/pkg/ctl/topic/errors/errors_topic.go +++ b/pkg/ctl/topic/errors/errors_topic.go @@ -34,6 +34,11 @@ var TopicAlreadyExistError = Output{ Out: "[✖] code: 409 reason: Partitioned topic already exists", } +var TopicNotFoundError = Output{ + Desc: "the specified topic does not found", + Out: "[✖] code: 404 reason: Topic not found", +} + var TenantNotExistError = Output{ Desc: "the tenant of the namespace is not exist", Out: "[✖] code: 404 reason: Tenant does not exist", diff --git a/pkg/ctl/topic/terminate/teminate.go b/pkg/ctl/topic/terminate/teminate.go new file mode 100644 index 000000000..d7c3486a1 --- /dev/null +++ b/pkg/ctl/topic/terminate/teminate.go @@ -0,0 +1,87 @@ +// 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 teminate + +import ( + "github.com/pkg/errors" + "github.com/streamnative/pulsarctl/pkg/cmdutils" + . "github.com/streamnative/pulsarctl/pkg/ctl/topic/errors" + . "github.com/streamnative/pulsarctl/pkg/pulsar" +) + +func TerminateCmd(vc *cmdutils.VerbCmd) { + var desc LongDescription + desc.CommandUsedFor = "This command is used for terminating a non-partitioned topic and not allow any more " + + "messages to be published." + desc.CommandPermission = "This command requires tenant admin permissions." + + var examples []Example + terminate := Example{ + Desc: "Terminate a non-partitioned topic and not allow any messages to be published", + Command: "pulsarctl topic terminate ", + } + desc.CommandExamples = append(examples, terminate) + + var out []Output + successOut := Output{ + Desc: "normal output", + Out: "Topic successfully terminated at ", + } + + partitionError := Output{ + Desc: "the specified is a partitioned topic", + Out: "[✖] code: 405 reason: Termination of a partitioned topic is not allowed", + } + out = append(out, successOut, ArgError, TopicNotFoundError, partitionError) + out = append(out, TopicNameErrors...) + out = append(out, NamespaceErrors...) + desc.CommandOutput = out + + vc.SetDescription( + "terminate", + "Terminate a non-partitioned topic", + desc.ToString()) + + vc.SetRunFuncWithNameArg(func() error { + return doTerminate(vc) + }) +} + +func doTerminate(vc *cmdutils.VerbCmd) error { + // for testing + if vc.NameError != nil { + return vc.NameError + } + + topic, err := GetTopicName(vc.NameArg) + if err != nil { + return err + } + + if !topic.IsPersistent() { + return errors.New("need to provide a persistent topic") + } + + admin := cmdutils.NewPulsarClient() + messageId, err :=admin.Topics().Terminate(*topic) + if err == nil { + vc.Command.Printf("Topic %s successfully terminated at %+v", topic.String(), messageId) + } + + return err +} diff --git a/pkg/ctl/topic/terminate/teminate_test.go b/pkg/ctl/topic/terminate/teminate_test.go new file mode 100644 index 000000000..296ee2f0f --- /dev/null +++ b/pkg/ctl/topic/terminate/teminate_test.go @@ -0,0 +1,64 @@ +// 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 teminate + +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, _, _ := TestTopicCommands(CreateTopicCmd, args) + assert.Nil(t, execErr) + + args = []string{"terminate", "test-terminate-topic"} + out, execErr, _, _ := TestTopicCommands(TerminateCmd, args) + assert.Nil(t, execErr) + assert.True(t, strings.HasPrefix(out.String(), + "Topic persistent://public/default/test-terminate-topic successfully terminated at")) +} + +func TestTerminatePartitionedTopicError(t *testing.T) { + args := []string{"create", "test-terminate-partitioned-topic", "2"} + _, execErr, _, _ := TestTopicCommands(CreateTopicCmd, args) + assert.Nil(t, execErr) + + args = []string{"terminate", "test-terminate-partitioned-topic"} + _, execErr, _, _ = TestTopicCommands(TerminateCmd, 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, _ := TestTopicCommands(TerminateCmd, 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, _, _ := TestTopicCommands(TerminateCmd, 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 0b1260a50..726363a96 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) } type topics struct { @@ -137,7 +138,7 @@ func (t *topics) GetInternalInfo(topic TopicName) (ManagedLedgerInfo, error) { endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "internal-info") var info ManagedLedgerInfo err := t.client.get(endpoint, &info) - return info, err + return info, err } func (t *topics) GetPermissions(topic TopicName) (map[string][]AuthAction, error) { @@ -204,3 +205,10 @@ func (t *topics) GetPartitionedStats(topic TopicName, perPartition bool) (Partit _, err := t.client.getWithQueryParams(endpoint, &stats, params, true) 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.post(endpoint, "", &messageId) + return messageId, err +} diff --git a/pkg/pulsar/topic_name.go b/pkg/pulsar/topic_name.go index 5c27eb150..40a37bd30 100644 --- a/pkg/pulsar/topic_name.go +++ b/pkg/pulsar/topic_name.go @@ -104,6 +104,10 @@ func (t *TopicName) GetDomain() TopicDomain { return t.domain } +func (t *TopicName) IsPersistent() bool { + return t.domain == persistent +} + func (t *TopicName) GetRestPath() string { return fmt.Sprintf("%s/%s/%s/%s", t.domain, t.tenant, t.namespace, t.GetEncodedTopic()) } From 12dbc9115738484eec18be8ebf6b26d8928225a4 Mon Sep 17 00:00:00 2001 From: Yong Zhang Date: Tue, 24 Sep 2019 17:22:36 +0800 Subject: [PATCH 2/4] Add to command group --- pkg/ctl/topic/topic.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pkg/ctl/topic/topic.go b/pkg/ctl/topic/topic.go index 723a7301a..f70cc1a0e 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/terminate" ) func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { @@ -49,6 +50,7 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { GetLastMessageIdCmd, GetStatsCmd, GetInternalStatsCmd, + teminate.TerminateCmd, } cmdutils.AddVerbCmds(flagGrouping, resourceCmd, commands...) From d380a6e022ca983224082824bb8f9bce7ae44de5 Mon Sep 17 00:00:00 2001 From: Yong Zhang Date: Thu, 26 Sep 2019 16:33:41 +0800 Subject: [PATCH 3/4] Format file --- pkg/ctl/topic/terminate/teminate.go | 40 +++++++++++++----------- pkg/ctl/topic/terminate/teminate_test.go | 23 +++++++------- 2 files changed, 33 insertions(+), 30 deletions(-) diff --git a/pkg/ctl/topic/terminate/teminate.go b/pkg/ctl/topic/terminate/teminate.go index d7c3486a1..1d28b7fc0 100644 --- a/pkg/ctl/topic/terminate/teminate.go +++ b/pkg/ctl/topic/terminate/teminate.go @@ -18,38 +18,40 @@ package teminate import ( - "github.com/pkg/errors" "github.com/streamnative/pulsarctl/pkg/cmdutils" - . "github.com/streamnative/pulsarctl/pkg/ctl/topic/errors" - . "github.com/streamnative/pulsarctl/pkg/pulsar" + e "github.com/streamnative/pulsarctl/pkg/ctl/topic/errors" + "github.com/streamnative/pulsarctl/pkg/pulsar" + + "github.com/pkg/errors" ) func TerminateCmd(vc *cmdutils.VerbCmd) { - var desc LongDescription - desc.CommandUsedFor = "This command is used for terminating a non-partitioned topic and not allow any more " + + var desc pulsar.LongDescription + desc.CommandUsedFor = "This command is used for terminating a non-partitioned topic and not allow any more " + "messages to be published." desc.CommandPermission = "This command requires tenant admin permissions." - var examples []Example - terminate := Example{ - Desc: "Terminate a non-partitioned topic and not allow any messages to be published", + var examples []pulsar.Example + terminate := pulsar.Example{ + Desc: "Terminate a non-partitioned topic and not allow any messages to be published", Command: "pulsarctl topic terminate ", } - desc.CommandExamples = append(examples, terminate) + examples = append(examples, terminate) + desc.CommandExamples = examples - var out []Output - successOut := Output{ + var out []pulsar.Output + successOut := pulsar.Output{ Desc: "normal output", - Out: "Topic successfully terminated at ", + Out: "Topic successfully terminated at ", } - partitionError := Output{ + partitionError := pulsar.Output{ Desc: "the specified is a partitioned topic", - Out: "[✖] code: 405 reason: Termination of a partitioned topic is not allowed", + Out: "[✖] code: 405 reason: Termination of a partitioned topic is not allowed", } - out = append(out, successOut, ArgError, TopicNotFoundError, partitionError) - out = append(out, TopicNameErrors...) - out = append(out, NamespaceErrors...) + out = append(out, successOut, e.ArgError, e.TopicNotFoundError, partitionError) + out = append(out, e.TopicNameErrors...) + out = append(out, e.NamespaceErrors...) desc.CommandOutput = out vc.SetDescription( @@ -68,7 +70,7 @@ func doTerminate(vc *cmdutils.VerbCmd) error { return vc.NameError } - topic, err := GetTopicName(vc.NameArg) + topic, err := pulsar.GetTopicName(vc.NameArg) if err != nil { return err } @@ -78,7 +80,7 @@ func doTerminate(vc *cmdutils.VerbCmd) error { } admin := cmdutils.NewPulsarClient() - messageId, err :=admin.Topics().Terminate(*topic) + messageId, err := admin.Topics().Terminate(*topic) if err == nil { vc.Command.Printf("Topic %s successfully terminated at %+v", topic.String(), messageId) } diff --git a/pkg/ctl/topic/terminate/teminate_test.go b/pkg/ctl/topic/terminate/teminate_test.go index 296ee2f0f..9effa86e5 100644 --- a/pkg/ctl/topic/terminate/teminate_test.go +++ b/pkg/ctl/topic/terminate/teminate_test.go @@ -21,44 +21,45 @@ import ( "strings" "testing" - . "github.com/streamnative/pulsarctl/pkg/ctl/topic/crud" - . "github.com/streamnative/pulsarctl/pkg/ctl/topic/test" + "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, _, _ := TestTopicCommands(CreateTopicCmd, args) + _, execErr, _, _ := test.TestTopicCommands(crud.CreateTopicCmd, args) assert.Nil(t, execErr) args = []string{"terminate", "test-terminate-topic"} - out, execErr, _, _ := TestTopicCommands(TerminateCmd, args) + out, execErr, _, _ := test.TestTopicCommands(TerminateCmd, args) assert.Nil(t, execErr) assert.True(t, strings.HasPrefix(out.String(), "Topic persistent://public/default/test-terminate-topic successfully terminated at")) } -func TestTerminatePartitionedTopicError(t *testing.T) { - args := []string{"create", "test-terminate-partitioned-topic", "2"} - _, execErr, _, _ := TestTopicCommands(CreateTopicCmd, args) +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, _, _ = TestTopicCommands(TerminateCmd, args) + _, execErr, _, _ = test.TestTopicCommands(TerminateCmd, 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, _ := TestTopicCommands(TerminateCmd, args) + _, _, nameErr, _ := test.TestTopicCommands(TerminateCmd, 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) { +func TestTerminateNonExistingTopic(t *testing.T) { args := []string{"terminate", "non-existing-topic"} - _, execErr, _, _ := TestTopicCommands(TerminateCmd, args) + _, execErr, _, _ := test.TestTopicCommands(TerminateCmd, args) assert.NotNil(t, execErr) assert.Equal(t, "code: 404 reason: Topic not found", execErr.Error()) } From b50fbdd7ac15b54a1f67e25681ec90688a38a58d Mon Sep 17 00:00:00 2001 From: Yong Zhang Date: Tue, 8 Oct 2019 14:08:09 +0800 Subject: [PATCH 4/4] Address comments --- pkg/ctl/topic/errors/errors_topic.go | 6 +-- pkg/ctl/topic/{terminate => stop}/teminate.go | 51 +++++++++++++------ .../{terminate => stop}/teminate_test.go | 12 ++--- pkg/ctl/topic/topic.go | 4 +- pkg/pulsar/admin.go | 9 ++++ pkg/pulsar/topic.go | 10 ++-- 6 files changed, 61 insertions(+), 31 deletions(-) rename pkg/ctl/topic/{terminate => stop}/teminate.go (60%) rename pkg/ctl/topic/{terminate => stop}/teminate_test.go (84%) diff --git a/pkg/ctl/topic/errors/errors_topic.go b/pkg/ctl/topic/errors/errors_topic.go index 410fdd5b6..257239b4e 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 does not exist", 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/terminate/teminate.go b/pkg/ctl/topic/stop/teminate.go similarity index 60% rename from pkg/ctl/topic/terminate/teminate.go rename to pkg/ctl/topic/stop/teminate.go index 1d28b7fc0..ccc593b22 100644 --- a/pkg/ctl/topic/terminate/teminate.go +++ b/pkg/ctl/topic/stop/teminate.go @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -package teminate +package stop import ( "github.com/streamnative/pulsarctl/pkg/cmdutils" @@ -23,30 +23,36 @@ import ( "github.com/streamnative/pulsarctl/pkg/pulsar" "github.com/pkg/errors" + "github.com/spf13/pflag" ) -func TerminateCmd(vc *cmdutils.VerbCmd) { +func TopicTerminateCmd(vc *cmdutils.VerbCmd) { var desc pulsar.LongDescription - desc.CommandUsedFor = "This command is used for terminating a non-partitioned topic and not allow any more " + - "messages to be published." + 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 and not allow any messages to be published", - Command: "pulsarctl topic terminate ", + Desc: "Terminate a non-partitioned topic (topic-name)", + Command: "pulsarctl topic terminate (topic-name)", } - examples = append(examples, terminate) + + 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 successfully terminated at ", + Out: "Topic (topic-name) is successfully terminated at (message-id)", } partitionError := pulsar.Output{ - Desc: "the specified is a partitioned topic", + 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) @@ -57,14 +63,22 @@ func TerminateCmd(vc *cmdutils.VerbCmd) { vc.SetDescription( "terminate", "Terminate a non-partitioned topic", - desc.ToString()) + desc.ToString(), + desc.ExampleToString()) + + var partition int vc.SetRunFuncWithNameArg(func() error { - return doTerminate(vc) + 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) error { +func doTerminate(vc *cmdutils.VerbCmd, partition int) error { // for testing if vc.NameError != nil { return vc.NameError @@ -76,13 +90,20 @@ func doTerminate(vc *cmdutils.VerbCmd) error { } if !topic.IsPersistent() { - return errors.New("need to provide a persistent topic") + 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) + messageID, err := admin.Topics().Terminate(*topic) if err == nil { - vc.Command.Printf("Topic %s successfully terminated at %+v", topic.String(), messageId) + vc.Command.Printf("Topic %s is successfully terminated at %+v", topic.String(), messageID) } return err diff --git a/pkg/ctl/topic/terminate/teminate_test.go b/pkg/ctl/topic/stop/teminate_test.go similarity index 84% rename from pkg/ctl/topic/terminate/teminate_test.go rename to pkg/ctl/topic/stop/teminate_test.go index 9effa86e5..e9ea17170 100644 --- a/pkg/ctl/topic/terminate/teminate_test.go +++ b/pkg/ctl/topic/stop/teminate_test.go @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -package teminate +package stop import ( "strings" @@ -33,10 +33,10 @@ func TestTerminateCmd(t *testing.T) { assert.Nil(t, execErr) args = []string{"terminate", "test-terminate-topic"} - out, execErr, _, _ := test.TestTopicCommands(TerminateCmd, args) + out, execErr, _, _ := test.TestTopicCommands(TopicTerminateCmd, args) assert.Nil(t, execErr) assert.True(t, strings.HasPrefix(out.String(), - "Topic persistent://public/default/test-terminate-topic successfully terminated at")) + "Topic persistent://public/default/test-terminate-topic is successfully terminated at")) } func TestTerminatePartitionedTopicError(t *testing.T) { @@ -45,21 +45,21 @@ func TestTerminatePartitionedTopicError(t *testing.T) { assert.Nil(t, execErr) args = []string{"terminate", "test-terminate-partitioned-topic"} - _, execErr, _, _ = test.TestTopicCommands(TerminateCmd, args) + _, 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(TerminateCmd, args) + _, _, 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(TerminateCmd, args) + _, 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 01610741c..9d72f60e4 100644 --- a/pkg/ctl/topic/topic.go +++ b/pkg/ctl/topic/topic.go @@ -19,12 +19,12 @@ package topic import ( "github.com/streamnative/pulsarctl/pkg/cmdutils" - "github.com/streamnative/pulsarctl/pkg/ctl/topic/terminate" "github.com/streamnative/pulsarctl/pkg/ctl/topic/crud" "github.com/streamnative/pulsarctl/pkg/ctl/topic/info" "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/stop" "github.com/spf13/cobra" ) @@ -37,7 +37,7 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { "topic") commands := []func(*cmdutils.VerbCmd){ - teminate.TerminateCmd, + stop.TopicTerminateCmd, crud.CreateTopicCmd, crud.DeleteTopicCmd, crud.GetTopicCmd, 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 446ae504f..6308620d0 100644 --- a/pkg/pulsar/topic.go +++ b/pkg/pulsar/topic.go @@ -38,7 +38,7 @@ type Topics interface { GetStats(TopicName) (TopicStats, error) GetInternalStats(TopicName) (PersistentTopicInternalStats, error) GetPartitionedStats(TopicName, bool) (PartitionedTopicStats, error) - Terminate(TopicName) (MessageId, error) + Terminate(TopicName) (MessageID, error) } type topics struct { @@ -206,9 +206,9 @@ func (t *topics) GetPartitionedStats(topic TopicName, perPartition bool) (Partit return stats, err } -func (t *topics) Terminate(topic TopicName) (MessageId, error) { +func (t *topics) Terminate(topic TopicName) (MessageID, error) { endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "terminate") - var messageId MessageId - err := t.client.post(endpoint, "", &messageId) - return messageId, err + var messageID MessageID + err := t.client.postWithObj(endpoint, "", &messageID) + return messageID, err }