diff --git a/serv.go b/serv.go index 20c96ea..35939b8 100644 --- a/serv.go +++ b/serv.go @@ -6,6 +6,7 @@ import ( "net/http" "os" "os/signal" + "strings" "syscall" "time" @@ -33,7 +34,6 @@ const ( var ( newline = []byte{'\n'} - space = []byte{' '} addr = flag.String("addr", ":9000", "http service address") bootstrapServers = flag.String("bootstrap-servers", "localhost:9092", "kafka bootstrap servers") partition = flag.Int("partition", 0, "kafka tipic partion") @@ -76,21 +76,17 @@ func (c *Client) readPump(topic string) { log.WithFields(log.Fields{ "topic": topic, "bootstrap servers": *bootstrapServers, + "partition": partition, "group": group}).Info("create kafka client") - // consumer, err := kafka.NewConsumer(&kafka.ConfigMap{ - // "bootstrap.servers": *bootstrapServers, - // "group.id": group, - // "session.timeout.ms": 6000, - // "auto.offset.reset": "latest"}) reader := kafka.NewReader(kafka.ReaderConfig{ - Brokers: []string{*bootstrapServers}, + Brokers: strings.Split(*bootstrapServers, ","), Topic: topic, - Partition: 0, + Partition: *partition, MinBytes: 10e1, MaxBytes: 10e6, // 10MB }) - reader.SetOffset(-1) //lastoffset + reader.SetOffset(-1) //latest offset log.WithFields(log.Fields{ "reader": reader,