You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
package main
import (
"context""fmt"
"github.com/twmb/franz-go/pkg/kgo"
)
funcmain() {
seeds:= []string{"localhost:9092"}
varclientOpts []kgo.Opt//clientOpts = append(clientOpts, kgo.WithLogger(kgo.BasicLogger(os.Stdout, kgo.LogLevelDebug, nil)))clientOpts=append(clientOpts, kgo.SeedBrokers(seeds...))
client, err:=kgo.NewClient(clientOpts...)
iferr!=nil {
panic(err)
}
deferclient.Close()
ctx:=context.Background()
vargroupOpts []kgo.GroupOptgroupOpts=append(groupOpts, kgo.GroupTopics("topic1", "topic2"))
varbalancer kgo.GroupBalancerbalancer=kgo.CooperativeStickyBalancer()
groupOpts=append(groupOpts, kgo.Balancers(balancer))
groupOpts=append(groupOpts, kgo.InstanceID("client-0"))
groupOpts=append(groupOpts, kgo.DisableAutoCommit())
// sigs := make(chan os.Signal, 2)// signal.Notify(sigs, os.Interrupt, syscall.SIGTERM, syscall.SIGHUP)
client.AssignGroup("my-group", groupOpts...)
for {
fetches:=client.PollFetches(ctx)
iferrs:=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.Println(string(record.Value), "from an iterator!")
}
// or a callback function.fetches.EachPartition(func(p kgo.FetchTopicPartition) {
for_, record:=rangep.Partition.Records {
fmt.Println(string(record.Value), "from range inside a callback!")
}
// We can even use a second callback!p.EachRecord(func(record*kgo.Record) {
fmt.Println(string(record.Value), "from a second callback!")
})
})
}
}
The text was updated successfully, but these errors were encountered:
A user shared this use case around using static group membership:
Using a K8s statefulset you can easily generate unique, deterministic consumer instance IDs that won’t change as you upgrade. But the default way for K8s to deploy a stateful set is go through the statefulset first shutting down, then replacing each pod. So every partition being consumed must wait for the entire shutdown and start-up time. This is probably fine for a lot for workloads. But I have near real-time constraints and even a 15 or 20 second delay is too much if we can avoid it.
There’s another feature of stateful sets that we’ll take advantage of. You can delete them without cascading the delete to the pods. So we’re able to orphan the running consumers while they continue to do their thing.
The implementation of static group member also has a nice detail we’ll take advantage of. When a new consumer makes a join group request with an existing instance ID and no member ID (so a brand new consumer trying to re-use an existing (and in-use) ID), Kafka will fence the old consumer instance to prevent it from continuing to consume while the new instance is allowed to replace it. So we get to bring up new pods which end up causing the old ones to terminate as they take over processing.
The text was updated successfully, but these errors were encountered: