forked from pd/sarama-cluster
/
main.go
97 lines (84 loc) · 2.67 KB
/
main.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
package main
import (
"flag"
"fmt"
"log"
"os"
"os/signal"
"strings"
"syscall"
"github.com/Shopify/sarama"
"github.com/bsm/sarama-cluster"
)
var (
groupID = flag.String("group", "", "REQUIRED: The shared consumer group name")
brokerList = flag.String("brokers", os.Getenv("KAFKA_PEERS"), "The comma separated list of brokers in the Kafka cluster")
topicList = flag.String("topics", "", "REQUIRED: The comma separated list of topics to consume")
offset = flag.String("offset", "newest", "The offset to start with. Can be `oldest`, `newest`")
verbose = flag.Bool("verbose", false, "Whether to turn on sarama logging")
logger = log.New(os.Stderr, "", log.LstdFlags)
)
func main() {
flag.Parse()
if *groupID == "" {
printUsageErrorAndExit("You have to provide a -group name.")
} else if *brokerList == "" {
printUsageErrorAndExit("You have to provide -brokers as a comma-separated list, or set the KAFKA_PEERS environment variable.")
} else if *topicList == "" {
printUsageErrorAndExit("You have to provide -topics as a comma-separated list.")
}
// Init config
config := cluster.NewConfig()
if *verbose {
sarama.Logger = logger
} else {
config.Consumer.Return.Errors = true
config.Group.Return.Notifications = true
}
switch *offset {
case "oldest":
config.Consumer.Offsets.Initial = sarama.OffsetOldest
case "newest":
config.Consumer.Offsets.Initial = sarama.OffsetNewest
default:
printUsageErrorAndExit("-offset should be `oldest` or `newest`")
}
// Init consumer, consume errors & messages
consumer, err := cluster.NewConsumer(strings.Split(*brokerList, ","), *groupID, strings.Split(*topicList, ","), config)
if err != nil {
printErrorAndExit(69, "Failed to start consumer: %s", err)
}
defer consumer.Close()
// Create signal channel
sigchan := make(chan os.Signal, 1)
signal.Notify(sigchan, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM)
// Consume all channels, wait for signal to exit
for {
select {
case msg, more := <-consumer.Messages():
if more {
fmt.Fprintf(os.Stdout, "%s/%d/%d\t%s\n", msg.Topic, msg.Partition, msg.Offset, msg.Value)
consumer.MarkOffset(msg, "")
}
case ntf, more := <-consumer.Notifications():
if more {
logger.Printf("Rebalanced: %+v\n", ntf)
}
case err, more := <-consumer.Errors():
if more {
logger.Printf("Error: %s\n", err.Error())
}
case <-sigchan:
return
}
}
}
func printErrorAndExit(code int, format string, values ...interface{}) {
fmt.Fprintf(os.Stderr, "ERROR: "+format+"\n\n", values...)
os.Exit(code)
}
func printUsageErrorAndExit(format string, values ...interface{}) {
fmt.Fprintf(os.Stderr, "ERROR: "+format+"\n\n", values...)
flag.Usage()
os.Exit(64)
}