kafka

Documentation

Index

• type AdminClient
  • func (a *AdminClient) CreateKafkaTopic(topic string, topicConfig map[string]*string, partitions int32, ...) error
  • func (a *AdminClient) IsTopicExists(topic string) (bool, error)
• type Client
  • func (c *Client) CheckKafkaBrokers(numOfBrokers int32) error
  • func (c *Client) DeleteKafkaTopics(topics ...string)
  • func (c *Client) GetPartitionLeaderAddress(partition int32, topic string) (string, error)
  • func (c *Client) IsDBConnected() (bool, error)
  • func (c *Client) RefreshTopicMetadata(topics ...string) error
• type ConsumerClient
  • func (c *ConsumerClient) ConsumeMessages(partition int32, topic string, offset int64, signal *chan bool, ...) error
• type KubeDBClientBuilder
  • func NewKubeDBClientBuilder(kc client.Client, db *dbapi.Kafka) *KubeDBClientBuilder
  • func (o *KubeDBClientBuilder) GetConfig() (*kafkago.Config, error)
  • func (o *KubeDBClientBuilder) GetKafkaAdminClient() (*AdminClient, error)
  • func (o *KubeDBClientBuilder) GetKafkaClient() (*Client, error)
  • func (o *KubeDBClientBuilder) GetKafkaConsumerClient() (*ConsumerClient, error)
  • func (o *KubeDBClientBuilder) GetKafkaProducerClient() (*ProducerClient, error)
  • func (o *KubeDBClientBuilder) WithContext(ctx context.Context) *KubeDBClientBuilder
  • func (o *KubeDBClientBuilder) WithPod(podName string) *KubeDBClientBuilder
  • func (o *KubeDBClientBuilder) WithURL(url string) *KubeDBClientBuilder
• type MessageMetadata
• type ProducerClient
  • func (p *ProducerClient) PublishMessages(partition int32, topic, key, message string) (*MessageMetadata, error)

Types

type AdminClient

type AdminClient struct {
	kafkago.ClusterAdmin
}

func (*AdminClient) CreateKafkaTopic

func (a *AdminClient) CreateKafkaTopic(topic string, topicConfig map[string]*string, partitions int32, replicationFactor int16) error

func (*AdminClient) IsTopicExists

func (a *AdminClient) IsTopicExists(topic string) (bool, error)

type Client

type Client struct {
	kafkago.Client
}

func (*Client) CheckKafkaBrokers

func (c *Client) CheckKafkaBrokers(numOfBrokers int32) error

func (*Client) DeleteKafkaTopics

func (c *Client) DeleteKafkaTopics(topics ...string)

func (*Client) GetPartitionLeaderAddress

func (c *Client) GetPartitionLeaderAddress(partition int32, topic string) (string, error)

func (*Client) IsDBConnected

func (c *Client) IsDBConnected() (bool, error)

func (*Client) RefreshTopicMetadata

func (c *Client) RefreshTopicMetadata(topics ...string) error

type ConsumerClient

type ConsumerClient struct {
	kafkago.Consumer
}

func (*ConsumerClient) ConsumeMessages

func (c *ConsumerClient) ConsumeMessages(partition int32, topic string, offset int64, signal *chan bool, message *chan MessageMetadata) error

type KubeDBClientBuilder

type KubeDBClientBuilder struct {
	// contains filtered or unexported fields
}

func NewKubeDBClientBuilder

func NewKubeDBClientBuilder(kc client.Client, db *dbapi.Kafka) *KubeDBClientBuilder

func (*KubeDBClientBuilder) GetConfig

func (o *KubeDBClientBuilder) GetConfig() (*kafkago.Config, error)

func (*KubeDBClientBuilder) GetKafkaAdminClient

func (o *KubeDBClientBuilder) GetKafkaAdminClient() (*AdminClient, error)

func (*KubeDBClientBuilder) GetKafkaClient

func (o *KubeDBClientBuilder) GetKafkaClient() (*Client, error)

func (*KubeDBClientBuilder) GetKafkaConsumerClient

func (o *KubeDBClientBuilder) GetKafkaConsumerClient() (*ConsumerClient, error)

func (*KubeDBClientBuilder) GetKafkaProducerClient

func (o *KubeDBClientBuilder) GetKafkaProducerClient() (*ProducerClient, error)

func (*KubeDBClientBuilder) WithContext

func (o *KubeDBClientBuilder) WithContext(ctx context.Context) *KubeDBClientBuilder

func (*KubeDBClientBuilder) WithPod

func (o *KubeDBClientBuilder) WithPod(podName string) *KubeDBClientBuilder

func (*KubeDBClientBuilder) WithURL

func (o *KubeDBClientBuilder) WithURL(url string) *KubeDBClientBuilder

type MessageMetadata

type MessageMetadata struct {
	Key       string
	Value     string
	Partition int32
	Offset    int64
}

type ProducerClient

type ProducerClient struct {
	kafkago.SyncProducer
}

func (*ProducerClient) PublishMessages

func (p *ProducerClient) PublishMessages(partition int32, topic, key, message string) (*MessageMetadata, error)