调整kafka默认配置参数, 去除new client

Author: yumaojun03

bus/broker/kafka/publisher.go

@@ -56,27 +56,20 @@ func (b *Publisher) Connect() error {
 	b.mux.Lock()
 	defer b.mux.Unlock()
 
-	b.l.Debugf("try connect: %v ...", b.conf.Hosts)
-
-	client, err := sarama.NewClient(b.conf.Hosts, b.kc)
-	if err != nil {
-		b.l.Errorf("new kafka client error, %s", err)
-		return err
-	}
-	b.l.Debugf("connect %v success", b.conf.Hosts)
-
 	// try to connect
-	producer, err := sarama.NewAsyncProducerFromClient(client)
+	b.l.Debugf("try connect: %v ...", b.conf.Hosts)
+	producer, err := sarama.NewAsyncProducer(b.conf.Hosts, b.kc)
 	if err != nil {
 		b.l.Errorf("new kafka producer fails with: %+v", err)
 		return err
 	}
 
 	b.producer = producer
 	b.pubChan = producer.Input()
+	b.l.Debugf("connect %v success", b.conf.Hosts)
+
 	go b.watchSuccess(producer.Successes())
 	go b.watchFailed(producer.Errors())
-
 	return nil
 }
 

bus/broker/kafka/subscriber.go

@@ -54,20 +54,15 @@ func (s *Subscriber) Connect() error {
 	s.mux.Lock()
 	defer s.mux.Unlock()
 
+	// try to connect
 	s.l.Debugf("try connect: %v ...", s.conf.Hosts)
-	client, err := sarama.NewClient(s.conf.Hosts, s.kc)
-	if err != nil {
-		s.l.Errorf("new kafka client error, %s", err)
-		return err
-	}
-	s.l.Debugf("connect %v success", s.conf.Hosts)
-
-	consumer, err := sarama.NewConsumerGroupFromClient(s.conf.GroupID, client)
+	consumer, err := sarama.NewConsumerGroup(s.conf.Hosts, s.conf.GroupID, s.kc)
 	if err != nil {
 		s.l.Errorf("Kafka consummer connect fails with: %+v", err)
 		return err
 	}
 	s.comsummer = consumer
+	s.l.Debugf("connect %v success", s.conf.Hosts)
 	return nil
 }