kafkaclient

Documentation

Index

• type CreateTopicOptions
• type KafkaClient
  • func New(opts *KafkaConfig) (client KafkaClient, err error)
  • func NewFromCluster(k8sclient client.Client, cluster *banzaicloudv1beta1.KafkaCluster) (client KafkaClient, err error)
• type KafkaConfig
  • func ClusterConfig(client client.Client, cluster *banzaicloudv1beta1.KafkaCluster) (*KafkaConfig, error)

Types

type CreateTopicOptions

type CreateTopicOptions struct {
	Name              string
	Partitions        int32
	ReplicationFactor int16
	Config            map[string]*string
}

CreateTopicOptions holds info about topic configuration

type KafkaClient

type KafkaClient interface {
	NumBrokers() int
	ListTopics() (map[string]sarama.TopicDetail, error)
	CreateTopic(*CreateTopicOptions) error
	EnsurePartitionCount(string, int32) (bool, error)
	EnsureTopicConfig(string, map[string]*string) error
	DeleteTopic(string) error
	GetTopic(string) (*sarama.TopicDetail, error)
	DescribeTopic(string) (*sarama.TopicMetadata, error)
	CreateUserACLs(banzaicloudv1alpha1.KafkaAccessType, string, string) error
	DeleteUserACLs(string) error

	ResolveBrokerID(int32) string
	DescribeCluster() ([]*sarama.Broker, error)
	GetCA() (string, string)
	GetBroker(int32) *sarama.Broker

	OfflineReplicaCount() (int, error)
	AllReplicaInSync() (bool, error)

	AlterPerBrokerConfig(int32, map[string]*string) error
	DescribePerBrokerConfig(int32, []string) ([]*sarama.ConfigEntry, error)

	AlterClusterWideConfig(map[string]*string) error
	DescribeClusterWideConfig() ([]sarama.ConfigEntry, error)

	Close() error
}

KafkaClient is the exported interface for kafka operations

func New

func New(opts *KafkaConfig) (client KafkaClient, err error)

New creates a new KafkaClient instance

func NewFromCluster

func NewFromCluster(k8sclient client.Client, cluster *banzaicloudv1beta1.KafkaCluster) (client KafkaClient, err error)

NewFromCluster is a convenience wrapper around New() and ClusterConfig()

type KafkaConfig

type KafkaConfig struct {
	BrokerURI string
	UseSSL    bool
	TLSConfig *tls.Config

	IssueCA     string
	IssueCAKind string

	OperationTimeout int64
}

KafkaConfig are the options to creating a new ClusterAdmin client

func ClusterConfig

func ClusterConfig(client client.Client, cluster *banzaicloudv1beta1.KafkaCluster) (*KafkaConfig, error)

ClusterConfig creates connection options from a KafkaCluster CR