func Consumer(s bandmaster.Service) *sarama_cluster.Consumer {
	return s.(*Service).c  
}
-----------------------------------------------------------------------------
Consumer returns the underlying Kafka consumer of the given service.

It assumes that the service is ready; i.e. it might return nil if it's
actually not.

NOTE: This will panic if `s` is not a `kafka.Service`.
package kafka
import (
	"context"
	"github.com/Shopify/sarama"
	sarama_cluster "github.com/bsm/sarama-cluster"
	"github.com/znly/bandmaster"
)
type Service struct {
	*bandmaster.ServiceBase  
	ctx       context.Context  
	canceller context.CancelFunc
	conf *Config
	c *sarama_cluster.Consumer
	p sarama.AsyncProducer
}
type Config struct {
	ClusterConf *sarama_cluster.Config
	Addrs           []string
	ConsumerTopics  []string
	ConsumerGroupID string
}
func New(conf *Config) bandmaster.Service {
	ctx, canceller := context.WithCancel(context.Background())
	return &Service{
		ServiceBase: bandmaster.NewServiceBase(),  
		ctx:       ctx,
		canceller: canceller,
		conf: conf,
	}
}
func (s *Service) Start(context.Context, map[string]bandmaster.Service) error {
	var err error
	if err = s.conf.ClusterConf.Validate(); err != nil {
		return err
	}
	if s.c == nil {  
		if len(s.conf.ConsumerGroupID) > 0 && len(s.conf.ConsumerTopics) > 0 {
			s.c, err = sarama_cluster.NewConsumer(s.conf.Addrs,
				s.conf.ConsumerGroupID, s.conf.ConsumerTopics, s.conf.ClusterConf,
			)
			if err != nil {
				return err
			}
		}
	}
	if s.p == nil {  
		s.p, err = sarama.NewAsyncProducer(
			s.conf.Addrs, &s.conf.ClusterConf.Config)
		if err != nil {
			return err
		}
	}
	return nil
}
func (s *Service) Stop(context.Context) error {
	s.canceller()
	if s.c != nil {
		if err := s.c.Close(); err != nil {
			return err
		}
		s.c = nil  
	}
	if s.p != nil {
		if err := s.p.Close(); err != nil {
			return err
		}
		s.c = nil  
	}
	return nil
}