[server] Support group-aware tiering table assignment

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -2181,6 +2181,16 @@ public class ConfigOptions {
                             "Specifies the table name of the datalake table. This option is currently supported only for Paimon. "
                                     + "If not set, the Fluss table name is used. The option may be configured before the Paimon table is created and cannot be changed after creation.");
 
+    public static final ConfigOption<String> TABLE_DATALAKE_TIERING_GROUP =
+            key("table.datalake.tiering-group")
+                    .stringType()
+                    .defaultValue("default")
+                    .withDescription(
+                            "The tiering service group responsible for this table. "
+                                    + "Only tiering services in the same group receive the table. "
+                                    + "Group names are case-sensitive and must not be empty or have leading or trailing whitespace. "
+                                    + "Changes through ALTER TABLE apply asynchronously to subsequent tiering rounds.");
+
     public static final ConfigOption<Duration> TABLE_DATALAKE_FRESHNESS =
             key("table.datalake.freshness")
                     .durationType()

fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java

@@ -56,6 +56,7 @@ public class FlussConfigUtils {
                         ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_LOOKUP_MODE.key(),
                         ConfigOptions.TABLE_DATALAKE_DATABASE_NAME.key(),
                         ConfigOptions.TABLE_DATALAKE_TABLE_NAME.key(),
+                        ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.key(),
                         ConfigOptions.TABLE_DATALAKE_FRESHNESS.key(),
                         ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION.key(),
                         ConfigOptions.TABLE_LOG_TTL.key(),
@@ -76,6 +77,14 @@ public static boolean isAlterableTableOption(String key) {
         return ALTERABLE_TABLE_OPTIONS.contains(key);
     }
 
+    /** Validates a tiering group shared by table configuration and tiering service heartbeats. */
+    public static void validateTieringGroup(String group) {
+        if (group == null || group.isEmpty() || !group.equals(group.trim())) {
+            throw new IllegalConfigurationException(
+                    "Tiering group must not be empty or have leading or trailing whitespace.");
+        }
+    }
+
     /**
      * Returns the default remote data directory from the configuration. Used as a fallback for
      * tables or partitions that do not contain remote data directory metadata.

fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java

@@ -148,6 +148,11 @@ public Optional<DataLakeFormat> getDataLakeFormat() {
         return config.getOptional(ConfigOptions.TABLE_DATALAKE_FORMAT);
     }
 
+    /** Gets the tiering service group responsible for this table. */
+    public String getDataLakeTieringGroup() {
+        return config.get(ConfigOptions.TABLE_DATALAKE_TIERING_GROUP);
+    }
+
     /**
      * Gets the data lake freshness of the table. It defines the maximum amount of time that the
      * datalake table's content should lag behind updates to the Fluss table.

fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java

@@ -28,6 +28,18 @@
 /** Tests for {@link TableConfig}. */
 class TableConfigTest {
 
+    @Test
+    void testDataLakeTieringGroup() {
+        Configuration conf = new Configuration();
+        TableConfig tableConfig = new TableConfig(conf);
+
+        assertThat(tableConfig.getDataLakeTieringGroup()).isEqualTo("default");
+        conf.set(ConfigOptions.TABLE_DATALAKE_TIERING_GROUP, "large");
+        assertThat(tableConfig.getDataLakeTieringGroup()).isEqualTo("large");
+        conf.removeConfig(ConfigOptions.TABLE_DATALAKE_TIERING_GROUP);
+        assertThat(tableConfig.getDataLakeTieringGroup()).isEqualTo("default");
+    }
+
     @Test
     void testDeleteBehavior() {
         Configuration conf = new Configuration();

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceOptions.java

@@ -18,6 +18,7 @@
 package org.apache.fluss.flink.tiering.source;
 
 import org.apache.fluss.config.ConfigOption;
+import org.apache.fluss.config.ConfigOptions;
 
 import java.time.Duration;
 
@@ -28,6 +29,13 @@ public class TieringSourceOptions {
 
     public static final String DATA_LAKE_CONFIG_PREFIX = "datalake.";
 
+    public static final ConfigOption<String> TIERING_GROUP =
+            key("tiering.group")
+                    .stringType()
+                    .defaultValue(ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.defaultValue())
+                    .withDescription(
+                            "The tiering group of tables handled by this tiering service.");
+
     public static final ConfigOption<Duration> POLL_TIERING_TABLE_INTERVAL =
             key("tiering.poll.table.interval")
                     .durationType()

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java

@@ -24,10 +24,12 @@
 import org.apache.fluss.client.metadata.MetadataUpdater;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.FlussConfigUtils;
 import org.apache.fluss.flink.metrics.FlinkMetricRegistry;
 import org.apache.fluss.flink.tiering.event.FailedTieringEvent;
 import org.apache.fluss.flink.tiering.event.FinishedTieringEvent;
 import org.apache.fluss.flink.tiering.event.TieringReachMaxDurationEvent;
+import org.apache.fluss.flink.tiering.source.TieringSourceOptions;
 import org.apache.fluss.flink.tiering.source.split.TieringLogSplit;
 import org.apache.fluss.flink.tiering.source.split.TieringSplit;
 import org.apache.fluss.flink.tiering.source.split.TieringSplitGenerator;
@@ -114,6 +116,7 @@ public class TieringSourceEnumerator
     private final ScheduledExecutorService timerService;
     private final SplitEnumeratorMetricGroup enumeratorMetricGroup;
     private final long pollTieringTableIntervalMs;
+    private final String tieringGroup;
     // Also inspected by asynchronous heartbeat requests.
     private final Deque<TieringSplit> pendingSplits;
     private final Set<Integer> readersAwaitingSplit;
@@ -182,6 +185,8 @@ public TieringSourceEnumerator(
         this.timerService = timerService;
         this.enumeratorMetricGroup = context.metricGroup();
         this.pollTieringTableIntervalMs = pollTieringTableIntervalMs;
+        this.tieringGroup = flussConf.get(TieringSourceOptions.TIERING_GROUP);
+        FlussConfigUtils.validateTieringGroup(tieringGroup);
         this.pendingSplits = new ConcurrentLinkedDeque<>();
         this.readersAwaitingSplit = Collections.synchronizedSet(new TreeSet<>());
         this.tieringTableEpochs = new ConcurrentHashMap<>();
@@ -218,7 +223,7 @@ void start(
         try {
             LakeTieringHeartbeatResponse heartbeatResponse =
                     waitHeartbeatResponse(
-                            coordinatorGateway.lakeTieringHeartbeat(basicHeartBeat()));
+                            coordinatorGateway.lakeTieringHeartbeat(basicHeartBeat(tieringGroup)));
             this.flussCoordinatorEpoch = heartbeatResponse.getCoordinatorEpoch();
             LOG.info(
                     "Register Tiering Service to Fluss Coordinator(epoch={}) success.",
@@ -452,7 +457,7 @@ private void assignSplits() {
         Map<Long, Long> currentFailedTableEpochs = new HashMap<>(this.failedTableEpochs);
         LakeTieringHeartbeatRequest tieringHeartbeatRequest =
                 tieringTableHeartBeat(
-                        basicHeartBeat(),
+                        basicHeartBeat(tieringGroup),
                         this.tieringTableEpochs,
                         currentFinishedTables,
                         currentFailedTableEpochs,
@@ -624,7 +629,7 @@ public void close() throws IOException {
             failedTableEpochs.putAll(tieringTableEpochs);
             tieringTableEpochs.clear();
             if (!failedTableEpochs.isEmpty()) {
-                reportFailedTable(basicHeartBeat(), failedTableEpochs);
+                reportFailedTable(basicHeartBeat(tieringGroup), failedTableEpochs);
             }
             try {
                 LOG.info("Closing Tiering Source Enumerator of at {}.", System.currentTimeMillis());
@@ -677,8 +682,8 @@ private void reportFailedTable(
     @VisibleForTesting
     static class HeartBeatHelper {
 
-        static LakeTieringHeartbeatRequest basicHeartBeat() {
-            return new LakeTieringHeartbeatRequest();
+        static LakeTieringHeartbeatRequest basicHeartBeat(String tieringGroup) {
+            return new LakeTieringHeartbeatRequest().setTieringGroup(tieringGroup);
         }
 
         static LakeTieringHeartbeatRequest heartBeatWithRequestNewTieringTable(

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorSchedulingTest.java

@@ -20,6 +20,9 @@
 import org.apache.fluss.client.admin.Admin;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.flink.tiering.TestingLakeTieringFactory;
+import org.apache.fluss.flink.tiering.event.FailedTieringEvent;
+import org.apache.fluss.flink.tiering.event.FinishedTieringEvent;
+import org.apache.fluss.flink.tiering.source.TieringSourceOptions;
 import org.apache.fluss.flink.tiering.source.split.TieringLogSplit;
 import org.apache.fluss.flink.tiering.source.split.TieringSplit;
 import org.apache.fluss.flink.tiering.source.split.TieringSplitGenerator;
@@ -35,6 +38,8 @@
 import org.apache.fluss.types.DataTypes;
 
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.IOException;
 import java.util.ArrayDeque;
@@ -55,6 +60,44 @@
 /** Tests for split assignment and table scheduling in {@link TieringSourceEnumerator}. */
 class TieringSourceEnumeratorSchedulingTest {
 
+    @ParameterizedTest
+    @ValueSource(strings = {"default", "large"})
+    void testTieringGroupIncludedInEveryHeartbeat(String group) throws Throwable {
+        Configuration configuration = new Configuration();
+        if (!group.equals(TieringSourceOptions.TIERING_GROUP.defaultValue())) {
+            configuration.set(TieringSourceOptions.TIERING_GROUP, group);
+        }
+        try (Fixture fixture = new Fixture(1, configuration)) {
+            configuration.set(TieringSourceOptions.TIERING_GROUP, "changed-after-start");
+            fixture.addTable(1, false);
+            fixture.requestTable();
+            fixture.context.runPeriodicCallable(0);
+            assertThat(fixture.requests.get(2).getTieringTablesList())
+                    .singleElement()
+                    .satisfies(table -> assertThat(table.getTableId()).isEqualTo(1));
+
+            fixture.enumerator.handleSourceEvent(0, new FinishedTieringEvent(1));
+            fixture.context.runNextOneTimeCallable();
+            assertThat(fixture.requests.get(3).getFinishedTablesList())
+                    .singleElement()
+                    .satisfies(table -> assertThat(table.getTableId()).isEqualTo(1));
+
+            fixture.addTable(2, false);
+            fixture.requestTable();
+            fixture.enumerator.handleSourceEvent(0, new FailedTieringEvent(2, "test failure"));
+            fixture.context.runNextOneTimeCallable();
+            assertThat(fixture.requests.get(5).getFailedTablesList())
+                    .singleElement()
+                    .satisfies(table -> assertThat(table.getTableId()).isEqualTo(2));
+
+            assertThat(fixture.requests).hasSize(6);
+            assertThat(fixture.claims()).hasSize(2);
+            assertThat(fixture.requests)
+                    .extracting(LakeTieringHeartbeatRequest::getTieringGroup)
+                    .containsOnly(group);
+        }
+    }
+
     @Test
     void testPendingSplitsAssignedInOrder() throws Exception {
         try (Fixture fixture = new Fixture()) {
@@ -244,13 +287,17 @@ private Fixture() {
         }
 
         private Fixture(int parallelism) {
+            this(parallelism, new Configuration());
+        }
+
+        private Fixture(int parallelism, Configuration configuration) {
             this.context = new TestingContext(parallelism);
             CoordinatorGateway gateway = mock(CoordinatorGateway.class);
             when(gateway.lakeTieringHeartbeat(any()))
                     .thenAnswer(invocation -> respondToHeartbeat(invocation.getArgument(0)));
             enumerator =
                     new TieringSourceEnumerator(
-                            new Configuration(),
+                            configuration,
                             new Configuration(),
                             context,
                             new TestingLakeTieringFactory(),

fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java

@@ -23,9 +23,11 @@
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.LakeTableSnapshotNotExistException;
+import org.apache.fluss.flink.tiering.LakeTieringJobBuilder;
 import org.apache.fluss.lake.paimon.testutils.FlinkPaimonTieringTestBase;
 import org.apache.fluss.lake.paimon.tiering.markdone.PartitionMarkDoneState;
 import org.apache.fluss.lake.paimon.tiering.markdone.PartitionMarkDoneStateJsonSerde;
+import org.apache.fluss.metadata.DataLakeFormat;
 import org.apache.fluss.metadata.PartitionInfo;
 import org.apache.fluss.metadata.PartitionSpec;
 import org.apache.fluss.metadata.Schema;
@@ -79,6 +81,8 @@
 import java.util.stream.Stream;
 
 import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.collectRowsUntilEndWithTimeout;
+import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.POLL_TIERING_TABLE_INTERVAL;
+import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.TIERING_GROUP;
 import static org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
 import static org.apache.fluss.lake.paimon.testutils.PaimonTestUtils.adjustToLegacyV1Table;
 import static org.apache.fluss.lake.paimon.tiering.markdone.PaimonPartitionMarkDone.MARK_DONE_STATE_PROPERTY;
@@ -205,6 +209,71 @@ void testTiering() throws Exception {
         }
     }
 
+    @Test
+    void testTieringWithGroups() throws Exception {
+        TablePath groupedTable = TablePath.of(DEFAULT_DB, "largeGroupTable");
+        TablePath defaultTable = TablePath.of(DEFAULT_DB, "defaultGroupTable");
+        try (AutoCloseable defaultTableCleanup = () -> dropTable(defaultTable);
+                AutoCloseable groupedTableCleanup = () -> dropTable(groupedTable)) {
+            long groupedTableId =
+                    createLogTable(
+                            groupedTable,
+                            1,
+                            false,
+                            Collections.singletonMap(
+                                    ConfigOptions.TABLE_DATALAKE_TIERING_GROUP.key(), "large"),
+                            Collections.emptyMap());
+            TableBucket groupedBucket = new TableBucket(groupedTableId, 0);
+            List<InternalRow> groupedRows = Arrays.asList(row(1, "large1"), row(2, "large2"));
+            writeRows(groupedTable, groupedRows, true);
+
+            TableBucket defaultBucket = new TableBucket(createLogTable(defaultTable), 0);
+            List<InternalRow> defaultRows =
+                    Arrays.asList(row(1, "default1"), row(2, "default2"), row(3, "default3"));
+            writeRows(defaultTable, defaultRows.subList(0, 2), true);
+
+            JobClient defaultJob = buildTieringJob(execEnv);
+            try (AutoCloseable defaultJobCleanup = () -> defaultJob.cancel().get()) {
+                assertReplicaStatus(defaultBucket, 2);
+                // Complete a second round while the large-group table is still waiting.
+                writeRows(defaultTable, defaultRows.subList(2, 3), true);
+                assertReplicaStatus(defaultBucket, 3);
+                checkDataInPaimonAppendOnlyTable(defaultTable, defaultRows, 0);
+                FileStoreTable groupedLakeTable =
+                        (FileStoreTable)
+                                paimonCatalog.getTable(
+                                        Identifier.create(
+                                                groupedTable.getDatabaseName(),
+                                                groupedTable.getTableName()));
+                assertThat(groupedLakeTable.snapshotManager().latestSnapshot())
+                        .as("The large-group table must wait for a matching tiering service")
+                        .isNull();
+
+                Configuration groupedConf = new Configuration(clientConf);
+                groupedConf.set(TIERING_GROUP, "large");
+                groupedConf.set(POLL_TIERING_TABLE_INTERVAL, Duration.ofMillis(500));
+                StreamExecutionEnvironment groupedEnv =
+                        StreamExecutionEnvironment.getExecutionEnvironment();
+                groupedEnv.setParallelism(2);
+                groupedEnv.enableCheckpointing(1000);
+                JobClient groupedJob =
+                        LakeTieringJobBuilder.newBuilder(
+                                        groupedEnv,
+                                        groupedConf,
+                                        Configuration.fromMap(getPaimonCatalogConf()),
+                                        new Configuration(),
+                                        DataLakeFormat.PAIMON.toString())
+                                .build();
+                try (AutoCloseable groupedJobCleanup = () -> groupedJob.cancel().get()) {
+                    assertReplicaStatus(groupedBucket, 2);
+                    checkDataInPaimonAppendOnlyTable(groupedTable, groupedRows, 0);
+                    checkFlussOffsetsInSnapshot(
+                            groupedTable, Collections.singletonMap(groupedBucket, 2L));
+                }
+            }
+        }
+    }
+
     @Test
     void testTieringForLegacyTable() throws Exception {
         // FIP-27: verify that tiering still works for a legacy table that carries the three

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java

@@ -20,6 +20,7 @@
 import org.apache.fluss.annotation.VisibleForTesting;
 import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.cluster.ServerType;
+import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.exception.DisconnectException;
 import org.apache.fluss.exception.FlussRuntimeException;
 import org.apache.fluss.exception.InvalidServerTypeException;
@@ -32,6 +33,7 @@
 import org.apache.fluss.rpc.messages.ApiVersionsResponse;
 import org.apache.fluss.rpc.messages.AuthenticateRequest;
 import org.apache.fluss.rpc.messages.AuthenticateResponse;
+import org.apache.fluss.rpc.messages.LakeTieringHeartbeatRequest;
 import org.apache.fluss.rpc.messages.ProduceLogRequest;
 import org.apache.fluss.rpc.messages.PutKvRequest;
 import org.apache.fluss.rpc.metrics.ClientMetricGroup;
@@ -77,6 +79,7 @@ final class ServerConnection {
     private static final short HISTORICAL_PRODUCE_LOG_MIN_VERSION = 1;
     private static final short HISTORICAL_PUT_KV_MIN_VERSION = 3;
     private static final short ALTER_BUCKET_COUNT_MIN_VERSION = 1;
+    private static final short TIERING_GROUP_MIN_VERSION = 1;
 
     private final ServerNode node;
 
@@ -426,6 +429,23 @@ void validateVersionCompatibility(ApiKeys apiKey, short version, ApiMessage rawR
                                 + '.');
             }
         }
+
+        if (apiKey == ApiKeys.LAKE_TIERING_HEARTBEAT && version < TIERING_GROUP_MIN_VERSION) {
+            LakeTieringHeartbeatRequest heartbeatRequest = (LakeTieringHeartbeatRequest) rawRequest;
+            if (heartbeatRequest.hasTieringGroup()
+                    && !ConfigOptions.TABLE_DATALAKE_TIERING_GROUP
+                            .defaultValue()
+                            .equals(heartbeatRequest.getTieringGroup())) {
+                throw new UnsupportedVersionException(
+                        "Non-default tiering groups require LAKE_TIERING_HEARTBEAT version "
+                                + TIERING_GROUP_MIN_VERSION
+                                + " or newer, but server "
+                                + node
+                                + " negotiated version "
+                                + version
+                                + '.');
+            }
+        }
     }
 
     private void handleApiVersionsResponse(ApiMessage response, Throwable cause) {