From 3c9b5cb3fd62b0e6bd7d4cd408acf74609426eb7 Mon Sep 17 00:00:00 2001 From: pisceslj Date: Tue, 10 Nov 2026 21:19:10 +0800 Subject: [PATCH] [server] Implement consumer offset fetches 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. --- .../fluss/server/tablet/TabletService.java | 32 +- .../consumer/group/ConsumerGroupManager.java | 97 +++++- .../consumer/group/ConsumerOffsetSchema.java | 55 +++- .../server/utils/ServerRpcMessageUtils.java | 74 +++++ .../server/tablet/TabletServiceTest.java | 24 ++ .../group/ConsumerGroupManagerFetchTest.java | 157 ++++++++++ ...onsumerOffsetFetchAuthorizationITCase.java | 242 +++++++++++++++ .../group/ConsumerOffsetFetchITCase.java | 287 ++++++++++++++++++ .../group/ConsumerOffsetSchemaTest.java | 51 +++- .../group/ConsumerOffsetTableITCase.java | 2 +- .../utils/ServerRpcMessageUtilsTest.java | 87 ++++++ website/docs/security/authorization.md | 1 + 12 files changed, 1083 insertions(+), 26 deletions(-) create mode 100644 fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerGroupManagerFetchTest.java create mode 100644 fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetFetchAuthorizationITCase.java create mode 100644 fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetFetchITCase.java diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java index 1115b00e067..4fa367856f7 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java @@ -162,6 +162,7 @@ import static org.apache.fluss.server.coordinator.CoordinatorContext.INITIAL_COORDINATOR_EPOCH; import static org.apache.fluss.server.log.FetchParams.DEFAULT_MAX_WAIT_MS_WHEN_MIN_BYTES_ENABLE; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.decodeCommitConsumerOffsetsRequest; +import static org.apache.fluss.server.utils.ServerRpcMessageUtils.decodeGetConsumerOffsetsRequest; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getFetchLogData; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getListOffsetsData; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getNotifyLakeTableOffset; @@ -850,10 +851,33 @@ public CompletableFuture commitConsumerOffsets( @Override public CompletableFuture getConsumerOffsets( GetConsumerOffsetsRequest request) { - CompletableFuture response = new CompletableFuture<>(); - response.completeExceptionally( - new UnsupportedOperationException("getConsumerOffsets is not supported yet")); - return response; + final String groupId = request.getGroupId(); + final Replica offsetsReplica; + final List requestedBuckets; + try { + offsetsReplica = groupManager.validateGroupId(groupId); + requestedBuckets = decodeGetConsumerOffsetsRequest(request); + } catch (ApiException | IllegalArgumentException e) { + return FutureUtils.completedExceptionally(e); + } + + if (requestedBuckets != null && requestedBuckets.isEmpty()) { + return CompletableFuture.completedFuture(new GetConsumerOffsetsResponse()); + } + + // The session is bound to the RPC thread; authorize all requested sources before reading. + if (authorizer != null && requestedBuckets != null) { + requestedBuckets.stream() + .map(TableBucket::getTableId) + .distinct() + .forEach(tableId -> authorizeTable(DESCRIBE, tableId)); + } + + // TODO: Once consumer-group ACLs are merged, authorize DESCRIBE on the group before + // reading offsets. + return groupManager + .handleGetConsumerOffsets(groupId, offsetsReplica, requestedBuckets) + .thenApply(ServerRpcMessageUtils::makeGetConsumerOffsetsResponse); } @Override diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/consumer/group/ConsumerGroupManager.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/consumer/group/ConsumerGroupManager.java index e529668b6b2..dd58e40c47c 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/consumer/group/ConsumerGroupManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/consumer/group/ConsumerGroupManager.java @@ -22,6 +22,7 @@ import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.cluster.ServerType; import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.exception.ApiException; import org.apache.fluss.exception.ConsumerGroupOwnerLoadingException; import org.apache.fluss.exception.ConsumerGroupOwnerNotAvailableException; import org.apache.fluss.exception.InvalidConsumerGroupIdException; @@ -30,13 +31,16 @@ import org.apache.fluss.metadata.DatabaseDescriptor; import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.metadata.LogFormat; +import org.apache.fluss.metadata.SchemaGetter; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableDescriptor; import org.apache.fluss.metadata.TableInfo; import org.apache.fluss.record.DefaultKvRecordBatch; import org.apache.fluss.record.KvRecordBatch; import org.apache.fluss.record.KvRecordBatchBuilder; +import org.apache.fluss.row.InternalRow; import org.apache.fluss.row.compacted.CompactedRow; +import org.apache.fluss.row.encode.ValueDecoder; import org.apache.fluss.rpc.entity.PutKvResultForBucket; import org.apache.fluss.rpc.gateway.CoordinatorGateway; import org.apache.fluss.rpc.messages.CreateDatabaseRequest; @@ -57,6 +61,7 @@ import org.apache.fluss.server.replica.ReplicaManager.OnlineReplica; import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.server.zk.data.TabletServerRegistration; +import org.apache.fluss.utils.ByteArraySlice; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -64,6 +69,7 @@ import javax.annotation.Nullable; import javax.annotation.concurrent.ThreadSafe; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -81,7 +87,7 @@ import static org.apache.fluss.server.zk.data.LeaderAndIsr.NO_LEADER; import static org.apache.fluss.utils.Preconditions.checkNotNull; -/** Discovers consumer group owners and commits offsets to the local leader replica. */ +/** Discovers consumer group owners and reads and commits their offsets on local leaders. */ @ThreadSafe @Internal public final class ConsumerGroupManager { @@ -93,6 +99,8 @@ public final class ConsumerGroupManager { private final ReplicaManager replicaManager; private final ConsumerOffsetTableConfig offsetsConfig; private final ConsumerOffsetSchema offsetSchema = new ConsumerOffsetSchema(); + private volatile SchemaGetter schemaGetter; + private volatile ValueDecoder valueDecoder; private final ZooKeeperClient zkClient; private final MetadataManager metadataManager; private final TabletServerMetadataCache metadataCache; @@ -136,6 +144,48 @@ public CompletableFuture findOwner(String groupId, String listenerNa listenerName)); } + /** Reads all or selected committed offsets from the local leader replica. */ + public CompletableFuture> handleGetConsumerOffsets( + String groupId, Replica offsetsReplica, List requestedBuckets) { + return CompletableFuture.supplyAsync( + () -> { + Map offsets = new HashMap<>(); + ValueDecoder decoder = getValueDecoder(offsetsReplica); + try { + if (requestedBuckets == null) { + populateAllOffsets( + offsetsReplica.prefixLookup( + offsetSchema.encodeOffsetCommitPrefix(groupId)), + decoder, + offsets); + } else { + List keys = new ArrayList<>(requestedBuckets.size()); + for (TableBucket tableBucket : requestedBuckets) { + keys.add( + offsetSchema.encodeOffsetKey( + groupId, + tableBucket.getTableId(), + tableBucket.getPartitionId(), + tableBucket.getBucket())); + } + populateSelectedOffsets( + requestedBuckets, + offsetsReplica.lookups(keys), + decoder, + offsets); + } + return offsets; + } catch (Exception e) { + LOG.warn("Failed to read offsets for group {}.", groupId, e); + if (e instanceof ApiException) { + throw (ApiException) e; + } + throw Errors.forException(e).exception(e.getMessage()); + } + }, + ioExecutor); + } + /** Updates local offsets table metadata even when another tablet served the find request. */ public void onTableMetadataUpdate(List tables) { for (TableMetadata table : tables) { @@ -431,6 +481,51 @@ public CompletableFuture> handleCommitConsumerOffsets( writeResults -> handleWriteCompletion(groupId, results, writeResults)); } + @VisibleForTesting + static void populateAllOffsets( + List values, + ValueDecoder valueDecoder, + Map offsets) { + for (ByteArraySlice value : values) { + InternalRow row = valueDecoder.decodeValue(value.toByteArray()).row; + offsets.put( + ConsumerOffsetSchema.toTableBucket(row), + ConsumerOffsetSchema.toOffsetAndMetadata(row)); + } + } + + @VisibleForTesting + static void populateSelectedOffsets( + List requestedBuckets, + List values, + ValueDecoder valueDecoder, + Map offsets) { + for (int i = 0; i < requestedBuckets.size(); i++) { + ByteArraySlice value = values.get(i); + if (value == null) { + offsets.put(requestedBuckets.get(i), null); + continue; + } + InternalRow row = valueDecoder.decodeValue(value.toByteArray()).row; + offsets.put(requestedBuckets.get(i), ConsumerOffsetSchema.toOffsetAndMetadata(row)); + } + } + + private ValueDecoder getValueDecoder(Replica offsetsReplica) { + SchemaGetter currentSchemaGetter = offsetsReplica.getSchemaGetter(); + ValueDecoder currentDecoder = valueDecoder; + if (currentDecoder == null || schemaGetter != currentSchemaGetter) { + synchronized (this) { + if (schemaGetter != currentSchemaGetter || valueDecoder == null) { + schemaGetter = currentSchemaGetter; + valueDecoder = ConsumerOffsetSchema.createValueDecoder(currentSchemaGetter); + } + currentDecoder = valueDecoder; + } + } + return currentDecoder; + } + @VisibleForTesting static Map handleWriteCompletion( String groupId, diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetSchema.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetSchema.java index 603f699a5bb..5fddc8bf07f 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetSchema.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetSchema.java @@ -19,13 +19,18 @@ import org.apache.fluss.annotation.Internal; import org.apache.fluss.bucketing.FlussBucketingFunction; +import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.SchemaGetter; +import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.row.BinaryString; import org.apache.fluss.row.GenericRow; import org.apache.fluss.row.InternalRow; import org.apache.fluss.row.compacted.CompactedRow; import org.apache.fluss.row.encode.CompactedRowEncoder; import org.apache.fluss.row.encode.KeyEncoder; +import org.apache.fluss.row.encode.KvValueLayout; +import org.apache.fluss.row.encode.ValueDecoder; import org.apache.fluss.types.DataType; import org.apache.fluss.types.DataTypes; import org.apache.fluss.types.RowType; @@ -58,12 +63,15 @@ public final class ConsumerOffsetSchema { private static final List PRIMARY_KEYS = Collections.unmodifiableList( Arrays.asList( - GROUP_ID_COLUMN, RECORD_TYPE_COLUMN, + GROUP_ID_COLUMN, TABLE_ID_COLUMN, PARTITION_ID_COLUMN, BUCKET_ID_COLUMN)); + private static final List OFFSET_COMMIT_PREFIX_KEYS = + Collections.unmodifiableList(Arrays.asList(RECORD_TYPE_COLUMN, GROUP_ID_COLUMN)); + private static final Schema TABLE_SCHEMA = createTableSchema(); private static final RowType ROW_TYPE = TABLE_SCHEMA.getRowType(); private static final DataType[] DATA_TYPES = ROW_TYPE.getChildren().toArray(new DataType[0]); @@ -76,11 +84,14 @@ public final class ConsumerOffsetSchema { private final ThreadLocal primaryKeyEncoder = ThreadLocal.withInitial( () -> KeyEncoder.ofBucketKeyEncoder(ROW_TYPE, PRIMARY_KEYS, null)); - private final ThreadLocal groupIdEncoder = + private final ThreadLocal groupBucketKeyEncoder = ThreadLocal.withInitial( () -> KeyEncoder.ofBucketKeyEncoder( ROW_TYPE, Collections.singletonList(GROUP_ID_COLUMN), null)); + private final ThreadLocal offsetCommitPrefixEncoder = + ThreadLocal.withInitial( + () -> KeyEncoder.ofBucketKeyEncoder(ROW_TYPE, OFFSET_COMMIT_PREFIX_KEYS, null)); /** Returns the schema of the consumer-offset system table. */ public static Schema getTableSchema() { @@ -108,11 +119,39 @@ public byte[] encodeOffsetKey(String groupId, long tableId, Long partitionId, in bucketId)); } - /** Encodes the {@code group_id} prefix used to enumerate all rows belonging to a group. */ - public byte[] encodeGroupPrefix(String groupId) { - // TODO: When record_type moves before group_id in the table schema, construct this row - // with (record_type, group_id) so the encoder reads group_id at index 1. - return groupIdEncoder.get().encodeKey(GenericRow.of(BinaryString.fromString(groupId))); + /** Creates a decoder for values stored with the consumer-offset schema. */ + public static ValueDecoder createValueDecoder(SchemaGetter schemaGetter) { + return new ValueDecoder(schemaGetter, KvFormat.COMPACTED, KvValueLayout.PLAIN); + } + + /** Converts a decoded offset row into its committed offset and metadata. */ + static ConsumerOffsetAndMetadata toOffsetAndMetadata(InternalRow row) { + return new ConsumerOffsetAndMetadata( + row.getLong(5), row.isNullAt(6) ? null : row.getString(6).toString()); + } + + /** Converts a decoded offset row into its source table bucket. */ + static TableBucket toTableBucket(InternalRow row) { + long partitionId = row.getLong(3); + return new TableBucket(row.getLong(2), partitionId < 0 ? null : partitionId, row.getInt(4)); + } + + /** Encodes the bucket key used to route a group to its offsets-table bucket. */ + public byte[] encodeGroupBucketKey(String groupId) { + return groupBucketKeyEncoder + .get() + .encodeKey(GenericRow.of(BinaryString.fromString(groupId))); + } + + /** + * Encodes the {@code record_type} and {@code group_id} prefix used to enumerate a group's + * committed offsets. + */ + public byte[] encodeOffsetCommitPrefix(String groupId) { + return offsetCommitPrefixEncoder + .get() + .encodeKey( + GenericRow.of(BinaryString.fromString(groupId), OFFSET_COMMIT_RECORD_TYPE)); } /** Encodes the complete primary key of an already materialized offset row. */ @@ -121,7 +160,7 @@ byte[] encodeOffsetKey(InternalRow row) { } int bucketFor(String groupId, int bucketCount) { - return FlussBucketingFunction.bucketForRowKey(encodeGroupPrefix(groupId), bucketCount); + return FlussBucketingFunction.bucketForRowKey(encodeGroupBucketKey(groupId), bucketCount); } /** The returned row uses this thread's buffer until its next call to this method. */ diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java index e92d6b745fd..f6ea30d319e 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java @@ -17,6 +17,7 @@ package org.apache.fluss.server.utils; +import org.apache.fluss.annotation.VisibleForTesting; import org.apache.fluss.cluster.Endpoint; import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.cluster.ServerType; @@ -55,6 +56,7 @@ import org.apache.fluss.record.MemoryLogRecords; import org.apache.fluss.remote.RemoteLogFetchInfo; import org.apache.fluss.remote.RemoteLogSegment; +import org.apache.fluss.row.BinaryString; import org.apache.fluss.rpc.entity.FetchLogResultForBucket; import org.apache.fluss.rpc.entity.LimitScanResultForBucket; import org.apache.fluss.rpc.entity.ListOffsetsResultForBucket; @@ -80,6 +82,8 @@ import org.apache.fluss.rpc.messages.FetchLogRequest; import org.apache.fluss.rpc.messages.FetchLogResponse; import org.apache.fluss.rpc.messages.FindConsumerGroupOwnerResponse; +import org.apache.fluss.rpc.messages.GetConsumerOffsetsRequest; +import org.apache.fluss.rpc.messages.GetConsumerOffsetsResponse; import org.apache.fluss.rpc.messages.GetFileSystemSecurityTokenResponse; import org.apache.fluss.rpc.messages.GetKvSnapshotMetadataResponse; import org.apache.fluss.rpc.messages.GetLakeSnapshotResponse; @@ -127,6 +131,10 @@ import org.apache.fluss.rpc.messages.PbFetchLogReqForTable; import org.apache.fluss.rpc.messages.PbFetchLogRespForBucket; import org.apache.fluss.rpc.messages.PbFetchLogRespForTable; +import org.apache.fluss.rpc.messages.PbGetConsumerOffsetsRequestBucket; +import org.apache.fluss.rpc.messages.PbGetConsumerOffsetsRequestTable; +import org.apache.fluss.rpc.messages.PbGetConsumerOffsetsResponseBucket; +import org.apache.fluss.rpc.messages.PbGetConsumerOffsetsResponseTable; import org.apache.fluss.rpc.messages.PbKeyValue; import org.apache.fluss.rpc.messages.PbKvSnapshot; import org.apache.fluss.rpc.messages.PbKvSnapshotLeaseForBucket; @@ -1255,6 +1263,72 @@ public static Map decodeCommitConsumerOf return offsets; } + /** Decodes the requested table buckets; returns {@code null} for an all-offsets request. */ + @VisibleForTesting + public static List decodeGetConsumerOffsetsRequest( + GetConsumerOffsetsRequest request) { + boolean allOffsets = request.hasAllOffsets() && request.isAllOffsets(); + if (allOffsets && !request.getTablesList().isEmpty()) { + throw new IllegalArgumentException( + "GetConsumerOffsets request cannot specify both all_offsets and buckets"); + } + if (allOffsets) { + return null; + } + + List requestedBuckets = new ArrayList<>(); + Set seenBuckets = new HashSet<>(); + for (PbGetConsumerOffsetsRequestTable table : request.getTablesList()) { + for (PbGetConsumerOffsetsRequestBucket bucket : table.getBucketsList()) { + Long partitionId = bucket.hasPartitionId() ? bucket.getPartitionId() : null; + TableBucket tableBucket = + new TableBucket(table.getTableId(), partitionId, bucket.getBucketId()); + if (!seenBuckets.add(tableBucket)) { + throw new IllegalArgumentException( + "Duplicate table bucket in GetConsumerOffsets request: " + tableBucket); + } + requestedBuckets.add(tableBucket); + } + } + return requestedBuckets; + } + + /** Builds a response grouped by table ID from decoded consumer offsets. */ + @VisibleForTesting + public static GetConsumerOffsetsResponse makeGetConsumerOffsetsResponse( + Map offsets) { + Map> bucketsByTable = new HashMap<>(); + offsets.forEach( + (tableBucket, offsetAndMetadata) -> { + PbGetConsumerOffsetsResponseBucket bucket = + new PbGetConsumerOffsetsResponseBucket() + .setBucketId(tableBucket.getBucket()); + if (tableBucket.getPartitionId() != null) { + bucket.setPartitionId(tableBucket.getPartitionId()); + } + if (offsetAndMetadata != null) { + bucket.setOffset(offsetAndMetadata.getOffset()); + BinaryString metadata = offsetAndMetadata.getMetadata(); + if (metadata != null) { + bucket.setMetadata(metadata.toString()); + } + } + bucketsByTable + .computeIfAbsent(tableBucket.getTableId(), ignored -> new ArrayList<>()) + .add(bucket); + }); + + GetConsumerOffsetsResponse response = new GetConsumerOffsetsResponse(); + bucketsByTable.forEach( + (tableId, buckets) -> + response.addAllTables( + Collections.singletonList( + new PbGetConsumerOffsetsResponseTable() + .setTableId(tableId) + .addAllBuckets(buckets)))); + return response; + } + /** * Converts put-KV bucket requests while preserving their historical partition context. * diff --git a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceTest.java b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceTest.java index 4e49546a081..4e5de696960 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceTest.java @@ -34,6 +34,8 @@ import org.apache.fluss.rpc.messages.CommitConsumerOffsetsRequest; import org.apache.fluss.rpc.messages.CommitConsumerOffsetsResponse; import org.apache.fluss.rpc.messages.FindConsumerGroupOwnerRequest; +import org.apache.fluss.rpc.messages.GetConsumerOffsetsRequest; +import org.apache.fluss.rpc.messages.GetConsumerOffsetsResponse; import org.apache.fluss.rpc.messages.PbCommitConsumerOffsetsResponseBucket; import org.apache.fluss.rpc.netty.server.Session; import org.apache.fluss.rpc.protocol.ApiKeys; @@ -192,6 +194,28 @@ void testConsumerOffsetAppendsDoNotWaitForReplicationAcknowledgement() throws Ex assertStoredOffset(GROUP_ID, 10L); } + @Test + void testGetConsumerOffsetsAfterFindOwnerUsesInitializedMetadata() throws Exception { + registerOffsetsTable(); + TabletService service = createService(); + + CompletableFuture find = + service.findConsumerGroupOwner( + new FindConsumerGroupOwnerRequest().setGroupId(GROUP_ID)); + assertThat(find).isNotDone(); + executor.triggerAll(); + assertThat(find).isCompleted(); + + GetConsumerOffsetsRequest request = + new GetConsumerOffsetsRequest().setGroupId(GROUP_ID).setAllOffsets(true); + CompletableFuture response = + service.getConsumerOffsets(request); + executor.triggerAll(); + + assertThat(response).isCompleted(); + assertThat(zkClient.tableExist(SystemTables.CONSUMER_OFFSETS)).isTrue(); + } + @Test void testCommitBeforeFindDoesNotScheduleInitialization() throws Exception { TabletService service = createService(); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerGroupManagerFetchTest.java b/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerGroupManagerFetchTest.java new file mode 100644 index 00000000000..341469a9804 --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerGroupManagerFetchTest.java @@ -0,0 +1,157 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.server.tablet.consumer.group; + +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.SchemaInfo; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.record.TestingSchemaGetter; +import org.apache.fluss.row.BinaryString; +import org.apache.fluss.row.compacted.CompactedRow; +import org.apache.fluss.row.encode.CompactedRowEncoder; +import org.apache.fluss.row.encode.ValueDecoder; +import org.apache.fluss.row.encode.ValueEncoder; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.DataTypes; +import org.apache.fluss.utils.ByteArraySlice; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for the consumer-offset fetch path. */ +class ConsumerGroupManagerFetchTest { + + private static final short OLD_SCHEMA_ID = 1; + private static final short NEW_SCHEMA_ID = 2; + + @Test + void testReadSelectedOffsetsPreservesOrderAndReturnsNoOffset() throws Exception { + String groupId = "group-1"; + TableBucket committedBucket = new TableBucket(10L, 3); + TableBucket missingBucket = new TableBucket(10L, 1); + Map offsets = new LinkedHashMap<>(); + ConsumerGroupManager.populateSelectedOffsets( + Arrays.asList(committedBucket, missingBucket), + Arrays.asList( + ByteArraySlice.wrap( + encodedOffset(groupId, committedBucket, 42L, "metadata")), + null), + valueDecoder(), + offsets); + + assertThat(offsets.keySet()).containsExactly(committedBucket, missingBucket); + assertThat(offsets.get(committedBucket).getOffset()).isEqualTo(42L); + assertThat(offsets.get(committedBucket).getMetadata().toString()).isEqualTo("metadata"); + assertThat(offsets.get(missingBucket)).isNull(); + } + + @Test + void testReadAllOffsetsRestoresPartitionedAndNonPartitionedBuckets() throws Exception { + String groupId = "group-1"; + TableBucket nonPartitioned = new TableBucket(10L, 0); + TableBucket partitioned = new TableBucket(11L, 7L, 1); + Map offsets = new LinkedHashMap<>(); + ConsumerGroupManager.populateAllOffsets( + Arrays.asList( + ByteArraySlice.wrap(encodedOffset(groupId, nonPartitioned, 10L, "")), + ByteArraySlice.wrap(encodedOffset(groupId, partitioned, 20L, "p"))), + valueDecoder(), + offsets); + + assertThat(new ArrayList<>(offsets.keySet())).containsExactly(nonPartitioned, partitioned); + assertThat(offsets.get(nonPartitioned).getOffset()).isEqualTo(10L); + assertThat(offsets.get(partitioned).getOffset()).isEqualTo(20L); + } + + @Test + void testReadAllOffsetsSupportsMixedSchemaVersions() { + String groupId = "group-1"; + TableBucket oldBucket = new TableBucket(10L, 0); + TableBucket newBucket = new TableBucket(11L, 1); + Map offsets = new LinkedHashMap<>(); + ConsumerGroupManager.populateAllOffsets( + Arrays.asList( + ByteArraySlice.wrap(encodedOffset(groupId, oldBucket, 10L, "old")), + ByteArraySlice.wrap( + encodedEvolvedOffset(groupId, newBucket, 20L, "new", "extra"))), + valueDecoder(), + offsets); + + assertThat(offsets.keySet()).containsExactly(oldBucket, newBucket); + assertThat(offsets.get(oldBucket).getOffset()).isEqualTo(10L); + assertThat(offsets.get(oldBucket).getMetadata().toString()).isEqualTo("old"); + assertThat(offsets.get(newBucket).getOffset()).isEqualTo(20L); + assertThat(offsets.get(newBucket).getMetadata().toString()).isEqualTo("new"); + } + + private static ValueDecoder valueDecoder() { + TestingSchemaGetter schemaGetter = + new TestingSchemaGetter(OLD_SCHEMA_ID, ConsumerOffsetSchema.getTableSchema()); + schemaGetter.updateLatestSchemaInfo(new SchemaInfo(evolvedSchema(), NEW_SCHEMA_ID)); + return ConsumerOffsetSchema.createValueDecoder(schemaGetter); + } + + private static Schema evolvedSchema() { + return Schema.newBuilder() + .fromSchema(ConsumerOffsetSchema.getTableSchema()) + .column("extra", DataTypes.STRING()) + .build(); + } + + private static byte[] encodedOffset( + String groupId, TableBucket tableBucket, long offset, String metadata) { + CompactedRow row = + new ConsumerOffsetSchema() + .encodeOffsetRow( + groupId, + ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE, + tableBucket.getTableId(), + tableBucket.getPartitionId(), + tableBucket.getBucket(), + new ConsumerOffsetAndMetadata(offset, metadata)); + return ValueEncoder.encodeValue(OLD_SCHEMA_ID, row); + } + + private static byte[] encodedEvolvedOffset( + String groupId, TableBucket tableBucket, long offset, String metadata, String extra) { + Schema schema = evolvedSchema(); + DataType[] fieldTypes = schema.getRowType().getChildren().toArray(new DataType[0]); + CompactedRowEncoder encoder = new CompactedRowEncoder(fieldTypes); + encoder.startNewRow(); + encoder.encodeField(0, BinaryString.fromString(groupId)); + encoder.encodeField(1, ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE); + encoder.encodeField(2, tableBucket.getTableId()); + encoder.encodeField( + 3, + tableBucket.getPartitionId() == null + ? ConsumerOffsetSchema.NON_PARTITIONED_PARTITION_ID + : tableBucket.getPartitionId()); + encoder.encodeField(4, tableBucket.getBucket()); + encoder.encodeField(5, offset); + encoder.encodeField(6, BinaryString.fromString(metadata)); + encoder.encodeField(7, BinaryString.fromString(extra)); + CompactedRow row = encoder.finishRow(); + return ValueEncoder.encodeValue(NEW_SCHEMA_ID, row); + } +} diff --git a/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetFetchAuthorizationITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetFetchAuthorizationITCase.java new file mode 100644 index 00000000000..810f6fdd675 --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetFetchAuthorizationITCase.java @@ -0,0 +1,242 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.server.tablet.consumer.group; + +import org.apache.fluss.cluster.ServerNode; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.exception.AuthorizationException; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.rpc.GatewayClientProxy; +import org.apache.fluss.rpc.gateway.TabletServerGateway; +import org.apache.fluss.rpc.messages.CommitConsumerOffsetsRequest; +import org.apache.fluss.rpc.messages.CommitConsumerOffsetsResponse; +import org.apache.fluss.rpc.messages.CreateAclsRequest; +import org.apache.fluss.rpc.messages.FindConsumerGroupOwnerRequest; +import org.apache.fluss.rpc.messages.GetConsumerOffsetsRequest; +import org.apache.fluss.rpc.messages.GetConsumerOffsetsResponse; +import org.apache.fluss.rpc.messages.PbCommitConsumerOffsetsResponseBucket; +import org.apache.fluss.rpc.messages.PbGetConsumerOffsetsRequestBucket; +import org.apache.fluss.rpc.messages.PbGetConsumerOffsetsRequestTable; +import org.apache.fluss.rpc.messages.PbGetConsumerOffsetsResponseBucket; +import org.apache.fluss.rpc.messages.PbGetConsumerOffsetsResponseTable; +import org.apache.fluss.rpc.metrics.TestingClientMetricGroup; +import org.apache.fluss.rpc.netty.client.NettyClient; +import org.apache.fluss.security.acl.AccessControlEntry; +import org.apache.fluss.security.acl.AclBinding; +import org.apache.fluss.security.acl.FlussPrincipal; +import org.apache.fluss.security.acl.OperationType; +import org.apache.fluss.security.acl.PermissionType; +import org.apache.fluss.security.acl.Resource; +import org.apache.fluss.server.metadata.SystemTables; +import org.apache.fluss.server.testutils.FlussClusterExtension; +import org.apache.fluss.server.zk.data.TableRegistration; +import org.apache.fluss.types.DataTypes; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.toPbAclInfos; +import static org.apache.fluss.security.acl.OperationType.DESCRIBE; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.createTable; +import static org.apache.fluss.testutils.common.CommonTestUtils.retry; +import static org.apache.fluss.testutils.common.CommonTestUtils.waitValue; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Integration tests for source-table authorization of consumer-offset fetches. */ +class ConsumerOffsetFetchAuthorizationITCase { + + private static final String GROUP_ID = "fetch-authorization"; + private static final Duration WAIT_TIMEOUT = Duration.ofMinutes(1); + private static final TablePath FIRST_TABLE = TablePath.of("auth_db", "first_source"); + private static final TablePath SECOND_TABLE = TablePath.of("auth_db", "second_source"); + + private final FlussClusterExtension cluster = + FlussClusterExtension.builder() + .setCoordinatorServerListeners("FLUSS://localhost:0, CLIENT://localhost:0") + .setTabletServerListeners("FLUSS://localhost:0, CLIENT://localhost:0") + .setClusterConf( + new Configuration() + .set(ConfigOptions.AUTHORIZER_ENABLED, true) + .set(ConfigOptions.OFFSETS_TABLE_NUM_BUCKETS, 1) + .set(ConfigOptions.OFFSETS_TABLE_REPLICATION_FACTOR, 1)) + .build(); + + private NettyClient client; + private TabletServerGateway clientGateway; + private TabletServerGateway internalGateway; + private TableBucket firstBucket; + private TableBucket secondBucket; + + @BeforeEach + void setUp() throws Exception { + cluster.start(); + TableDescriptor descriptor = + TableDescriptor.builder() + .schema(Schema.newBuilder().column("id", DataTypes.INT()).build()) + .distributedBy(2) + .build(); + long firstTableId = createTable(cluster, FIRST_TABLE, descriptor); + long secondTableId = createTable(cluster, SECOND_TABLE, descriptor); + firstBucket = new TableBucket(firstTableId, 0); + secondBucket = new TableBucket(secondTableId, 0); + cluster.waitUntilAllReplicaReady(firstBucket); + cluster.waitUntilAllReplicaReady(secondBucket); + + ServerNode clientNode = cluster.getTabletServerNodes("CLIENT").get(0); + // Use a separate client so the external endpoint cannot reuse an internal connection. + client = new NettyClient(new Configuration(), TestingClientMetricGroup.newInstance()); + clientGateway = + GatewayClientProxy.createGatewayProxy( + () -> clientNode, client, TabletServerGateway.class); + internalGateway = cluster.newTabletServerClientForNode(clientNode.id()); + retry( + WAIT_TIMEOUT, + () -> + assertThat( + internalGateway.findConsumerGroupOwner( + new FindConsumerGroupOwnerRequest() + .setGroupId(GROUP_ID))) + .succeedsWithin(WAIT_TIMEOUT) + .extracting(response -> response.getOwner().getNodeId()) + .isEqualTo(clientNode.id())); + + TableRegistration offsetsTable = + waitValue( + () -> cluster.getZooKeeperClient().getTable(SystemTables.CONSUMER_OFFSETS), + WAIT_TIMEOUT, + "Consumer offsets table was not created for group " + GROUP_ID); + cluster.waitUntilAllReplicaReady(new TableBucket(offsetsTable.tableId, 0)); + } + + @AfterEach + void tearDown() throws Exception { + try { + if (client != null) { + client.close(); + } + } finally { + cluster.close(); + } + } + + @Test + void testSelectedFetchRequiresSourceDescribePermission() throws Exception { + grant(DESCRIBE, Resource.table(FIRST_TABLE)); + commitOffset(internalGateway, firstBucket, 10L); + commitOffset(internalGateway, secondBucket, 20L); + + assertThatThrownBy( + () -> + clientGateway + .getConsumerOffsets( + selectedOffsetsRequest(firstBucket, secondBucket)) + .get()) + .rootCause() + .isInstanceOf(AuthorizationException.class) + .hasMessageContaining("DESCRIBE") + .hasMessageContaining(SECOND_TABLE.toString()); + + assertThat(fetchOffsets(clientGateway, selectedOffsetsRequest(firstBucket))) + .containsOnlyKeys(firstBucket) + .containsEntry(firstBucket, new ConsumerOffsetAndMetadata(10L, null)); + } + + private void grant(OperationType operation, Resource resource) throws Exception { + List bindings = + Collections.singletonList( + new AclBinding( + resource, + new AccessControlEntry( + FlussPrincipal.ANONYMOUS, + "*", + operation, + PermissionType.ALLOW))); + cluster.newCoordinatorClient() + .createAcls(new CreateAclsRequest().addAllAcls(toPbAclInfos(bindings))) + .get(); + cluster.waitUntilAuthenticationSync(bindings, true); + } + + private static GetConsumerOffsetsRequest selectedOffsetsRequest(TableBucket... buckets) { + GetConsumerOffsetsRequest request = + new GetConsumerOffsetsRequest().setGroupId(GROUP_ID).setAllOffsets(false); + for (TableBucket bucket : buckets) { + PbGetConsumerOffsetsRequestTable table = request.addTable(); + table.setTableId(bucket.getTableId()); + PbGetConsumerOffsetsRequestBucket requestBucket = table.addBucket(); + if (bucket.getPartitionId() != null) { + requestBucket.setPartitionId(bucket.getPartitionId()); + } + requestBucket.setBucketId(bucket.getBucket()); + } + return request; + } + + private static void commitOffset( + TabletServerGateway gateway, TableBucket sourceBucket, long offset) throws Exception { + CommitConsumerOffsetsRequest request = + new CommitConsumerOffsetsRequest().setGroupId(GROUP_ID); + request.addTable() + .setTableId(sourceBucket.getTableId()) + .addBucket() + .setBucketId(sourceBucket.getBucket()) + .setOffset(offset); + + CommitConsumerOffsetsResponse response = gateway.commitConsumerOffsets(request).get(); + assertThat(response.getTablesList()).hasSize(1); + assertThat(response.getTableAt(0).getBucketsList()).hasSize(1); + PbCommitConsumerOffsetsResponseBucket responseBucket = + response.getTableAt(0).getBucketAt(0); + assertThat(responseBucket.hasErrorCode()).isFalse(); + } + + private static Map fetchOffsets( + TabletServerGateway gateway, GetConsumerOffsetsRequest request) throws Exception { + GetConsumerOffsetsResponse response = gateway.getConsumerOffsets(request).get(); + Map offsets = new LinkedHashMap<>(); + for (PbGetConsumerOffsetsResponseTable table : response.getTablesList()) { + for (PbGetConsumerOffsetsResponseBucket bucket : table.getBucketsList()) { + TableBucket tableBucket = + new TableBucket( + table.getTableId(), + bucket.hasPartitionId() ? bucket.getPartitionId() : null, + bucket.getBucketId()); + offsets.put( + tableBucket, + bucket.hasOffset() + ? new ConsumerOffsetAndMetadata( + bucket.getOffset(), + bucket.hasMetadata() ? bucket.getMetadata() : null) + : null); + } + } + return offsets; + } +} diff --git a/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetFetchITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetFetchITCase.java new file mode 100644 index 00000000000..f38405a477b --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetFetchITCase.java @@ -0,0 +1,287 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information regarding copyright + * ownership. The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.server.tablet.consumer.group; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.exception.ConsumerGroupOwnerLoadingException; +import org.apache.fluss.exception.ConsumerGroupOwnerNotAvailableException; +import org.apache.fluss.metadata.PartitionSpec; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.rpc.gateway.TabletServerGateway; +import org.apache.fluss.rpc.messages.CommitConsumerOffsetsRequest; +import org.apache.fluss.rpc.messages.CommitConsumerOffsetsResponse; +import org.apache.fluss.rpc.messages.FindConsumerGroupOwnerRequest; +import org.apache.fluss.rpc.messages.GetConsumerOffsetsRequest; +import org.apache.fluss.rpc.messages.GetConsumerOffsetsResponse; +import org.apache.fluss.rpc.messages.PbCommitConsumerOffsetsRequestBucket; +import org.apache.fluss.rpc.messages.PbCommitConsumerOffsetsRequestTable; +import org.apache.fluss.rpc.messages.PbCommitConsumerOffsetsResponseBucket; +import org.apache.fluss.rpc.messages.PbGetConsumerOffsetsRequestBucket; +import org.apache.fluss.rpc.messages.PbGetConsumerOffsetsRequestTable; +import org.apache.fluss.rpc.messages.PbGetConsumerOffsetsResponseBucket; +import org.apache.fluss.rpc.messages.PbGetConsumerOffsetsResponseTable; +import org.apache.fluss.server.metadata.SystemTables; +import org.apache.fluss.server.testutils.FlussClusterExtension; +import org.apache.fluss.server.zk.data.TableRegistration; +import org.apache.fluss.types.DataTypes; +import org.apache.fluss.utils.ExceptionUtils; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import java.time.Duration; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ExecutionException; + +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.createPartition; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.createTable; +import static org.apache.fluss.testutils.common.CommonTestUtils.retry; +import static org.apache.fluss.testutils.common.CommonTestUtils.waitValue; +import static org.assertj.core.api.Assertions.assertThat; + +/** Integration tests for reading committed consumer offsets. */ +class ConsumerOffsetFetchITCase { + + private static final Duration WAIT_TIMEOUT = Duration.ofMinutes(1); + + @RegisterExtension + public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION = + FlussClusterExtension.builder() + .setNumOfTabletServers(1) + .setClusterConf( + new Configuration() + .set(ConfigOptions.OFFSETS_TABLE_NUM_BUCKETS, 1) + .set(ConfigOptions.OFFSETS_TABLE_REPLICATION_FACTOR, 1)) + .build(); + + @Test + void testFetchAllAndSelectedOffsetsForNonPartitionedTable() throws Exception { + String groupId = "fetch-non-partitioned-group"; + TablePath tablePath = TablePath.of("fetch_non_partitioned_db", "source_table"); + long tableId = + createTable( + FLUSS_CLUSTER_EXTENSION, + tablePath, + TableDescriptor.builder() + .schema( + Schema.newBuilder() + .column("id", DataTypes.INT()) + .primaryKey("id") + .build()) + .distributedBy(3, "id") + .build()); + TableBucket committedBucket0 = new TableBucket(tableId, 0); + TableBucket committedBucket1 = new TableBucket(tableId, 1); + TableBucket missingBucket = new TableBucket(tableId, 2); + + TabletServerGateway ownerGateway = ownerGateway(groupId); + commitOffset(ownerGateway, groupId, committedBucket0, 10L, null); + commitOffset(ownerGateway, groupId, committedBucket1, 20L, "metadata"); + + retry( + WAIT_TIMEOUT, + () -> + assertThat(fetchOffsets(ownerGateway, allOffsetsRequest(groupId))) + .containsOnlyKeys(committedBucket0, committedBucket1) + .containsEntry( + committedBucket0, new ConsumerOffsetAndMetadata(10L, null)) + .containsEntry( + committedBucket1, + new ConsumerOffsetAndMetadata(20L, "metadata"))); + + retry( + WAIT_TIMEOUT, + () -> + assertThat( + fetchOffsets( + ownerGateway, + selectedOffsetsRequest( + groupId, committedBucket0, missingBucket))) + .containsOnlyKeys(committedBucket0, missingBucket) + .containsEntry( + committedBucket0, new ConsumerOffsetAndMetadata(10L, null)) + .containsEntry(missingBucket, null)); + + String emptyGroupId = "fetch-empty-group"; + retry( + WAIT_TIMEOUT, + () -> + assertThat(fetchOffsets(ownerGateway, allOffsetsRequest(emptyGroupId))) + .isEmpty()); + } + + @Test + void testFetchOffsetsForPartitionedTable() throws Exception { + String groupId = "fetch-partitioned-group"; + TablePath tablePath = TablePath.of("fetch_partitioned_db", "source_table"); + long tableId = + createTable( + FLUSS_CLUSTER_EXTENSION, + tablePath, + TableDescriptor.builder() + .schema( + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("region", DataTypes.STRING()) + .primaryKey("id", "region") + .build()) + .distributedBy(1, "id") + .partitionedBy("region") + .build()); + long partitionId = + createPartition( + FLUSS_CLUSTER_EXTENSION, + tablePath, + new PartitionSpec(Collections.singletonMap("region", "cn")), + false); + TableBucket sourceBucket = new TableBucket(tableId, partitionId, 0); + + TabletServerGateway ownerGateway = ownerGateway(groupId); + commitOffset(ownerGateway, groupId, sourceBucket, 30L, "partitioned"); + + retry( + WAIT_TIMEOUT, + () -> + assertThat(fetchOffsets(ownerGateway, allOffsetsRequest(groupId))) + .containsOnlyKeys(sourceBucket) + .containsEntry( + sourceBucket, + new ConsumerOffsetAndMetadata(30L, "partitioned"))); + } + + private static TabletServerGateway ownerGateway(String groupId) throws Exception { + int nodeId = FLUSS_CLUSTER_EXTENSION.getTabletServerNodes().get(0).id(); + TabletServerGateway discoveryGateway = + FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(nodeId); + int ownerId = findOwner(discoveryGateway, groupId); + + TableRegistration offsetsTable = + waitValue( + () -> + FLUSS_CLUSTER_EXTENSION + .getZooKeeperClient() + .getTable(SystemTables.CONSUMER_OFFSETS), + WAIT_TIMEOUT, + "Consumer offsets table was not created for group " + groupId); + TableBucket offsetsBucket = new TableBucket(offsetsTable.tableId, 0); + FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(offsetsBucket); + return FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(ownerId); + } + + private static int findOwner(TabletServerGateway gateway, String groupId) { + return waitValue( + () -> { + try { + return Optional.of( + gateway.findConsumerGroupOwner( + new FindConsumerGroupOwnerRequest() + .setGroupId(groupId)) + .get() + .getOwner() + .getNodeId()); + } catch (ExecutionException e) { + Throwable cause = ExceptionUtils.stripExecutionException(e); + if (cause instanceof ConsumerGroupOwnerNotAvailableException + || cause instanceof ConsumerGroupOwnerLoadingException) { + return Optional.empty(); + } + throw e; + } + }, + WAIT_TIMEOUT, + "Consumer group owner was not found for group " + groupId); + } + + private static GetConsumerOffsetsRequest allOffsetsRequest(String groupId) { + return new GetConsumerOffsetsRequest().setGroupId(groupId).setAllOffsets(true); + } + + private static GetConsumerOffsetsRequest selectedOffsetsRequest( + String groupId, TableBucket... buckets) { + GetConsumerOffsetsRequest request = + new GetConsumerOffsetsRequest().setGroupId(groupId).setAllOffsets(false); + for (TableBucket bucket : buckets) { + PbGetConsumerOffsetsRequestTable table = request.addTable(); + table.setTableId(bucket.getTableId()); + PbGetConsumerOffsetsRequestBucket requestBucket = table.addBucket(); + if (bucket.getPartitionId() != null) { + requestBucket.setPartitionId(bucket.getPartitionId()); + } + requestBucket.setBucketId(bucket.getBucket()); + } + return request; + } + + private static void commitOffset( + TabletServerGateway gateway, + String groupId, + TableBucket sourceBucket, + long offset, + String metadata) + throws Exception { + CommitConsumerOffsetsRequest request = + new CommitConsumerOffsetsRequest().setGroupId(groupId); + PbCommitConsumerOffsetsRequestTable table = request.addTable(); + table.setTableId(sourceBucket.getTableId()); + PbCommitConsumerOffsetsRequestBucket bucket = table.addBucket(); + if (sourceBucket.getPartitionId() != null) { + bucket.setPartitionId(sourceBucket.getPartitionId()); + } + bucket.setBucketId(sourceBucket.getBucket()).setOffset(offset); + if (metadata != null) { + bucket.setMetadata(metadata); + } + + CommitConsumerOffsetsResponse response = gateway.commitConsumerOffsets(request).get(); + assertThat(response.getTablesList()).hasSize(1); + assertThat(response.getTableAt(0).getBucketsList()).hasSize(1); + PbCommitConsumerOffsetsResponseBucket responseBucket = + response.getTableAt(0).getBucketAt(0); + assertThat(responseBucket.hasErrorCode()).isFalse(); + } + + private static Map fetchOffsets( + TabletServerGateway gateway, GetConsumerOffsetsRequest request) throws Exception { + GetConsumerOffsetsResponse response = gateway.getConsumerOffsets(request).get(); + Map offsets = new LinkedHashMap<>(); + for (PbGetConsumerOffsetsResponseTable table : response.getTablesList()) { + for (PbGetConsumerOffsetsResponseBucket bucket : table.getBucketsList()) { + TableBucket tableBucket = + new TableBucket( + table.getTableId(), + bucket.hasPartitionId() ? bucket.getPartitionId() : null, + bucket.getBucketId()); + ConsumerOffsetAndMetadata offsetAndMetadata = + bucket.hasOffset() + ? new ConsumerOffsetAndMetadata( + bucket.getOffset(), + bucket.hasMetadata() ? bucket.getMetadata() : null) + : null; + offsets.put(tableBucket, offsetAndMetadata); + } + } + return offsets; + } +} diff --git a/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetSchemaTest.java b/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetSchemaTest.java index 1227d1ca766..59dd1b271aa 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetSchemaTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetSchemaTest.java @@ -28,6 +28,7 @@ import org.junit.jupiter.api.Test; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch; @@ -49,10 +50,10 @@ class ConsumerOffsetSchemaTest { @Test void testSchemaMatchesFip() { - // PK: (group_id, record_type, table_id, partition_id, bucket_id) + // PK: (record_type, group_id, table_id, partition_id, bucket_id) assertThat(ConsumerOffsetSchema.getPrimaryKeys()) .containsExactly( - "group_id", "record_type", "table_id", "partition_id", "bucket_id"); + "record_type", "group_id", "table_id", "partition_id", "bucket_id"); // Value: (offset, metadata) — FIP-53 has no commit_timestamp. assertThat(ConsumerOffsetSchema.getTableSchema().getColumnNames()) .containsExactly( @@ -153,12 +154,12 @@ void testEncodeOffsetKeyMatchesRowKeyEncoding() { } @Test - void testGroupPrefixIsBytePrefixOfFullKey() { + void testOffsetCommitPrefixIsBytePrefixOfFullKey() { // The prefix scan correctness of the zero-memory ConsumerGroupManager depends on this // property: - // encodeGroupPrefix(g) must be a strict byte prefix of every encodeOffsetKey of group g. + // encodeOffsetCommitPrefix(g) must be a strict byte prefix of every offset-commit key. for (String group : new String[] {"a", "my-group", "group-with-ünicode", "x"}) { - byte[] prefix = offsetSchema.encodeGroupPrefix(group); + byte[] prefix = offsetSchema.encodeOffsetCommitPrefix(group); assertThat(prefix).isNotEmpty(); byte[] keyNonPartitioned = offsetSchema.encodeOffsetKey(group, 1L, null, 0); @@ -173,7 +174,7 @@ void testGroupPrefixIsBytePrefixOfFullKey() { @Test void testDifferentGroupsPrefixesDoNotCrossMatch() { - byte[] prefixA = offsetSchema.encodeGroupPrefix("group-a"); + byte[] prefixA = offsetSchema.encodeOffsetCommitPrefix("group-a"); byte[] keyB = offsetSchema.encodeOffsetKey("group-b", 1L, null, 0); // "group-a" and "group-b" share a common byte prefix ("group-"), but the full encoded // group prefixes must diverge before the end — otherwise a prefix scan for group-a @@ -184,9 +185,9 @@ void testDifferentGroupsPrefixesDoNotCrossMatch() { } @Test - void testEncodeGroupPrefixMatchesStandardBucketKeyEncoder() { + void testGroupBucketKeyMatchesStandardBucketKeyEncoder() { // The standard KV path creates a bucket key encoder from the table's schema and - // the distributedBy column ("group_id"). ConsumerOffsetSchema.encodeGroupPrefix must + // the distributedBy column ("group_id"). ConsumerOffsetSchema.encodeGroupBucketKey must // produce // identical bytes — this is the guarantee that FindConsumerGroupOwner routes to the // same bucket where the data actually resides. @@ -214,10 +215,36 @@ void testEncodeGroupPrefixMatchesStandardBucketKeyEncoder() { -1, new ConsumerOffsetAndMetadata(-1L, "")); byte[] standardBytes = standardEncoder.encodeKey(row); - byte[] consumerOffsetSchemaBytes = offsetSchema.encodeGroupPrefix(group); + byte[] consumerOffsetSchemaBytes = offsetSchema.encodeGroupBucketKey(group); assertThat(consumerOffsetSchemaBytes) - .as("encodeGroupPrefix(%s) must match the standard bucket key encoder", group) + .as( + "encodeGroupBucketKey(%s) must match the standard bucket key encoder", + group) + .isEqualTo(standardBytes); + } + } + + @Test + void testOffsetCommitPrefixMatchesRecordTypeAndGroupEncoder() { + KeyEncoder standardEncoder = + KeyEncoder.ofBucketKeyEncoder( + ConsumerOffsetSchema.getTableSchema().getRowType(), + Arrays.asList("record_type", "group_id"), + null); + + for (String group : new String[] {"a", "my-group", "group-with-ünicode", "x"}) { + CompactedRow row = + offsetSchema.encodeOffsetRow( + group, + ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE, + -1L, + -1L, + -1, + new ConsumerOffsetAndMetadata(-1L, "")); + byte[] standardBytes = standardEncoder.encodeKey(row); + assertThat(offsetSchema.encodeOffsetCommitPrefix(group)) + .as("offset-commit prefix for %s must encode record_type then group_id", group) .isEqualTo(standardBytes); } } @@ -249,7 +276,7 @@ void testSharedSchemaEncodesRowsAndKeysAcrossThreads() throws Exception { for (int i = 0; i < threads; i++) { String groupId = "group-" + i; byte[] expectedKey = offsetSchema.encodeOffsetKey(groupId, 42L, 7L, 3); - byte[] expectedPrefix = offsetSchema.encodeGroupPrefix(groupId); + byte[] expectedPrefix = offsetSchema.encodeOffsetCommitPrefix(groupId); futures.add( executor.submit( () -> { @@ -268,7 +295,7 @@ void testSharedSchemaEncodesRowsAndKeysAcrossThreads() throws Exception { assertThat(rowsReady.await(30, TimeUnit.SECONDS)) .isTrue(); } - assertThat(offsetSchema.encodeGroupPrefix(groupId)) + assertThat(offsetSchema.encodeOffsetCommitPrefix(groupId)) .isEqualTo(expectedPrefix); assertThat(offsetSchema.encodeOffsetKey(row)) .isEqualTo(expectedKey); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetTableITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetTableITCase.java index d1adcb9c3dd..9aac1e69720 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetTableITCase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetTableITCase.java @@ -89,7 +89,7 @@ void testInternalCreationIgnoresKvRestrictionAndClusterDataLakeFormat() throws E int standardBucket = BucketingFunction.of(tableConfig.getDataLakeFormat().orElse(null)) .bucketing( - offsetSchema.encodeGroupPrefix(groupId), + offsetSchema.encodeGroupBucketKey(groupId), tableRegistration.bucketCount); assertThat(offsetSchema.bucketFor(groupId, tableRegistration.bucketCount)) .isEqualTo(standardBucket); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java b/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java index b35ff256048..2271d1bc853 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java @@ -36,6 +36,8 @@ import org.apache.fluss.rpc.messages.CommitConsumerOffsetsRequest; import org.apache.fluss.rpc.messages.CommitConsumerOffsetsResponse; import org.apache.fluss.rpc.messages.FetchLogResponse; +import org.apache.fluss.rpc.messages.GetConsumerOffsetsRequest; +import org.apache.fluss.rpc.messages.GetConsumerOffsetsResponse; import org.apache.fluss.rpc.messages.LookupResponse; import org.apache.fluss.rpc.messages.PbBucketMetadata; import org.apache.fluss.rpc.messages.PbCommitConsumerOffsetsRequestBucket; @@ -43,6 +45,9 @@ import org.apache.fluss.rpc.messages.PbCommitConsumerOffsetsResponseBucket; import org.apache.fluss.rpc.messages.PbCommitConsumerOffsetsResponseTable; import org.apache.fluss.rpc.messages.PbFetchLogRespForBucket; +import org.apache.fluss.rpc.messages.PbGetConsumerOffsetsRequestBucket; +import org.apache.fluss.rpc.messages.PbGetConsumerOffsetsRequestTable; +import org.apache.fluss.rpc.messages.PbGetConsumerOffsetsResponseBucket; import org.apache.fluss.rpc.messages.PbPartitionMetadata; import org.apache.fluss.rpc.messages.PbProduceLogReqForBucket; import org.apache.fluss.rpc.messages.PbProduceLogRespForBucket; @@ -72,6 +77,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.UUID; @@ -144,6 +150,87 @@ void testCommitConsumerOffsetsRequestRejectsReservedPartitionId() { .hasMessageContaining("omit partition_id"); } + @Test + void testDecodeGetConsumerOffsetsRequest() { + GetConsumerOffsetsRequest allOffsets = + new GetConsumerOffsetsRequest().setGroupId("group-1").setAllOffsets(true); + assertThat(ServerRpcMessageUtils.decodeGetConsumerOffsetsRequest(allOffsets)).isNull(); + + GetConsumerOffsetsRequest empty = + new GetConsumerOffsetsRequest().setGroupId("group-1").setAllOffsets(false); + assertThat(ServerRpcMessageUtils.decodeGetConsumerOffsetsRequest(empty)).isEmpty(); + + PbGetConsumerOffsetsRequestBucket bucket = + new PbGetConsumerOffsetsRequestBucket().setBucketId(2); + PbGetConsumerOffsetsRequestTable table = + new PbGetConsumerOffsetsRequestTable() + .setTableId(3L) + .addAllBuckets(Collections.singletonList(bucket)); + GetConsumerOffsetsRequest selected = + new GetConsumerOffsetsRequest() + .setGroupId("group-1") + .setAllOffsets(false) + .addAllTables(Collections.singletonList(table)); + assertThat(ServerRpcMessageUtils.decodeGetConsumerOffsetsRequest(selected)) + .containsExactly(new TableBucket(3L, 2)); + } + + @Test + void testGetConsumerOffsetsRequestRejectsInvalidCombinations() { + PbGetConsumerOffsetsRequestBucket bucket = + new PbGetConsumerOffsetsRequestBucket().setBucketId(2); + PbGetConsumerOffsetsRequestTable table = + new PbGetConsumerOffsetsRequestTable() + .setTableId(3L) + .addAllBuckets(Arrays.asList(bucket, bucket)); + + GetConsumerOffsetsRequest allOffsetsWithBuckets = + new GetConsumerOffsetsRequest() + .setGroupId("group-1") + .setAllOffsets(true) + .addAllTables(Collections.singletonList(table)); + assertThatThrownBy( + () -> + ServerRpcMessageUtils.decodeGetConsumerOffsetsRequest( + allOffsetsWithBuckets)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("all_offsets"); + + GetConsumerOffsetsRequest duplicateBuckets = + new GetConsumerOffsetsRequest() + .setGroupId("group-1") + .setAllOffsets(false) + .addAllTables(Collections.singletonList(table)); + assertThatThrownBy( + () -> + ServerRpcMessageUtils.decodeGetConsumerOffsetsRequest( + duplicateBuckets)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Duplicate table bucket"); + } + + @Test + void testBuildGetConsumerOffsetsResponsePreservesOrderAndRequiredMetadata() { + TableBucket committedBucket = new TableBucket(2L, 3); + TableBucket missingBucket = new TableBucket(2L, 1); + Map offsets = new LinkedHashMap<>(); + offsets.put(committedBucket, new ConsumerOffsetAndMetadata(42L, "metadata")); + offsets.put(missingBucket, null); + + GetConsumerOffsetsResponse response = + ServerRpcMessageUtils.makeGetConsumerOffsetsResponse(offsets); + + List buckets = + response.getTablesList().get(0).getBucketsList(); + assertThat(buckets) + .extracting(PbGetConsumerOffsetsResponseBucket::getBucketId) + .containsExactly(3, 1); + assertThat(buckets.get(0).getOffset()).isEqualTo(42L); + assertThat(buckets.get(0).getMetadata()).isEqualTo("metadata"); + assertThat(buckets.get(1).hasOffset()).isFalse(); + assertThat(buckets.get(1).hasMetadata()).isFalse(); + } + @ParameterizedTest @ValueSource(booleans = {false, true}) void testCommitConsumerOffsetsRequestRejectsPartitionSentinelAlias(boolean sentinelFirst) { diff --git a/website/docs/security/authorization.md b/website/docs/security/authorization.md index 0287bbe23bf..7c71407ccf6 100644 --- a/website/docs/security/authorization.md +++ b/website/docs/security/authorization.md @@ -124,6 +124,7 @@ Below is a summary of the currently public protocols and their relationship with | GET_CLUSTER_HEALTH | DESCRIBE | Cluster | | | REGISTER_PRODUCER_OFFSETS | WRITE | Table | Requires `WRITE` on all tables in the request. | | GET_PRODUCER_OFFSETS | READ | Table | Offsets for tables without `READ` permission are filtered out. | +| GET_CONSUMER_OFFSETS | DESCRIBE | Table | Requires `DESCRIBE` on every source table in the request. `allOffsets` requests do not perform table-level ACL checks yet. | | DELETE_PRODUCER_OFFSETS | WRITE | Table | Requires `WRITE` on all tables in the producer offset snapshot. | | ACQUIRE_KV_SNAPSHOT_LEASE | READ | Table | Requires `READ` on all tables in the request. | | RELEASE_KV_SNAPSHOT_LEASE | READ | Table | Requires `READ` on all tables in the request. |