README
¶
KafkaAdmin Library
Is used to ensure that a topic exists. It's API doesn't support configuration of the topics to avoid conflicts between different services using the same topic. Each service, should call this lib on booting. It doesn't matter whether it reads or writes to the topic.
Every topic created has the cleanup.policy compact, a replication of 3 and 32 partitions, as this is a sane default for our platform.
Usage
In code you can use it like this:
err := kafkaadmin.EnsureCompactedTopicExists(ctx, kafkaURL, tlsConfig, topicName)
if err != nil {
return fmt.Errorf("ensuring topic %q failed: %w", topicName, err)
}
Made with ♥ by Team Hasselhoff
Documentation
¶
Index ¶
- func CompactedTopicConfig(topicName string) kafka.TopicConfig
- func EnsureCompactedTopicExists(ctx context.Context, kafkaURL string, tlsConfig *tls.Config, topicName string) error
- func EnsureTopicExistsWithCleanupPolicyDelete(ctx context.Context, kafkaURL string, tlsConfig *tls.Config, topicName string) error
- func EnsureTopicExistsWithConfig(ctx context.Context, kafkaURL string, tlsConfig *tls.Config, ...) error
- func TopicConfigCleanupPolicyDelete(topicName string) kafka.TopicConfig
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func CompactedTopicConfig ¶
func CompactedTopicConfig(topicName string) kafka.TopicConfig
func EnsureCompactedTopicExists ¶
func EnsureCompactedTopicExists(ctx context.Context, kafkaURL string, tlsConfig *tls.Config, topicName string) error
EnsureCompactedTopicExists checks if the topic with the given name is missing if it doesn't exist, it creates it with the default configuration The configuration can't be altered currently, because we want to avoid conflicts by different services creating the same topic
func EnsureTopicExistsWithCleanupPolicyDelete ¶ added in v0.5.0
func EnsureTopicExistsWithCleanupPolicyDelete(ctx context.Context, kafkaURL string, tlsConfig *tls.Config, topicName string) error
EnsureTopicExistsWithCleanupPolicyDelete checks if the topic with the given name is missing if it doesn't exist, it creates it with the default configuration The configuration can't be altered currently, because we want to avoid conflicts by different services creating the same topic
func TopicConfigCleanupPolicyDelete ¶ added in v0.5.0
func TopicConfigCleanupPolicyDelete(topicName string) kafka.TopicConfig
Types ¶
This section is empty.
Source Files