[server] Add consumer group owner discovery by pisceslj · Pull Request #4716 · apache/fluss

pisceslj · 2026-10-19T04:00:41Z

Purpose

Linked issue: close #4714

Part of #4608.

Implement the server-side consumer-group owner discovery required by
FIP-53.

Brief change log

Add ConsumerOffsetTableManager to lazily create and resolve the
sys.consumer_offsets system table.
Configure changelog retention and last-write RocksDB row TTL.
Use the tagged KV value layout required by row TTL.
Add GroupManager to validate group IDs and map them to buckets.
Resolve the bucket leader from the TabletServer metadata cache.
Implement FindConsumerGroupOwner in TabletService.
Wire the group manager into the TabletServer lifecycle.
Add owner-discovery, routing, and metadata-cache tests.

Tests

GroupManagerTest
TabletServerMetadataCacheTest#testGetCachedBucketMetadata
TabletServiceITCase#testFindConsumerGroupOwner
./mvnw verify -pl fluss-server -am

API and Format

This PR implements the previously introduced
FIND_CONSUMER_GROUP_OWNER RPC.

It creates the internal sys.consumer_offsets primary-key table using
the tagged KV value layout for row-level TTL metadata. No new public API
keys or protobuf messages are introduced.

Documentation

No additional user-facing documentation is required in this PR. The
consumer-offset configuration options were introduced in the table
foundation PR.

gyang94 · 2026-10-19T11:16:41Z

+
+    private FindConsumerGroupOwnerResponse buildFindConsumerGroupOwnerResponse(
+            BucketMetadata bucketMetadata, String listenerName) {
+        int leaderId = bucketMetadata.getLeaderId().getAsInt();


It may throw exception if getLeaderId() returns Optional.empty()

Thanks for pointing this out. The leader ID is checked immediately before
calling buildFindConsumerGroupOwnerResponse:

if (!bucketMetadata.getLeaderId().isPresent()) { throw new ConsumerGroupOwnerLoadingException(...); }

BucketMetadata is immutable and its leaderId field is final. The helper
is also private and currently has only this call site, so the leader ID
cannot become empty between the presence check and getAsInt(). Therefore,
getAsInt() is safe here and I prefer to keep the current code.

gyang94 · 2026-10-19T11:50:05Z

+     * access). Empty if this server's snapshot does not know the table or bucket yet.
+     */
+    public Optional<BucketMetadata> getCachedBucketMetadata(TablePath tablePath, int bucketId) {
+        ServerMetadataSnapshot snapshot = serverMetadataSnapshot;


This snapshot may lag behind the current bucket leadership.
But the owner discovery now has no Zookeeper fallback on a cache miss. Retries may keep returning owner unavailable.

Good point. The TabletServer metadata cache is eventually updated when bucket
leadership changes, so a stale cached owner is temporary. The actual
consumer-offset operations will validate local leadership and return the
retriable NOT_CONSUMER_GROUP_OWNER error when necessary, allowing the client
to invalidate its owner cache and rediscover the owner.

For the cache-miss case, I will add an asynchronous ZooKeeper fallback so
retries can make progress before the local metadata snapshot is populated.
The fallback will also refresh the local metadata cache.

gyang94 · 2026-10-19T12:04:03Z

+                            bucketCount, replicationFactor, servers);
+
+            try {
+                metadataManager.createTable(


Multiple TS can pass the initial existence checks and call createTableconcurrently; the atomic creation of the first schema is inside this function zookeeperClient.registerFirstSchema().
I think we catch the NodeExistsException will be better than the existing generic FlussRuntimeException, and it can avoid an additional tableExists check.

gyang94 · 2026-10-19T12:24:11Z

+                            bucketCount, replicationFactor, servers);
+
+            try {
+                metadataManager.createTable(


If the TS crashes after registering the first schema but before completing table registration, tableExist() remains false, while every subsequent creation attempt fails with NodeExistsException on the existing schema. This leaves automatic table creation stuck, even after a TS restart。

swuferhong

Hi, @pisceslj thanks for your contribution! I left some comments. Could you also rebase onto the latest main to help avoid flaky CI failures? Thanks!

swuferhong · 2026-10-21T12:07:50Z

+            Optional<ServerNode> cachedOwner =
+                    metadataCache.getTabletServer(
+                            cachedBucket.get().getLeaderId().getAsInt(), listenerName);
+            if (cachedOwner.isPresent()) {


Cache miss fallback covers missing metadata, but not stale metadata. If leadership moves from A to B while A remains alive and this TS misses the metadata update, this branch keeps returning A. How will repeated discovery refresh this server-side cache?

The TabletServer metadata cache is eventually updated when bucket
leadership changes, so a stale cached owner is temporary. The actual
consumer-offset operations will validate local leadership and return the
retriable NOT_CONSUMER_GROUP_OWNER error when necessary, allowing the client
to invalidate its owner cache and rediscover the owner.

swuferhong · 2026-10-21T12:08:52Z

+                    .setTablePath()
+                    .setDatabaseName(CONSUMER_OFFSET_TABLE_PATH.getDatabaseName())
+                    .setTableName(CONSUMER_OFFSET_TABLE_PATH.getTableName());
+            coordinatorGateway.createTable(createTableRequest).get();


Routing creation through the coordinator still reaches the same non-atomic MetadataManager.createTable implementation. If schema registration succeeds but table registration fails, subsequent retries still fail on the existing schema. Could we cover recovery from this partially created state?

swuferhong · 2026-10-21T12:10:11Z

+                    .setTablePath()
+                    .setDatabaseName(CONSUMER_OFFSET_TABLE_PATH.getDatabaseName())
+                    .setTableName(CONSUMER_OFFSET_TABLE_PATH.getTableName());
+            coordinatorGateway.createTable(createTableRequest).get();


This now goes through validateTableCreationPermission, so allow.create.kv.tables=false also prevents creating sys.consumer_offsets. This limitation should be explicit rather than reported as a transient owner-unavailable error for this case.

Thanks for pointing this out. I updated the validation so internally created system tables bypass the user-facing allow.create.kv.tables restriction, while external requests cannot create system tables. Regular KV tables remain subject to the setting.

I also added an integration test with allow.create.kv.tables=false that triggers findConsumerGroupOwner and verifies that sys.consumer_offsets is created successfully, so this case no longer surfaces as a transient owner-unavailable error.

swuferhong · 2026-10-21T12:34:46Z

+    public int bucketFor(String groupId, int bucketCount) {
+        validateGroupId(groupId);
+        byte[] bucketKey = ConsumerOffsetSchema.encodeGroupPrefix(groupId);
+        return FlussBucketingFunction.bucketForRowKey(bucketKey, bucketCount);


Could we avoid inheriting the cluster's lake format for sys.consumer_offsets? here uses Fluss bucketing, while standard writers/lookups follow the table's lake format, even when tiering is disabled. This mismatch could cause future offset lookups to miss data. Please keep both routing paths consistent and add a test with datalake.format configured.

Thanks for catching this. I updated the creation defaults so system tables, including sys.consumer_offsets, do not inherit the cluster’s lake format. This keeps GroupManager and standard KV writers/lookups consistently using Fluss bucketing.

I also added an integration test with datalake.format=PAIMON and tiering disabled. The test verifies that sys.consumer_offsets has no lake format property and that both routing paths calculate the same bucket.

wuchong

I left some comments.

AI-Contributed/Feature: 0/6 AI-Contributed/UT: 0/2

wuchong

LGTM.

I pushed a commit to revert using offsets retention for log ttl.

pisceslj mentioned this pull request Oct 19, 2026

FIP-53: Log Scanner Offset Management #4608

Open

2 tasks done

gyang94 reviewed Oct 19, 2026

View reviewed changes

swuferhong reviewed Oct 21, 2026

View reviewed changes

pisceslj force-pushed the feature/fip-53-consumer-group-owner branch from b47ee83 to b4672e6 Compare October 22, 2026 12:56

wuchong reviewed Oct 22, 2026

View reviewed changes

pisceslj added 3 commits October 23, 2026 10:39

[server] Add consumer group owner discovery

66687de

[server] Route offset table creation through coordinator

a43eab8

[server] Fix consumer offset system table creation

3f05111

pisceslj force-pushed the feature/fip-53-consumer-group-owner branch from b4672e6 to d258eb3 Compare October 23, 2026 03:33

[server] Reserve sys namespace and refine consumer offset table

197c5a5

pisceslj force-pushed the feature/fip-53-consumer-group-owner branch from d258eb3 to 197c5a5 Compare October 23, 2026 03:38

use offsets retention for log ttl.

63d2c89

AI-Contributed/Feature: 0/6 AI-Contributed/UT: 0/2

wuchong approved these changes Oct 23, 2026

View reviewed changes

wuchong merged commit 212c4d4 into apache:main Oct 23, 2026
20 checks passed

Conversation

pisceslj commented Oct 19, 2026

Purpose

Brief change log

Tests

API and Format

Documentation

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

swuferhong left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

swuferhong Oct 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

swuferhong Oct 21, 2026 •

edited

Loading