kafka

Documentation

Index

• Constants
• Variables
• func GatherBrokerConfigIfAvailable(kafkaClusterSpec v1beta1.KafkaClusterSpec, brokerID int) (*v1beta1.BrokerConfig, error)
• func GetBootstrapServers(cluster *v1beta1.KafkaCluster) (string, error)
• func GetBootstrapServersService(cluster *v1beta1.KafkaCluster) (string, error)
• func GetBrokerServiceFqdn(cluster *v1beta1.KafkaCluster, broker *v1beta1.Broker) string
• func GetClusterServiceDomainName(cluster *v1beta1.KafkaCluster) string
• func GetClusterServiceFqdn(cluster *v1beta1.KafkaCluster) string
• func GrantsToACLStrings(dn string, grants []v1alpha1.UserTopicGrant) []string
• func LabelsForKafka(name string) map[string]string
• func ShouldRefreshOnlyPerBrokerConfigs(currentConfigs, desiredConfigs *properties.Properties, log logr.Logger) bool

Constants

const (
	// AllBrokerServiceTemplate template for Kafka all broker service
	AllBrokerServiceTemplate = "%s-all-broker"
	// HeadlessServiceTemplate template for Kafka headless service
	HeadlessServiceTemplate = "%s-headless"
	// NodePortServiceTemplate template for Kafka nodeport service
	NodePortServiceTemplate = "%s-%d-%s"

	//ConfigPropertyName name in the ConfigMap's Data field for the broker configuration
	ConfigPropertyName = "broker-config"
)

const (
	// BrokerHostnameTemplate defines the hostname template for Kafka brokers in the following format:
	// 	<KAFKA_CLUSTER_NAME>-<BROKER_ID>
	BrokerHostnameTemplate = "%s-%d"
	// BrokerHostnameTemplate defines the domain template for Kafka brokers in the following format:
	// 	<K8S_NAMESPACE>.svc.<K8S_CLUSTER_DOMAIN>
	ServiceDomainNameTemplate = "%s.svc.%s"
)

Variables

var PerBrokerConfigs = []string{

	"ssl.client.auth",

	"listeners",
	"advertised.listeners",

	securityProtocolMapConfigName,
}

PerBrokerConfigs configurations will not trigger rolling upgrade when updated

Functions

func GatherBrokerConfigIfAvailable

func GatherBrokerConfigIfAvailable(kafkaClusterSpec v1beta1.KafkaClusterSpec, brokerID int) (*v1beta1.BrokerConfig, error)

GatherBrokerConfigIfAvailable return the brokerConfig for a specific ID if available

func GetBootstrapServers

func GetBootstrapServers(cluster *v1beta1.KafkaCluster) (string, error)

func GetBootstrapServersService

func GetBootstrapServersService(cluster *v1beta1.KafkaCluster) (string, error)

func GetBrokerServiceFqdn

func GetBrokerServiceFqdn(cluster *v1beta1.KafkaCluster, broker *v1beta1.Broker) string

func GetClusterServiceDomainName

func GetClusterServiceDomainName(cluster *v1beta1.KafkaCluster) string

func GetClusterServiceFqdn

func GetClusterServiceFqdn(cluster *v1beta1.KafkaCluster) string

func GrantsToACLStrings

func GrantsToACLStrings(dn string, grants []v1alpha1.UserTopicGrant) []string

GrantsToACLStrings converts a user DN and a list of topic grants to raw strings for a CR status

func LabelsForKafka

func LabelsForKafka(name string) map[string]string

LabelsForKafka returns the labels for selecting the resources belonging to the given kafka CR name.

func ShouldRefreshOnlyPerBrokerConfigs

func ShouldRefreshOnlyPerBrokerConfigs(currentConfigs, desiredConfigs *properties.Properties, log logr.Logger) bool