client

Documentation

Index

• func MakeAdminClient(ctx context.Context, env *KafkaEnvConfig) (sarama.ClusterAdmin, error)
• func NewConfigFromEnv(ctx context.Context) ([]string, *sarama.Config, error)
• func NewConfigFromSpec(ctx context.Context, kc kubernetes.Interface, obj *sourcesv1beta1.KafkaSource) ([]string, *sarama.Config, error)
• func NewConfigWithEnv(ctx context.Context, env *KafkaEnvConfig) ([]string, *sarama.Config, error)
• func NewProducer(ctx context.Context) (sarama.Client, error)
• type AdapterNet
• type AdapterSASL
• type AdapterTLS
• type KafkaConfig
• type KafkaEnvConfig
  • func NewEnvConfigFromSpec(ctx context.Context, kc kubernetes.Interface, obj *sourcesv1beta1.KafkaSource) (KafkaEnvConfig, error)

Functions

func MakeAdminClient

func MakeAdminClient(ctx context.Context, env *KafkaEnvConfig) (sarama.ClusterAdmin, error)

NewProducer is a helper method for constructing an admin client

func NewConfigFromEnv

func NewConfigFromEnv(ctx context.Context) ([]string, *sarama.Config, error)

NewConfig extracts the Kafka configuration from the environment.

func NewConfigFromSpec

func NewConfigFromSpec(ctx context.Context, kc kubernetes.Interface, obj *sourcesv1beta1.KafkaSource) ([]string, *sarama.Config, error)

NewConfig extracts the Kafka configuration from a KafkaSource spec.

func NewConfigWithEnv

func NewConfigWithEnv(ctx context.Context, env *KafkaEnvConfig) ([]string, *sarama.Config, error)

NewConfig extracts the Kafka configuration from the environment.

func NewProducer

func NewProducer(ctx context.Context) (sarama.Client, error)

NewProducer is a helper method for constructing a client for producing kafka methods.

Types

type AdapterNet

type AdapterNet struct {
	SASL AdapterSASL
	TLS  AdapterTLS
}

type AdapterSASL

type AdapterSASL struct {
	Enable   bool   `envconfig:"KAFKA_NET_SASL_ENABLE" required:"false"`
	User     string `envconfig:"KAFKA_NET_SASL_USER" required:"false"`
	Password string `envconfig:"KAFKA_NET_SASL_PASSWORD" required:"false"`
	Type     string `envconfig:"KAFKA_NET_SASL_TYPE" required:"false"`
}

type AdapterTLS

type AdapterTLS struct {
	Enable bool   `envconfig:"KAFKA_NET_TLS_ENABLE" required:"false"`
	Cert   string `envconfig:"KAFKA_NET_TLS_CERT" required:"false"`
	Key    string `envconfig:"KAFKA_NET_TLS_KEY" required:"false"`
	CACert string `envconfig:"KAFKA_NET_TLS_CA_CERT" required:"false"`
}

type KafkaConfig

type KafkaConfig struct {
	SaramaYamlString string
}

type KafkaEnvConfig

type KafkaEnvConfig struct {
	// KafkaConfigJson is the environment variable that's passed to adapter by the controller.
	// It contains configuration from the Kafka configmap.
	KafkaConfigJson  string                `envconfig:"K_KAFKA_CONFIG"`
	BootstrapServers []string              `envconfig:"KAFKA_BOOTSTRAP_SERVERS" required:"true"`
	InitialOffset    sourcesv1beta1.Offset `envconfig:"KAFKA_INITIAL_OFFSET" `
	Net              AdapterNet
}

func NewEnvConfigFromSpec

func NewEnvConfigFromSpec(ctx context.Context, kc kubernetes.Interface, obj *sourcesv1beta1.KafkaSource) (KafkaEnvConfig, error)

NewEnvConfigFromSpec validates and creates a KafkaEnvConfig from a KafkaSource