Re: [DISCUSS] FIP-53: Log Scanner Offset Management

[jie lu <pi...@gmail.com>]

> Hi Giannis,
>
> Thank you for the thorough and high-signal review — this is exactly the kind of feedback that makes the design better. Let me take your points in order.
>
> 1. Splitting the FIP — agreed, and already done.
>
> I've split the proposal into three independent FIPs:
>
> - FIP-53 — Offset storage & manual assignment: the `scanner_offsets` table, `OffsetCommit`/`OffsetFetch`, manual `assign()`, and the Admin surface (describe/list/reset/delete). This delivers durable resume for a single consumer, server-side queryable progress, and lets us wire up the existing Kafka offset handlers — all with no group machinery.
> - FIP-xx — Group membership & rebalance protocol.
> - FIP-xx — Flink and Kafka-compat integration.
>
> FIP-53 is the low-risk core and I'll drive it first, so reviewers can evaluate the storage design separately from the rebalance protocol.
>
> 2. Offset-storage core — making the guarantees explicit.
>
> - Placement: All offset rows for a given group land in the same bucket of the offsets system table; the bucket is determined by hashing the groupId via `bucketFor(groupId)`.
> - Read consistency: `OffsetFetch` is served by the group's coordinator (the offsets-bucket leader) from the in-memory materialized state; under `requireStable`, encountering an in-flight commit returns an `UNSTABLE` error code rather than a stale value.
>
> - Atomicity of multi-bucket commits: Since all buckets of the same group are written to the same offsets-bucket, `storeOffsets` packs them into a single `KvRecordBatch` and performs one `append`. The storage layer is therefore all-or-nothing for the whole request: on success, all buckets are materialized together via `onOffsetCommitAppend`; on failure, they all go through `failPendingOffsetWrite` and return the same error code.
>
> 3. Group protocol — agreed we shouldn't just mirror Kafka.
>
> The current rebalance design is fairly eager; I'll move this into the separate membership/rebalance FIP for deeper discussion.
>
> 4. Membership state placement
>
> I think the write-amplification cost is smaller than it looks, and I'd gently push back on the premise that Kafka keeps membership purely in memory:
>
> Kafka also persists a `GroupMetadata` record to `__consumer_offsets` on each generation stabilization — precisely so a new coordinator can restore a `Stable` group and its assignment without forcing a full rebalance.
>
> Our design matches that cadence rather than exceeding it:
>
> - The group snapshot is persisted only at generation boundaries (`SyncGroup → Stable`, or `group → Empty`), not on every join/heartbeat.
> - It's a single compacted-KV upsert keyed by `groupId`, so after compaction the steady-state footprint is one row per group, not one per generation.
>
> What "recover group state via replay" means concretely:
>
> it's latest-snapshot recovery plus aborting any in-flight rebalance — not a true transition replay.
>
> On gaining ownership of the offsets bucket, the new coordinator scans the log to the HWM,  applies last-write-wins per key (keep the latest group snapshot; tombstones remove), and reconstructs each group directly into `Stable` (members present) or `Empty`.
>
>
> I'll update the wiki with the split and these clarifications and follow up here. Thanks again — this materially improved the design.
>
> Best,
> Jie
>
>
> On Wed, Sep 16, 2026 at 3:32 PM Giannis Polyzos <ip...@gmail.com> wrote:
>>
>> Hi Jie,
>>
>> Thanks a lot for putting this FIP-53. This is a genuinely important gap to
>> close and something I've been thinking about for a long time now, so I'm
>> more than happy to help with the effort here. Fluss today has no
>> server-side committed offsets and no consumer groups (offsets live only in
>> LogScannerStatus in memory, with durable resume available solely through
>> the Flink connector's checkpointed split state), so bringing native offset
>> management and group semantics is real, high-value work. I especially
>> appreciate how much thought went into the design; the write-up is thorough
>> and easy to follow.
>>
>> I have some feedback I hope is useful. I've tried to separate "I'd love to
>> see this before I'm comfortable +1'ing" from "smaller polish."
>>
>> *1.* *Consider splitting the FIP.* The title says "offset management," but
>> the proposal actually covers three fairly independent things:
>>
>>    - a) offset commit/fetch storage + admin/reset,
>>    - b) consumer group membership and rebalancing, and
>>    - c) the Flink and Kafka-compat integration.
>>
>> Committed offsets are valuable on their own, and even a single consumer wanting
>> durable resume, or wiring up the existing  Kafka offset handlers, benefits
>> without any group machinery.
>> Landing (a) first would deliver most of the practical value at a fraction
>> of the risk, and would let reviewers evaluate the low-risk storage design
>> separately from the more involved rebalance protocol. Do you think it makes
>> sense to break it up this way?
>>
>> *2.* *The offset-storage design is the strong core, and I really like
>> it. *Offsets
>> as a system KV table, keyed by (groupId, tableId, partitionId, bucketId),
>> upsert on commit, point-lookup on fetch, with tombstones and retention, is
>> a natural fit and very on-brand with "everything is a table." Rejecting
>> ZooKeeper for high-frequency writes and Kafka's bespoke *__consumer_offsets*
>> format both make sense to me. One thing worth stating explicitly:
>> OffsetFetch should read from the bucket leader so a consumer reliably reads
>> back what it just committed. It'd also help to spell out that multi-bucket
>> commits are per-bucket atomic (matching Kafka), and how a client reconciles
>> a partial commit.
>>
>> *3. For the group protocol.* I think we have a chance here to improve on
>> Kafka rather than mirror it. Two areas I'd love your thoughts on:
>>
>>    - *Assignment ownership.* The proposal follows Kafka's model where a
>>    "leader member" computes the assignment plan. Fluss already has a central
>>    CoordinatorServer and authoritative server-side metadata, which Kafka never
>>    had. Server-side assignment could remove the SyncGroup round trip and
>>    simplify every client. Could we add a rejected-alternatives entry weighing
>>    server-side vs. client-side assignment?
>>    - Rebalance style. The core protocol looks eager (revoke-all on any
>>    membership change), which is the behavior *Kafka later worked hard to
>>    soften with cooperative/incremental rebalancing*. Since this is
>>    greenfield, it may be worth designing for cooperative rebalancing from the
>>    start, or at least keeping the assignment protocol version-negotiable so it
>>    can be added without a wire break.
>>
>> *4. Membership state placement.* Offsets fit the KV table beautifully, but
>> membership/generation state churns on every deploy or scale event and is
>> more event-log-shaped. Persisting full group snapshots as upserts into the
>> same compacted KV table may cost more in write amplification than keeping
>> membership in memory and rebuilding it on coordinator failover via member
>> rejoin (closer to how Kafka operates). It'd help to clarify what "*recover
>> group state via replay"* means concretely: recovering the latest snapshot
>> and aborting any in-flight rebalance, or a true transition replay.
>>
>> *Again, the direction is right, and the storage design in particular is
>> solid. *
>>
>> Thanks again for driving this forward.
>>
>> Best,
>> Giannis
>>
>> On Tue, Sep 15, 2026 at 1:43 PM jie lu <pi...@gmail.com> wrote:
>>
>> > Hi all, I would like to start a discussion for FIP-53: Log Scanner Offset
>> > Management
>> >
>> >
>> > https://cwiki.apache.org/confluence/spaces/FLUSS/pages/449282458/FIP-53+Log+Scanner+Offset+Management
>> >
>> >
>> > Today a LogScanner has no server-side persistent consumption offset: if the
>> > client crashes without checkpointing, the offset is lost and you can only
>> > restart from earliest, latest or timestamp. FIP-53 gives the LogScanner
>> > Kafka-style consumer-group semantics so that offsets are managed by the
>> > server.
>> >
>> > Scanners sharing a group id form a logical group with a single server-side
>> > committed offset per group, so a new member resumes where a crashed one
>> > left off. Two subscription modes are supported: a manual-assignment mode,
>> > where the user subscribes to specific buckets and no rebalance happens; and
>> > a group-managed mode, where the cluster assigns buckets to members and
>> > member join/leave triggers an automatic rebalance, with each bucket
>> > consumed by exactly one member. This is handled by a GroupCoordinator
>> > running in-process on each TabletServer, with offsets and group metadata
>> > persisted in an internal KV system table.
>> >
>> > Looking forward to your feedback.
>> >