You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "pisceslj (via GitHub)" <gi...@apache.org> on 2026/10/13 14:09:03 UTC
[PR] [server] Add consumer offset table foundations [fluss]
pisceslj opened a new pull request, #4626:
URL: https://github.com/apache/fluss/pull/4626
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink], [rust], [python], [c++], [elixir]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
- **Generative AI disclosure:** Indicate whether generative AI tools were used in authoring this PR. If yes, specify the tool below.
- [ ] No generative AI tools used
- [ ] Yes (please specify the tool below)
**(The sections below can be removed for hotfixes or typos)**
-->
<!--
Generated-by: [Tool Name and Version] following [the guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
-->
### Purpose
Linked issue: close #4625
Part of #4608
This is the first step of the consumer-offset system table work. It adds only the
foundational configuration, schema, and value model needed by later subtasks; it does not
wire the table into the server or add RPC handlers.
### Brief change log
- Add server-side configuration options for the consumer-offset system table:
- `server.consumer-offsets.metadata.max-size`
- `server.consumer-offsets.ttl`
- `server.consumer-offsets.table.bucket-count`
- `server.consumer-offsets.table.replication-factor`
- `server.consumer-offsets.commit-timeout`
- Add `ConsumerOffsetTableConfig` as the strongly typed configuration model and table identity
for `sys.consumer_offsets`.
- Add `ConsumerOffsetSchema` with the table schema, primary-key layout, and
row/key/prefix encoders.
- Add `ConsumerOffsetAndMetadata` as the immutable committed-offset and metadata value object.
- Add unit tests for configuration loading and schema encoding compatibility.
- Document the new server-side consumer-offset options.
### Tests
- `ConsumerOffsetTableConfigTest`
- `ConsumerOffsetSchemaTest`
- `ConsumerOffsetAndMetadataTest`
- `./mvnw validate -pl fluss-common,fluss-server -am`
### API and Format
No public API or storage format change in this subtask.
This change only introduces internal configuration options and classes. The
`sys.consumer_offsets` schema is added as a constant and is not yet used by the server;
wiring and RPC changes will be introduced in later subtasks.
### Documentation
Adds the new `server.consumer-offsets.*` configuration options to
`website/docs/maintenance/configuration.md`.
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add consumer offset table foundations [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on PR #4626:
URL: https://github.com/apache/fluss/pull/4626#issuecomment-6213004845
> Looks good to me. One concern is that, now we always create a new encoder `createFullKeyEncoder()` when we call `ConsumerOffsetSchema.encodeOffsetKey(row)`. Depending on how frequently this method is called, especially in a loop over a batch of offsets, this could add unnecessary allocation and initialization overhead. We can evaluate whether reuse would be worthwhile once the call sites are introduced in follow-up PRs, and make any needed optimization there. It may not block the current PR.
+1 for this. We should reuse the key encoder (but not shared globally as it's not thread safe), we can make key encoder as a parameter of `encodeOffsetKey`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add consumer offset table foundations [fluss]
Posted by "pisceslj (via GitHub)" <gi...@apache.org>.
pisceslj commented on code in PR #4626:
URL: https://github.com/apache/fluss/pull/4626#discussion_r4266717476
##########
fluss-server/src/test/java/org/apache/fluss/server/tablet/coordinator/group/ConsumerOffsetSchemaTest.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * 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.coordinator.group;
+
+import org.apache.fluss.bucketing.FlussBucketingFunction;
+import org.apache.fluss.memory.MemorySegment;
+import org.apache.fluss.metadata.KvFormat;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.compacted.CompactedRow;
+import org.apache.fluss.row.decode.FixedSchemaDecoder;
+import org.apache.fluss.row.encode.KeyEncoder;
+import org.apache.fluss.row.encode.ValueEncoder;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for the sys.consumer_offsets schema and key/value encode/decode. */
+class ConsumerOffsetSchemaTest {
+
+ private static final short SCHEMA_ID = (short) 1;
+
+ private final FixedSchemaDecoder decoder =
+ new FixedSchemaDecoder(KvFormat.COMPACTED, ConsumerOffsetSchema.getTableSchema());
+
+ @Test
+ void testSchemaMatchesFip() {
+ // PK: (group_id, record_type, table_id, partition_id, bucket_id)
+ assertThat(ConsumerOffsetSchema.getPrimaryKeys())
+ .containsExactly(
+ "group_id", "record_type", "table_id", "partition_id", "bucket_id");
+ // Value: (offset, metadata) — FIP-53 has no commit_timestamp.
+ assertThat(ConsumerOffsetSchema.getTableSchema().getColumnNames())
+ .containsExactly(
+ "group_id",
+ "record_type",
+ "table_id",
+ "partition_id",
+ "bucket_id",
+ "offset",
+ "metadata");
+ }
+
+ @Test
+ void testOffsetCommitValueRoundTrip() {
+ CompactedRow row =
+ ConsumerOffsetSchema.encodeOffsetRow(
+ "my-group",
+ ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE,
+ 42L,
+ 7L,
+ 3,
+ new ConsumerOffsetAndMetadata(100L, "wm=5"));
+
+ InternalRow decoded = decode(row);
+ assertThat(decoded.getString(0).toString()).isEqualTo("my-group");
+ assertThat(decoded.getInt(1)).isEqualTo(ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE);
+ assertThat(decoded.getLong(2)).isEqualTo(42L);
+ assertThat(decoded.getLong(3)).isEqualTo(7L);
+ assertThat(decoded.getInt(4)).isEqualTo(3);
+ assertThat(decoded.getLong(5)).isEqualTo(100L);
+ assertThat(decoded.getString(6).toString()).isEqualTo("wm=5");
+ }
+
+ @Test
+ void testNullPartitionIdEncodesAsMinusOneSentinel() {
+ // Non-partitioned table: partitionId == null materializes as -1 (PK is NOT NULL).
+ CompactedRow row =
+ ConsumerOffsetSchema.encodeOffsetRow(
+ "g",
+ ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE,
+ 1L,
+ null,
+ 0,
+ new ConsumerOffsetAndMetadata(5L, ""));
+ assertThat(decode(row).getLong(3)).isEqualTo(-1L);
+
+ // Partition id 0 is a valid partition id and must survive the round-trip unchanged.
+ CompactedRow zeroPartition =
+ ConsumerOffsetSchema.encodeOffsetRow(
+ "g",
+ ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE,
+ 1L,
+ 0L,
+ 0,
+ new ConsumerOffsetAndMetadata(5L, ""));
+ assertThat(decode(zeroPartition).getLong(3)).isEqualTo(0L);
+ }
+
+ @Test
+ void testEncodeOffsetKeyMatchesRowKeyEncoding() {
+ CompactedRow row =
+ ConsumerOffsetSchema.encodeOffsetRow(
+ "my-group",
+ ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE,
+ 42L,
+ 7L,
+ 3,
+ new ConsumerOffsetAndMetadata(100L, ""));
+
+ byte[] keyFromRow = ConsumerOffsetSchema.encodeOffsetKey(row);
+ byte[] keyFromFields = ConsumerOffsetSchema.encodeOffsetKey("my-group", 42L, 7L, 3);
+ assertThat(keyFromFields).isEqualTo(keyFromRow);
+ }
+
+ @Test
+ void testGroupPrefixIsBytePrefixOfFullKey() {
+ // The prefix scan correctness of the zero-memory GroupManager depends on this property:
+ // encodeGroupPrefix(g) must be a strict byte prefix of every encodeOffsetKey of group g.
+ for (String group : new String[] {"a", "my-group", "group-with-ünicode", "x"}) {
+ byte[] prefix = ConsumerOffsetSchema.encodeGroupPrefix(group);
+ assertThat(prefix).isNotEmpty();
+
+ byte[] keyNonPartitioned = ConsumerOffsetSchema.encodeOffsetKey(group, 1L, null, 0);
+ byte[] keyPartitioned = ConsumerOffsetSchema.encodeOffsetKey(group, 2L, 99L, 5);
+
+ assertThat(isPrefix(prefix, keyNonPartitioned))
+ .as("prefix of group %s must be a byte prefix of its full keys", group)
+ .isTrue();
+ assertThat(isPrefix(prefix, keyPartitioned)).isTrue();
+ }
+ }
+
+ @Test
+ void testDifferentGroupsPrefixesDoNotCrossMatch() {
+ byte[] prefixA = ConsumerOffsetSchema.encodeGroupPrefix("group-a");
+ byte[] keyB = ConsumerOffsetSchema.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
+ // would also see group-b's rows.
+ assertThat(isPrefix(prefixA, keyB))
+ .as("group-a's prefix must not match group-b's key")
+ .isFalse();
+ }
+
+ @Test
+ void testEncodeGroupPrefixMatchesStandardBucketKeyEncoder() {
+ // The standard KV path creates a bucket key encoder from the table's schema and
+ // the distributedBy column ("group_id"). ConsumerOffsetSchema.encodeGroupPrefix must
+ // produce
+ // identical bytes — this is the guarantee that FindConsumerGroupOwner routes to the
+ // same bucket where the data actually resides.
+ KeyEncoder standardEncoder =
+ KeyEncoder.ofBucketKeyEncoder(
+ ConsumerOffsetSchema.getTableSchema().getRowType(),
+ Collections.singletonList("group_id"),
+ null);
+
+ for (String group :
+ new String[] {
+ "a",
+ "my-group",
+ "test-group-123",
+ "group-with-\u00fcnicode",
+ "x",
+ "g" + "y".repeat(200)
+ }) {
+ CompactedRow row =
+ ConsumerOffsetSchema.encodeOffsetRow(
+ group,
+ ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE,
+ -1L,
+ -1L,
+ -1,
+ new ConsumerOffsetAndMetadata(-1L, ""));
+ byte[] standardBytes = standardEncoder.encodeKey(row);
+ byte[] consumerOffsetSchemaBytes = ConsumerOffsetSchema.encodeGroupPrefix(group);
+
+ assertThat(consumerOffsetSchemaBytes)
+ .as("encodeGroupPrefix(%s) must match the standard bucket key encoder", group)
+ .isEqualTo(standardBytes);
+ }
+ }
+
+ @Test
+ void testBucketForIsDeterministicAndInRange() {
+ // GroupManager.bucketFor uses FlussBucketingFunction on the encoded group_id;
+ // verify the result is deterministic and always within [0, bucketCount).
+ int bucketCount = 50;
+ for (String group : new String[] {"a", "my-group", "test-group-123", "another-group"}) {
+ byte[] bucketKey = ConsumerOffsetSchema.encodeGroupPrefix(group);
+ int bucket1 = FlussBucketingFunction.bucketForRowKey(bucketKey, bucketCount);
+ int bucket2 = FlussBucketingFunction.bucketForRowKey(bucketKey, bucketCount);
+
+ assertThat(bucket1)
+ .as("bucket for group %s must be in [0, %d)", group, bucketCount)
+ .isBetween(0, bucketCount - 1);
+ assertThat(bucket1)
+ .as("bucketing must be deterministic for group %s", group)
+ .isEqualTo(bucket2);
+ }
+ }
+
+ // ------------------------------------------------------------------
+
+ /** Decodes a row the same way the zero-memory GroupManager decodes RocksDB values. */
+ private InternalRow decode(CompactedRow row) {
+ byte[] stored = ValueEncoder.encodeValue(SCHEMA_ID, row);
+ return decoder.decode(MemorySegment.wrap(stored));
+ }
+
+ private static boolean isPrefix(byte[] prefix, byte[] bytes) {
+ return bytes.length >= prefix.length
+ && Arrays.equals(prefix, 0, prefix.length, bytes, 0, prefix.length);
Review Comment:
I’ve updated the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add consumer offset table foundations [fluss]
Posted by "gyang94 (via GitHub)" <gi...@apache.org>.
gyang94 commented on PR #4626:
URL: https://github.com/apache/fluss/pull/4626#issuecomment-6187203821
Looks good to me.
One concern is that, now we always create a new encoder `createFullKeyEncoder()` when we call `ConsumerOffsetSchema.encodeOffsetKey(row)`. Depending on how frequently this method is called, especially in a loop over a batch of offsets, this could add unnecessary allocation and initialization overhead.
We can evaluate whether reuse would be worthwhile once the call sites are introduced in follow-up PRs, and make any needed optimization there. It may not block the current PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add consumer offset table foundations [fluss]
Posted by "gyang94 (via GitHub)" <gi...@apache.org>.
gyang94 commented on code in PR #4626:
URL: https://github.com/apache/fluss/pull/4626#discussion_r4266209076
##########
fluss-server/src/test/java/org/apache/fluss/server/tablet/coordinator/group/ConsumerOffsetSchemaTest.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * 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.coordinator.group;
+
+import org.apache.fluss.bucketing.FlussBucketingFunction;
+import org.apache.fluss.memory.MemorySegment;
+import org.apache.fluss.metadata.KvFormat;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.compacted.CompactedRow;
+import org.apache.fluss.row.decode.FixedSchemaDecoder;
+import org.apache.fluss.row.encode.KeyEncoder;
+import org.apache.fluss.row.encode.ValueEncoder;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for the sys.consumer_offsets schema and key/value encode/decode. */
+class ConsumerOffsetSchemaTest {
+
+ private static final short SCHEMA_ID = (short) 1;
+
+ private final FixedSchemaDecoder decoder =
+ new FixedSchemaDecoder(KvFormat.COMPACTED, ConsumerOffsetSchema.getTableSchema());
+
+ @Test
+ void testSchemaMatchesFip() {
+ // PK: (group_id, record_type, table_id, partition_id, bucket_id)
+ assertThat(ConsumerOffsetSchema.getPrimaryKeys())
+ .containsExactly(
+ "group_id", "record_type", "table_id", "partition_id", "bucket_id");
+ // Value: (offset, metadata) — FIP-53 has no commit_timestamp.
+ assertThat(ConsumerOffsetSchema.getTableSchema().getColumnNames())
+ .containsExactly(
+ "group_id",
+ "record_type",
+ "table_id",
+ "partition_id",
+ "bucket_id",
+ "offset",
+ "metadata");
+ }
+
+ @Test
+ void testOffsetCommitValueRoundTrip() {
+ CompactedRow row =
+ ConsumerOffsetSchema.encodeOffsetRow(
+ "my-group",
+ ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE,
+ 42L,
+ 7L,
+ 3,
+ new ConsumerOffsetAndMetadata(100L, "wm=5"));
+
+ InternalRow decoded = decode(row);
+ assertThat(decoded.getString(0).toString()).isEqualTo("my-group");
+ assertThat(decoded.getInt(1)).isEqualTo(ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE);
+ assertThat(decoded.getLong(2)).isEqualTo(42L);
+ assertThat(decoded.getLong(3)).isEqualTo(7L);
+ assertThat(decoded.getInt(4)).isEqualTo(3);
+ assertThat(decoded.getLong(5)).isEqualTo(100L);
+ assertThat(decoded.getString(6).toString()).isEqualTo("wm=5");
+ }
+
+ @Test
+ void testNullPartitionIdEncodesAsMinusOneSentinel() {
+ // Non-partitioned table: partitionId == null materializes as -1 (PK is NOT NULL).
+ CompactedRow row =
+ ConsumerOffsetSchema.encodeOffsetRow(
+ "g",
+ ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE,
+ 1L,
+ null,
+ 0,
+ new ConsumerOffsetAndMetadata(5L, ""));
+ assertThat(decode(row).getLong(3)).isEqualTo(-1L);
+
+ // Partition id 0 is a valid partition id and must survive the round-trip unchanged.
+ CompactedRow zeroPartition =
+ ConsumerOffsetSchema.encodeOffsetRow(
+ "g",
+ ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE,
+ 1L,
+ 0L,
+ 0,
+ new ConsumerOffsetAndMetadata(5L, ""));
+ assertThat(decode(zeroPartition).getLong(3)).isEqualTo(0L);
+ }
+
+ @Test
+ void testEncodeOffsetKeyMatchesRowKeyEncoding() {
+ CompactedRow row =
+ ConsumerOffsetSchema.encodeOffsetRow(
+ "my-group",
+ ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE,
+ 42L,
+ 7L,
+ 3,
+ new ConsumerOffsetAndMetadata(100L, ""));
+
+ byte[] keyFromRow = ConsumerOffsetSchema.encodeOffsetKey(row);
+ byte[] keyFromFields = ConsumerOffsetSchema.encodeOffsetKey("my-group", 42L, 7L, 3);
+ assertThat(keyFromFields).isEqualTo(keyFromRow);
+ }
+
+ @Test
+ void testGroupPrefixIsBytePrefixOfFullKey() {
+ // The prefix scan correctness of the zero-memory GroupManager depends on this property:
+ // encodeGroupPrefix(g) must be a strict byte prefix of every encodeOffsetKey of group g.
+ for (String group : new String[] {"a", "my-group", "group-with-ünicode", "x"}) {
+ byte[] prefix = ConsumerOffsetSchema.encodeGroupPrefix(group);
+ assertThat(prefix).isNotEmpty();
+
+ byte[] keyNonPartitioned = ConsumerOffsetSchema.encodeOffsetKey(group, 1L, null, 0);
+ byte[] keyPartitioned = ConsumerOffsetSchema.encodeOffsetKey(group, 2L, 99L, 5);
+
+ assertThat(isPrefix(prefix, keyNonPartitioned))
+ .as("prefix of group %s must be a byte prefix of its full keys", group)
+ .isTrue();
+ assertThat(isPrefix(prefix, keyPartitioned)).isTrue();
+ }
+ }
+
+ @Test
+ void testDifferentGroupsPrefixesDoNotCrossMatch() {
+ byte[] prefixA = ConsumerOffsetSchema.encodeGroupPrefix("group-a");
+ byte[] keyB = ConsumerOffsetSchema.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
+ // would also see group-b's rows.
+ assertThat(isPrefix(prefixA, keyB))
+ .as("group-a's prefix must not match group-b's key")
+ .isFalse();
+ }
+
+ @Test
+ void testEncodeGroupPrefixMatchesStandardBucketKeyEncoder() {
+ // The standard KV path creates a bucket key encoder from the table's schema and
+ // the distributedBy column ("group_id"). ConsumerOffsetSchema.encodeGroupPrefix must
+ // produce
+ // identical bytes — this is the guarantee that FindConsumerGroupOwner routes to the
+ // same bucket where the data actually resides.
+ KeyEncoder standardEncoder =
+ KeyEncoder.ofBucketKeyEncoder(
+ ConsumerOffsetSchema.getTableSchema().getRowType(),
+ Collections.singletonList("group_id"),
+ null);
+
+ for (String group :
+ new String[] {
+ "a",
+ "my-group",
+ "test-group-123",
+ "group-with-\u00fcnicode",
+ "x",
+ "g" + "y".repeat(200)
Review Comment:
failed to compile `repeat()` with Java 8.
##########
fluss-server/src/test/java/org/apache/fluss/server/tablet/coordinator/group/ConsumerOffsetSchemaTest.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * 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.coordinator.group;
+
+import org.apache.fluss.bucketing.FlussBucketingFunction;
+import org.apache.fluss.memory.MemorySegment;
+import org.apache.fluss.metadata.KvFormat;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.compacted.CompactedRow;
+import org.apache.fluss.row.decode.FixedSchemaDecoder;
+import org.apache.fluss.row.encode.KeyEncoder;
+import org.apache.fluss.row.encode.ValueEncoder;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for the sys.consumer_offsets schema and key/value encode/decode. */
+class ConsumerOffsetSchemaTest {
+
+ private static final short SCHEMA_ID = (short) 1;
+
+ private final FixedSchemaDecoder decoder =
+ new FixedSchemaDecoder(KvFormat.COMPACTED, ConsumerOffsetSchema.getTableSchema());
+
+ @Test
+ void testSchemaMatchesFip() {
+ // PK: (group_id, record_type, table_id, partition_id, bucket_id)
+ assertThat(ConsumerOffsetSchema.getPrimaryKeys())
+ .containsExactly(
+ "group_id", "record_type", "table_id", "partition_id", "bucket_id");
+ // Value: (offset, metadata) — FIP-53 has no commit_timestamp.
+ assertThat(ConsumerOffsetSchema.getTableSchema().getColumnNames())
+ .containsExactly(
+ "group_id",
+ "record_type",
+ "table_id",
+ "partition_id",
+ "bucket_id",
+ "offset",
+ "metadata");
+ }
+
+ @Test
+ void testOffsetCommitValueRoundTrip() {
+ CompactedRow row =
+ ConsumerOffsetSchema.encodeOffsetRow(
+ "my-group",
+ ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE,
+ 42L,
+ 7L,
+ 3,
+ new ConsumerOffsetAndMetadata(100L, "wm=5"));
+
+ InternalRow decoded = decode(row);
+ assertThat(decoded.getString(0).toString()).isEqualTo("my-group");
+ assertThat(decoded.getInt(1)).isEqualTo(ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE);
+ assertThat(decoded.getLong(2)).isEqualTo(42L);
+ assertThat(decoded.getLong(3)).isEqualTo(7L);
+ assertThat(decoded.getInt(4)).isEqualTo(3);
+ assertThat(decoded.getLong(5)).isEqualTo(100L);
+ assertThat(decoded.getString(6).toString()).isEqualTo("wm=5");
+ }
+
+ @Test
+ void testNullPartitionIdEncodesAsMinusOneSentinel() {
+ // Non-partitioned table: partitionId == null materializes as -1 (PK is NOT NULL).
+ CompactedRow row =
+ ConsumerOffsetSchema.encodeOffsetRow(
+ "g",
+ ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE,
+ 1L,
+ null,
+ 0,
+ new ConsumerOffsetAndMetadata(5L, ""));
+ assertThat(decode(row).getLong(3)).isEqualTo(-1L);
+
+ // Partition id 0 is a valid partition id and must survive the round-trip unchanged.
+ CompactedRow zeroPartition =
+ ConsumerOffsetSchema.encodeOffsetRow(
+ "g",
+ ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE,
+ 1L,
+ 0L,
+ 0,
+ new ConsumerOffsetAndMetadata(5L, ""));
+ assertThat(decode(zeroPartition).getLong(3)).isEqualTo(0L);
+ }
+
+ @Test
+ void testEncodeOffsetKeyMatchesRowKeyEncoding() {
+ CompactedRow row =
+ ConsumerOffsetSchema.encodeOffsetRow(
+ "my-group",
+ ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE,
+ 42L,
+ 7L,
+ 3,
+ new ConsumerOffsetAndMetadata(100L, ""));
+
+ byte[] keyFromRow = ConsumerOffsetSchema.encodeOffsetKey(row);
+ byte[] keyFromFields = ConsumerOffsetSchema.encodeOffsetKey("my-group", 42L, 7L, 3);
+ assertThat(keyFromFields).isEqualTo(keyFromRow);
+ }
+
+ @Test
+ void testGroupPrefixIsBytePrefixOfFullKey() {
+ // The prefix scan correctness of the zero-memory GroupManager depends on this property:
+ // encodeGroupPrefix(g) must be a strict byte prefix of every encodeOffsetKey of group g.
+ for (String group : new String[] {"a", "my-group", "group-with-ünicode", "x"}) {
+ byte[] prefix = ConsumerOffsetSchema.encodeGroupPrefix(group);
+ assertThat(prefix).isNotEmpty();
+
+ byte[] keyNonPartitioned = ConsumerOffsetSchema.encodeOffsetKey(group, 1L, null, 0);
+ byte[] keyPartitioned = ConsumerOffsetSchema.encodeOffsetKey(group, 2L, 99L, 5);
+
+ assertThat(isPrefix(prefix, keyNonPartitioned))
+ .as("prefix of group %s must be a byte prefix of its full keys", group)
+ .isTrue();
+ assertThat(isPrefix(prefix, keyPartitioned)).isTrue();
+ }
+ }
+
+ @Test
+ void testDifferentGroupsPrefixesDoNotCrossMatch() {
+ byte[] prefixA = ConsumerOffsetSchema.encodeGroupPrefix("group-a");
+ byte[] keyB = ConsumerOffsetSchema.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
+ // would also see group-b's rows.
+ assertThat(isPrefix(prefixA, keyB))
+ .as("group-a's prefix must not match group-b's key")
+ .isFalse();
+ }
+
+ @Test
+ void testEncodeGroupPrefixMatchesStandardBucketKeyEncoder() {
+ // The standard KV path creates a bucket key encoder from the table's schema and
+ // the distributedBy column ("group_id"). ConsumerOffsetSchema.encodeGroupPrefix must
+ // produce
+ // identical bytes — this is the guarantee that FindConsumerGroupOwner routes to the
+ // same bucket where the data actually resides.
+ KeyEncoder standardEncoder =
+ KeyEncoder.ofBucketKeyEncoder(
+ ConsumerOffsetSchema.getTableSchema().getRowType(),
+ Collections.singletonList("group_id"),
+ null);
+
+ for (String group :
+ new String[] {
+ "a",
+ "my-group",
+ "test-group-123",
+ "group-with-\u00fcnicode",
+ "x",
+ "g" + "y".repeat(200)
+ }) {
+ CompactedRow row =
+ ConsumerOffsetSchema.encodeOffsetRow(
+ group,
+ ConsumerOffsetSchema.OFFSET_COMMIT_RECORD_TYPE,
+ -1L,
+ -1L,
+ -1,
+ new ConsumerOffsetAndMetadata(-1L, ""));
+ byte[] standardBytes = standardEncoder.encodeKey(row);
+ byte[] consumerOffsetSchemaBytes = ConsumerOffsetSchema.encodeGroupPrefix(group);
+
+ assertThat(consumerOffsetSchemaBytes)
+ .as("encodeGroupPrefix(%s) must match the standard bucket key encoder", group)
+ .isEqualTo(standardBytes);
+ }
+ }
+
+ @Test
+ void testBucketForIsDeterministicAndInRange() {
+ // GroupManager.bucketFor uses FlussBucketingFunction on the encoded group_id;
+ // verify the result is deterministic and always within [0, bucketCount).
+ int bucketCount = 50;
+ for (String group : new String[] {"a", "my-group", "test-group-123", "another-group"}) {
+ byte[] bucketKey = ConsumerOffsetSchema.encodeGroupPrefix(group);
+ int bucket1 = FlussBucketingFunction.bucketForRowKey(bucketKey, bucketCount);
+ int bucket2 = FlussBucketingFunction.bucketForRowKey(bucketKey, bucketCount);
+
+ assertThat(bucket1)
+ .as("bucket for group %s must be in [0, %d)", group, bucketCount)
+ .isBetween(0, bucketCount - 1);
+ assertThat(bucket1)
+ .as("bucketing must be deterministic for group %s", group)
+ .isEqualTo(bucket2);
+ }
+ }
+
+ // ------------------------------------------------------------------
+
+ /** Decodes a row the same way the zero-memory GroupManager decodes RocksDB values. */
+ private InternalRow decode(CompactedRow row) {
+ byte[] stored = ValueEncoder.encodeValue(SCHEMA_ID, row);
+ return decoder.decode(MemorySegment.wrap(stored));
+ }
+
+ private static boolean isPrefix(byte[] prefix, byte[] bytes) {
+ return bytes.length >= prefix.length
+ && Arrays.equals(prefix, 0, prefix.length, bytes, 0, prefix.length);
Review Comment:
failed to compile `Arrays.equals` in Java 8
##########
fluss-server/src/main/java/org/apache/fluss/server/tablet/coordinator/group/ConsumerOffsetSchema.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.coordinator.group;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.row.BinaryString;
+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.RowEncoder;
+import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.types.RowType;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.fluss.row.BinaryString.EMPTY_UTF8;
+
+/** Schema and encoding helpers for the {@code sys.consumer_offsets} system table. */
+@Internal
+public final class ConsumerOffsetSchema {
+
+ private static final String GROUP_ID_COLUMN = "group_id";
+ private static final String RECORD_TYPE_COLUMN = "record_type";
+ private static final String TABLE_ID_COLUMN = "table_id";
+ private static final String PARTITION_ID_COLUMN = "partition_id";
+ private static final String BUCKET_ID_COLUMN = "bucket_id";
+ private static final String OFFSET_COLUMN = "offset";
+ private static final String METADATA_COLUMN = "metadata";
+
+ /** Record type for consumer offset commits. */
+ public static final int OFFSET_COMMIT_RECORD_TYPE = 1;
+
+ /** Record type reserved for consumer group metadata. */
+ public static final int GROUP_METADATA_RECORD_TYPE = 2;
+
+ /** Sentinel used for non-partitioned tables because primary-key columns cannot be null. */
+ public static final long NON_PARTITIONED_PARTITION_ID = -1L;
+
+ private static final ConsumerOffsetAndMetadata EMPTY_OFFSET =
+ new ConsumerOffsetAndMetadata(-1L, "");
+
+ private static final List<String> PRIMARY_KEYS =
+ Collections.unmodifiableList(
+ Arrays.asList(
+ GROUP_ID_COLUMN,
+ RECORD_TYPE_COLUMN,
+ TABLE_ID_COLUMN,
+ PARTITION_ID_COLUMN,
+ BUCKET_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]);
+
+ private static final KeyEncoder FULL_KEY_ENCODER =
Review Comment:
The CompactedKeyEncoder reuses a mutable writer internally. Here we define it as a static shared singleton object, there will be a problem that the concurrent calls to either shared instance can corrupt the encoded bytes.
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -2269,6 +2269,45 @@ public class ConfigOptions {
+ "Downstream consumers must be upgraded to Fluss v1.0+ before enabling this option, "
+ "as older versions cannot parse the extended batch format.");
+ // ------------------------------------------------------------------------
+ // ConfigOptions for Consumer Offsets
+ // ------------------------------------------------------------------------
+
+ public static final ConfigOption<Integer> OFFSET_METADATA_MAX_BYTES =
+ key("server.consumer-offsets.metadata.max-size")
+ .intType()
+ .defaultValue(4096)
+ .withDescription(
+ "The maximum size for a metadata entry associated with an offset commit.");
+
+ public static final ConfigOption<Long> OFFSETS_RETENTION_MS =
+ key("server.consumer-offsets.ttl")
+ .longType()
+ .defaultValue(7L * 24L * 60L * 60L * 1000L)
+ .withDescription(
+ "The retention period for committed consumer offsets after they become inactive.");
+
+ public static final ConfigOption<Integer> OFFSETS_TABLE_NUM_BUCKETS =
+ key("server.consumer-offsets.table.bucket-count")
+ .intType()
+ .defaultValue(50)
Review Comment:
Maybe we can have a fluss-specific sizing value here? Kafka is 50 defaultly, but in fluss, each KV bucket leader maintains a separate RocksDB instance, a large bucket number will introduce too much overhead for small and medium clusters.
##########
fluss-server/src/main/java/org/apache/fluss/server/tablet/coordinator/group/ConsumerOffsetSchema.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.coordinator.group;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.row.BinaryString;
+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.RowEncoder;
+import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.types.RowType;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.fluss.row.BinaryString.EMPTY_UTF8;
+
+/** Schema and encoding helpers for the {@code sys.consumer_offsets} system table. */
+@Internal
+public final class ConsumerOffsetSchema {
+
+ private static final String GROUP_ID_COLUMN = "group_id";
+ private static final String RECORD_TYPE_COLUMN = "record_type";
+ private static final String TABLE_ID_COLUMN = "table_id";
+ private static final String PARTITION_ID_COLUMN = "partition_id";
+ private static final String BUCKET_ID_COLUMN = "bucket_id";
+ private static final String OFFSET_COLUMN = "offset";
+ private static final String METADATA_COLUMN = "metadata";
+
+ /** Record type for consumer offset commits. */
+ public static final int OFFSET_COMMIT_RECORD_TYPE = 1;
+
+ /** Record type reserved for consumer group metadata. */
+ public static final int GROUP_METADATA_RECORD_TYPE = 2;
+
+ /** Sentinel used for non-partitioned tables because primary-key columns cannot be null. */
+ public static final long NON_PARTITIONED_PARTITION_ID = -1L;
+
+ private static final ConsumerOffsetAndMetadata EMPTY_OFFSET =
+ new ConsumerOffsetAndMetadata(-1L, "");
+
+ private static final List<String> PRIMARY_KEYS =
+ Collections.unmodifiableList(
+ Arrays.asList(
+ GROUP_ID_COLUMN,
+ RECORD_TYPE_COLUMN,
+ TABLE_ID_COLUMN,
+ PARTITION_ID_COLUMN,
+ BUCKET_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]);
+
+ private static final KeyEncoder FULL_KEY_ENCODER =
+ KeyEncoder.ofBucketKeyEncoder(ROW_TYPE, PRIMARY_KEYS, null);
+
+ private static final KeyEncoder GROUP_PREFIX_ENCODER =
Review Comment:
same as above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add consumer offset table foundations [fluss]
Posted by "pisceslj (via GitHub)" <gi...@apache.org>.
pisceslj closed pull request #4626: [server] Add consumer offset table foundations
URL: https://github.com/apache/fluss/pull/4626
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add consumer offset table foundations [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #4626:
URL: https://github.com/apache/fluss/pull/4626#discussion_r4292479109
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -2269,6 +2269,48 @@ public class ConfigOptions {
+ "Downstream consumers must be upgraded to Fluss v1.0+ before enabling this option, "
+ "as older versions cannot parse the extended batch format.");
+ // ------------------------------------------------------------------------
+ // ConfigOptions for Consumer Offsets
+ // ------------------------------------------------------------------------
+
+ public static final ConfigOption<Integer> OFFSET_METADATA_MAX_BYTES =
+ key("server.consumer-offsets.metadata.max-size")
+ .intType()
+ .defaultValue(4096)
+ .withDescription(
+ "The maximum size for a metadata entry associated with an offset commit.");
+
+ public static final ConfigOption<Long> OFFSETS_RETENTION_MS =
+ key("server.consumer-offsets.ttl")
+ .longType()
+ .defaultValue(7L * 24L * 60L * 60L * 1000L)
Review Comment:
Please use Duration type for ttl.
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -2269,6 +2269,48 @@ public class ConfigOptions {
+ "Downstream consumers must be upgraded to Fluss v1.0+ before enabling this option, "
+ "as older versions cannot parse the extended batch format.");
+ // ------------------------------------------------------------------------
+ // ConfigOptions for Consumer Offsets
+ // ------------------------------------------------------------------------
+
+ public static final ConfigOption<Integer> OFFSET_METADATA_MAX_BYTES =
+ key("server.consumer-offsets.metadata.max-size")
+ .intType()
+ .defaultValue(4096)
+ .withDescription(
+ "The maximum size for a metadata entry associated with an offset commit.");
+
+ public static final ConfigOption<Long> OFFSETS_RETENTION_MS =
+ key("server.consumer-offsets.ttl")
+ .longType()
+ .defaultValue(7L * 24L * 60L * 60L * 1000L)
+ .withDescription(
+ "The retention period for committed consumer offsets after they become inactive.");
+
+ public static final ConfigOption<Integer> OFFSETS_TABLE_NUM_BUCKETS =
+ key("server.consumer-offsets.table.bucket-count")
+ .intType()
+ .defaultValue(16)
+ .withDescription(
+ "The number of buckets for the consumer-offset system table. This value should not change after deployment. "
+ + "Unlike log tables, each KV bucket leader maintains a separate RocksDB instance, so a large value "
+ + "introduces overhead for small and medium clusters. Increase it only if the consumer-offset write "
+ + "load cannot be handled by the default number of bucket leaders.");
+
+ public static final ConfigOption<Integer> OFFSETS_TABLE_REPLICATION_FACTOR =
+ key("server.consumer-offsets.table.replication-factor")
+ .intType()
+ .defaultValue(3)
+ .withDescription(
+ "The replication factor for the consumer-offset system table. Table creation waits until the cluster has enough tablet servers.");
+
+ public static final ConfigOption<Integer> OFFSET_COMMIT_TIMEOUT_MS =
+ key("server.consumer-offsets.commit-timeout")
+ .intType()
+ .defaultValue(5000)
Review Comment:
Please use Duration type for timeout.
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -2269,6 +2269,48 @@ public class ConfigOptions {
+ "Downstream consumers must be upgraded to Fluss v1.0+ before enabling this option, "
+ "as older versions cannot parse the extended batch format.");
+ // ------------------------------------------------------------------------
+ // ConfigOptions for Consumer Offsets
+ // ------------------------------------------------------------------------
+
+ public static final ConfigOption<Integer> OFFSET_METADATA_MAX_BYTES =
+ key("server.consumer-offsets.metadata.max-size")
+ .intType()
+ .defaultValue(4096)
Review Comment:
Please use `memoryType()`, so that users can easily define memory size (e.g, `1MB`, `8KB`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add consumer offset table foundations [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #4626:
URL: https://github.com/apache/fluss/pull/4626#discussion_r4293250217
##########
fluss-server/src/main/java/org/apache/fluss/server/tablet/coordinator/group/ConsumerOffsetAndMetadata.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.coordinator.group;
Review Comment:
Since it's not coordinator group now, we should rename the package to `consumer.group`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add consumer offset table foundations [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong merged PR #4626:
URL: https://github.com/apache/fluss/pull/4626
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org