add delete schemas force option

pkg/ctl/schemas/delete.go

@@ -18,6 +18,8 @@
 package schemas
 
 import (
+	"github.com/spf13/pflag"
+
 	"github.com/streamnative/pulsarctl/pkg/cmdutils"
 )
 
@@ -32,6 +34,12 @@ func deleteSchema(vc *cmdutils.VerbCmd) {
 		Command: "pulsarctl schemas delete (topic name)",
 	}
 	examples = append(examples, del)
+
+	delForce := cmdutils.Example{
+		Desc:    "Force delete the latest schema for a topic",
+		Command: "pulsarctl schemas delete (topic name) --force",
+	}
+	examples = append(examples, delForce)
 	desc.CommandExamples = examples
 
 	var out []cmdutils.Output
@@ -55,15 +63,27 @@ func deleteSchema(vc *cmdutils.VerbCmd) {
 		"delete",
 	)
 
+	var force bool
+	vc.FlagSetGroup.InFlagSet("DeleteSchema", func(set *pflag.FlagSet) {
+		set.BoolVarP(&force, "force", "f", false,
+			"Force delete the schema")
+	})
+	vc.EnableOutputFlagSet()
+
 	vc.SetRunFuncWithNameArg(func() error {
-		return doDeleteSchema(vc)
+		return doDeleteSchema(vc, force)
 	}, "the topic name is not specified or the topic name is specified more than one")
 }
 
-func doDeleteSchema(vc *cmdutils.VerbCmd) error {
+func doDeleteSchema(vc *cmdutils.VerbCmd, force bool) error {
 	topic := vc.NameArg
 	admin := cmdutils.NewPulsarClient()
-	err := admin.Schemas().DeleteSchema(topic)
+	var err error
+	if force {
+		err = admin.Schemas().ForceDeleteSchema(topic)
+	} else {
+		err = admin.Schemas().DeleteSchema(topic)
+	}
 	if err == nil {
 		vc.Command.Printf("Deleted %s successfully\n", topic)
 	}

pkg/ctl/schemas/schema_test.go

@@ -77,3 +77,71 @@ func TestFailSchema(t *testing.T) {
 	assert.NotNil(t, execErr)
 	assert.Equal(t, execErr.Error(), uploadErr)
 }
+
+func TestSchemaForceDelete(t *testing.T) {
+	fileName := "avro-schema-force"
+	f, err := os.Create(fileName)
+	assert.Nil(t, err)
+	defer func(name string) {
+		_ = os.Remove(name)
+	}(fileName)
+
+	_, err = f.WriteString("{\n" +
+		"   \"type\": \"AVRO\",\n" +
+		"   \"schema\": " +
+		"\"{\\\"type\\\":\\\"record\\\"," +
+		"\\\"name\\\":\\\"TestForce\\\"," +
+		"\\\"fields\\\":[{" +
+		"\\\"name\\\":\\\"id\\\"," +
+		"\\\"type\\\":[\\\"null\\\",\\\"int\\\"]}," +
+		"{\\\"name\\\":\\\"name\\\",\\\"type\\\":[\\\"null\\\",\\\"string\\\"]}]}\",\n" +
+		"   \"properties\": {}\n" +
+		"}\n")
+	assert.Nil(t, err)
+
+	// Upload schema first
+	args := []string{"upload", "test-schema-force", "-f", fileName}
+	out, _, err := TestSchemasCommands(uploadSchema, args)
+	assert.NoError(t, err)
+	assert.Equal(t, "Upload test-schema-force successfully\n", out.String())
+
+	// Test force delete with -f flag
+	delArgs := []string{"delete", "test-schema-force", "-f"}
+	delOut, _, err := TestSchemasCommands(deleteSchema, delArgs)
+	assert.Nil(t, err)
+	assert.Equal(t, delOut.String(), "Deleted test-schema-force successfully\n")
+}
+
+func TestSchemaForceDeleteLongFlag(t *testing.T) {
+	fileName := "avro-schema-force-long"
+	f, err := os.Create(fileName)
+	assert.Nil(t, err)
+	defer func(name string) {
+		_ = os.Remove(name)
+	}(fileName)
+
+	_, err = f.WriteString("{\n" +
+		"   \"type\": \"AVRO\",\n" +
+		"   \"schema\": " +
+		"\"{\\\"type\\\":\\\"record\\\"," +
+		"\\\"name\\\":\\\"TestForceLong\\\"," +
+		"\\\"fields\\\":[{" +
+		"\\\"name\\\":\\\"id\\\"," +
+		"\\\"type\\\":[\\\"null\\\",\\\"int\\\"]}," +
+		"{\\\"name\\\":\\\"name\\\",\\\"type\\\":[\\\"null\\\",\\\"string\\\"]}]}\",\n" +
+		"   \"properties\": {}\n" +
+		"}\n")
+	assert.Nil(t, err)
+
+	// Upload schema first
+	args := []string{"upload", "test-schema-force-long", "-f", fileName}
+	out, _, err := TestSchemasCommands(uploadSchema, args)
+	assert.NoError(t, err)
+	assert.Equal(t, "Upload test-schema-force-long successfully\n", out.String())
+
+	// Test force delete with --force flag
+	delArgs := []string{"delete", "test-schema-force-long", "--force"}
+	delOut, _, err := TestSchemasCommands(deleteSchema, delArgs)
+	assert.Nil(t, err)
+	assert.Equal(t, delOut.String(), "Deleted test-schema-force-long successfully\n")
+}