Repository navigation
Conversation
Implement the server side of GET_CONSUMER_OFFSETS. Decode all-offset and selected-bucket requests, read committed offsets from the local leader replica, and group the response by source table. Move record_type ahead of group_id in the consumer-offset primary key and scan committed offsets by the (record_type, group_id) prefix. Decode values through the schema getter so old and new value schemas can be read together. Require DESCRIBE permission on every source table named by a selected-bucket request. All-offset requests defer group-level authorization until the consumer-group ACL work lands. Add unit, RPC, and integration tests for selected and all-offset reads, mixed schema versions, authorization, and partitioned and non-partitioned buckets. Update the authorization documentation.
| } | ||
|
|
||
| /** Reads all or selected committed offsets from the local leader replica. */ | ||
| public CompletableFuture<Map<TableBucket, ConsumerOffsetAndMetadata>> handleGetConsumerOffsets( |
There was a problem hiding this comment.
move it after handleCommitConsumerOffsets
| GetConsumerOffsetsRequest request) { | ||
| boolean allOffsets = request.hasAllOffsets() && request.isAllOffsets(); | ||
| if (allOffsets && !request.getTablesList().isEmpty()) { | ||
| throw new IllegalArgumentException( |
There was a problem hiding this comment.
Throwing IllegalArgumentException here and for duplicate buckets makes both validation failures surface as UNKNOWN_SERVER_ERROR over RPC. Could we introduce a non-retriable InvalidRequestException mapped to Errors.INVALID_REQUEST and use it for both checks? Other RPCs could then reuse the same error for invalid request parameters.
| /** Reads all or selected committed offsets from the local leader replica. */ | ||
| public CompletableFuture<Map<TableBucket, ConsumerOffsetAndMetadata>> handleGetConsumerOffsets( | ||
| String groupId, Replica offsetsReplica, List<TableBucket> requestedBuckets) { | ||
| return CompletableFuture.supplyAsync( |
There was a problem hiding this comment.
Could we run the offset fetch directly on the RPC worker thread and drop supplyAsync(..., ioExecutor)? The existing KV lookup and prefix lookup handlers already call Replica.lookups and Replica.prefixLookup on that thread, so we can avoid queueing the same reads on a second executor here.
| RECORD_TYPE_COLUMN, | ||
| GROUP_ID_COLUMN, |
There was a problem hiding this comment.
Could we order the columns in createTableSchema() to match the primary key, with RECORD_TYPE_COLUMN first and GROUP_ID_COLUMN second? That way the offset-commit prefix row can use the same (record_type, group_id) order as the key. Please update the positional row encoders to match.
| if (currentDecoder == null || schemaGetter != currentSchemaGetter) { | ||
| synchronized (this) { | ||
| if (schemaGetter != currentSchemaGetter || valueDecoder == null) { | ||
| schemaGetter = currentSchemaGetter; |
There was a problem hiding this comment.
Could we drop the cached schemaGetter field from ConsumerGroupManager and pass offsetsReplica.getSchemaGetter() directly to ConsumerOffsetSchema.createValueDecoder(...)? The replica already owns the getter for this bucket.
| PARTITION_ID_COLUMN, | ||
| BUCKET_ID_COLUMN)); | ||
|
|
||
| private static final List<String> OFFSET_COMMIT_PREFIX_KEYS = |
| () -> | ||
| KeyEncoder.ofBucketKeyEncoder( | ||
| ROW_TYPE, Collections.singletonList(GROUP_ID_COLUMN), null)); | ||
| private final ThreadLocal<KeyEncoder> offsetCommitPrefixEncoder = |
Purpose
Linked issue: close #5137
Part of #4608.
Implement server-side consumer-offset fetches so clients can read committed offsets for all buckets or selected source-table buckets.
Brief change log
GET_CONSUMER_OFFSETSinTabletServer.record_type, group_id, table_id, partition_id, bucket_idand scan byrecord_type + group_id.ValueDecoderto read old and new value schemas together.DESCRIBEpermission on every source table named in a selected-bucket request.Tests
Targeted Maven verification completed for
fluss-serverand its reactor dependencies. Checkstyle, Spotless, RAT, andgit diff --checkpassed.ConsumerGroupManagerFetchTest: 3 casesConsumerOffsetFetchITCase: 2 casesConsumerOffsetFetchAuthorizationITCase: 1 caseConsumerOffsetTableITCase: 2 casesConsumerOffsetCommitITCase: 1 caseConsumerOffsetSchemaTest: 11 casesTabletServiceTest: 8 casesServerRpcMessageUtilsTest: 19 casesAll 47 distinct cases passed.
API and Format
No RPC schema changes are introduced by this PR. The consumer-offset table primary-key order changes to
record_type, group_id, table_id, partition_id, bucket_id, and prefix scans userecord_type + group_id.Documentation
Updates
website/docs/security/authorization.mdwithGET_CONSUMER_OFFSETSpermission semantics.