func SetupConsumer(config ConsumerConfig) (Consumer, error) {
	// default error handler
	if config.OnError == nil {
		config.OnError = func(err error) {
			log.Error(err)
		}
	}

	if config.OnNotification == nil {
		config.OnNotification = func(notification *cluster.Notification) {
			log.Infof("(Rebalance) Notification: [%+v]", notification)
		}
	}

	brokerList := strings.Split(config.Brokers, ",")
	for k, v := range brokerList {
		brokerList[k] = strings.TrimSpace(v)
	}
	if len(brokerList) == 0 {
		return nil, errors.New("The Kafka consumer requires at least 1 broker to resolve topics.")
	}

	topicList := strings.Split(config.Topics, ",")
	for k, v := range topicList {
		topicList[k] = strings.TrimSpace(v)
	}
	if len(topicList) == 0 {
		return nil, errors.New("The Kafka consumer requires at least 1 topic to consume.")
	}

	clusterConfig := cluster.NewConfig()
	clusterConfig.Consumer.Return.Errors = true

	if config.ParitionStrategy == "" {
		clusterConfig.Group.PartitionStrategy = cluster.StrategyRoundRobin
	} else {
		clusterConfig.Group.PartitionStrategy = config.ParitionStrategy
	}

	if config.InitialOffset == 0 {
		clusterConfig.Consumer.Offsets.Initial = OffsetOldest
	} else {
		clusterConfig.Consumer.Offsets.Initial = config.InitialOffset
	}

	if config.SessionTimeout > 0 {
		clusterConfig.Group.Session.Timeout = time.Duration(config.SessionTimeout) * time.Second
	}

	if config.MetricsRegistry != nil {
		clusterConfig.MetricRegistry = metrics.NewPrefixedChildRegistry(config.MetricsRegistry, "kafka-")
	}

	clusterConsumer, err := cluster.NewConsumer(brokerList, config.ConsumerGroup, topicList, clusterConfig)
	if err != nil {
		return nil, err
	}

	// handle messages
	msgs := make(chan *ConsumerMessage, config.BufferSize)
	go func() {
		defer close(msgs)

		for {
			msg, ok := <-clusterConsumer.Messages()
			if !ok {
				log.Infof("Consumer message channel for [%s] closed", config.Topics)
				return
			}

			msgs <- &ConsumerMessage{
				Key:       msg.Key,
				Value:     msg.Value,
				Topic:     msg.Topic,
				Partition: msg.Partition,
				Offset:    msg.Offset,
				Timestamp: msg.Timestamp,
			}
		}
	}()

	// handle errors
	go func() {
		for {
			err, ok := <-clusterConsumer.Errors()
			if !ok {
				log.Infof("Consumer error channel for [%s] closed", config.Topics)
				return
			}
			config.OnError(err)
		}
	}()

	// handle notifications
	go func() {
		for {
			notification, ok := <-clusterConsumer.Notifications()
			if !ok {
				log.Infof("Consumer notifications channel for [%s] closed", config.Topics)
				return
			}
			config.OnNotification(notification)
		}
	}()
	return &consumerImp{config: config, clusterConsumer: clusterConsumer, messages: msgs}, nil
}
// MustSetupConsumer creates a consumer for the config.InputTopics on config.InputBrokers
MustSetupConsumer creates a consumer for the config.InputTopics on config.InputBrokers
// MustSetupConsumer creates a consumer for the config . InputTopics on config . InputBrokers
func SetupConsumer(config ConsumerConfig) (Consumer, error) {
	if config.OnError == nil {
		config.OnError = func(err error) {
			log.Error(err)
		}
	}
	if config.OnNotification == nil {
		config.OnNotification = func(notification *cluster.Notification) {
			log.Infof("(Rebalance) Notification: [%+v]", notification)
		}
	}
	brokerList := strings.Split(config.Brokers, ",")
	for k, v := range brokerList {
		brokerList[k] = strings.TrimSpace(v)
	}
	if len(brokerList) == 0 {
		return nil, errors.New("The Kafka consumer requires at least 1 broker to resolve topics.")
	}
	topicList := strings.Split(config.Topics, ",")
	for k, v := range topicList {
		topicList[k] = strings.TrimSpace(v)
	}
	if len(topicList) == 0 {
		return nil, errors.New("The Kafka consumer requires at least 1 topic to consume.")
	}
	clusterConfig := cluster.NewConfig()
	clusterConfig.Consumer.Return.Errors = true
	if config.ParitionStrategy == "" {
		clusterConfig.Group.PartitionStrategy = cluster.StrategyRoundRobin
	} else {
		clusterConfig.Group.PartitionStrategy = config.ParitionStrategy
	}
	if config.InitialOffset == 0 {
		clusterConfig.Consumer.Offsets.Initial = OffsetOldest
	} else {
		clusterConfig.Consumer.Offsets.Initial = config.InitialOffset
	}
	if config.SessionTimeout > 0 {
		clusterConfig.Group.Session.Timeout = time.Duration(config.SessionTimeout) * time.Second
	}
	if config.MetricsRegistry != nil {
		clusterConfig.MetricRegistry = metrics.NewPrefixedChildRegistry(config.MetricsRegistry, "kafka-")
	}
	clusterConsumer, err := cluster.NewConsumer(brokerList, config.ConsumerGroup, topicList, clusterConfig)
	if err != nil {
		return nil, err
	}
	msgs := make(chan *ConsumerMessage, config.BufferSize)
	go func() {
		defer close(msgs)
		for {
			msg, ok := <-clusterConsumer.Messages()
			if !ok {
				log.Infof("Consumer message channel for [%s] closed", config.Topics)
				return
			}
			msgs <- &ConsumerMessage{
				Key:       msg.Key,
				Value:     msg.Value,
				Topic:     msg.Topic,
				Partition: msg.Partition,
				Offset:    msg.Offset,
				Timestamp: msg.Timestamp,
			}
		}
	}()
	go func() {
		for {
			err, ok := <-clusterConsumer.Errors()
			if !ok {
				log.Infof("Consumer error channel for [%s] closed", config.Topics)
				return
			}
			config.OnError(err)
		}
	}()
	go func() {
		for {
			notification, ok := <-clusterConsumer.Notifications()
			if !ok {
				log.Infof("Consumer notifications channel for [%s] closed", config.Topics)
				return
			}
			config.OnNotification(notification)
		}
	}()
	return &consumerImp{config: config, clusterConsumer: clusterConsumer, messages: msgs}, nil
}
func SetupConsumer ( config ConsumerConfig ) ( Consumer , error ) { if config . OnError == nil { config . OnError = func ( err error ) { log . Error ( err ) } } if config . OnNotification == nil { config . OnNotification = func ( notification * cluster . Notification ) { log . Infof ( " " , notification ) } } brokerList := strings . Split ( config . Brokers , " " ) for k , v := range brokerList { brokerList [ k ] = strings . TrimSpace ( v ) } if len ( brokerList ) == 0 { return nil , errors . New ( " " ) } topicList := strings . Split ( config . Topics , " " ) for k , v := range topicList { topicList [ k ] = strings . TrimSpace ( v ) } if len ( topicList ) == 0 { return nil , errors . New ( " " ) } clusterConfig := cluster . NewConfig ( ) clusterConfig . Consumer . Return . Errors = true if config . ParitionStrategy == " " { clusterConfig . Group . PartitionStrategy = cluster . StrategyRoundRobin } else { clusterConfig . Group . PartitionStrategy = config . ParitionStrategy } if config . InitialOffset == 0 { clusterConfig . Consumer . Offsets . Initial = OffsetOldest } else { clusterConfig . Consumer . Offsets . Initial = config . InitialOffset } if config . SessionTimeout > 0 { clusterConfig . Group . Session . Timeout = time . Duration ( config . SessionTimeout ) * time . Second } if config . MetricsRegistry != nil { clusterConfig . MetricRegistry = metrics . NewPrefixedChildRegistry ( config . MetricsRegistry , " " ) } clusterConsumer , err := cluster . NewConsumer ( brokerList , config . ConsumerGroup , topicList , clusterConfig ) if err != nil { return nil , err } msgs := make ( chan * ConsumerMessage , config . BufferSize ) go func ( ) { defer close ( msgs ) for { msg , ok := <- clusterConsumer . Messages ( ) if ! ok { log . Infof ( " " , config . Topics ) return } msgs <- & ConsumerMessage { Key : msg . Key , Value : msg . Value , Topic : msg . Topic , Partition : msg . Partition , Offset : msg . Offset , Timestamp : msg . Timestamp , } } } ( ) go func ( ) { for { err , ok := <- clusterConsumer . Errors ( ) if ! ok { log . Infof ( " " , config . Topics ) return } config . OnError ( err ) } } ( ) go func ( ) { for { notification , ok := <- clusterConsumer . Notifications ( ) if ! ok { log . Infof ( " " , config . Topics ) return } config . OnNotification ( notification ) } } ( ) return & consumerImp { config : config , clusterConsumer : clusterConsumer , messages : msgs } , nil }
MustSetupConsumer creates a consumer for the config.InputTopics on config.InputBrokers
MustSetupConsumer creates a consumer for the config . InputTopics on config . InputBrokers
// default error handler // handle messages // handle errors // handle notifications