common

Documentation

Index

• Variables
• type Config
  • func (c Config) FranzClientOpts(logger *zerolog.Logger) []kgo.Opt
  • func (c Config) TryDial(ctx context.Context) error
  • func (c Config) Validate(ctx context.Context) error
  • func (c Config) WithFranzClientOpts(opts ...kgo.Opt) Config
• type ConfigSASL
  • func (c ConfigSASL) SASL() sasl.Mechanism
  • func (c ConfigSASL) Validate(context.Context) error
• type ConfigTLS
  • func (c ConfigTLS) TLS() *tls.Config
  • func (c ConfigTLS) Validate(context.Context) error

Variables

var ErrSASLInvalidAuth = errors.New("invalid sasl auth, please specify both saslUsername and saslPassword")

Types

type Config

type Config struct {
	// Servers is a list of Kafka bootstrap servers, which will be used to
	// discover all the servers in a cluster.
	Servers []string `json:"servers" validate:"required"`

	// ClientID is a unique identifier for client connections established by
	// this connector.
	ClientID string `json:"clientID" default:"conduit-connector-kafka"`

	ConfigSASL
	ConfigTLS
	// contains filtered or unexported fields
}

Config contains common configuration parameters.

func (Config) FranzClientOpts

func (c Config) FranzClientOpts(logger *zerolog.Logger) []kgo.Opt

FranzClientOpts returns the kafka client options derived from the common config.

func (Config) TryDial

func (c Config) TryDial(ctx context.Context) error

TryDial tries to establish a connection to brokers and returns nil if it succeeds to connect to at least one broker.

func (Config) Validate

func (c Config) Validate(ctx context.Context) error

Validate executes manual validations beyond what is defined in struct tags.

func (Config) WithFranzClientOpts

func (c Config) WithFranzClientOpts(opts ...kgo.Opt) Config

WithFranzClientOpts lets you specify custom kafka client options (meant for test purposes).

type ConfigSASL

type ConfigSASL struct {
	// Mechanism configures the connector to use SASL authentication. If
	// empty, no authentication will be performed.
	Mechanism string `json:"saslMechanism" validate:"inclusion=PLAIN|SCRAM-SHA-256|SCRAM-SHA-512"`
	// Username sets up the username used with SASL authentication.
	Username string `json:"saslUsername"`
	// Password sets up the password used with SASL authentication.
	Password string `json:"saslPassword"`
}

func (ConfigSASL) SASL

func (c ConfigSASL) SASL() sasl.Mechanism

SASL returns the SASL mechanism or nil.

func (ConfigSASL) Validate

func (c ConfigSASL) Validate(context.Context) error

Validate executes manual validations beyond what is defined in struct tags.

type ConfigTLS

type ConfigTLS struct {
	// TLSEnabled defines whether TLS is needed to communicate with the Kafka cluster.
	TLSEnabled bool `json:"tls.enabled"`
	// ClientCert is the Kafka client's certificate.
	ClientCert string `json:"clientCert"`
	// ClientKey is the Kafka client's private key.
	ClientKey string `json:"clientKey"`
	// CACert is the Kafka broker's certificate.
	CACert string `json:"caCert"`
	// InsecureSkipVerify defines whether to validate the broker's certificate
	// chain and host name. If 'true', accepts any certificate presented by the
	// server and any host name in that certificate.
	InsecureSkipVerify bool `json:"insecureSkipVerify"`
}

func (ConfigTLS) TLS

func (c ConfigTLS) TLS() *tls.Config

TLS returns the TLS config or nil.

func (ConfigTLS) Validate

func (c ConfigTLS) Validate(context.Context) error

Validate executes manual validations beyond what is defined in struct tags.