kafka

Documentation

Index

• type AdminClient
  • func NewAdminClient(ctx context.Context, caFactory ClusterAdminFactory) (AdminClient, error)
• type AdminClientManager
  • func (c *AdminClientManager) ListConsumerGroups() ([]string, error)
• type ClusterAdminFactory

Types

type AdminClient

type AdminClient interface {
	// ListConsumerGroups Lists the consumer groups
	ListConsumerGroups() ([]string, error)
}

func NewAdminClient

func NewAdminClient(ctx context.Context, caFactory ClusterAdminFactory) (AdminClient, error)

type AdminClientManager

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

AdminClientManager manages a ClusterAdmin connection and recreates one when needed it is made to overcome https://github.com/Shopify/sarama/issues/1162

func (*AdminClientManager) ListConsumerGroups

func (c *AdminClientManager) ListConsumerGroups() ([]string, error)

ListConsumerGroups Returns a list of the consumer groups.

In the occasion of errors, there will be a retry with an exponential backoff. Due to a known issue in Sarama ClusterAdmin https://github.com/Shopify/sarama/issues/1162, a new ClusterAdmin will be created with every retry until the call succeeds or the timeout is reached.

type ClusterAdminFactory

type ClusterAdminFactory func() (sarama.ClusterAdmin, error)