kafka

Dec 8, 2020

Changes in this version

type Config

+ TopicReplicationFactor int16

Nov 29, 2020

Changes in this version

+ const SASLMechanismGSSAPI

+ const SASLMechanismOAuthBearer

+ const SASLMechanismPlain

+ const SASLMechanismScramSHA256

+ const SASLMechanismScramSHA512

+ const TimestampEarliest

+ const TimestampLatest

+ func NewKgoConfig(cfg *Config, logger *zap.Logger) ([]kgo.Opt, error)

+ func SerializeJson(data interface{}) ([]byte, error)

+ type Config struct

+ Brokers []string

+ ClientID string

+ SASL SASLConfig

+ TLS TLSConfig

+ func (c *Config) RegisterFlags(f *flag.FlagSet)

+ func (c *Config) SetDefaults()

+ func (c *Config) Validate() error

+ type KgoZapLogger struct

+ func (k KgoZapLogger) Level() kgo.LogLevel

+ func (k KgoZapLogger) Log(level kgo.LogLevel, msg string, keyvals ...interface{})

+ type PartitionMarks struct

+ High int64

+ Low int64

+ PartitionID int32

+ type SASLConfig struct

+ Enabled bool

+ GSSAPIConfig SASLGSSAPIConfig

+ Mechanism string

+ Password string

+ Username string

+ func (c *SASLConfig) RegisterFlags(f *flag.FlagSet)

+ func (c *SASLConfig) SetDefaults()

+ func (c *SASLConfig) Validate() error

+ type SASLGSSAPIConfig struct

+ AuthType string

+ KerberosConfigPath string

+ KeyTabPath string

+ Password string

+ Realm string

+ ServiceName string

+ Username string

+ func (c *SASLGSSAPIConfig) RegisterFlags(f *flag.FlagSet)

+ type Service struct

+ Config Config

+ KafkaClient *kgo.Client

+ Logger *zap.Logger

+ func NewService(cfg Config, logger *zap.Logger) (*Service, error)

+ func (s *Service) GetAverageMessageSize(ctx context.Context, topicName string) (int64, error)

+ func (s *Service) GetMetadata(ctx context.Context) (*kmsg.MetadataResponse, error)

+ func (s *Service) GetPartitionMarks(ctx context.Context, topic string, partitionIDs []int32) (map[int32]PartitionMarks, error)

+ func (s *Service) GetTopicMessageCount(ctx context.Context, topicName string) (int64, error)

+ func (s *Service) GetTopicMetadata(ctx context.Context, topicName string) (*kmsg.MetadataResponseTopic, error)

+ func (s *Service) GetTopicSize(ctx context.Context, topicName string, partitionIDs []int32) (TopicLogDirSize, error)

+ func (s *Service) ListOffsets(ctx context.Context, topicPartitions map[string][]int32, timestamp int64) (TopicPartitionOffsets, error)

+ func (s *Service) NewKafkaClient() (*kgo.Client, error)

+ type TLSConfig struct

+ CaFilepath string

+ CertFilepath string

+ Enabled bool

+ InsecureSkipTLSVerify bool

+ KeyFilepath string

+ Passphrase string

+ func (c *TLSConfig) RegisterFlags(f *flag.FlagSet)

+ type TopicLogDirSize struct

+ LeaderLogDirSize int64

+ ReplicaLogDirSize int64

+ TotalSize int64

+ type TopicPartitionOffsets = map[string]map[int32]int64