Skip to content

[server] Implement consumer offset fetches #5138

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pisceslj wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[server] Implement consumer offset fetches #5138

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 28 additions & 4 deletions fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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 | 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
Expand Down
97 changes: 96 additions & 1 deletion ...ver/src/main/java/org/apache/fluss/server/tablet/consumer/group/ConsumerGroupManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -57,13 +61,15 @@
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;

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;
Expand All @@ -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 {
Expand All @@ -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;
Expand Down Expand Up @@ -136,6 +144,48 @@ public CompletableFuture<ServerNode> findOwner(String groupId, String listenerNa
listenerName));
}

/** Reads all or selected committed offsets from the local leader replica. */
public CompletableFuture<Map<TableBucket, ConsumerOffsetAndMetadata>> handleGetConsumerOffsets(

@wuchong wuchong Nov 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

move it after handleCommitConsumerOffsets

String groupId, Replica offsetsReplica, List<TableBucket> requestedBuckets) {
return CompletableFuture.supplyAsync(

@wuchong wuchong Nov 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we run the offset fetch directly on the RPC worker thread and drop supplyAsync(..., ioExecutor)? The existing KV lookup and prefix lookup handlers already call Replica.lookups and Replica.prefixLookup on that thread, so we can avoid queueing the same reads on a second executor here.

() -> {
Map<TableBucket, ConsumerOffsetAndMetadata> offsets = new HashMap<>();
ValueDecoder decoder = getValueDecoder(offsetsReplica);
try {
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 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<TableMetadata> tables) {
for (TableMetadata table : tables) {
Expand Down Expand Up @@ -431,6 +481,51 @@ public CompletableFuture<Map<TableBucket, Errors>> handleCommitConsumerOffsets(
writeResults -> handleWriteCompletion(groupId, results, writeResults));
}

@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));
}
}

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;

@wuchong wuchong Nov 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we drop the cached schemaGetter field from ConsumerGroupManager and pass offsetsReplica.getSchemaGetter() directly to ConsumerOffsetSchema.createValueDecoder(...)? The replica already owns the getter for this bucket.

valueDecoder = ConsumerOffsetSchema.createValueDecoder(currentSchemaGetter);
}
currentDecoder = valueDecoder;
}
}
return currentDecoder;
}

@VisibleForTesting
static Map<TableBucket, Errors> handleWriteCompletion(
String groupId,
Expand Down
55 changes: 47 additions & 8 deletions ...ver/src/main/java/org/apache/fluss/server/tablet/consumer/group/ConsumerOffsetSchema.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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,
Comment on lines 66 to +67

@wuchong wuchong Nov 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we order the columns in createTableSchema() to match the primary key, with RECORD_TYPE_COLUMN first and GROUP_ID_COLUMN second? That way the offset-commit prefix row can use the same (record_type, group_id) order as the key. Please update the positional row encoders to match.

TABLE_ID_COLUMN,
PARTITION_ID_COLUMN,
BUCKET_ID_COLUMN));

private static final List<String> OFFSET_COMMIT_PREFIX_KEYS =

@wuchong wuchong Nov 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: 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]);
Expand All @@ -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> offsetCommitPrefixEncoder =

@wuchong wuchong Nov 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: groupLookupPrefixEncoder

ThreadLocal.withInitial(
() -> KeyEncoder.ofBucketKeyEncoder(ROW_TYPE, OFFSET_COMMIT_PREFIX_KEYS, null));

/** Returns the schema of the consumer-offset system table. */
public static Schema getTableSchema() {
Expand Down Expand Up @@ -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. */
Expand All @@ -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. */
Expand Down
Loading
Loading