go/coordinator/internal/utils/pulsar_admin.go

package utils

import (
	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/apache/pulsar-client-go/pulsaradmin"
	pulsar_utils "github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
)

// This function creates topics in Pulsar. It takes in a list of topics and creates them in pulsar.
// It assumes that the tenant and namespace already exist in Pulsar.
func CreateTopics(pulsarAdminURL string, tenant string, namespace string, topics []string) error {
	cfg := &pulsaradmin.Config{
		WebServiceURL: pulsarAdminURL,
	}
	admin, err := pulsaradmin.NewClient(cfg)
	if err != nil {
		log.Error("Failed to create pulsar admin client", zap.Error(err))
		return err
	}

	for _, topic := range topics {
		topicSchema := "persistent://" + tenant + "/" + namespace + "/" + topic
		topicName, err := pulsar_utils.GetTopicName(topicSchema)
		if err != nil {
			log.Error("Failed to get topic name", zap.Error(err))
			return err
		}
		metadata, err := admin.Topics().GetMetadata(*topicName)
		if err != nil {
			log.Info("Failed to get topic metadata, needs to create", zap.Error(err))
		} else {
			log.Info("Topic already exists", zap.String("topic", topic), zap.Any("metadata", metadata))
			continue
		}
		err = admin.Topics().Create(*topicName, 0)
		if err != nil {
			log.Error("Failed to create topic", zap.Error(err))
			return err
		}
	}
	return nil
}