[pulsaradmin] Add GetAllSchemas command (#1289)

* Add GetAllSchemas command

* Fix doc comment

* Fix typo

* Fix linter error: line too long

* Remove unnecessary print

pulsaradmin/pkg/admin/schema.go

@@ -35,6 +35,9 @@ type Schema interface {
 	// GetSchemaInfoByVersion retrieves the schema of a topic at a given <tt>version</tt>
 	GetSchemaInfoByVersion(topic string, version int64) (*utils.SchemaInfo, error)
 
+	// GetAllSchemas retrieves all schemas of a topic
+	GetAllSchemas(topic string) ([]*utils.SchemaInfoWithVersion, error)
+
 	// DeleteSchema deletes the schema associated with a given <tt>topic</tt>
 	DeleteSchema(topic string) error
 
@@ -130,6 +133,24 @@ func (s *schemas) GetSchemaInfoByVersion(topic string, version int64) (*utils.Sc
 	return info, nil
 }
 
+func (s *schemas) GetAllSchemas(topic string) ([]*utils.SchemaInfoWithVersion, error) {
+	topicName, err := utils.GetTopicName(topic)
+	if err != nil {
+		return nil, err
+	}
+	var response utils.GetAllSchemasResponse
+	endpoint := s.pulsar.endpoint(s.basePath, topicName.GetTenant(), topicName.GetNamespace(),
+		topicName.GetLocalName(), "schemas")
+
+	err = s.pulsar.Client.Get(endpoint, &response)
+	if err != nil {
+		return nil, err
+	}
+
+	infos := utils.ConvertGetAllSchemasResponseToSchemaInfosWithVersion(topicName, response)
+	return infos, nil
+}
+
 func (s *schemas) DeleteSchema(topic string) error {
 	return s.delete(topic, false)
 }

pulsaradmin/pkg/admin/schema_test.go

@@ -27,6 +27,31 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+func TestSchemas_GetAllSchemas(t *testing.T) {
+	cfg := &config.Config{}
+	admin, err := New(cfg)
+	assert.NoError(t, err)
+	assert.NotNil(t, admin)
+
+	topic := fmt.Sprintf("my-topic-%v", time.Now().Nanosecond())
+	schemaPayload := utils.PostSchemaPayload{
+		SchemaType: "STRING",
+		Schema:     "",
+	}
+	err = admin.Schemas().CreateSchemaByPayload(topic, schemaPayload)
+	assert.NoError(t, err)
+
+	infos, err := admin.Schemas().GetAllSchemas(topic)
+	assert.NoError(t, err)
+	assert.Len(t, infos, 1)
+
+	err = admin.Schemas().ForceDeleteSchema(topic)
+	assert.NoError(t, err)
+
+	_, err = admin.Schemas().GetSchemaInfo(topic)
+	assert.Errorf(t, err, "Schema not found")
+}
+
 func TestSchemas_DeleteSchema(t *testing.T) {
 	cfg := &config.Config{}
 	admin, err := New(cfg)

pulsaradmin/pkg/utils/schema_util.go

@@ -44,6 +44,10 @@ type GetSchemaResponse struct {
 	Properties map[string]string `json:"properties"`
 }
 
+type GetAllSchemasResponse struct {
+	Schemas []GetSchemaResponse `json:"getSchemaResponses"`
+}
+
 type IsCompatibility struct {
 	IsCompatibility             bool                        `json:"compatibility"`
 	SchemaCompatibilityStrategy SchemaCompatibilityStrategy `json:"schemaCompatibilityStrategy"`
@@ -90,3 +94,16 @@ func ConvertGetSchemaResponseToSchemaInfoWithVersion(tn *TopicName, response Get
 	info.Version = response.Version
 	return info
 }
+
+func ConvertGetAllSchemasResponseToSchemaInfosWithVersion(
+	tn *TopicName,
+	response GetAllSchemasResponse,
+) []*SchemaInfoWithVersion {
+	infos := make([]*SchemaInfoWithVersion, len(response.Schemas))
+
+	for i, schema := range response.Schemas {
+		infos[i] = ConvertGetSchemaResponseToSchemaInfoWithVersion(tn, schema)
+	}
+
+	return infos
+}