# Share Groups (/connectors/kafka/how-to/share-groups)



Kafka share groups (KIP-932) give the Kafka connector a second, queue-style way to consume a
topic. Instead of assigning whole partitions to consumers, individual records are acquired,
processed, and acknowledged one at a time — closer to how a KubeMQ Queue behaves than to a
classic consumer group.

<Callout type="warn">
  **Share groups are supported in preview, not GA.** The data plane —
  `ShareGroupHeartbeat`(76), `ShareFetch`(78), `ShareAcknowledge`(79), plus the admin/observability
  keys `ShareGroupDescribe`(77), `DescribeShareGroupOffsets`(90), `AlterShareGroupOffsets`(91), and
  `DeleteShareGroupOffsets`(92) — is implemented and advertised: acquire, `Accept`/`Release`/
  `Reject` acknowledgements, multi-record batches, and cluster follower→leader `ShareFetch`
  forwarding are all proven against a real client. What hasn't run yet is the full multi-client
  share-group conformance matrix — the share-group analogue of the transactions/EOS conformance
  matrix — so this carries a **preview** verdict, not a GA guarantee. Never treat share groups as
  "fully supported" until that matrix lands. Track status on the
  [fitness matrix](/connectors/kafka/reference/fitness-matrix) and the
  [capabilities reference](/connectors/kafka/reference/capabilities).
</Callout>

## How share groups differ from classic consumer groups [#how-share-groups-differ-from-classic-consumer-groups]

A classic [consumer group](/connectors/kafka/concepts/consumer-groups) assigns whole
partitions to members via Join/Sync/Heartbeat — one partition, one owning consumer at a time,
with a rebalance whenever membership changes. A share group throws that model out: every member
can receive records from every partition, and the unit of ownership is a **single record** (or a
contiguous batch), not a partition. There's no partition-assignment protocol to reason about —
just heartbeat-based membership (`ShareGroupHeartbeat`) plus per-record acquisition on fetch.

That makes a share group behave much more like a KubeMQ **Queue**: multiple workers pull from a
shared backlog, a record goes to exactly one worker at a time, and a worker that fails to
process it releases the record back for someone else to pick up — see the acquire/acknowledge
cycle below.

## Acquire, deliver, acknowledge [#acquire-deliver-acknowledge]

Where a classic consumer commits offsets in bulk, a share consumer acknowledges **per record**
(or per contiguous batch) with one of three outcomes:

| Acknowledgement | Effect                                                                                                            |
| --------------- | ----------------------------------------------------------------------------------------------------------------- |
| **Accept**      | Terminal — the record is durably consumed; the group's start-offset advances past it.                             |
| **Release**     | The record becomes available for redelivery (to this or another member), with its delivery count incremented.     |
| **Reject**      | Terminal, like Accept, but signals "skip this record" rather than "processed successfully" — it never redelivers. |

A record the client never acknowledges is released automatically once the **30-second
acquisition lock** — the `AcquisitionLockTimeoutMillis` the connector advertises in every
`ShareFetch` response — expires. That's the same outcome as an explicit Release, and it also
counts as a delivery attempt.

<Callout type="info">
  **Redelivery has a limit.** A record redelivered (via Release or a lock timeout) more than
  **5** times is **archived** — the group's cursor advances past it permanently, so a single
  poison record can never wedge the partition for everyone else. The record's bytes are
  untouched; a plain `Fetch` consumer on the same topic still sees it.
</Callout>

## Produce and share-consume [#produce-and-share-consume]

Both client libraries below drive the full flow — produce onto a plain topic, then acquire,
process, and acknowledge from a share group:

<Tabs groupId="language" items="['Go', 'Java']">
  <Tab value="Go">
    ```go
    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)
            }
        }
    }
    ```
  </Tab>

  <Tab value="Java">
    ```java
    // 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
        }
    }
    ```
  </Tab>
</Tabs>

<Callout type="info">
  **Only these two clients have a share-consumer API today.** `kcat`, `confluent-kafka` (Python),
  `kafkajs`, `Confluent.Kafka` (C#), and `rdkafka` (Ruby/Rust) have no KIP-932 share-consumer
  surface yet, so there's no gap-fallback tab to show for them. franz-go is the client the
  connector's own share-group support was validated against; Java's `KafkaShareConsumer` is Apache
  Kafka's own early-access preview client.
</Callout>

## Related [#related]

<Cards>
  <Card title="Consumer Groups" href="/connectors/kafka/concepts/consumer-groups" description="The classic Join/Sync/Heartbeat protocol, durable per-group offsets, and static membership — what a share group replaces the partition-assignment model with." />

  <Card title="Capabilities" href="/connectors/kafka/reference/capabilities" description="Every implemented Kafka API and its version range, including the share-group keys and their preview status." />

  <Card title="Fitness matrix" href="/connectors/kafka/reference/fitness-matrix" description="What's drop-in, preview, roadmap, and unsupported when running Kafka workloads on KubeMQ." />
</Cards>