[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;
@@ -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,13 +61,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 +88,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 {
@@ -101,6 +108,7 @@ public final class ConsumerGroupManager {
     private final Executor ioExecutor;
     private final AtomicReference<CompletableFuture<TableInfo>> tableInitFuture =
             new AtomicReference<>();
+    @Nullable private volatile CachedValueDecoder cachedValueDecoder;
 
     /** Creates a manager for consumer groups on this tablet server. */
     public ConsumerGroupManager(
@@ -431,6 +439,84 @@ 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 = getValueDecoder(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
+    ValueDecoder getValueDecoder(SchemaGetter schemaGetter) {
+        CachedValueDecoder cached = cachedValueDecoder;
+        if (cached == null || cached.schemaGetter != schemaGetter) {
+            // Concurrent misses may create two decoders. Each caller keeps its matching pair.
+            cached = new CachedValueDecoder(schemaGetter);
+            cachedValueDecoder = cached;
+        }
+        return cached.decoder;
+    }
+
+    @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,
@@ -527,4 +613,14 @@ static Errors mapCommitError(Errors error) {
                 return error;
         }
     }
+
+    private static final class CachedValueDecoder {
+        private final SchemaGetter schemaGetter;
+        private final ValueDecoder decoder;
+
+        private CachedValueDecoder(SchemaGetter schemaGetter) {
+            this.schemaGetter = schemaGetter;
+            this.decoder = ConsumerOffsetSchema.createValueDecoder(schemaGetter);
+        }
+    }
 }