spring_boot_kafka_consumer

Documentation

Index

• type KafkaConsumerConfig
• type KafkaConsumerStarter
  • func (k *KafkaConsumerStarter) GetPartitionConsumeChan(partition int32) (ch <-chan *sarama.ConsumerMessage, err error)
  • func (k *KafkaConsumerStarter) GetPartitions() []int32
  • func (k *KafkaConsumerStarter) GetUserTopic() string

Types

type KafkaConsumerConfig

type KafkaConsumerConfig struct {
	ConnectTimeOut     int      `value:"${kafka.socket_connect_timeout:=30}"`
	ReadTimeOut        int      `value:"${kafka.socket_read_timeout:=30}"`
	WriteTimeOut       int      `value:"${kafka.socket_write_timeout:=30}"`
	Brokers            []string `value:"${kafka.brokers:=127.0.0.1:9092}"`
	Version            string   `value:"${kafka.version:=2.4.0}"`
	ChannelBufferSize  int      `value:"${kafka.message_cache:=256}"`
	ClientId           string   `value:"${kafka.client_id:=kafka_client}"`
	IsGroup            bool     `value:"${kafka.consumer.is_group:=false}"`
	AutoCommitInterval int      `value:"${kafka.consumer.auto_commit_interval_ms:=1000}"`
	UserTopic          string   `value:"${kafka.consumer.user_topic:=}"`
	UserPartitions     []int    `value:"${kafka.consumer.user_partitions:=}"`
	OffsetLevel        string   `value:"${kafka.consumer.offset_level:=offset_newest}"`
}

type KafkaConsumerStarter

type KafkaConsumerStarter struct {
	KafkaConsumerConfig *KafkaConsumerConfig `autowire:""`
	// contains filtered or unexported fields
}

func (*KafkaConsumerStarter) GetPartitionConsumeChan

func (k *KafkaConsumerStarter) GetPartitionConsumeChan(partition int32) (ch <-chan *sarama.ConsumerMessage, err error)

func (*KafkaConsumerStarter) GetPartitions

func (k *KafkaConsumerStarter) GetPartitions() []int32

func (*KafkaConsumerStarter) GetUserTopic

func (k *KafkaConsumerStarter) GetUserTopic() string