util

Documentation

Index

• func GetEncoderConfig(changefeedID model.ChangeFeedID, sinkURI *url.URL, protocol config.Protocol, ...) (*common.Config, error)
• func GetFileExtension(protocol config.Protocol) string
• func GetProtocol(protocolStr string) (config.Protocol, error)
• func GetTopic(sinkURI *url.URL) (string, error)
• func GetTopicManagerAndTryCreateTopic(ctx context.Context, changefeedID model.ChangeFeedID, topic string, ...) (manager.TopicManager, error)
• func IsPulsarSupportedProtocols(p config.Protocol) bool

Functions

func GetEncoderConfig

func GetEncoderConfig(
	changefeedID model.ChangeFeedID,
	sinkURI *url.URL,
	protocol config.Protocol,
	replicaConfig *config.ReplicaConfig,
	maxMsgBytes int,
) (*common.Config, error)

GetEncoderConfig returns the encoder config and validates the config.

func GetFileExtension

func GetFileExtension(protocol config.Protocol) string

GetFileExtension returns the extension for specific protocol

func GetProtocol

func GetProtocol(protocolStr string) (config.Protocol, error)

GetProtocol returns the protocol from the sink URI.

func GetTopic

func GetTopic(sinkURI *url.URL) (string, error)

GetTopic returns the topic name from the sink URI.

func GetTopicManagerAndTryCreateTopic

func GetTopicManagerAndTryCreateTopic(
	ctx context.Context,
	changefeedID model.ChangeFeedID,
	topic string,
	topicCfg *kafka.AutoCreateTopicConfig,
	adminClient kafka.ClusterAdminClient,
) (manager.TopicManager, error)

GetTopicManagerAndTryCreateTopic returns the topic manager and try to create the topic.

func IsPulsarSupportedProtocols

func IsPulsarSupportedProtocols(p config.Protocol) bool

IsPulsarSupportedProtocols returns whether the protocol is supported by pulsar.