Add command topic Persistence Policy. (#246)

pkg/ctl/topic/get_persistence.go

@@ -0,0 +1,82 @@
+// 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 topic
+
+import (
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+
+	"github.com/streamnative/pulsarctl/pkg/pulsar/utils"
+)
+
+func GetPersistenceCmd(vc *cmdutils.VerbCmd) {
+	var desc cmdutils.LongDescription
+	desc.CommandUsedFor = "Get persistence for a topic"
+	desc.CommandPermission = "This command requires namespace admin persistence."
+
+	var examples []cmdutils.Example
+	get := cmdutils.Example{
+		Desc:    "Get persistence for a topic",
+		Command: "pulsarctl topic get-persistence (topic-name)",
+	}
+	examples = append(examples, get)
+	desc.CommandExamples = examples
+
+	var out []cmdutils.Output
+	successOut := cmdutils.Output{
+		Desc: "normal output",
+		Out:  "(persistence-value)",
+	}
+	out = append(out, successOut, ArgError)
+	out = append(out, TopicNameErrors...)
+	out = append(out, NamespaceErrors...)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"get-persistence",
+		"Get the persistence of a topic",
+		desc.ToString(),
+		desc.ExampleToString(),
+	)
+
+	vc.SetRunFuncWithNameArg(func() error {
+		return doGetPersistence(vc)
+	}, "the topic name is not specified or the topic name is specified more than one")
+
+	vc.EnableOutputFlagSet()
+}
+
+func doGetPersistence(vc *cmdutils.VerbCmd) error {
+	// for testing
+	if vc.NameError != nil {
+		return vc.NameError
+	}
+
+	topic, err := utils.GetTopicName(vc.NameArg)
+	if err != nil {
+		return err
+	}
+
+	admin := cmdutils.NewPulsarClient()
+	persistence, err := admin.Topics().GetPersistence(*topic)
+	if err == nil {
+		oc := cmdutils.NewOutputContent().WithObject(persistence)
+		err = vc.OutputConfig.WriteOutput(vc.Command.OutOrStdout(), oc)
+	}
+
+	return err
+}

pkg/ctl/topic/persistence_test.go

@@ -0,0 +1,77 @@
+// 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 topic
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/streamnative/pulsarctl/pkg/pulsar/utils"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPersistence(t *testing.T) {
+	topicName := "persistent://public/default/test-persistence-topic-10"
+	args := []string{"create", topicName, "1"}
+	_, execErr, _, _ := TestTopicCommands(CreateTopicCmd, args)
+	assert.Nil(t, execErr)
+
+	setArgs := []string{"set-persistence", topicName, "-e", "5", "-w", "4", "-a", "3", "-r", "2.2"}
+	setOut, execErr, _, _ := TestTopicCommands(SetPersistenceCmd, setArgs)
+	assert.Nil(t, execErr)
+	assert.Equal(t, setOut.String(), "Set persistence successfully for ["+topicName+"]\n")
+
+	time.Sleep(time.Duration(1) * time.Second)
+	getArgs := []string{"get-persistence", topicName}
+	getOut, execErr, _, _ := TestTopicCommands(GetPersistenceCmd, getArgs)
+	var persistenceData utils.PersistenceData
+	err := json.Unmarshal(getOut.Bytes(), &persistenceData)
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.Nil(t, execErr)
+	assert.Equal(t, persistenceData.BookkeeperEnsemble, int64(5))
+	assert.Equal(t, persistenceData.BookkeeperWriteQuorum, int64(4))
+	assert.Equal(t, persistenceData.BookkeeperAckQuorum, int64(3))
+	assert.Equal(t, persistenceData.ManagedLedgerMaxMarkDeleteRate, float64(2.2))
+
+	setArgs = []string{"remove-persistence", topicName}
+	setOut, execErr, _, _ = TestTopicCommands(RemovePersistenceCmd, setArgs)
+	assert.Nil(t, execErr)
+	assert.Equal(t, setOut.String(), "Remove persistence successfully for ["+topicName+"]\n")
+
+	time.Sleep(time.Duration(1) * time.Second)
+	getArgs = []string{"get-persistence", topicName}
+	getOut, execErr, _, _ = TestTopicCommands(GetPersistenceCmd, getArgs)
+	err = json.Unmarshal(getOut.Bytes(), &persistenceData)
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.Nil(t, execErr)
+	assert.Equal(t, persistenceData.BookkeeperEnsemble, int64(0))
+	assert.Equal(t, persistenceData.BookkeeperWriteQuorum, int64(0))
+	assert.Equal(t, persistenceData.BookkeeperAckQuorum, int64(0))
+	assert.Equal(t, persistenceData.ManagedLedgerMaxMarkDeleteRate, float64(0))
+
+	// test value
+	setArgs = []string{"set-persistence", topicName, "-e", "1", "-w", "4", "-a", "3", "-r", "2.2"}
+	_, execErr, _, _ = TestTopicCommands(SetPersistenceCmd, setArgs)
+	assert.NotNil(t, execErr)
+	assert.Equal(t, execErr.Error(), "code: 400 reason: Bookkeeper Ensemble (1) >= WriteQuorum (4) >= AckQuoru (3)")
+}

pkg/ctl/topic/remove_persistence.go

@@ -0,0 +1,76 @@
+// 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 topic
+
+import (
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/streamnative/pulsarctl/pkg/pulsar/utils"
+)
+
+func RemovePersistenceCmd(vc *cmdutils.VerbCmd) {
+	desc := cmdutils.LongDescription{}
+	desc.CommandUsedFor = "Remove persistence for a topic"
+	desc.CommandPermission = "This command requires tenant admin permissions."
+
+	var examples []cmdutils.Example
+	msg := cmdutils.Example{
+		Desc:    "Remove persistence for a topic",
+		Command: "pulsarctl topics remove-persistence topic",
+	}
+	examples = append(examples, msg)
+	desc.CommandExamples = examples
+
+	var out []cmdutils.Output
+	successOut := cmdutils.Output{
+		Desc: "normal output",
+		Out:  "Remove persistence successfully for [topic]",
+	}
+	out = append(out, successOut, ArgError)
+	out = append(out, TopicNameErrors...)
+	out = append(out, NamespaceErrors...)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"remove-persistence",
+		"Remove persistence for a topic",
+		desc.ToString(),
+		desc.ExampleToString(),
+		"remove-persistence",
+	)
+	vc.SetRunFuncWithNameArg(func() error {
+		return doRemovePersistence(vc)
+	}, "the topic name is not specified or the topic name is specified more than one")
+}
+
+func doRemovePersistence(vc *cmdutils.VerbCmd) error {
+	// for testing
+	if vc.NameError != nil {
+		return vc.NameError
+	}
+
+	topic, err := utils.GetTopicName(vc.NameArg)
+	if err != nil {
+		return err
+	}
+	admin := cmdutils.NewPulsarClient()
+	err = admin.Topics().RemovePersistence(*topic)
+	if err == nil {
+		vc.Command.Printf("Remove persistence successfully for [%s]\n", topic.String())
+	}
+	return err
+}

pkg/ctl/topic/set_persistence.go

@@ -0,0 +1,106 @@
+// 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 topic
+
+import (
+	"github.com/spf13/pflag"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/streamnative/pulsarctl/pkg/pulsar/utils"
+)
+
+func SetPersistenceCmd(vc *cmdutils.VerbCmd) {
+	desc := cmdutils.LongDescription{}
+	desc.CommandUsedFor = "Set persistence for a topic"
+	desc.CommandPermission = "This command requires tenant admin permissions."
+
+	var examples []cmdutils.Example
+	msg := cmdutils.Example{
+		Desc:    "Set persistence for a topic",
+		Command: "pulsarctl topics set-persistence topic -e 4 -w 3 -a 2 -r 1.0",
+	}
+	examples = append(examples, msg)
+	desc.CommandExamples = examples
+
+	var out []cmdutils.Output
+	successOut := cmdutils.Output{
+		Desc: "normal output",
+		Out:  "Set persistence successfully for [topic]",
+	}
+	out = append(out, successOut, ArgError)
+	out = append(out, TopicNameErrors...)
+	out = append(out, NamespaceErrors...)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"set-persistence",
+		"Set persistence for a topic",
+		desc.ToString(),
+		desc.ExampleToString(),
+		"set-persistence",
+	)
+	persistenceData := &utils.PersistenceData{}
+	vc.SetRunFuncWithNameArg(func() error {
+		return doSetPersistence(vc, persistenceData)
+	}, "the topic name is not specified or the topic name is specified more than one")
+
+	vc.FlagSetGroup.InFlagSet("Persistence", func(set *pflag.FlagSet) {
+		set.Int64VarP(
+			&persistenceData.BookkeeperEnsemble,
+			"bookkeeper-ensemble",
+			"e",
+			0,
+			"Number of bookies to use for a topic")
+		set.Int64VarP(
+			&persistenceData.BookkeeperWriteQuorum,
+			"bookkeeper-write-quorum",
+			"w",
+			0,
+			"How many writes to make of each entry")
+		set.Int64VarP(
+			&persistenceData.BookkeeperAckQuorum,
+			"bookkeeper-ack-quorum",
+			"a",
+			0,
+			"Number of acks (guaranteed copies) to wait for each entry")
+		set.Float64VarP(
+			&persistenceData.ManagedLedgerMaxMarkDeleteRate,
+			"ml-mark-delete-max-rate",
+			"r",
+			0.0,
+			"Throttling rate of mark-delete operation (0 means no throttle)")
+	})
+	vc.EnableOutputFlagSet()
+}
+
+func doSetPersistence(vc *cmdutils.VerbCmd, persistenceData *utils.PersistenceData) error {
+	// for testing
+	if vc.NameError != nil {
+		return vc.NameError
+	}
+
+	topic, err := utils.GetTopicName(vc.NameArg)
+	if err != nil {
+		return err
+	}
+	admin := cmdutils.NewPulsarClient()
+	err = admin.Topics().SetPersistence(*topic, *persistenceData)
+	if err == nil {
+		vc.Command.Printf("Set persistence successfully for [%s]\n", topic.String())
+	}
+	return err
+}

pkg/ctl/topic/topic.go

@@ -65,6 +65,9 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command {
 		GetMaxUnackMessagesPerSubscriptionCmd,
 		SetMaxUnackMessagesPerSubscriptionCmd,
 		RemoveMaxUnackMessagesPerSubscriptionCmd,
+		GetPersistenceCmd,
+		SetPersistenceCmd,
+		RemovePersistenceCmd,
 	}
 
 	cmdutils.AddVerbCmds(flagGrouping, resourceCmd, commands...)

pkg/pulsar/topic.go

@@ -143,6 +143,15 @@ type Topics interface {
 
 	// RemoveMaxUnackMessagesPerSubscription Remove max unacked messages policy on subscription for a topic
 	RemoveMaxUnackMessagesPerSubscription(utils.TopicName) error
+
+	// GetPersistence Get the persistence policies for a topic
+	GetPersistence(utils.TopicName) (*utils.PersistenceData, error)
+
+	// SetPersistence Set the persistence policies for a topic
+	SetPersistence(utils.TopicName, utils.PersistenceData) error
+
+	// RemovePersistence Remove the persistence policies for a topic
+	RemovePersistence(utils.TopicName) error
 }
 
 type topics struct {
@@ -441,3 +450,20 @@ func (t *topics) RemoveMaxUnackMessagesPerSubscription(topic utils.TopicName) er
 	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "maxUnackedMessagesOnSubscription")
 	return t.pulsar.Client.Delete(endpoint)
 }
+
+func (t *topics) GetPersistence(topic utils.TopicName) (*utils.PersistenceData, error) {
+	var persistenceData utils.PersistenceData
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "persistence")
+	err := t.pulsar.Client.Get(endpoint, &persistenceData)
+	return &persistenceData, err
+}
+
+func (t *topics) SetPersistence(topic utils.TopicName, persistenceData utils.PersistenceData) error {
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "persistence")
+	return t.pulsar.Client.Post(endpoint, &persistenceData)
+}
+
+func (t *topics) RemovePersistence(topic utils.TopicName) error {
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "persistence")
+	return t.pulsar.Client.Delete(endpoint)
+}