From 17cdc58d800356741f47b505eaba4df397c984c8 Mon Sep 17 00:00:00 2001 From: Jacob Brinlee <66123601+jrbrinlee1@users.noreply.github.com> Date: Wed, 18 Jan 2023 17:42:19 -0600 Subject: [PATCH] SUP-288 (#2414) * adding kafka consumer config options (--kafka-max-poll-interval, --kafka-session-timeout, --kafka-group-instance-id, --kafka-socket-keepalive-enable, and --consumer-close-timeout) * wrapping consumer.Close() in timeout. Will wait consumer-close-timeout seconds before forcing consumer to exit * clean up logs --- batch/batch.go | 4 ++-- idk/common/configMap.go | 29 +++++++++++++++++++++++++++++ idk/ingest.go | 10 +++++++--- idk/kafka/cmd.go | 20 +++++++++++++++----- idk/kafka/source.go | 41 ++++++++++++++++++++++++++++++----------- 5 files changed, 83 insertions(+), 21 deletions(-) diff --git a/batch/batch.go b/batch/batch.go index 8ee02d41c..c33093474 100644 --- a/batch/batch.go +++ b/batch/batch.go @@ -1245,7 +1245,7 @@ func (b *Batch) doImport(frags, clearFrags fragments) error { } ferr := b.importer.ImportRoaringBitmap(ctx, b.tbl.ID, fld, shard, viewMap, false) - b.log.Debugf("imp-roar %s,shard:%d,views:%d %v", field, shard, len(clearViewMap), time.Since(starty)) + b.log.Debugf("imp-roar field: %s, shard:%d, views:%d %v", field, shard, len(clearViewMap), time.Since(starty)) return errors.Wrapf(ferr, "importing data for %s", field) }) } @@ -1703,7 +1703,7 @@ func (b *Batch) importValueData() error { start := time.Now() fld := featurebase.FieldInfoToField(field) err := b.importer.DoImport(ctx, b.tbl.ID, fld, shard, path, data) - b.log.Debugf("imp-vals %s,shard:%d,data:%d %v", field, shard, len(data), time.Since(start)) + b.log.Debugf("imp-vals field: %s, shard: %d, data: %d %v", field.Name, shard, len(data), time.Since(start)) return errors.Wrapf(err, "importing values for field = %s", field.Name) }) startIdx = i diff --git a/idk/common/configMap.go b/idk/common/configMap.go index 9b6b80e24..82aa8386a 100644 --- a/idk/common/configMap.go +++ b/idk/common/configMap.go @@ -44,6 +44,8 @@ func LaunchKafkaEventConfirmer(producer *confluent.Producer, finished *int32, it return doneChan } +// For a list of confluent consumer configuraiton options, go here: +// https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md func SetupConfluent(m *idk.ConfluentCommand) (*confluent.ConfigMap, error) { var err error configMap := &confluent.ConfigMap{} @@ -188,5 +190,32 @@ func SetupConfluent(m *idk.ConfluentCommand) (*confluent.ConfigMap, error) { } } + if m.KafkaGroupInstanceId != "" { + err = configMap.SetKey("group.instance.id", m.KafkaGroupInstanceId) + if err != nil { + return nil, err + } + } + + if m.KafkaMaxPollInterval != "" { + err = configMap.SetKey("max.poll.interval.ms", m.KafkaMaxPollInterval) + if err != nil { + return nil, err + } + } + + if m.KafkaSessionTimeout != "" { + err = configMap.SetKey("session.timeout.ms", m.KafkaSessionTimeout) + if err != nil { + return nil, err + } + } + + if m.KafkaSocketKeepaliveEnable != "" { + err = configMap.SetKey("socket.keepalive.enable", m.KafkaSocketKeepaliveEnable) + if err != nil { + return nil, err + } + } return configMap, nil } diff --git a/idk/ingest.go b/idk/ingest.go index a6829a874..9081d0e3e 100644 --- a/idk/ingest.go +++ b/idk/ingest.go @@ -181,9 +181,13 @@ type ConfluentCommand struct { KafkaSslEndpointIdentificationAlgorithm string `help:"The endpoint identification algorithm used by clients to validate server host name (ssl.endpoint.identification.algorithm) "` KafkaEnableSslCertificateVerification bool `help:"(enable.ssl.certificate.verification)"` KafkaSocketTimeoutMs int `help:"(socket.timeout.ms)"` + KafkaSocketKeepaliveEnable string `help:"The (socket.keepalive.enable) kafka consumer configuration"` - KafkaClientId string `help:"(client.id)"` - KafkaDebug string `help:"Kafka debug string (debug)"` + KafkaClientId string `help:"(client.id)"` + KafkaDebug string `help:"The (debug) kafka consumer configuration. A comma-separated list of debug contexts to enable. Detailed Consumer: consumer,cgrp,topic,fetch. Set to 'all' for most verbose option."` + KafkaMaxPollInterval string `help:"The (max.poll.interval.ms) kafka consumer configuration. The max time the consumer can go without polling the broker. Consumer exits after this timeout."` + KafkaSessionTimeout string `help:"The (session.timeout.ms) kafka consumer configuration. The max time the consumer can go without sending a heartbeat to the broker"` + KafkaGroupInstanceId string `help:"The (group.instance.id) kafka consumer configuration."` KafkaSaslUsername string `help:"SASL authentication username (sasl.username)"` KafkaSaslPassword string `help:"SASL authentication password (sasl.password)"` @@ -309,7 +313,7 @@ func (m *Main) runIngester(c int, l *msgCounter) error { err = source.Close() if err != nil { if m.log != nil { - m.log.Printf("error on close %v", err) + m.log.Errorf("Closing source: %v", err) } } }() diff --git a/idk/kafka/cmd.go b/idk/kafka/cmd.go index 44efbcebd..f1aebad8b 100644 --- a/idk/kafka/cmd.go +++ b/idk/kafka/cmd.go @@ -13,7 +13,8 @@ type Main struct { Group string `help:"Kafka group."` Topics []string `help:"Kafka topics to read from."` Timeout time.Duration `help:"Time to wait for more records from Kafka before flushing a batch. 0 to disable."` - SkipOld bool `short:"" help:"Skip to the most recent Kafka message rather than starting at the beginning."` + SkipOld bool `short:"" help:"False sets kafka consumer configuration auto.offset.reset to earliest, True sets it to latest."` + ConsumerCloseTimeout int `help:"The amount of time in seconds to wait for the consumer to close properly."` } func NewMain() (*Main, error) { @@ -22,13 +23,16 @@ func NewMain() (*Main, error) { ConfluentCommand: idk.ConfluentCommand{ KafkaBootstrapServers: []string{"localhost:9092"}, }, - Group: "defaultgroup", - Topics: []string{"defaulttopic"}, - Timeout: time.Second, + Group: "defaultgroup", + Topics: []string{"defaulttopic"}, + Timeout: time.Second, + ConsumerCloseTimeout: 30, } + m.SchemaRegistryURL = "http://" + defaultRegistryHost - m.OffsetMode = true m.Main.Namespace = "ingester_kafka" + //m.Main.OffsetMode = m.OffsetMode + m.OffsetMode = true m.NewSource = func() (idk.Source, error) { source := NewSource() source.KafkaBootstrapServers = m.KafkaBootstrapServers @@ -43,6 +47,12 @@ func NewMain() (*Main, error) { source.SchemaRegistryUsername = m.SchemaRegistryUsername source.SchemaRegistryPassword = m.SchemaRegistryPassword source.Verbose = m.Verbose + source.KafkaMaxPollInterval = m.KafkaMaxPollInterval + source.KafkaSessionTimeout = m.KafkaSessionTimeout + source.KafkaGroupInstanceId = m.KafkaGroupInstanceId + source.KafkaDebug = m.KafkaDebug + source.KafkaSocketKeepaliveEnable = m.KafkaSocketKeepaliveEnable + source.consumerCloseTimeout = m.ConsumerCloseTimeout if err := source.Open(); err != nil { return nil, errors.Wrap(err, "opening source") diff --git a/idk/kafka/source.go b/idk/kafka/source.go index 7ac92c479..a2f4ac3c9 100644 --- a/idk/kafka/source.go +++ b/idk/kafka/source.go @@ -33,14 +33,15 @@ import ( // achieve concurrency, create multiple Sources. type Source struct { idk.ConfluentCommand - Topics []string - Group string - Log logger.Logger - Timeout time.Duration - SkipOld bool - Verbose bool - schema Schema - TLS idk.TLSConfig + Topics []string + Group string + Log logger.Logger + Timeout time.Duration + SkipOld bool + Verbose bool + schema Schema + TLS idk.TLSConfig + consumerCloseTimeout int spoolBase uint64 spool []confluent.TopicPartition @@ -262,7 +263,9 @@ func (s *Source) CommitMessages(recs []confluent.TopicPartition) ([]confluent.To return s.client.CommitOffsets(recs) } -// Open initializes the kafka source. +// Open initializes the kafka source. (i.e. creating and configuring a consumer) +// The configuration options for the confluentinc/confluent-kafka-go/kafka +// libarary are: https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md func (s *Source) Open() error { cfg, err := common.SetupConfluent(&s.ConfluentCommand) if err != nil { @@ -444,10 +447,26 @@ func (c *Source) generator() { func (s *Source) Close() error { if s.client != nil { if s.opened { // only close opened sources + var err error + closedReturned := make(chan error, 1) + // send quit message to polling routine & wait for it to exit s.quit <- struct{}{} s.wg.Wait() - err := s.client.Close() - s.opened = false + s.Log.Debugf("Trying to close consumer %s...", s.client.String()) + go func() { + closedReturned <- s.client.Close() + }() + start := time.Now() + select { + case err = <-closedReturned: + if err == nil { + s.Log.Debugf("Successfully closed consumer %s!", s.client.String()) + s.opened = false + } + case <-time.After(time.Duration(s.consumerCloseTimeout * 1000 * 1000 * 1000)): + err = fmt.Errorf("Unable to properly close consumer %s after %f seconds", s.client.String(), time.Since(start).Seconds()) + } + return errors.Wrap(err, "closing kafka consumer") } }