input

package

v0.8.0 Latest Latest Go to latest Published: Oct 20, 2025 License: Apache-2.0 Imports: 20 Imported by: 0

Details

View Source

const (
	Krb5KeytabAuth = 2
	CommitRetries  = 6
	RetryBackoff   = 5 * time.Second
)

This section is empty.

func GetFranzConfig(kfkCfg *config.KafkaConfig) (opts []kgo.Opt, err error)

func Leave(consumerId string)

func NewConsumerPoller(consumerId, consumerName string, client *kgo.Client)

func OnConsumerPoll(consumerId string)

func Walk(maxPollInterval int) string

type Fetches struct {
	Fetch   *kgo.Fetches
	TraceId string
}

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

KafkaFranz implements input.Inputer refers to examples/group_consuming/main.go

func NewKafkaFranz() *KafkaFranz

NewKafkaFranz get instance of kafka reader

func (k *KafkaFranz) CommitMessages(msg *model.InputMessage) error

func (k *KafkaFranz) Description() string

Description of this kafka consumer, consumer group name

func (k *KafkaFranz) Init(cfg *config.Config, gCfg *config.GroupConfig, f chan Fetches, cleanupFn func()) (err error)

Init Initialise the kafka instance with configuration

func (k *KafkaFranz) Run()

kafka main loop

func (k *KafkaFranz) Stop()

Stop kafka consumer and close all connections

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