Share Groups

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/twmb/franz-go/pkg/kgo"
)

func main() {
    ctx := context.Background()

    // A share group has no partition assignment — every member shares the
    // same pool of records, each one acquired individually.
    sc, err := kgo.NewClient(
        kgo.SeedBrokers("localhost:9092"),
        kgo.ShareGroup("orders-share-group"),
        kgo.ConsumeTopics("orders"),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer sc.Close()

    for {
        fetches := sc.PollFetches(ctx)
        if errs := fetches.Errors(); len(errs) > 0 {
            log.Fatal(errs[0].Err)
        }

        var accepted []*kgo.Record
        fetches.EachRecord(func(r *kgo.Record) {
            fmt.Printf("acquired offset=%d attempt=%d: %s\n", r.Offset, r.DeliveryCount(), r.Value)
            // process the record here — on failure, MarkAcks with AckRelease
            // or AckReject instead of AckAccept below.
            accepted = append(accepted, r)
        })

        sc.MarkAcks(kgo.AckAccept, accepted...)
        if err := sc.FlushAcks(ctx); err != nil {
            log.Printf("flush acknowledgements: %v", err)
        }
    }
}

// KafkaShareConsumer is Apache Kafka's own KIP-932 preview client (early
// access) — illustrative only. Its API is still evolving, so verify
// the exact surface against the Apache Kafka client version you pin.
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "orders-share-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

try (KafkaShareConsumer<String, String> consumer = new KafkaShareConsumer<>(props)) {
    consumer.subscribe(Collections.singleton("orders"));

    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(5000));
        for (ConsumerRecord<String, String> record : records) {
            // process the record here — on failure, acknowledge RELEASE or
            // REJECT instead of ACCEPT below.
            consumer.acknowledge(record, AcknowledgeType.ACCEPT);
        }
        consumer.commitSync(); // flushes the pending acknowledgements
    }
}