client

Documentation

Index

• func InitOffsets(ctx context.Context, kafkaClient sarama.Client, topics []string, ...) error
• 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 InitOffsets

func InitOffsets(ctx context.Context, kafkaClient sarama.Client, topics []string, consumerGroup string) error

We want to make sure that ALL consumer group offsets are set before marking the source as ready, to avoid "losing" events in case the consumer group session is closed before at least one message is consumed from ALL partitions. Without InitOffsets, an event sent to a partition with an uninitialized offset will not be forwarded when the session is closed (or a rebalancing is in progress).

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"`
	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