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 ShouldRefreshOnlyPerBrokerConfigs(currentConfigs, desiredConfigs *properties.Properties, log logr.Logger) bool

Constants

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"
)

const (
	KafkaConfigSuperUsers = "super.users"

	KafkaConfigBoostrapServers    = "bootstrap.servers"
	KafkaConfigZooKeeperConnect   = "zookeeper.connect"
	KafkaConfigBrokerId           = "broker.id"
	KafkaConfigBrokerLogDirectory = "log.dirs"

	KafkaConfigListeners                   = "listeners"
	KafkaConfigListenerName                = "listener.name"
	KafkaConfigListenerSecurityProtocolMap = "listener.security.protocol.map"
	KafkaConfigInterBrokerListenerName     = "inter.broker.listener.name"
	KafkaConfigAdvertisedListeners         = "advertised.listeners"
	KafkaConfigControlPlaneListener        = "control.plane.listener.name"

	KafkaConfigSecurityProtocol      = "security.protocol"
	KafkaConfigSSLClientAuth         = "ssl.client.auth"
	KafkaConfigSSLTrustStoreType     = "ssl.truststore.type"
	KafkaConfigSSLTrustStoreLocation = "ssl.truststore.location"
	KafkaConfigSSLTrustStorePassword = "ssl.truststore.password"
	KafkaConfigSSLKeystoreType       = "ssl.keystore.type"
	KafkaConfigSSLKeyStoreLocation   = "ssl.keystore.location"
	KafkaConfigSSLKeyStorePassword   = "ssl.keystore.password"
)

used for Kafka configurations

const (
	CruiseControlConfigMetricsReporters                 = "metric.reporters"
	CruiseControlConfigMetricsReportersBootstrapServers = "cruise.control.metrics.reporter.bootstrap.servers"
	CruiseControlConfigMetricsReporterK8sMode           = "cruise.control.metrics.reporter.kubernetes.mode"
)

used for Cruise Control configurations

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"
)

const ConfigPropertyName = "broker-config"

ConfigPropertyName name in the ConfigMap's Data field for the broker configuration

Variables

var PerBrokerConfigs = []string{

	KafkaConfigSSLClientAuth,

	KafkaConfigListeners,
	KafkaConfigAdvertisedListeners,
	KafkaConfigListenerSecurityProtocolMap,
}

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 ShouldRefreshOnlyPerBrokerConfigs

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