kafkaadmin

README

Documentation

Overview

Package kafkaadmin provides Kafka administrative functionality.

Index

• Variables
• func NewConsumer(cfg Config) (*kafka.Consumer, error)
• type Client
  • func NewClientWithFactory(cfg Config, factory FactoryFunc) (*Client, error)
  • func (c Client) Close()
  • func (c Client) CreateTopic(ctx context.Context, cfg CreateTopicConfig) error
  • func (c Client) DeleteTopic(ctx context.Context, name string) error
• type Config
• type CreateTopicConfig
• type FactoryFunc
• type KafkaAdmin
  • func NewClient(cfg Config) (KafkaAdmin, error)
• type ReplicaAssignment

Variables

var (

	// SecurityProtocolSet is the set of protocols supported to communicate with brokers
	SecurityProtocolSet = map[string]struct{}{"PLAINTEXT": empty, "SSL": empty, "SASL_PLAINTEXT": empty, "SASL_SSL": empty}
	// SASLMechanismSet is the set of mechanisms supported for client to broker authentication
	SASLMechanismSet = map[string]struct{}{"PLAIN": empty, "SCRAM-SHA-256": empty, "SCRAM-SHA-512": empty}
)

Functions

func NewConsumer

func NewConsumer(cfg Config) (*kafka.Consumer, error)

Types

type Client

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

Client implements a KafkaAdmin.

func NewClientWithFactory

func NewClientWithFactory(cfg Config, factory FactoryFunc) (*Client, error)

NewClientWithFactory returns a new admin Client using a factory func for the kafkaAdminClient

func (Client) Close

func (c Client) Close()

Close closes the Client.

func (Client) CreateTopic

func (c Client) CreateTopic(ctx context.Context, cfg CreateTopicConfig) error

CreateTopic creates a topic.

func (Client) DeleteTopic

func (c Client) DeleteTopic(ctx context.Context, name string) error

DeleteTopic deletes a topic.

type Config

type Config struct {
	BootstrapServers string
	GroupId          string
	SSLCALocation    string
	SecurityProtocol string
	SASLMechanism    string
	SASLUsername     string
	SASLPassword     string
}

Config holds Client configuration parameters.

type CreateTopicConfig

type CreateTopicConfig struct {
	Name              string
	Partitions        int
	ReplicationFactor int
	Config            map[string]string
	ReplicaAssignment ReplicaAssignment
}

CreateTopicConfig holds CreateTopic parameters.

type FactoryFunc

type FactoryFunc func(conf *kafka.ConfigMap) (*kafka.AdminClient, error)

type KafkaAdmin

type KafkaAdmin interface {
	Close()
	CreateTopic(context.Context, CreateTopicConfig) error
	DeleteTopic(context.Context, string) error
}

KafkaAdmin interface.

func NewClient

func NewClient(cfg Config) (KafkaAdmin, error)

NewClient returns a KafkaAdmin.

type ReplicaAssignment

type ReplicaAssignment [][]int32