fmt kafka.go
package queuev2
import (
- "sync"
- "github.com/Shopify/sarama"
- "os"
- "time"
"errors"
- "github.com/wvanbergen/kafka/consumergroup"
+ "github.com/Shopify/sarama"
"github.com/qiniu/log"
+ "github.com/wvanbergen/kafka/consumergroup"
"github.com/wvanbergen/kazoo-go"
+ "os"
+ "sync"
+ "time"
)
const (
@@ -47,7 +47,6 @@ func NewKafkaQueue(name string, hosts []string, exitChan chan struct{}) BackendQ
return nil
}
-
kq := &kafkaQueue{
name: name,
channel: make(chan []byte),
@@ -112,7 +111,7 @@ func createConsumer(hosts []string) (*consumergroup.ConsumerGroup, error) {
config.Zookeeper.Chroot = ""
config.Zookeeper.Timeout = time.Duration(30) * time.Second
config.Offsets.Initial = sarama.OffsetOldest
- topic := []string{Topic,}
+ topic := []string{Topic}
Consumer, consumerErr := consumergroup.JoinConsumerGroup(
Group,
