associate partitions with consumer threads

Author: replay

docker/docker-chaos/metrictank.ini

@@ -188,6 +188,8 @@ metrics-buffer-max = 100ms
 channel-buffer-size = 1000
 # The minimum number of message bytes to fetch in a request
 fetch-min = 1
+# Initial maximum number of bytes per topic+partition to request when fetching messages from the broker
+fetch-message-max = 32768
 # The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
 max-wait = 1s
 # Time to wait between attempts to fetch metadata
@@ -302,6 +304,8 @@ channel-buffer-size = 1000
 backlog-process-timeout = 60s
 # The minimum number of message bytes to fetch in a request
 fetch-min = 1
+# Initial maximum number of bytes per topic+partition to request when fetching messages from the broker
+fetch-message-max = 32768
 # The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
 max-wait = 1s
 # Time to wait between attempts to fetch metadata

docker/docker-cluster/metrictank.ini

@@ -188,6 +188,8 @@ metrics-buffer-max = 100ms
 channel-buffer-size = 1000
 # The minimum number of message bytes to fetch in a request
 fetch-min = 1
+# Initial maximum number of bytes per topic+partition to request when fetching messages from the broker
+fetch-message-max = 32768
 # The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
 max-wait = 1s
 # Time to wait between attempts to fetch metadata
@@ -302,6 +304,8 @@ channel-buffer-size = 1000
 backlog-process-timeout = 60s
 # The minimum number of message bytes to fetch in a request
 fetch-min = 1
+# Initial maximum number of bytes per topic+partition to request when fetching messages from the broker
+fetch-message-max = 32768
 # The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
 max-wait = 1s
 # Time to wait between attempts to fetch metadata

docker/docker-dev-custom-cfg-kafka/metrictank.ini

@@ -188,6 +188,8 @@ metrics-buffer-max = 100ms
 channel-buffer-size = 1000
 # The minimum number of message bytes to fetch in a request
 fetch-min = 1
+# Initial maximum number of bytes per topic+partition to request when fetching messages from the broker
+fetch-message-max = 32768
 # The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
 max-wait = 1s
 # Time to wait between attempts to fetch metadata
@@ -302,6 +304,8 @@ channel-buffer-size = 1000
 backlog-process-timeout = 60s
 # The minimum number of message bytes to fetch in a request
 fetch-min = 1
+# Initial maximum number of bytes per topic+partition to request when fetching messages from the broker
+fetch-message-max = 32768
 # The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
 max-wait = 1s
 # Time to wait between attempts to fetch metadata

docs/config.md

@@ -236,6 +236,8 @@ metrics-buffer-max = 100ms
 channel-buffer-size = 1000
 # The minimum number of message bytes to fetch in a request
 fetch-min = 1
+# Initial maximum number of bytes per topic+partition to request when fetching messages from the broker
+fetch-message-max = 32768
 # The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
 max-wait = 1s
 # Time to wait between attempts to fetch metadata
@@ -359,6 +361,8 @@ channel-buffer-size = 1000
 backlog-process-timeout = 60s
 # The minimum number of message bytes to fetch in a request
 fetch-min = 1
+# Initial maximum number of bytes per topic+partition to request when fetching messages from the broker
+fetch-message-max = 32768
 # The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
 max-wait = 1s
 # Time to wait between attempts to fetch metadata

input/kafkamdm/kafkamdm.go

@@ -53,6 +53,7 @@ func ConfigSetup() {
 	inKafkaMdm.DurationVar(&clientConf.BufferMax, "metrics-buffer-max", time.Millisecond*100, "Delay in milliseconds to wait for messages in the producer queue to accumulate before constructing message batches (MessageSets) to transmit to brokers")
 	inKafkaMdm.IntVar(&clientConf.ChannelBufferSize, "channel-buffer-size", 1000, "Maximum number of messages allowed on the producer queue")
 	inKafkaMdm.IntVar(&clientConf.FetchMin, "fetch-min", 1, "Minimum number of bytes the broker responds with. If fetch.wait.max.ms expires the accumulated data will be sent to the client regardless of this setting")
+	inKafkaMdm.IntVar(&clientConf.FetchMessageMax, "fetch-message-max", 32768, "Initial maximum number of bytes per topic+partition to request when fetching messages from the broker.")
 	inKafkaMdm.DurationVar(&clientConf.MaxWait, "max-wait", time.Millisecond*100, "Maximum time the broker may wait to fill the response with fetch.min.bytes")
 	inKafkaMdm.DurationVar(&clientConf.MetadataBackoffTime, "metadata-backoff-time", time.Millisecond*500, "Time to wait between attempts to fetch metadata")
 	inKafkaMdm.IntVar(&clientConf.MetadataRetries, "metadata-retries", 5, "Number of retries to fetch metadata in case of failure")

kafka/consumer.go

@@ -16,17 +16,18 @@ import (
 var LogLevel int
 
 type Consumer struct {
-	conf             ClientConf
-	wg               sync.WaitGroup
-	consumer         *confluent.Consumer
-	Partitions       []int32
-	currentOffsets   map[int32]*int64
-	bootTimeOffsets  map[int32]int64
-	partitionOffset  map[int32]*stats.Gauge64
-	partitionLogSize map[int32]*stats.Gauge64
-	partitionLag     map[int32]*stats.Gauge64
-	LagMonitor       *LagMonitor
-	stopChan         chan struct{}
+	conf               ClientConf
+	wg                 sync.WaitGroup
+	consumer           *confluent.Consumer
+	partitionConsumers map[int32]*confluent.Consumer
+	Partitions         []int32
+	currentOffsets     map[int32]*int64
+	bootTimeOffsets    map[int32]int64
+	partitionOffset    map[int32]*stats.Gauge64
+	partitionLogSize   map[int32]*stats.Gauge64
+	partitionLag       map[int32]*stats.Gauge64
+	LagMonitor         *LagMonitor
+	stopChan           chan struct{}
 }
 
 type ClientConf struct {
@@ -41,6 +42,7 @@ type ClientConf struct {
 	BufferMax             time.Duration
 	ChannelBufferSize     int
 	FetchMin              int
+	FetchMessageMax       int
 	NetMaxOpenRequests    int
 	MaxWait               time.Duration
 	SessionTimeout        time.Duration
@@ -63,47 +65,49 @@ func (c *ClientConf) OffsetIsValid() bool {
 	return true
 }
 
-func (c *ClientConf) GetConfluentConfig() *confluent.ConfigMap {
-	return GetConfig(c.Broker, c.ClientID, "snappy", c.BatchNumMessages, int(c.BufferMax/time.Millisecond), c.ChannelBufferSize, c.FetchMin, c.NetMaxOpenRequests, int(c.MaxWait/time.Millisecond), int(c.SessionTimeout/time.Millisecond))
-}
-
 func NewConfig() *ClientConf {
 	return &ClientConf{
 		GaugePrefix:           "default.kafka.partition",
 		BatchNumMessages:      10000,
 		BufferMax:             time.Millisecond * 100,
 		ChannelBufferSize:     1000000,
 		FetchMin:              1,
-		NetMaxOpenRequests:    100,
-		MaxWait:               time.Millisecond * 100,
+		FetchMessageMax:       32768,
+		MaxWait:               time.Second * 1,
 		SessionTimeout:        time.Second * 30,
+		NetMaxOpenRequests:    1000,
 		MetadataRetries:       5,
 		MetadataBackoffTime:   time.Millisecond * 500,
 		MetadataTimeout:       time.Second * 10,
 		LagCollectionInterval: time.Second * 5,
 	}
 }
 
+func (c *ClientConf) GetConfluentConfig(clientId string) *confluent.ConfigMap {
+	conf := GetConfig(c.Broker, "snappy", c.BatchNumMessages, int(c.BufferMax/time.Millisecond), c.ChannelBufferSize, c.FetchMin, c.FetchMessageMax, c.NetMaxOpenRequests, int(c.MaxWait/time.Millisecond), int(c.SessionTimeout/time.Millisecond))
+	conf.SetKey("group.id", clientId)
+	return conf
+}
+
 func NewConsumer(conf *ClientConf) (*Consumer, error) {
 	if len(conf.Topics) < 1 {
 		return nil, fmt.Errorf("kafka-consumer: Requiring at least 1 topic")
 	}
 
-	clientConf := conf.GetConfluentConfig()
-	clientConf.SetKey("retries", 10)
-	clientConf.SetKey("enable.partition.eof", false)
-	clientConf.SetKey("enable.auto.offset.store", false)
-	clientConf.SetKey("enable.auto.commit", false)
-	clientConf.SetKey("go.events.channel.enable", true)
-	clientConf.SetKey("go.application.rebalance.enable", true)
-	consumer, err := confluent.NewConsumer(clientConf)
-	if err != nil {
-		return nil, err
+	getConsumer := func(clientId string) (*confluent.Consumer, error) {
+		clientConf := conf.GetConfluentConfig(clientId)
+		clientConf.SetKey("retries", 10)
+		clientConf.SetKey("enable.partition.eof", false)
+		clientConf.SetKey("enable.auto.offset.store", false)
+		clientConf.SetKey("enable.auto.commit", false)
+		clientConf.SetKey("go.events.channel.enable", true)
+		clientConf.SetKey("go.events.channel.size", 100000)
+		clientConf.SetKey("go.application.rebalance.enable", true)
+		return confluent.NewConsumer(clientConf)
 	}
 
 	c := Consumer{
 		conf:             *conf,
-		consumer:         consumer,
 		currentOffsets:   make(map[int32]*int64),
 		bootTimeOffsets:  make(map[int32]int64),
 		partitionOffset:  make(map[int32]*stats.Gauge64),
@@ -112,6 +116,12 @@ func NewConsumer(conf *ClientConf) (*Consumer, error) {
 		stopChan:         make(chan struct{}),
 	}
 
+	var err error
+	c.consumer, err = getConsumer(conf.ClientID + "-metadata")
+	if err != nil {
+		return nil, err
+	}
+
 	availParts, err := GetPartitions(c.consumer, c.conf.Topics, c.conf.MetadataRetries, int(c.conf.MetadataTimeout/time.Millisecond), c.conf.MetadataBackoffTime)
 	if err != nil {
 		return nil, err
@@ -135,8 +145,14 @@ func NewConsumer(conf *ClientConf) (*Consumer, error) {
 		}
 	}
 
+	c.partitionConsumers = make(map[int32]*confluent.Consumer, len(c.Partitions))
 	for _, part := range c.Partitions {
-		_, offset, err := c.consumer.QueryWatermarkOffsets(c.conf.Topics[0], part, int(c.conf.MetadataTimeout/time.Millisecond))
+		c.partitionConsumers[part], err = getConsumer(fmt.Sprintf("%s-partition-%d", conf.ClientID, part))
+		if err != nil {
+			return nil, err
+		}
+
+		_, offset, err := c.partitionConsumers[part].QueryWatermarkOffsets(c.conf.Topics[0], part, int(c.conf.MetadataTimeout/time.Millisecond))
 		if err != nil {
 			return nil, fmt.Errorf("Failed to get newest offset for topic %s part %d: %s", c.conf.Topics[0], part, err)
 		}
@@ -164,8 +180,8 @@ func (c *Consumer) Start(processBacklog *sync.WaitGroup) error {
 
 	go c.monitorLag(processBacklog)
 
-	for range c.Partitions {
-		go c.consume()
+	for _, partition := range c.Partitions {
+		go c.consume(partition)
 	}
 
 	return nil
@@ -202,13 +218,18 @@ func (c *Consumer) StartAndAwaitBacklog(backlogProcessTimeout time.Duration) err
 	return nil
 }
 
-func (c *Consumer) consume() {
+func (c *Consumer) consume(partition int32) {
 	c.wg.Add(1)
 	defer c.wg.Done()
 
-	var ok bool
 	var offsetPtr *int64
-	events := c.consumer.Events()
+	var ok bool
+	if offsetPtr, ok = c.currentOffsets[partition]; !ok || offsetPtr == nil {
+		log.Fatal(3, "kafka-consumer: Failed to get currentOffset for partition %d", partition)
+	}
+
+	log.Info("kafka-consumer: Consumer started for partition %d of topics %+v", partition, c.conf.Topics)
+	events := c.partitionConsumers[partition].Events()
 	for {
 		select {
 		case ev := <-events:
@@ -221,23 +242,18 @@ func (c *Consumer) consume() {
 				log.Info("kafka-consumer: Revoked partitions: %+v", e)
 			case *confluent.Message:
 				tp := e.TopicPartition
-				if LogLevel < 2 {
-					log.Debug("kafka-consumer: Received message: Topic %s, Partition: %d, Offset: %d, Key: %x", tp.Topic, tp.Partition, tp.Offset, e.Key)
-				}
-
-				if offsetPtr, ok = c.currentOffsets[tp.Partition]; !ok || offsetPtr == nil {
-					log.Error(3, "kafka-consumer: Received message of unexpected partition: %s:%d", tp.Topic, tp.Partition)
+				if tp.Partition != partition {
+					fmt.Println("received unexpected partition")
 					continue
 				}
-
 				c.conf.MessageHandler(e.Value, tp.Partition)
 				atomic.StoreInt64(offsetPtr, int64(tp.Offset))
 			case *confluent.Error:
 				log.Error(3, "kafka-consumer: Kafka consumer error: %s", e.String())
 				return
 			}
 		case <-c.stopChan:
-			log.Info("kafka-consumer: Consumer ended.")
+			log.Info("kafka-consumer: Consumer ended for partition %d topics %+v", partition, c.conf.Topics)
 			return
 		}
 	}
@@ -302,12 +318,14 @@ func (c *Consumer) monitorLag(processBacklog *sync.WaitGroup) {
 func (c *Consumer) startConsumer() error {
 	var offset confluent.Offset
 	var err error
-	var topicPartitions confluent.TopicPartitions
 	c.currentOffsets = make(map[int32]*int64, len(c.Partitions))
 
-	for i, topic := range c.conf.Topics {
-		for _, partition := range c.Partitions {
-			var currentOffset int64
+	for _, partition := range c.Partitions {
+		var currentOffset int64
+		var topicPartitions confluent.TopicPartitions
+		c.currentOffsets[partition] = &currentOffset
+
+		for _, topic := range c.conf.Topics {
 			switch c.conf.StartAtOffset {
 			case "oldest":
 				currentOffset, err = c.tryGetOffset(topic, partition, int64(confluent.OffsetBeginning), 3, time.Second)
@@ -345,14 +363,15 @@ func (c *Consumer) startConsumer() error {
 				Partition: partition,
 				Offset:    offset,
 			})
+		}
 
-			if i == 0 {
-				c.currentOffsets[partition] = &currentOffset
-			}
+		err := c.partitionConsumers[partition].Assign(topicPartitions)
+		if err != nil {
+			return err
 		}
 	}
 
-	return c.consumer.Assign(topicPartitions)
+	return nil
 }
 
 func (c *Consumer) tryGetOffset(topic string, partition int32, offsetI int64, attempts int, sleep time.Duration) (int64, error) {
@@ -403,4 +422,7 @@ func (c *Consumer) Stop() {
 	close(c.stopChan)
 	c.wg.Wait()
 	c.consumer.Close()
+	for i := range c.partitionConsumers {
+		c.partitionConsumers[i].Close()
+	}
 }

kafka/utils.go

@@ -4,12 +4,12 @@ import (
 	confluent "github.com/confluentinc/confluent-kafka-go/kafka"
 )
 
-func GetConfig(broker, client, compression string, batchNumMessages, bufferMaxMs, bufferSize, fetchMin, maxOpenRequests, maxWaitMs, sessionTimeoutMs int) *confluent.ConfigMap {
+func GetConfig(broker, compression string, batchNumMessages, bufferMaxMs, bufferSize, fetchMin, fetchMessageMax, maxOpenRequests, maxWaitMs, sessionTimeoutMs int) *confluent.ConfigMap {
 	return &confluent.ConfigMap{
 		"bootstrap.servers":                     broker,
 		"compression.codec":                     "snappy",
-		"group.id":                              client,
 		"fetch.min.bytes":                       fetchMin,
+		"fetch.message.max.bytes":               fetchMessageMax,
 		"fetch.wait.max.ms":                     maxWaitMs,
 		"max.in.flight.requests.per.connection": maxOpenRequests,
 		"queue.buffering.max.messages":          bufferSize,

mdata/notifierKafka/cfg.go

@@ -31,6 +31,7 @@ func init() {
 	fs.DurationVar(&clientConf.BufferMax, "metrics-buffer-max", time.Millisecond*100, "Delay to wait for messages in the producer queue to accumulate before constructing message batches (MessageSets) to transmit to brokers")
 	fs.IntVar(&clientConf.ChannelBufferSize, "channel-buffer-size", 1000000, "Maximum number of messages allowed on the producer queue")
 	fs.IntVar(&clientConf.FetchMin, "fetch-min", 1, "Minimum number of bytes the broker responds with. If fetch.wait.max.ms expires the accumulated data will be sent to the client regardless of this setting")
+	fs.IntVar(&clientConf.FetchMessageMax, "fetch-message-max", 32768, "Initial maximum number of bytes per topic+partition to request when fetching messages from the broker.")
 	fs.DurationVar(&clientConf.MaxWait, "max-wait", time.Millisecond*100, "Maximum time the broker may wait to fill the response with fetch.min.bytes")
 	fs.DurationVar(&clientConf.MetadataBackoffTime, "metadata-backoff-time", time.Millisecond*500, "Time to wait between attempts to fetch metadata")
 	fs.IntVar(&clientConf.MetadataRetries, "metadata-retries", 5, "Number of retries to fetch metadata in case of failure")

mdata/notifierKafka/notifierKafka.go

@@ -34,7 +34,7 @@ type NotifierKafka struct {
 
 func New(instance string, metrics mdata.Metrics, idx idx.MetricIndex) *NotifierKafka {
 	clientConf.ClientID = instance + "-notifier-producer"
-	producer, err := confluent.NewProducer(clientConf.GetConfluentConfig())
+	producer, err := confluent.NewProducer(clientConf.GetConfluentConfig("notifier"))
 
 	if err != nil {
 		log.Fatal(2, "kafka-cluster failed to initialize producer: %s", err)

metrictank-sample.ini

@@ -191,6 +191,8 @@ metrics-buffer-max = 100ms
 channel-buffer-size = 1000
 # The minimum number of message bytes to fetch in a request
 fetch-min = 1
+# Initial maximum number of bytes per topic+partition to request when fetching messages from the broker
+fetch-message-max = 32768
 # The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
 max-wait = 1s
 # Time to wait between attempts to fetch metadata
@@ -305,6 +307,8 @@ channel-buffer-size = 1000
 backlog-process-timeout = 60s
 # The minimum number of message bytes to fetch in a request
 fetch-min = 1
+# Initial maximum number of bytes per topic+partition to request when fetching messages from the broker
+fetch-message-max = 32768
 # The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
 max-wait = 1s
 # Time to wait between attempts to fetch metadata

scripts/config/metrictank-docker.ini

@@ -188,6 +188,8 @@ metrics-buffer-max = 100ms
 channel-buffer-size = 1000
 # The minimum number of message bytes to fetch in a request
 fetch-min = 1
+# Initial maximum number of bytes per topic+partition to request when fetching messages from the broker
+fetch-message-max = 32768
 # The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
 max-wait = 1s
 # Time to wait between attempts to fetch metadata
@@ -302,6 +304,8 @@ channel-buffer-size = 1000
 backlog-process-timeout = 60s
 # The minimum number of message bytes to fetch in a request
 fetch-min = 1
+# Initial maximum number of bytes per topic+partition to request when fetching messages from the broker
+fetch-message-max = 32768
 # The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
 max-wait = 1s
 # Time to wait between attempts to fetch metadata