Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix kafka producer and consumer logs show empty objects (#1281)

Browse Source
This commit is contained in:
Suraj Banakar
2019-09-29 23:47:05 +08:00
committed by Ta-Ching Chen
parent 7cdc146515
commit b47b3ecb70
1 changed files with 20 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/mqtrigger/messageQueue/kafka.go
+20 -3
View File
@@ -63,7 +63,10 @@ func makeKafkaMessageQueue(logger *zap.Logger, routerUrl string, mqCfg MessageQu
brokers: strings.Split(mqCfg.Url, ","), brokers: strings.Split(mqCfg.Url, ","),
version: kafkaVersion, version: kafkaVersion,
} }
logger.Info("created kafka queue", zap.Any("kafka", kafka))
logger.Info("created kafka queue", zap.Any("kafka brokers", kafka.brokers),
zap.Any("kafka version", kafka.version))
return kafka, nil return kafka, nil
} }
@@ -81,7 +84,14 @@ func (kafka Kafka) subscribe(trigger *fv1.MessageQueueTrigger) (messageQueueSubs
consumerConfig.Group.Return.Notifications = true consumerConfig.Group.Return.Notifications = true
consumerConfig.Config.Version = kafka.version consumerConfig.Config.Version = kafka.version
consumer, err := cluster.NewConsumer(kafka.brokers, string(trigger.Metadata.UID), []string{trigger.Spec.Topic}, consumerConfig) consumer, err := cluster.NewConsumer(kafka.brokers, string(trigger.Metadata.UID), []string{trigger.Spec.Topic}, consumerConfig)
kafka.logger.Info("created a new consumer", zap.Any("consumer", consumer)) kafka.logger.Info("created a new consumer", zap.Strings("brokers", kafka.brokers),
zap.String("input topic", trigger.Spec.Topic),
zap.String("output topic", trigger.Spec.ResponseTopic),
zap.String("error topic", trigger.Spec.ErrorTopic),
zap.String("trigger name", trigger.Metadata.Name),
zap.String("function namespace", trigger.Metadata.Namespace),
zap.String("function name", trigger.Spec.FunctionReference.Name))
if err != nil { if err != nil {
panic(err) panic(err)
} }
@@ -93,7 +103,14 @@ func (kafka Kafka) subscribe(trigger *fv1.MessageQueueTrigger) (messageQueueSubs
producerConfig.Producer.Return.Successes = true producerConfig.Producer.Return.Successes = true
producerConfig.Version = kafka.version producerConfig.Version = kafka.version
producer, err := sarama.NewSyncProducer(kafka.brokers, producerConfig) producer, err := sarama.NewSyncProducer(kafka.brokers, producerConfig)
kafka.logger.Info("created a new producer", zap.Any("producer", producer)) kafka.logger.Info("created a new producer", zap.Strings("brokers", kafka.brokers),
zap.String("input topic", trigger.Spec.Topic),
zap.String("output topic", trigger.Spec.ResponseTopic),
zap.String("error topic", trigger.Spec.ErrorTopic),
zap.String("trigger name", trigger.Metadata.Name),
zap.String("function namespace", trigger.Metadata.Namespace),
zap.String("function name", trigger.Spec.FunctionReference.Name))
if err != nil { if err != nil {
panic(err) panic(err)
} }