|
|
|
@ -8,7 +8,6 @@ package kafka
|
|
|
|
|
|
|
|
|
|
import (
|
|
|
|
|
"context"
|
|
|
|
|
"fmt"
|
|
|
|
|
"github.com/Shopify/sarama"
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
@ -29,7 +28,7 @@ func NewMConsumerGroup(consumerConfig *MConsumerGroupConfig, topics, addrs []str
|
|
|
|
|
config.Version = consumerConfig.KafkaVersion
|
|
|
|
|
config.Consumer.Offsets.Initial = consumerConfig.OffsetsInitial
|
|
|
|
|
config.Consumer.Return.Errors = consumerConfig.IsReturnErr
|
|
|
|
|
fmt.Println("init address is ", addrs, "topics is ", topics)
|
|
|
|
|
//fmt.Println("init address is ", addrs, "topics is ", topics)
|
|
|
|
|
consumerGroup, err := sarama.NewConsumerGroup(addrs, groupID, config)
|
|
|
|
|
if err != nil {
|
|
|
|
|
panic(err.Error())
|
|
|
|
|