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

@@ -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

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
 }

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)
 			}
 		}
 	}()

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

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")
 		}
 	}