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 <[email protected]> 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. >
