franz

Documentation

Index

• Constants
• func GetMechanism(sasl SASL) sasl.Mechanism
• func NewTLSConfig(caCertFiles, clientCertFile, clientKeyFile string, insecureSkipVerify bool) (*tls.Config, error)
• type Config
  • func (c *Config) Validate() error
• type GSSAPI
• type Logger
  • func (*Logger) Level() kgo.LogLevel
  • func (l *Logger) Log(level kgo.LogLevel, msg string, keyvals ...interface{})
• type RenderTopicFail
• type SASL
• type Sink
  • func NewSink() *Sink
  • func (s *Sink) Category() api.Category
  • func (s *Sink) Config() interface{}
  • func (s *Sink) Consume(batch api.Batch) api.Result
  • func (s *Sink) Init(context api.Context) error
  • func (s *Sink) SetCodec(c codec.Codec)
  • func (s *Sink) Start() error
  • func (s *Sink) Stop()
  • func (s *Sink) String() string
  • func (s *Sink) Type() api.Type
• type TLS

Constants

const (
	Krb5KeytabAuth = 2
)

const Type = "franzKafka"

Functions

func GetMechanism

func GetMechanism(sasl SASL) sasl.Mechanism

func NewTLSConfig

func NewTLSConfig(caCertFiles, clientCertFile, clientKeyFile string, insecureSkipVerify bool) (*tls.Config, error)

NewTLSConfig Refers to: https://medium.com/processone/using-tls-authentication-for-your-go-kafka-client-3c5841f2a625 https://github.com/denji/golang-tls https://www.baeldung.com/java-keystore-truststore-difference

Types

type Config

type Config struct {
	Brokers                       []string          `yaml:"brokers,omitempty" validate:"required"`
	Topic                         string            `yaml:"topic,omitempty" validate:"required" default:"loggie"`
	IfRenderTopicFailed           RenderTopicFail   `yaml:"ifRenderTopicFailed,omitempty"`
	IgnoreUnknownTopicOrPartition bool              `yaml:"ignoreUnknownTopicOrPartition,omitempty"`
	Balance                       string            `yaml:"balance,omitempty" default:"roundRobin"`
	BatchSize                     int               `yaml:"batchSize,omitempty"`
	BatchBytes                    int32             `yaml:"batchBytes,omitempty"`
	RetryTimeout                  time.Duration     `yaml:"retryTimeout,omitempty"`
	WriteTimeout                  time.Duration     `yaml:"writeTimeout,omitempty"`
	Compression                   string            `yaml:"compression,omitempty" default:"gzip"`
	SASL                          SASL              `yaml:"sasl,omitempty"`
	TLS                           TLS               `yaml:"tls,omitempty"`
	Security                      map[string]string `yaml:"security,omitempty"`
	PartitionKey                  string            `yaml:"partitionKey,omitempty"`
}

func (*Config) Validate

func (c *Config) Validate() error

type GSSAPI

type GSSAPI struct {
	AuthType           int    `yaml:"authType,omitempty"`
	KeyTabPath         string `yaml:"keyTabPath,omitempty"`
	KerberosConfigPath string `yaml:"kerberosConfigPath,omitempty"`
	ServiceName        string `yaml:"serviceName,omitempty"`
	UserName           string `yaml:"username,omitempty"`
	Password           string `yaml:"password,omitempty"`
	Realm              string `yaml:"realm,omitempty"`
	DisablePAFXFAST    bool   `yaml:"disablePAFXFAST,omitempty"`
}

type Logger

type Logger struct {
}

func (*Logger) Level

func (*Logger) Level() kgo.LogLevel

func (*Logger) Log

func (l *Logger) Log(level kgo.LogLevel, msg string, keyvals ...interface{})

type RenderTopicFail

type RenderTopicFail struct {
	DropEvent    bool   `yaml:"dropEvent,omitempty" default:"true"`
	IgnoreError  bool   `yaml:"ignoreError,omitempty"`
	DefaultTopic string `yaml:"defaultTopic,omitempty"`
}

type SASL

type SASL struct {
	Mechanism string `yaml:"mechanism,omitempty"`
	Enabled   bool   `yaml:"enabled,omitempty"`
	UserName  string `yaml:"username,omitempty"`
	Password  string `yaml:"password,omitempty"`
	GSSAPI    GSSAPI `yaml:"gssapi,omitempty"`
}

type Sink

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

func NewSink

func NewSink() *Sink

func (*Sink) Category

func (s *Sink) Category() api.Category

func (*Sink) Config

func (s *Sink) Config() interface{}

func (*Sink) Consume

func (s *Sink) Consume(batch api.Batch) api.Result

func (*Sink) Init

func (s *Sink) Init(context api.Context) error

func (*Sink) SetCodec

func (s *Sink) SetCodec(c codec.Codec)

func (*Sink) Start

func (s *Sink) Start() error

func (*Sink) Stop

func (s *Sink) Stop()

func (*Sink) String

func (s *Sink) String() string

func (*Sink) Type

func (s *Sink) Type() api.Type

type TLS

type TLS struct {
	Enabled        bool   `yaml:"enabled,omitempty"`
	CaCertFiles    string `yaml:"caCertFiles,omitempty"`
	ClientCertFile string `yaml:"clientCertFile,omitempty"`
	ClientKeyFile  string `yaml:"clientKeyFile,omitempty"`

	TrustStoreLocation string `yaml:"trustStoreLocation,omitempty"`
	TrustStorePassword string `yaml:"trustStorePassword,omitempty"`
	KeystoreLocation   string `yaml:"keystoreLocation,omitempty"`
	KeystorePassword   string `yaml:"keystorePassword,omitempty"`
	EndpIdentAlgo      string `yaml:"endpIdentAlgo,omitempty"`
}