93 lines
2.1 KiB
Go
93 lines
2.1 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"flag"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/twmb/franz-go/pkg/kadm"
|
|
"github.com/twmb/franz-go/pkg/kgo"
|
|
"github.com/twmb/franz-go/pkg/sasl/plain"
|
|
)
|
|
|
|
func main() {
|
|
ctx := context.Background()
|
|
|
|
var password, hostname string
|
|
var replication int
|
|
flag.StringVar(&password, "p", "", "password to connect to kafka")
|
|
flag.StringVar(&hostname, "h", "kafka", "hostname to connect to kafka")
|
|
flag.IntVar(&replication, "r", 3, "number of replication")
|
|
flag.Parse()
|
|
|
|
seeds := []string{fmt.Sprintf("%s.zerops:9092", hostname)}
|
|
cl, err := kgo.NewClient(
|
|
kgo.SeedBrokers(seeds...),
|
|
kgo.SASL(plain.Auth{
|
|
User: "zerops",
|
|
Pass: password,
|
|
}.AsMechanism()),
|
|
)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
defer cl.Close()
|
|
|
|
adminClient := kadm.NewClient(cl)
|
|
topic, err := adminClient.ListTopics(ctx, "foo")
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
// if topic.Error() != nil {
|
|
// panic(topic.Error())
|
|
// }
|
|
|
|
if !topic.Has("foo") {
|
|
partitions := int32(6)
|
|
if replication == 1 {
|
|
partitions = 1
|
|
}
|
|
resp, err := adminClient.CreateTopic(ctx, partitions, int16(replication), nil, "foo")
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
if resp.Err != nil {
|
|
panic(resp.Err)
|
|
}
|
|
}
|
|
|
|
go func() {
|
|
for {
|
|
message := time.Now().Format(time.TimeOnly)
|
|
fmt.Printf("%s ==>\n", message)
|
|
// Alternatively, ProduceSync exists to synchronously produce a batch of records.
|
|
record := &kgo.Record{Topic: "foo", Value: []byte(message)}
|
|
if err := cl.ProduceSync(ctx, record).FirstErr(); err != nil {
|
|
fmt.Printf("record had a produce error while synchronously producing: %v\n", err)
|
|
}
|
|
time.Sleep(time.Second)
|
|
}
|
|
}()
|
|
|
|
cl.AddConsumeTopics("foo")
|
|
for {
|
|
fetches := cl.PollFetches(ctx)
|
|
if fetches.IsClientClosed() {
|
|
fmt.Println("closed ")
|
|
return
|
|
}
|
|
if errs := fetches.Errors(); len(errs) > 0 {
|
|
// All errors are retried internally when fetching, but non-retriable errors are
|
|
// returned from polls so that users can notice and take action.
|
|
panic(fmt.Sprint(errs))
|
|
}
|
|
// We can iterate through a record iterator...
|
|
iter := fetches.RecordIter()
|
|
for !iter.Done() {
|
|
record := iter.Next()
|
|
fmt.Printf("==> %s\n", string(record.Value))
|
|
}
|
|
}
|
|
}
|