[server] Implement consumer offset fetches

fluss-common/src/main/java/org/apache/fluss/exception/InvalidRequestException.java

@@ -0,0 +1,30 @@
+/*
+ * 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.exception;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+/** Thrown when a request contains an invalid combination or value of parameters. */
+@PublicEvolving
+public class InvalidRequestException extends ApiException {
+    private static final long serialVersionUID = 1L;
+
+    public InvalidRequestException(String message) {
+        super(message);
+    }
+}

fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java

@@ -48,6 +48,7 @@
 import org.apache.fluss.exception.InvalidPartitionException;
 import org.apache.fluss.exception.InvalidProducerIdException;
 import org.apache.fluss.exception.InvalidReplicationFactorException;
+import org.apache.fluss.exception.InvalidRequestException;
 import org.apache.fluss.exception.InvalidRequiredAcksException;
 import org.apache.fluss.exception.InvalidScanRequestException;
 import org.apache.fluss.exception.InvalidServerRackInfoException;
@@ -319,7 +320,11 @@ public enum Errors {
     CONSUMER_OFFSET_METADATA_TOO_LARGE(
             80,
             "The metadata field of the consumer offset request is too large.",
-            ConsumerOffsetMetadataTooLargeException::new);
+            ConsumerOffsetMetadataTooLargeException::new),
+    INVALID_REQUEST(
+            81,
+            "The request contains an invalid combination or value of parameters.",
+            InvalidRequestException::new);
 
     private static final Logger LOG = LoggerFactory.getLogger(Errors.class);
 

fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/ConsumerGroupErrorsTest.java

@@ -22,6 +22,7 @@
 import org.apache.fluss.exception.ConsumerGroupOwnerNotAvailableException;
 import org.apache.fluss.exception.ConsumerOffsetMetadataTooLargeException;
 import org.apache.fluss.exception.InvalidConsumerGroupIdException;
+import org.apache.fluss.exception.InvalidRequestException;
 import org.apache.fluss.exception.NotConsumerGroupOwnerException;
 import org.apache.fluss.exception.RetriableException;
 
@@ -40,6 +41,7 @@ void testErrorCodes() {
         assertThat(Errors.NOT_CONSUMER_GROUP_OWNER.code()).isEqualTo(78);
         assertThat(Errors.CONSUMER_GROUP_ID_NOT_FOUND.code()).isEqualTo(79);
         assertThat(Errors.CONSUMER_OFFSET_METADATA_TOO_LARGE.code()).isEqualTo(80);
+        assertThat(Errors.INVALID_REQUEST.code()).isEqualTo(81);
     }
 
     @Test
@@ -56,6 +58,9 @@ void testExceptionMappings() {
                 .isEqualTo(Errors.CONSUMER_GROUP_ID_NOT_FOUND);
         assertThat(Errors.forException(new ConsumerOffsetMetadataTooLargeException("large")))
                 .isEqualTo(Errors.CONSUMER_OFFSET_METADATA_TOO_LARGE);
+        assertThat(Errors.forException(new InvalidRequestException("invalid request")))
+                .isEqualTo(Errors.INVALID_REQUEST);
+        assertThat(Errors.INVALID_REQUEST.exception()).isNotInstanceOf(RetriableException.class);
     }
 
     @Test

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<CommitConsumerOffsetsResponse> commitConsumerOffsets(
     @Override
     public CompletableFuture<GetConsumerOffsetsResponse> getConsumerOffsets(
             GetConsumerOffsetsRequest request) {
-        CompletableFuture<GetConsumerOffsetsResponse> response = new CompletableFuture<>();
-        response.completeExceptionally(
-                new UnsupportedOperationException("getConsumerOffsets is not supported yet"));
-        return response;
+        final String groupId = request.getGroupId();
+        final Replica offsetsReplica;
+        final List<TableBucket> requestedBuckets;
+        try {
+            offsetsReplica = groupManager.validateGroupId(groupId);
+            requestedBuckets = decodeGetConsumerOffsetsRequest(request);
+        } catch (ApiException 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

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;
@@ -36,7 +37,9 @@
 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,13 +60,16 @@
 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.apache.fluss.utils.concurrent.FutureUtils;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 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 {
@@ -431,6 +437,74 @@ public CompletableFuture<Map<TableBucket, Errors>> handleCommitConsumerOffsets(
                 writeResults -> handleWriteCompletion(groupId, results, writeResults));
     }
 
+    /**
+     * Reads all or selected committed offsets from the local leader replica on the calling thread.
+     */
+    public CompletableFuture<Map<TableBucket, ConsumerOffsetAndMetadata>> handleGetConsumerOffsets(
+            String groupId, Replica offsetsReplica, List<TableBucket> requestedBuckets) {
+        Map<TableBucket, ConsumerOffsetAndMetadata> offsets = new HashMap<>();
+        try {
+            ValueDecoder decoder =
+                    ConsumerOffsetSchema.createValueDecoder(offsetsReplica.getSchemaGetter());
+            if (requestedBuckets == null) {
+                populateAllOffsets(
+                        offsetsReplica.prefixLookup(offsetSchema.encodeOffsetCommitPrefix(groupId)),
+                        decoder,
+                        offsets);
+            } else {
+                List<byte[]> 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 CompletableFuture.completedFuture(offsets);
+        } catch (Exception e) {
+            LOG.warn("Failed to read offsets for group {}.", groupId, e);
+            ApiException exception =
+                    e instanceof ApiException
+                            ? (ApiException) e
+                            : Errors.forException(e).exception(e.getMessage());
+            return FutureUtils.completedExceptionally(exception);
+        }
+    }
+
+    @VisibleForTesting
+    static void populateAllOffsets(
+            List<ByteArraySlice> values,
+            ValueDecoder valueDecoder,
+            Map<TableBucket, ConsumerOffsetAndMetadata> 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<TableBucket> requestedBuckets,
+            List<ByteArraySlice> values,
+            ValueDecoder valueDecoder,
+            Map<TableBucket, ConsumerOffsetAndMetadata> 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));
+        }
+    }
+
     @VisibleForTesting
     static Map<TableBucket, Errors> handleWriteCompletion(
             String groupId,

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<String> 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<String> GROUP_LOOKUP_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<KeyEncoder> primaryKeyEncoder =
             ThreadLocal.withInitial(
                     () -> KeyEncoder.ofBucketKeyEncoder(ROW_TYPE, PRIMARY_KEYS, null));
-    private final ThreadLocal<KeyEncoder> groupIdEncoder =
+    private final ThreadLocal<KeyEncoder> groupBucketKeyEncoder =
             ThreadLocal.withInitial(
                     () ->
                             KeyEncoder.ofBucketKeyEncoder(
                                     ROW_TYPE, Collections.singletonList(GROUP_ID_COLUMN), null));
+    private final ThreadLocal<KeyEncoder> groupLookupPrefixEncoder =
+            ThreadLocal.withInitial(
+                    () -> KeyEncoder.ofBucketKeyEncoder(ROW_TYPE, GROUP_LOOKUP_PREFIX_KEYS, null));
 
     /** Returns the schema of the consumer-offset system table. */
     public static Schema getTableSchema() {
@@ -101,18 +112,48 @@ public static String getBucketKeyColumn() {
     public byte[] encodeOffsetKey(String groupId, long tableId, Long partitionId, int bucketId) {
         return encodeOffsetKey(
                 GenericRow.of(
-                        BinaryString.fromString(groupId),
                         OFFSET_COMMIT_RECORD_TYPE,
+                        BinaryString.fromString(groupId),
                         tableId,
                         partitionId == null ? NON_PARTITIONED_PARTITION_ID : partitionId,
                         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) {
+        // The bucket-key encoder reads group_id from column 1; record_type is only a placeholder.
+        return groupBucketKeyEncoder
+                .get()
+                .encodeKey(
+                        GenericRow.of(OFFSET_COMMIT_RECORD_TYPE, 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 groupLookupPrefixEncoder
+                .get()
+                .encodeKey(
+                        GenericRow.of(OFFSET_COMMIT_RECORD_TYPE, BinaryString.fromString(groupId)));
     }
 
     /** Encodes the complete primary key of an already materialized offset row. */
@@ -121,7 +162,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. */
@@ -134,8 +175,8 @@ CompactedRow encodeOffsetRow(
             ConsumerOffsetAndMetadata offsetAndMetadata) {
         CompactedRowEncoder encoder = rowEncoder.get();
         encoder.startNewRow();
-        encoder.encodeField(0, BinaryString.fromString(groupId));
-        encoder.encodeField(1, recordType);
+        encoder.encodeField(0, recordType);
+        encoder.encodeField(1, BinaryString.fromString(groupId));
         encoder.encodeField(2, tableId);
         encoder.encodeField(3, partitionId == null ? NON_PARTITIONED_PARTITION_ID : partitionId);
         encoder.encodeField(4, bucketId);
@@ -146,8 +187,8 @@ CompactedRow encodeOffsetRow(
 
     private static Schema createTableSchema() {
         return Schema.newBuilder()
-                .column(GROUP_ID_COLUMN, DataTypes.STRING())
                 .column(RECORD_TYPE_COLUMN, DataTypes.INT())
+                .column(GROUP_ID_COLUMN, DataTypes.STRING())
                 .column(TABLE_ID_COLUMN, DataTypes.BIGINT())
                 .column(PARTITION_ID_COLUMN, DataTypes.BIGINT())
                 .column(BUCKET_ID_COLUMN, DataTypes.INT())